From 6dfd5993a29d279d034004a6b407ed1acb1645ca Mon Sep 17 00:00:00 2001 From: codeking Date: Sun, 18 Jan 2026 05:00:58 +0100 Subject: [PATCH] Update and refactor existing codebase --- core/Constants.py | 12 +++---- core/controllers/ApplicationController.py | 2 +- core/models/session/SessionState.py | 2 +- core/services/WebServiceApiService.py | 44 +++++++++++++++++------ 4 files changed, 41 insertions(+), 19 deletions(-) diff --git a/core/Constants.py b/core/Constants.py index ee1a514..2d20b14 100644 --- a/core/Constants.py +++ b/core/Constants.py @@ -6,8 +6,8 @@ import os @dataclass(frozen=True) class Constants: - SP_API_BASE_URL: Final[str] = os.environ.get('SP_API_BASE_URL') or 'https://api.hydraveil.net/api/v1' - PING_URL: Final[str] = os.environ.get('PING_URL') or 'https://api.hydraveil.net/api/v1/health' + SP_API_BASE_URL: Final[str] = os.environ.get('SP_API_BASE_URL', 'https://api.hydraveil.net/api/v1') + PING_URL: Final[str] = os.environ.get('PING_URL', 'https://api.hydraveil.net/api/v1/health') HV_CLIENT_PATH: Final[str] = os.environ.get('HV_CLIENT_PATH') HV_CLIENT_VERSION_NUMBER: Final[str] = os.environ.get('HV_CLIENT_VERSION_NUMBER') @@ -16,10 +16,10 @@ class Constants: SYSTEM_CONFIG_PATH: Final[str] = '/etc' - CACHE_HOME: Final[str] = os.environ.get('XDG_CACHE_HOME') or os.path.join(HOME, '.cache') - CONFIG_HOME: Final[str] = os.environ.get('XDG_CONFIG_HOME') or os.path.join(HOME, '.config') - DATA_HOME: Final[str] = os.environ.get('XDG_DATA_HOME') or os.path.join(HOME, '.local/share') - STATE_HOME: Final[str] = os.environ.get('XDG_STATE_HOME') or os.path.join(HOME, '.local/state') + CACHE_HOME: Final[str] = os.environ.get('XDG_CACHE_HOME', os.path.join(HOME, '.cache')) + CONFIG_HOME: Final[str] = os.environ.get('XDG_CONFIG_HOME', os.path.join(HOME, '.config')) + DATA_HOME: Final[str] = os.environ.get('XDG_DATA_HOME', os.path.join(HOME, '.local/share')) + STATE_HOME: Final[str] = os.environ.get('XDG_STATE_HOME', os.path.join(HOME, '.local/state')) HV_SYSTEM_CONFIG_PATH: Final[str] = f'{SYSTEM_CONFIG_PATH}/hydra-veil' diff --git a/core/controllers/ApplicationController.py b/core/controllers/ApplicationController.py index a14b296..3c1d1a3 100644 --- a/core/controllers/ApplicationController.py +++ b/core/controllers/ApplicationController.py @@ -152,7 +152,7 @@ class ApplicationController: return selected_items + available_preferred_items @staticmethod - def __run_process(initialization_file_path, profile, display, session_state): + def __run_process(initialization_file_path, profile, display, session_state: SessionState): if shutil.which('bwrap') is None: raise CommandNotFoundError('bwrap') diff --git a/core/models/session/SessionState.py b/core/models/session/SessionState.py index b1a38c4..0e8b232 100644 --- a/core/models/session/SessionState.py +++ b/core/models/session/SessionState.py @@ -71,7 +71,7 @@ class SessionState: shutil.rmtree(session_state_path, ignore_errors=True) @staticmethod - def __kill_associated_processes(session_state): + def __kill_associated_processes(session_state: 'SessionState'): associated_process_ids = list(session_state.process_ids) network_connections = psutil.net_connections() diff --git a/core/services/WebServiceApiService.py b/core/services/WebServiceApiService.py index 2dfd75c..b216915 100644 --- a/core/services/WebServiceApiService.py +++ b/core/services/WebServiceApiService.py @@ -18,10 +18,12 @@ class WebServiceApiService: @staticmethod def get_applications(proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/platforms/linux-x86_64/applications', None, proxies) applications = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for application in response.json()['data']: applications.append(Application(application['code'], application['name'], application['id'])) @@ -30,10 +32,12 @@ class WebServiceApiService: @staticmethod def get_application_versions(code: str, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get(f'/platforms/linux-x86_64/applications/{code}/application-versions', None, proxies) application_versions = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for application_version in response.json()['data']: application_versions.append(ApplicationVersion(code, application_version['version_number'], application_version['format_revision'], application_version['id'], application_version['download_path'], application_version['released_at'], application_version['file_hash'])) @@ -42,10 +46,12 @@ class WebServiceApiService: @staticmethod def get_client_versions(proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/platforms/linux-x86_64/appimage/client-versions', None, proxies) client_versions = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for client_version in response.json()['data']: client_versions.append(ClientVersion(client_version['version_number'], client_version['released_at'], client_version['id'], client_version['download_path'])) @@ -54,10 +60,12 @@ class WebServiceApiService: @staticmethod def get_operators(proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/operators', None, proxies) operators = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for operator in response.json()['data']: operators.append(Operator(operator['id'], operator['name'], operator['public_key'], operator['nostr_public_key'], operator['nostr_profile_reference'], operator['nostr_attestation']['event_reference'])) @@ -66,10 +74,12 @@ class WebServiceApiService: @staticmethod def get_locations(proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/locations', None, proxies) locations = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for location in response.json()['data']: locations.append(Location(location['country']['code'], location['code'], location['id'], location['country']['name'], location['name'], location['time_zone']['code'], location['operator_id'], location['provider']['name'], location['is_proxy_capable'], location['is_wireguard_capable'])) @@ -78,10 +88,12 @@ class WebServiceApiService: @staticmethod def get_subscription_plans(proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/subscription-plans', None, proxies) subscription_plans = [] - if response.status_code == 200: + if response.status_code == status_codes.OK: for subscription_plan in response.json()['data']: subscription_plans.append(SubscriptionPlan(subscription_plan['id'], subscription_plan['code'], subscription_plan['wireguard_session_limit'], subscription_plan['duration'], subscription_plan['price'], subscription_plan['features_proxy'], subscription_plan['features_wireguard'])) @@ -90,12 +102,14 @@ class WebServiceApiService: @staticmethod def post_subscription(subscription_plan_id, location_id, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__post('/subscriptions', None, { 'subscription_plan_id': subscription_plan_id, 'location_id': location_id }, proxies) - if response.status_code == 201: + if response.status_code == status_codes.CREATED: return Subscription(response.headers['X-Billing-Code']) else: @@ -104,13 +118,15 @@ class WebServiceApiService: @staticmethod def get_subscription(billing_code: str, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + billing_code = billing_code.replace('-', '').upper() billing_code_fragments = re.findall('....?', billing_code) billing_code = '-'.join(billing_code_fragments) response = WebServiceApiService.__get('/subscriptions/current', billing_code, proxies) - if response.status_code == 200: + if response.status_code == status_codes.OK: subscription = response.json()['data'] return Subscription(billing_code, Subscription.from_iso_format(subscription['expires_at'])) @@ -121,9 +137,11 @@ class WebServiceApiService: @staticmethod def get_invoice(billing_code: str, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/invoices/current', billing_code, proxies) - if response.status_code == 200: + if response.status_code == status_codes.OK: response_data = response.json()['data'] @@ -145,9 +163,11 @@ class WebServiceApiService: @staticmethod def get_proxy_configuration(billing_code: str, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__get('/proxy-configurations/current', billing_code, proxies) - if response.status_code == 200: + if response.status_code == status_codes.OK: proxy_configuration = response.json()['data'] return ProxyConfiguration(proxy_configuration['ip_address'], proxy_configuration['port'], proxy_configuration['username'], proxy_configuration['password'], proxy_configuration['location']['time_zone']['code']) @@ -158,11 +178,13 @@ class WebServiceApiService: @staticmethod def post_wireguard_session(country_code: str, location_code: str, billing_code: str, public_key: str, proxies: Optional[dict] = None): + from requests.status_codes import codes as status_codes + response = WebServiceApiService.__post(f'/countries/{country_code}/locations/{location_code}/wireguard-sessions', billing_code, { 'public_key': public_key, }, proxies) - if response.status_code == 201: + if response.status_code == status_codes.CREATED: return response.text else: return None