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

fix system-profile

Browse source
This commit is contained in:
Zenaku 2026-05-23 15:15:15 +00:00
parent 918c070f73
commit 866dda72e7
1 changed files with 1 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
core/models/session/SessionProfile.py
View file

@ -22,35 +22,24 @@ class SessionProfile(BaseProfile):
return self.connection is not None return self.connection is not None
def save(self): def save(self):
if 'application_version' in self._get_dirty_keys(): if 'application_version' in self._get_dirty_keys():
persistent_state_path = f'{self.get_data_path()}/persistent-state' persistent_state_path = f'{self.get_data_path()}/persistent-state'
if os.path.isdir(persistent_state_path): if os.path.isdir(persistent_state_path):
shutil.rmtree(persistent_state_path, ignore_errors=True) shutil.rmtree(persistent_state_path, ignore_errors=True)
if 'location' in self._get_dirty_keys(): if 'location' in self._get_dirty_keys():
self.__delete_proxy_configuration() self.__delete_proxy_configuration()
self.__delete_wireguard_configuration() self.__delete_wireguard_configuration()
super().save() super().save()
def attach_proxy_configuration(self, proxy_configuration): def attach_proxy_configuration(self, proxy_configuration):
proxy_configuration_file_contents = f'{proxy_configuration.to_json(indent=4)}\n' proxy_configuration_file_contents = f'{proxy_configuration.to_json(indent=4)}\n'
os.makedirs(Constants.HV_CONFIG_HOME, exist_ok=True) os.makedirs(Constants.HV_CONFIG_HOME, exist_ok=True)
proxy_configuration_file_path = self.get_proxy_configuration_path() proxy_configuration_file_path = self.get_proxy_configuration_path()
with open(proxy_configuration_file_path, 'w') as proxy_configuration_file: with open(proxy_configuration_file_path, 'w') as proxy_configuration_file:
proxy_configuration_file.write(proxy_configuration_file_contents) proxy_configuration_file.write(proxy_configuration_file_contents)
def attach_wireguard_configuration(self, wireguard_configuration): def attach_wireguard_configuration(self, wireguard_configuration):
wireguard_configuration_file_path = self.get_wireguard_configuration_path() wireguard_configuration_file_path = self.get_wireguard_configuration_path()
with open(wireguard_configuration_file_path, 'w') as wireguard_configuration_file: with open(wireguard_configuration_file_path, 'w') as wireguard_configuration_file:
wireguard_configuration_file.write(wireguard_configuration) wireguard_configuration_file.write(wireguard_configuration)
@ -61,19 +50,15 @@ class SessionProfile(BaseProfile):
return f'{self.get_config_path()}/wg.conf' return f'{self.get_config_path()}/wg.conf'
def get_proxy_configuration(self): def get_proxy_configuration(self):
try: try:
config_file_contents = open(self.get_proxy_configuration_path(), 'r').read() config_file_contents = open(self.get_proxy_configuration_path(), 'r').read()
except FileNotFoundError: except FileNotFoundError:
return None return None
try: try:
proxy_configuration = json.loads(config_file_contents) proxy_configuration = json.loads(config_file_contents)
except ValueError: except ValueError:
return None return None
proxy_configuration = ProxyConfiguration.from_dict(proxy_configuration) proxy_configuration = ProxyConfiguration.from_dict(proxy_configuration)
return proxy_configuration return proxy_configuration
def has_proxy_configuration(self): def has_proxy_configuration(self):
@ -83,32 +68,22 @@ class SessionProfile(BaseProfile):
return os.path.isfile(f'{self.get_config_path()}/wg.conf') return os.path.isfile(f'{self.get_config_path()}/wg.conf')
def address_security_incident(self): def address_security_incident(self):
super().address_security_incident() super().address_security_incident()
self.__delete_wireguard_configuration() self.__delete_wireguard_configuration()
def determine_timezone(self): def determine_timezone(self):
time_zone = None time_zone = None
if self.has_connection(): if self.has_connection():
if self.connection.needs_proxy_configuration(): if self.connection.needs_proxy_configuration():
if self.has_proxy_configuration(): if self.has_proxy_configuration():
time_zone = self.get_proxy_configuration().time_zone time_zone = self.get_proxy_configuration().time_zone
elif self.connection.needs_wireguard_configuration(): elif self.connection.needs_wireguard_configuration():
if self.has_wireguard_configuration(): if self.has_wireguard_configuration():
time_zone = self.get_wireguard_configuration_metadata('TZ') time_zone = self.get_wireguard_configuration_metadata('TZ')
if time_zone is None and self.has_location(): if time_zone is None and self.has_location():
time_zone = self.location.time_zone time_zone = self.location.time_zone
if time_zone is None: if time_zone is None:
raise UnknownTimeZoneError('The preferred time zone could not be determined.') raise UnknownTimeZoneError('The preferred time zone could not be determined.')
return time_zone return time_zone
def __delete_proxy_configuration(self): def __delete_proxy_configuration(self):
@ -116,34 +91,3 @@ class SessionProfile(BaseProfile):
def __delete_wireguard_configuration(self): def __delete_wireguard_configuration(self):
Path(self.get_wireguard_configuration_path()).unlink(missing_ok=True) Path(self.get_wireguard_configuration_path()).unlink(missing_ok=True)
def attach_operator_proxy_session(self, operator_proxy_session):
from core.models.OperatorProxySession import OperatorProxySession
operator_proxy_session_file_contents = f'{operator_proxy_session.to_json(indent=4)}\n'
os.makedirs(self.get_config_path(), exist_ok=True)
operator_proxy_session_file_path = self.get_operator_proxy_session_path()
with open(operator_proxy_session_file_path, 'w') as operator_proxy_session_file:
operator_proxy_session_file.write(operator_proxy_session_file_contents)
def get_operator_proxy_session_path(self):
return f'{self.get_config_path()}/operator_proxy_session.json'
def get_operator_proxy_session(self):
try:
config_file_contents = open(self.get_operator_proxy_session_path(), 'r').read()
except FileNotFoundError:
return None
try:
data = json.loads(config_file_contents)
except ValueError:
return None
from core.models.OperatorProxySession import OperatorProxySession
return OperatorProxySession.from_dict(data)
def has_operator_proxy_session(self):
return os.path.isfile(self.get_operator_proxy_session_path())