sp-hydra-veil-core/core/services/WebServiceApiService.py

from core.Constants import Constants
from core.models.ClientVersion import ClientVersion
from core.models.Location import Location
from core.models.Subscription import Subscription
from core.models.SubscriptionPlan import SubscriptionPlan
from core.models.invoice.Invoice import Invoice
from core.models.invoice.PaymentMethod import PaymentMethod
from core.models.session.Application import Application
from core.models.session.ApplicationVersion import ApplicationVersion
from core.models.session.ProxyConfiguration import ProxyConfiguration
from typing import Optional
import re
import requests


class WebServiceApiService:

    @staticmethod
    def get_applications(proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/platforms/linux-x86_64/applications', None, proxies)
        applications = []

        if response.status_code == requests.codes.ok:
            for application in response.json()['data']:
                applications.append(Application(application['code'], application['name'], application['id']))

        return applications

    @staticmethod
    def get_application_versions(code: str, proxies: Optional[dict] = None):

        response = WebServiceApiService.__get(f'/platforms/linux-x86_64/applications/{code}/application-versions', None, proxies)
        application_versions = []

        if response.status_code == requests.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']))

        return application_versions

    @staticmethod
    def get_client_versions(proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/platforms/linux-x86_64/client-versions', None, proxies)
        client_versions = []

        if response.status_code == requests.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']))

        return client_versions

    @staticmethod
    def get_locations(proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/locations', None, proxies)
        locations = []

        if response.status_code == requests.codes.ok:
            for location in response.json()['data']:
                locations.append(Location(location['code'], location['id'], location['name']))

        return locations

    @staticmethod
    def get_subscription_plans(proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/subscription-plans', None, proxies)
        subscription_plans = []

        if response.status_code == requests.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']))

        return subscription_plans

    @staticmethod
    def post_subscription(subscription_plan_id, location_id, proxies: Optional[dict] = None):

        response = WebServiceApiService.__post('/subscriptions', None, {
            'subscription_plan_id': subscription_plan_id,
            'location_id': location_id
        }, proxies)

        if response.status_code == requests.codes.created:
            return Subscription(response.headers['X-Billing-Code'])

    @staticmethod
    def get_subscription(billing_code: str, proxies: Optional[dict] = None):

        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 == requests.codes.ok:

            subscription = response.json()['data']
            return Subscription(billing_code, Subscription.from_iso_format(subscription['expires_at']))

    @staticmethod
    def get_invoice(billing_code: str, proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/invoices/current', billing_code, proxies)

        if response.status_code == requests.codes.ok:

            response_data = response.json()['data']

            invoice = {
                'status': response_data['status'],
                'expires_at': response_data['expires_at']
            }

            payment_methods = []

            for payment_method in response_data['payment_methods']:
                payment_methods.append(PaymentMethod(payment_method['code'], payment_method['name'], payment_method['address'], payment_method['payment_link'], payment_method['rate'], payment_method['amount'], payment_method['due']))

            return Invoice(billing_code, invoice['status'], invoice['expires_at'], tuple[PaymentMethod](payment_methods))

    @staticmethod
    def get_proxy_configuration(billing_code: str, proxies: Optional[dict] = None):

        response = WebServiceApiService.__get('/proxy-configurations/current', billing_code, proxies)

        if response.status_code == requests.codes.ok:

            proxy_configuration = response.json()['data']
            return ProxyConfiguration(proxy_configuration['ip_address'], proxy_configuration['port'], proxy_configuration['username'], proxy_configuration['password'])

        else:
            return None

    @staticmethod
    def post_wireguard_session(location_code: str, billing_code: str, public_key: str, proxies: Optional[dict] = None):

        response = WebServiceApiService.__post(f'/locations/{location_code}/wireguard-sessions', billing_code, {
            'public_key': public_key,
        }, proxies)

        if response.status_code == requests.codes.created:
            return response.text
        else:
            return None

    @staticmethod
    def __get(path, billing_code: Optional[str] = None, proxies: Optional[dict] = None):

        if billing_code is not None:
            headers = {'X-Billing-Code': billing_code}
        else:
            headers = None

        return requests.get(Constants.SP_API_BASE_URL + path, headers=headers, proxies=proxies)

    @staticmethod
    def __post(path, billing_code: Optional[str] = None, body: Optional[dict] = None, proxies: Optional[dict] = None):

        if billing_code is not None:
            headers = {'X-Billing-Code': billing_code}
        else:
            headers = None

        return requests.post(Constants.SP_API_BASE_URL + path, headers=headers, json=body, proxies=proxies)