128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
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)
|