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

Update and refactor existing codebase

Browse source
This commit is contained in:
codeking 2026-01-18 05:00:58 +01:00
parent f8304aebf2
commit 6dfd5993a2
4 changed files with 41 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
core/Constants.py
View file

@ -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'

2
core/controllers/ApplicationController.py
View file

@ -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')

2
core/models/session/SessionState.py
View file

@ -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()

44
core/services/WebServiceApiService.py
View file

@ -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