codeking/sp-hydra-veil-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improve exception override-related logic

Browse source
This commit is contained in:
codeking 2025-11-04 20:45:40 +01:00
parent 31a70eea5e
commit 70d3905117
2 changed files with 14 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
core/controllers/ConnectionController.py
View file

@ -39,7 +39,7 @@ class ConnectionController:
return None return None
@staticmethod @staticmethod
def establish_connection(profile: Union[SessionProfile, SystemProfile], force: bool = False, connection_observer: Optional[ConnectionObserver] = None): def establish_connection(profile: Union[SessionProfile, SystemProfile], ignore: tuple[type[Exception]] = (), connection_observer: Optional[ConnectionObserver] = None):
connection = profile.connection connection = profile.connection
@ -87,14 +87,14 @@ class ConnectionController:
if profile.is_session_profile(): if profile.is_session_profile():
try: try:
return ConnectionController.establish_session_connection(profile, force=force, connection_observer=connection_observer) return ConnectionController.establish_session_connection(profile, ignore=ignore, connection_observer=connection_observer)
except ConnectionError: except ConnectionError:
if ConnectionController.__should_renegotiate(profile): if ConnectionController.__should_renegotiate(profile):
ProfileController.register_wireguard_session(profile, connection_observer=connection_observer) ProfileController.register_wireguard_session(profile, connection_observer=connection_observer)
return ConnectionController.establish_session_connection(profile, force=force, connection_observer=connection_observer) return ConnectionController.establish_session_connection(profile, ignore=ignore, connection_observer=connection_observer)
else: else:
raise ConnectionError('The connection could not be established.') raise ConnectionError('The connection could not be established.')
@ -117,7 +117,7 @@ class ConnectionController:
return None return None
@staticmethod @staticmethod
def establish_session_connection(profile: SessionProfile, force: Optional[bool] = False, connection_observer: Optional[ConnectionObserver] = None): def establish_session_connection(profile: SessionProfile, ignore: tuple[type[Exception]] = (), connection_observer: Optional[ConnectionObserver] = None):
session_directory = tempfile.mkdtemp(prefix='hv-') session_directory = tempfile.mkdtemp(prefix='hv-')
session_state = SessionStateController.get_or_new(profile.id) session_state = SessionStateController.get_or_new(profile.id)
@ -129,7 +129,7 @@ class ConnectionController:
if not ConnectionController.system_uses_wireguard_interface(): if not ConnectionController.system_uses_wireguard_interface():
if not force: if not ConnectionUnprotectedError in ignore:
raise ConnectionUnprotectedError('Connection unprotected while the system is not using a WireGuard interface.') raise ConnectionUnprotectedError('Connection unprotected while the system is not using a WireGuard interface.')
else: else:
ProfileController.disable(profile) ProfileController.disable(profile)

16
core/controllers/ProfileController.py
View file

@ -44,13 +44,13 @@ class ProfileController:
profile_observer.notify('updated', profile) profile_observer.notify('updated', profile)
@staticmethod @staticmethod
def enable(profile: Union[SessionProfile, SystemProfile], force: bool = False, pristine: bool = False, asynchronous: bool = False, profile_observer: ProfileObserver = None, application_version_observer: ApplicationVersionObserver = None, connection_observer: ConnectionObserver = None): def enable(profile: Union[SessionProfile, SystemProfile], ignore: tuple[type[Exception]] = (), pristine: bool = False, asynchronous: bool = False, profile_observer: ProfileObserver = None, application_version_observer: ApplicationVersionObserver = None, connection_observer: ConnectionObserver = None):
from core.controllers.ConnectionController import ConnectionController from core.controllers.ConnectionController import ConnectionController
if ProfileController.is_enabled(profile): if ProfileController.is_enabled(profile):
if not force: if not ProfileStateConflictError in ignore:
raise ProfileStateConflictError('The profile is already enabled or its session was not properly terminated.') raise ProfileStateConflictError('The profile is already enabled or its session was not properly terminated.')
else: else:
ProfileController.disable(profile) ProfileController.disable(profile)
@ -66,7 +66,7 @@ class ProfileController:
ApplicationVersionController.install(application_version, application_version_observer=application_version_observer, connection_observer=connection_observer) ApplicationVersionController.install(application_version, application_version_observer=application_version_observer, connection_observer=connection_observer)
try: try:
port_number = ConnectionController.establish_connection(profile, force=force, connection_observer=connection_observer) port_number = ConnectionController.establish_connection(profile, ignore=ignore, connection_observer=connection_observer)
except ConnectionError: except ConnectionError:
raise ProfileActivationError('The profile could not be enabled.') raise ProfileActivationError('The profile could not be enabled.')
@ -78,7 +78,7 @@ class ProfileController:
if profile.is_system_profile(): if profile.is_system_profile():
try: try:
ConnectionController.establish_connection(profile, force=force, connection_observer=connection_observer) ConnectionController.establish_connection(profile, ignore=ignore, connection_observer=connection_observer)
except ConnectionError: except ConnectionError:
raise ProfileActivationError('The profile could not be enabled.') raise ProfileActivationError('The profile could not be enabled.')
@ -86,7 +86,7 @@ class ProfileController:
profile_observer.notify('enabled', profile) profile_observer.notify('enabled', profile)
@staticmethod @staticmethod
def disable(profile: Union[SessionProfile, SystemProfile], explicitly: bool = True, force: bool = False, profile_observer: ProfileObserver = None): def disable(profile: Union[SessionProfile, SystemProfile], explicitly: bool = True, ignore: tuple[type[Exception]] = (), profile_observer: ProfileObserver = None):
from core.controllers.ConnectionController import ConnectionController from core.controllers.ConnectionController import ConnectionController
@ -95,7 +95,9 @@ class ProfileController:
if SessionStateController.exists(profile.id): if SessionStateController.exists(profile.id):
session_state = SessionStateController.get(profile.id) session_state = SessionStateController.get(profile.id)
session_state.dissolve(session_state.id)
if session_state is not None:
session_state.dissolve(session_state.id)
if profile.is_system_profile(): if profile.is_system_profile():
@ -105,7 +107,7 @@ class ProfileController:
if subject.is_session_profile(): if subject.is_session_profile():
if subject.connection.is_unprotected() and ProfileController.is_enabled(subject) and not force: if subject.connection.is_unprotected() and ProfileController.is_enabled(subject) and not ConnectionUnprotectedError in ignore:
raise ConnectionUnprotectedError('Disabling this system connection would leave one or more sessions exposed.') raise ConnectionUnprotectedError('Disabling this system connection would leave one or more sessions exposed.')
if SystemStateController.exists(): if SystemStateController.exists():