hydraveil-gui/gui/v2/actions/settings_data.py

import json
import os

from core.Constants import Constants
from core.controllers.ConfigurationController import ConfigurationController
from core.controllers.PolicyController import PolicyController
from core.controllers.ProfileController import ProfileController
from core.services.networking.systemwide.systemwide_utils import get_firewall_setting

from gui.v2.actions.config import load_gui_config
from gui.v2.actions.profile_order import normalize_profile_order
from gui.v2.actions.ticket_failure import get_ticket_verification_failure



def read_firewall_setting():
    try:
        firewall_setting = get_firewall_setting()
        return firewall_setting
    except Exception:
        pass
    return False


def read_systemwide_enabled():
    try:
        privilege_policy = PolicyController.get('privilege')
        if privilege_policy is not None:
            return PolicyController.is_instated(privilege_policy)
    except Exception:
        pass
    return False


def read_bwrap_enabled():
    try:
        capability_policy = PolicyController.get('capability')
        if capability_policy is not None:
            return PolicyController.is_instated(capability_policy)
    except Exception:
        pass
    return False


def prepare(gui_config_file):
    profiles = ProfileController.get_all()
    profile_order = normalize_profile_order(gui_config_file, profiles.keys())
    try:
        endpoint_verification_enabled = ConfigurationController.get_endpoint_verification_enabled()
    except Exception:
        endpoint_verification_enabled = False
    return {
        "profiles": profiles,
        "profile_order": profile_order,
        "current_connection": ConfigurationController.get_connection(),
        "endpoint_verification_enabled": endpoint_verification_enabled,
        "systemwide_enabled": read_systemwide_enabled(),
        "bwrap_enabled": read_bwrap_enabled(),
        "firewall_setting": read_firewall_setting(),
        "gui_config": load_gui_config(gui_config_file),
        "ticket_failure": get_ticket_verification_failure(gui_config_file),
    }


def empty_payload():
    return {
        "profiles": {},
        "profile_order": [],
        "current_connection": None,
        "endpoint_verification_enabled": False,
        "systemwide_enabled": False,
        "bwrap_enabled": False,
        "firewall_setting": False,
        "gui_config": None,
        "ticket_failure": None,
    }


def extract_endpoint_ip(profile):
    try:
        profile_path = Constants.HV_PROFILE_CONFIG_HOME + f'/{profile.id}'
        wg_conf_path = f'{profile_path}/wg.conf.bak'
        if not os.path.exists(wg_conf_path):
            return None
        with open(wg_conf_path, 'r') as f:
            content = f.read()
        for line in content.split('\n'):
            if line.strip().startswith('Endpoint = '):
                endpoint = line.strip().split(' = ')[1]
                return endpoint.split(':')[0]
        return None
    except Exception:
        return None


def truncate_key(text, max_length=50):
    if not text or text == "N/A" or len(text) <= max_length:
        return text
    start_len = max_length // 2 - 2
    end_len = max_length // 2 - 2
    return text[:start_len] + "....." + text[-end_len:]


def build_verification_view(profile):
    operator = None
    if profile and getattr(profile, 'location', None) and profile.location.operator:
        operator = profile.location.operator
    if not operator:
        return {
            "operator_name": "N/A",
            "nostr_public_key": "N/A",
            "hydraveil_public_key": "N/A",
            "nostr_attestation_event_reference": "N/A",
        }
    return {
        "operator_name": operator.name or "N/A",
        "nostr_public_key": operator.nostr_public_key or "N/A",
        "hydraveil_public_key": operator.public_key or "N/A",
        "nostr_attestation_event_reference": operator.nostr_attestation_event_reference or "N/A",
    }


def build_subscription_view(profile):
    subscription = profile.subscription
    billing_code = str(subscription.billing_code)
    if hasattr(subscription, 'expires_at') and subscription.expires_at:
        expires_at = subscription.expires_at.strftime("%Y-%m-%d %H:%M:%S UTC")
    else:
        expires_at = "Not available"
    return {"billing_code": billing_code, "expires_at": expires_at}


def format_ticket_failure_status(failure):
    if not failure:
        return "No saved verification failure. If ticket preparation fails validation, recovery data will appear here."
    failed_validations = failure.get("failed_validations", [])
    how_many_failed = failure.get("how_many_failed", len(failed_validations))
    updated_at = failure.get("updated_at", "unknown time")
    failed_text = ", ".join(str(item) for item in failed_validations)
    return (
        f"Saved verification failure: {how_many_failed} failed. "
        f"Failed validation indices: {failed_text}. Saved: {updated_at}"
    )


def format_ticket_recovery_result(label, result):
    try:
        payload = json.dumps(result, indent=2, default=str)
    except TypeError:
        payload = str(result)
    return f"{label}:\n{payload}"