from core.Constants import Constants from core.Errors import UnknownTimeZoneError from core.models.BaseProfile import BaseProfile # from core.models.session.ApplicationVersion import ApplicationVersion from core.models.orm_models.ApplicationVersion import ApplicationVersion from core.models.session.ProxyConfiguration import ProxyConfiguration from core.models.session.SessionConnection import SessionConnection from dataclasses import dataclass from pathlib import Path from typing import Optional import json import os import shutil @dataclass class SessionProfile(BaseProfile): resolution: str application_version: Optional[ApplicationVersion] connection: Optional[SessionConnection] = None ticket: Optional[int] = None def has_connection(self): return self.connection is not None def save(self): print("We are able to trigger save on the child session") if 'application_version' in self._get_dirty_keys(): persistent_state_path = f'{self.get_data_path()}/persistent-state' if os.path.isdir(persistent_state_path): shutil.rmtree(persistent_state_path, ignore_errors=True) if 'location' in self._get_dirty_keys(): self.__delete_proxy_configuration() self.__delete_wireguard_configuration() # === APPLICATION === app_version_dict = self.application_version.convert_to_dict() print(f"session child got {app_version_dict} as dict..") super().save(app_version_dict=app_version_dict) def attach_proxy_configuration(self, proxy_configuration): proxy_configuration_file_contents = f'{proxy_configuration.to_json(indent=4)}\n' os.makedirs(Constants.HV_CONFIG_HOME, exist_ok=True) proxy_configuration_file_path = self.get_proxy_configuration_path() with open(proxy_configuration_file_path, 'w') as proxy_configuration_file: proxy_configuration_file.write(proxy_configuration_file_contents) def attach_wireguard_configuration(self, wireguard_configuration): wireguard_configuration_file_path = self.get_wireguard_configuration_path() with open(wireguard_configuration_file_path, 'w') as wireguard_configuration_file: wireguard_configuration_file.write(wireguard_configuration) def get_proxy_configuration_path(self): return f'{self.get_config_path()}/proxy.json' def get_wireguard_configuration_path(self): return f'{self.get_config_path()}/wg.conf' def get_proxy_configuration(self): try: config_file_contents = open(self.get_proxy_configuration_path(), 'r').read() except FileNotFoundError: return None try: proxy_configuration = json.loads(config_file_contents) except ValueError: return None proxy_configuration = ProxyConfiguration.from_dict(proxy_configuration) return proxy_configuration def has_proxy_configuration(self): return os.path.isfile(f'{self.get_config_path()}/proxy.json') def has_wireguard_configuration(self): return os.path.isfile(f'{self.get_config_path()}/wg.conf') def address_security_incident(self): super().address_security_incident() self.__delete_wireguard_configuration() def determine_timezone(self): time_zone = None if self.has_connection(): if self.connection.needs_proxy_configuration(): if self.has_proxy_configuration(): time_zone = self.get_proxy_configuration().time_zone elif self.connection.needs_wireguard_configuration(): if self.has_wireguard_configuration(): time_zone = self.get_wireguard_configuration_metadata('TZ') if time_zone is None and self.has_location(): time_zone = self.location.time_zone if time_zone is None: raise UnknownTimeZoneError('The preferred time zone could not be determined.') return time_zone def __delete_proxy_configuration(self): Path(self.get_proxy_configuration_path()).unlink(missing_ok=True) def __delete_wireguard_configuration(self): Path(self.get_wireguard_configuration_path()).unlink(missing_ok=True)