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

Improved UI feedback on connections. And new plumbing UI data structure and prompts for stacktrace errors

Browse source
This commit is contained in:
SimplifiedPrivacy 2026-10-24 13:13:06 -04:00
parent 68c47f9a8c
commit c4a3e6c948
8 changed files with 97 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
core/errors/ErrorStack.py Normal file
View file

@ -0,0 +1,9 @@
from dataclasses import dataclass
@dataclass
class ErrorStack:
file_name: str = False
line_number: int = False
func_name: str = False
message_for_user: str = False
full_stacktrace: str = False

35
core/errors/translate_error.py
View file

@ -1,4 +1,5 @@
from core.controllers.ConfigurationController import ConfigurationController
from core.errors.ErrorStack import ErrorStack
from core.models.Result import Result, ResultError
from core.errors.logger import logger
from core.Errors import (
@ -8,9 +9,9 @@ from core.Errors import (
)
from core.utils.type_checker import type_checker
from typing import Type, Optional
import traceback
import os
ERROR_TO_UI_MESSAGE = {
ProfileActivationError: "The profile could not be enabled",
@ -65,27 +66,43 @@ def translate_error(error: Type[Exception], profile_id: int) -> Result:
return Result(valid=False, error_type=ResultError.MISSING_SOFTWARE, message=error_msg)
else:
error_msg = get_traceback_line(error)
return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg)
error_stack = get_traceback_line(error)
return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_stack.message_for_user)
def strip_filename(path: str) -> str:
"""Returns only the filename from a full system path."""
return os.path.basename(path)
def get_traceback_line(error) -> str:
# Extract the traceback from the exception object
def get_traceback_line(error) -> ErrorStack:
"""Extract the traceback from the exception object"""
tb = error.__traceback__
# extract_tb returns a list of frames; the LAST one [-1] is where the error happened
summary = traceback.extract_tb(tb)
if summary:
last_frame = summary[-1]
func_name = last_frame.name # The function name
line_no = last_frame.lineno # The line number
file_name = last_frame.filename # The file path
file_name = strip_filename(last_frame.filename) # The file path
else:
func_name, line_no, file_name = "Unknown", "Unknown", "Unknown"
# Generate the user-facing message and the full formatted stacktrace
error_msg = f"Error in {func_name} at {file_name}:{line_no} -> {str(error)}"
full_trace = "".join(traceback.format_exception(type(error), error, tb))
logger.error(error_msg)
return error_msg
error_stack = ErrorStack(
file_name=file_name,
line_number=line_no,
func_name=func_name,
message_for_user=error_msg,
full_stacktrace=full_trace
)
return error_stack
@type_checker

13
core/infrastructure/error_popup.py
View file

@ -1,3 +1,5 @@
from core.errors.ErrorStack import ErrorStack
from typing import Protocol, Optional, Callable, Any, Type
class PopupProtocol(Protocol):
@ -15,3 +17,14 @@ class PopupProtocol(Protocol):
def show(self) -> None:
"""Framework-neutral method to display the popup"""
...
class StackErrorProtocol(Protocol):
def __init__(
self,
error_data: ErrorStack,
**kwargs
) -> None: ...
def show_window(self) -> None:
"""Framework-neutral method to display the popup"""
...

34
core/infrastructure/plumbing.py
View file

@ -1,21 +1,19 @@
from core.infrastructure.thread_dispatch import MainThreadDispatcher
from core.infrastructure.error_popup import PopupProtocol
from core.infrastructure.error_popup import PopupProtocol, StackErrorProtocol
from core.infrastructure.service_registry import ServiceLocator
from core.errors.translate_error import get_traceback_line
from core.errors.ErrorStack import ErrorStack
from core.models.Result import Result, ResultError
# from core.infrastructure.action_registry import ENUM_MAP
from core.errors.logger import logger
# from infrastructure.observer_service import notify
# generic:
from concurrent.futures import ThreadPoolExecutor, Future
from typing import Callable, Any, Optional, Type
import inspect
from functools import partial
import sys
MAX_ATTEMPTS = 3
MAX_ATTEMPTS = 2
executor = ThreadPoolExecutor(max_workers=1)
class ThreadPlumbing:
@ -23,11 +21,12 @@ class ThreadPlumbing:
Manage Execution of functions on threads, and orchestrate their retries,
then pass the data through to popup prompts for the user to decide how to solve any error, by enum type.
"""
def __init__(self, action: Callable, ui_dispatcher: MainThreadDispatcher, error_implementation: Type[PopupProtocol], ENUM_MAP: dict, auto_navigate_dict: dict, ui_caller: Optional[Callable] = None, *args, **kwargs):
def __init__(self, action: Callable, ui_dispatcher: MainThreadDispatcher, error_implementation: Type[PopupProtocol], stack_error_implementation: Type[StackErrorProtocol], ENUM_MAP: dict, auto_navigate_dict: dict, ui_caller: Optional[Callable] = None, *args, **kwargs):
self.action = action
self.ui_dispatcher = ui_dispatcher
self.ui_caller = ui_caller
self.error_implementation = error_implementation
self.stack_error_implementation = stack_error_implementation
self.ENUM_MAP = ENUM_MAP
self.auto_navigate_dict = auto_navigate_dict
self.navigator = ServiceLocator.get("navigator")
@ -67,9 +66,10 @@ class ThreadPlumbing:
except Exception as e:
error_msg = f"There was an error: {str(e)}"
logger.error(error_msg)
stack_trace = get_traceback_line(e)
logger.error(stack_trace)
result = Result(valid=False, error_type=ResultError.UNKNOWN, message=str(e))
error_stack = get_traceback_line(e)
error_msg = error_stack.message_for_user
logger.error(error_msg)
result = Result(valid=False, error_stack=error_stack, error_type=ResultError.UNKNOWN, message=error_msg)
finally:
# Back to main thread
@ -81,6 +81,20 @@ class ThreadPlumbing:
logger.info(f"We are on the main thread with result {result}")
#----------------------------
# OUTRIGHT ERROR STACK
#----------------------------
if result.error_stack:
error_window = self.stack_error_implementation(error_data=result.error_stack)
user_choice = error_window.show_window()
print(f"user_choice is {user_choice}")
if user_choice == 2:
sys.exit()
return
#----------------------------
# MAX TRIES
#----------------------------
if result.valid or self.attempt_count >= MAX_ATTEMPTS:
if self.ui_caller:
print("calling back with result")

2
core/models/Result.py
View file

@ -1,4 +1,5 @@
from core.services.networking.api_requests.ApiResponseModel import BackoffStrategy
from core.errors.ErrorStack import ErrorStack
from enum import Enum
from dataclasses import dataclass
@ -63,6 +64,7 @@ class Result():
valid: bool
goal_result: Optional[bool] = None
error_type: ResultError = ResultError.SUCCESS
error_stack: Optional[ErrorStack] = None
data: Optional[Any] = None
message: Optional[str] = None
goal: Optional[str] = None

6
core/services/networking/general_connection_tools/ip.py
View file

@ -32,6 +32,8 @@ API_ENDPOINT_FORMATS = {
MAX_ATTEMPTS = 3
def test(connection_observer: Optional[ConnectionObserver], port: Optional[int] = None) -> Result:
if connection_observer:
connection_observer.notify("connecting", "Testing IP..")
return asyncio.run(_test(connection_observer, port))
@ -50,8 +52,8 @@ async def _test(connection_observer: Optional[ConnectionObserver], port: Optiona
async def attempt_request(url: str, index: int):
logger.info(f"Attempt {index + 1}: Trying {url}")
if connection_observer:
connection_observer.notify("connecting", f"Testing IP. Attempt {index + 1}")
# if connection_observer:
# connection_observer.notify("connecting", f"Testing IP. Attempt {index + 1}")
try:
raw_response = await client.get(url)

19
core/services/networking/systemwide/systemwide_wireguard.py
View file

@ -316,8 +316,11 @@ def establish_system_connection(
# We allow 3 total attempts (Initial + 2 retries)
for attempt in range(1, 4):
try:
if connection_observer:
connection_observer.notify("connecting", f"Connecting attempt {attempt} of 3.. ")
# if connection_observer:
# connection_observer.notify("connecting", f"Connecting attempt {attempt} of 3.. ")
notification = f"Connecting attempt {attempt} of 3.. "
Notifications.emit(EventType.STATUS_UPDATE, notification)
# Attempt the actual system connection
connection_result = __establish_system_connection(
@ -331,7 +334,9 @@ def establish_system_connection(
return _finish_function(connection_result)
# If not valid, we treat it as a failure to trigger the except block logic
raise ConnectionError("Connection result invalid")
notification = f"{attempt} attempt failed.."
Notifications.emit(EventType.STATUS_UPDATE, notification)
raise ConnectionError(notification)
except (ConnectionError, CalledProcessError) as e:
_cleanup_on_error(firewall_setting, dns_setting)
@ -390,11 +395,15 @@ def establish_system_connection(
logger.info("Skipping MTU testing, it's not relevant here.")
# If MTU wasn't changed or wasn't applicable, don't bother with attempt 3
raise ConnectionError("Failed to connect.")
notification = "Failed to connect."
Notifications.emit(EventType.STATUS_UPDATE, notification)
raise ConnectionError(notification)
# return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")
else:
# Attempt 3 failed
logger.error("Maximum retries reached. Connection failed.")
raise ConnectionError("Failed to connect.")
notification = "Failed to connect."
Notifications.emit(EventType.STATUS_UPDATE, notification)
raise ConnectionError(notification)
# return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")

9
core/services/networking/systemwide/wireguard/mtu_tools.py
View file

@ -1,4 +1,4 @@
from core.observers.Notifications import Notifications, EventType
# from core.observers.Notifications import Notifications, EventType
from core.utils.run_commands import run_generic_command
from core.models.Result import Result, ResultError
from core.errors.logger import logger
@ -33,7 +33,8 @@ def mtu_is_fine(current_mtu: Optional[int] = None) -> bool:
testing_mtu = MAX_MTU
notification = f"Testing MTU {testing_mtu}..."
Notifications.emit(EventType.STATUS_UPDATE, notification)
logger.info(notification)
# Notifications.emit(EventType.STATUS_UPDATE, notification)
for url in URLS:
result = test_mtu_without_async(mtu_value=testing_mtu, url=url, timeout=6)
@ -74,7 +75,7 @@ async def find_optimal_mtu_binary_search(low: int, high: int, url: str) -> int:
while low <= high:
mid = (low + high) // 2
notification = f"Binary searching: Testing MTU {mid} against {url}..."
Notifications.emit(EventType.STATUS_UPDATE, notification)
# Notifications.emit(EventType.STATUS_UPDATE, notification)
logger.info(notification)
result = await test_an_mtu(mid, url, timeout=current_timeout)
@ -93,7 +94,7 @@ async def loop_through_testing_mtu():
for each_mtu in INITIAL_TEST_LIST:
notification = f"Probing MTU {each_mtu} across all IPs..."
Notifications.emit(EventType.STATUS_UPDATE, notification)
# Notifications.emit(EventType.STATUS_UPDATE, notification)
logger.info(notification)
# Create tasks for all URLs for the current MTU value