forked from Support/sp-hydra-veil-gui
138 lines
4.7 KiB
Python
Executable file
138 lines
4.7 KiB
Python
Executable file
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 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_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(),
|
|
"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,
|
|
"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}"
|