124 lines
4 KiB
Python
124 lines
4 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.session.ProxyConfiguration import ProxyConfiguration
|
|
from core.models.session.SessionConnection import SessionConnection
|
|
from dataclasses import dataclass
|
|
from json import JSONDecodeError
|
|
from typing import Optional
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
|
|
|
|
@dataclass
|
|
class SessionProfile(BaseProfile):
|
|
resolution: str
|
|
application_version: Optional[ApplicationVersion]
|
|
connection: Optional[SessionConnection]
|
|
|
|
def has_connection(self):
|
|
return self.connection is not None
|
|
|
|
def save(self):
|
|
|
|
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)
|
|
|
|
super().save()
|
|
|
|
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)
|
|
proxy_configuration_file.close()
|
|
|
|
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)
|
|
wireguard_configuration_file.close()
|
|
|
|
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 JSONDecodeError:
|
|
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 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 __get_wireguard_configuration(self):
|
|
|
|
try:
|
|
with open(self.get_wireguard_configuration_path(), 'r') as file:
|
|
return file.readlines()
|
|
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
def __get_wireguard_configuration_metadata(self, key):
|
|
|
|
for line in self.__get_wireguard_configuration():
|
|
match = re.match(r'^# {} = (.*)$'.format(re.escape(key)), line)
|
|
|
|
if match:
|
|
return re.sub(r'[^a-zA-Z0-9+\- /]', '', match.group(1).strip())
|