diff --git a/change_log.md b/change_log.md
index c90d71d..2567964 100644
--- a/change_log.md
+++ b/change_log.md
@@ -3,6 +3,11 @@
# 2.4.3
### Systemwide Killswitch Introduced for Enable
+
+### July 23, 2026
+Separation of performing subprocess commands, parsing them for error strings, and handling those errors as enum types. Systemwide error handling was introduced via a wrapper to all systemwide functions requiring the sudo scripts.
+
+
### July 20, 2026
Introduced IP-interface-based killswitch support for systemwide wireguard, on enable only. This version is stable and working for enable only, with a "test" script in the fileslot. And fixed GUI's object save creation. Additionally rolled back Connection data models to previous non-enum version.
diff --git a/core/assets/sudo_scripts/dns b/core/assets/sudo_scripts/dns
index d102c29..641f538 100644
--- a/core/assets/sudo_scripts/dns
+++ b/core/assets/sudo_scripts/dns
@@ -47,7 +47,11 @@ if [[ "$ACTION" == "set" ]]; then
_set_dns
fi
-if [[ "$ACTION" == "revert" ]]; then
- resolvectl revert "$IFACE" 2>/dev/null || true
+# if [[ "$ACTION" == "revert" ]]; then
+# resolvectl revert "$IFACE" 2>/dev/null || true
+# fi
+
+if [[ "$ACTION" == "revert" ]]; then
+ resolvectl revert "$IFACE" || true
fi
diff --git a/core/controllers/SystemStateController.py b/core/controllers/SystemStateController.py
index a5af5f0..de25700 100644
--- a/core/controllers/SystemStateController.py
+++ b/core/controllers/SystemStateController.py
@@ -13,7 +13,9 @@ class SystemStateController:
@staticmethod
def create(profile_id: int, firewalled: bool, dns_set: bool) -> SystemState:
- return SystemState(profile_id, firewalled, dns_set).save()
+ current_state = SystemState(profile_id, firewalled, dns_set)
+ current_state.save()
+ return current_state
@staticmethod
def update_or_create(system_state):
diff --git a/core/errors/exceptions.py b/core/errors/exceptions.py
index cd81195..133f8bd 100644
--- a/core/errors/exceptions.py
+++ b/core/errors/exceptions.py
@@ -1,3 +1,25 @@
+from core.models.Result import Result, ResultError
+
+class SudoScript(Exception):
+ """There are issues with the sudo scripts"""
+ def __init__(self, result: Result):
+ self.result = result
+ if result.message and result.message is not None:
+ error_msg = result.message
+ else:
+ error_msg = result.error_type
+ super().__init__(error_msg)
+
+class MissingPreReqs(Exception):
+ """User can't take an action because they lack the prereqs"""
+ def __init__(self, result: Result):
+ self.result = result
+ if result.message and result.message is not None:
+ error_msg = result.message
+ else:
+ error_msg = result.error_type
+ super().__init__(error_msg)
+
class TorServiceInitializationError(Exception):
pass
diff --git a/core/models/Result.py b/core/models/Result.py
index 796199d..bfccef2 100644
--- a/core/models/Result.py
+++ b/core/models/Result.py
@@ -9,6 +9,8 @@ class ResultError(Enum):
SUCCESS = "success"
NOT_SUPPORTED = "NOT_SUPPORTED"
INVALID_INPUT = "invalid_input"
+ MISSING_FILE = "missing_file"
+ MISSING_DEPENDENCY = "missing_dependency"
CONNECTION = "connection"
DATABASE = "database"
PERMISSION = "permission"
@@ -17,6 +19,8 @@ class ResultError(Enum):
FIREWALL = "firewall"
CLIENT_DNS = "client_dns"
EXTERNAL_DNS = "external_dns"
+ INTERFACE = "interface"
+ TIMEOUT = "timeout"
UNKNOWN = "unknown"
@dataclass
@@ -25,6 +29,7 @@ class Result():
error_type: ResultError = ResultError.SUCCESS
data: Optional[Any] = None
message: Optional[str] = None
+ goal: Optional[str] = None
def user_message(self) -> str:
"""Human-readable error for the UI."""
diff --git a/core/services/networking/systemwide/dns.py b/core/services/networking/systemwide/dns.py
index 85f9dd5..f1315b1 100644
--- a/core/services/networking/systemwide/dns.py
+++ b/core/services/networking/systemwide/dns.py
@@ -1,69 +1,49 @@
# custom
+from core.services.networking.systemwide.systemwide_errors import systemwide_hell_raiser
+from core.utils.basic_operations.wrap_with import wrap_with
+from core.utils.basic_operations.run_generic_command import run_generic_command
from core.Constants import Constants
from core.models.Result import Result, ResultError
from core.errors.logger import logger
-import subprocess
-from subprocess import CalledProcessError
-
DNS_SCRIPT = f"/opt/hydra-veil/dns"
-
# requires script to have sudo
+@wrap_with(systemwide_hell_raiser)
def set_on(network_interface: str, dns_should_be: str) -> Result:
- result = subprocess.run(
- ["sudo", DNS_SCRIPT, "set", network_interface, dns_should_be],
- capture_output=True,
- text=True,
- )
+ command = ["sudo", DNS_SCRIPT, "set", network_interface, dns_should_be]
+ human_readable_goal = "Set DNS"
+ return run_generic_command(command, human_readable_goal)
- if result.returncode == 0:
- logger.info("DNS setup successful")
- return Result(valid=True)
- else:
- logger.error(f"DNS setup failed: {result.stderr.strip()}")
- return Result(valid=False, message=result.stderr.strip())
# requires script to have sudo
+@wrap_with(systemwide_hell_raiser)
def revert() -> Result:
- try:
- result = subprocess.run(
- ["sudo", DNS_SCRIPT, "revert"],
- capture_output=True,
- text=True,
- )
-
- if result.returncode == 0:
- logger.info("DNS revert successful")
- return Result(valid=True)
- else:
- logger.error(f"DNS revert failed: {result.stderr.strip()}")
- return Result(valid=False, message=result.stderr.strip())
- except CalledProcessError as e:
- return Result(valid=False, message=e)
+ command = ["sudo", DNS_SCRIPT, "revert"]
+ human_readable_goal = "DNS revert"
+ return run_generic_command(command, human_readable_goal)
# does NOT need sudo
def is_systemd_enabled() -> Result:
- try:
- process = subprocess.Popen(('systemctl', 'is-enabled', 'systemd-resolved'), stdout=subprocess.PIPE)
- process_output = str(process.stdout.read())
- if 'enabled' in process_output:
+ command = ["systemctl", "is-enabled", "systemd-resolved"]
+ human_readable_goal = "Find out if SystemD is enabled"
+ result = run_generic_command(command, human_readable_goal, timeout=5)
+ if result.valid:
+ if "enabled" in result.data:
return Result(valid=True)
else:
- return Result(valid=False, message=process_output)
- except CalledProcessError as e:
- return Result(valid=False, message=e)
+ logger.error("Systemctl is not even enabled.")
+ return Result(valid=False) # this would be true in the original object just because the command ran.
+ else:
+ return result # send the error itself.
# does NOT need sudo
def get_resolvctl_data(which_command: str) -> Result:
- try:
- process_output = subprocess.check_output(('resolvectl', which_command), text=True)
- return Result(valid=True, data=process_output)
-
- except CalledProcessError as e:
- return Result(valid=False, message=e)
+ command = ['resolvectl', which_command]
+ human_readable_goal = "Get Resolvctl Data"
+ return run_generic_command(command, human_readable_goal, timeout=5)
# These are the raw Linux systemctl commands:
diff --git a/core/services/networking/systemwide/killswitch.py b/core/services/networking/systemwide/killswitch.py
index a12318c..53b1ddd 100644
--- a/core/services/networking/systemwide/killswitch.py
+++ b/core/services/networking/systemwide/killswitch.py
@@ -1,41 +1,37 @@
-import subprocess
+from core.services.networking.systemwide.systemwide_errors import systemwide_hell_raiser
+from core.utils.basic_operations.wrap_with import wrap_with
+from core.utils.basic_operations.run_generic_command import run_generic_command
from core.models.Result import Result, ResultError
from core.errors.logger import logger
-
+import subprocess
KILLSWITCH_WRAPPER = "/opt/hydra-veil/firewall" # needs chmod 755
+@wrap_with(systemwide_hell_raiser)
def arm(server_ip: str, tunnel_if: str, internal_subnet: str = None) -> Result:
- cmd = ["sudo", KILLSWITCH_WRAPPER, "arm", server_ip, tunnel_if]
+ command = ["sudo", KILLSWITCH_WRAPPER, "arm", server_ip, tunnel_if]
if internal_subnet:
- cmd.append(internal_subnet)
- result = subprocess.run(cmd, capture_output=True, text=True)
- if result.returncode != 0:
- error_log = f"Failed to arm firewall: {result.stderr.strip()}"
- logger.error(f"[KILLSWITCH] {error_log}")
- return Result(valid=False, error_type=ResultError.FIREWALL, message=error_log)
- else:
- return Result(valid=True)
-
+ command.append(internal_subnet)
+ human_readable_goal = "Arming the Firewall"
+ return run_generic_command(command, human_readable_goal)
+@wrap_with(systemwide_hell_raiser)
def disarm() -> bool:
- result = subprocess.run(
- ["sudo", KILLSWITCH_WRAPPER, "disarm"],
- capture_output=True,
- text=True,
- )
- if result.returncode != 0:
- error_msg = f"Failed to disarm: {result.stderr.strip()}"
- logger.error(f"[KILLSWITCH] {error_log}")
- return Result(valid=False, error_type=ResultError.FIREWALL, message=error_log)
- else:
- return Result(valid=True)
-
+ command = ["sudo", KILLSWITCH_WRAPPER, "disarm"]
+ human_readable_goal = "Disarming the Firewall"
+ return run_generic_command(command, human_readable_goal)
def status() -> bool:
- result = subprocess.run(
- ["sudo", KILLSWITCH_WRAPPER, "status"],
- capture_output=True,
- text=True,
- )
- return result.returncode == 0 and result.stdout.strip() == "armed"
\ No newline at end of file
+ command = ["sudo", KILLSWITCH_WRAPPER, "status"]
+ human_readable_goal = "Get the Firewall's status"
+ result = run_generic_command(command, human_readable_goal, timeout=5)
+ if result.valid:
+ if "armed" in result.data.strip():
+ return True
+ else:
+ logger.error("[FIREWALL STATUS] It is currently NOT armed.")
+ return False
+ else:
+ # process error with running the command itself,
+ # send the error object with details.
+ return result
diff --git a/core/services/networking/systemwide/systemwide_errors.py b/core/services/networking/systemwide/systemwide_errors.py
new file mode 100644
index 0000000..0d130b1
--- /dev/null
+++ b/core/services/networking/systemwide/systemwide_errors.py
@@ -0,0 +1,54 @@
+from core.errors.exceptions import SudoScript, MissingPreReqs
+from core.models.Result import Result, ResultError
+from core.errors.logger import logger
+
+"""
+Purpose:
+ Raises Errors for known problems with the systemwide connections
+
+Kinds of Raises:
+ SudoScript
+ MissingPreReqs
+"""
+
+def systemwide_hell_raiser(result: Result) -> Result:
+ if result.valid:
+ return result
+
+ logger.error(f"[HELL RAISER] We were trying to {result.goal} but got an error of {result.error_type} with a human message of {result.message}")
+
+ if result.error_type == ResultError.MISSING_FILE:
+ logger.error(f"[HELL RAISER] We are missing the setup scripts for {result.goal}")
+ result.message = f"You're missing the setup scripts for {result.goal}. Re-do the install, or check the setup you did."
+ raise SudoScript(result)
+
+ elif result.error_type == ResultError.PERMISSION:
+ error_msg = f"There's improper permissions on the scripts for {result.goal}. Re-do the install, or check the setup you did."
+ result.message = error_msg
+ logger.error(f"[HELL RAISER] {error_msg}")
+ raise SudoScript(result)
+
+ elif result.error_type == ResultError.MISSING_DEPENDENCY:
+ error_output = result.message
+ list_of_dependencies = ["nmcli", "resolvconf", "nftables", "nft"]
+ what_is_missing = []
+
+ for each_dependency in list_of_dependencies:
+ if each_dependency in error_output:
+ what_is_missing.append(each_dependency)
+
+ result.message = f"You're missing the prereq packages of {what_is_missing}. Re-do the install, or check the setup you did."
+ raise MissingPreReqs(result)
+
+ elif result.error_type == ResultError.INTERFACE:
+ logger.info("The Interface is not found.")
+ # might not be an error, if it's already down,
+ return result
+
+ elif result.error_type == ResultError.UNKNOWN:
+ logger.info(f"Unknown error type for {result.goal} with {result.message} and potentially output of {result.data}.")
+ return result
+
+ else:
+ logger.info("Unknown Result object")
+ return result
\ No newline at end of file
diff --git a/core/services/networking/systemwide/systemwide_wireguard.py b/core/services/networking/systemwide/systemwide_wireguard.py
index 3bc9da8..e909673 100644
--- a/core/services/networking/systemwide/systemwide_wireguard.py
+++ b/core/services/networking/systemwide/systemwide_wireguard.py
@@ -53,7 +53,7 @@ def revert_dns() -> Result:
turn_off = dns.revert()
logger.info(f"The result of the turn off is {turn_off.valid}")
if not turn_off.valid:
- logger.error(f"Critical DNS Error! Could NOT turn off DNS. {turn_off.message}")
+ logger.error(f"Critical DNS Error! Could NOT turn off DNS. Error type: {turn_off.error_type} with {turn_off.message}")
return turn_off.valid
@@ -100,6 +100,8 @@ def check_and_kill_firewall():
else:
error_msg = f"Firewall can't be disabled: {disable_result_object.message}"
logger.error(error_msg)
+
+ # if the systemwide_raiser didn't catch it with the reason, then..
raise ConnectionTerminationError(error_msg)
else:
logger.info("We are skipping disabling the firewall, because it's already off.")
@@ -235,6 +237,16 @@ def __establish_system_connection(
if not sinkhole_result:
raise ConnectionError('IPv6 Could not be protected.')
+ # ============= SETUP STATE =============
+ # Even if firewall is off, we want to save the fact we turned on the VPN tunnel, before we raise errors.
+
+ logger.info(f"Setting State JSON..")
+ current_state = SystemStateController.create(
+ profile_id=profile.id,
+ firewalled=firewall_setting, # intended
+ dns_set=dns_setting # intended
+ )
+
# ============= PREP SETTINGS =============
if firewall_setting or dns_setting:
wg_interface_result = extract_wg_interface_name(process_output)
@@ -258,6 +270,22 @@ def __establish_system_connection(
if turn_on_result.valid:
firewall_tracker = True
+ # ============= CONFIRM ON FIREWALL =============
+ # if firewall_setting:
+ logger.info(f"Evaluating Firewall Situation after we tried to turn on it on.")
+ firewall_status = killswitch.status()
+ logger.info(f"""
+The firewall setting is on.
+The result of arming it, {turn_on_result.valid}
+The result of the status {firewall_status}
+ """)
+ if firewall_status:
+ logger.info("Firewall is on. All good.")
+ else:
+ logger.error("Firewall FAILED TO ARM") # DON'T TRY TO RENEGOTIATE WITH DEAD INTERNET !!
+
+ # raise ConnectionError(f'Firewall failed to arm!! {turn_on_result.message}.') # if this triggers, it tries to renegotiate with dead internet.
+
# ============= DNS =============
if dns_setting:
set_dns_result = set_dns(str(profile.id), wg_interface_name)
@@ -274,39 +302,19 @@ def __establish_system_connection(
did_dns_work = False
logger.info("Skipping DNS setup, because it's off in settings")
- # ============= SETUP STATE =============
- # Even if firewall is off, we want to save the fact we turned on the VPN tunnel, before we raise errors.
-
- logger.info(f"Setting State JSON..")
- SystemStateController.create(
- profile_id=profile.id,
- firewalled=firewall_tracker,
- dns_set=did_dns_work
- )
-
- # ============= CONFIRM ON FIREWALL =============
- if firewall_setting:
- logger.info(f"Evaluating Firewall Situation after we tried to turn on it on.")
- firewall_status = killswitch.status()
- logger.info(f"""
-The firewall setting is on.
-The result of arming it, {turn_on_result.valid}
-The result of the status {firewall_status}
- """)
- if firewall_status:
- logger.info("Firewall is on. All good.")
- else:
- logger.error("Firewall FAILED TO ARM") # TRIES TO RENEGOTIATE WITH DEAD INTERNET !!
- # raise ConnectionError(f'Firewall failed to arm!! {turn_on_result.message}.') # if this triggers, it tries to renegotiate with dead internet.
-
- logger.info("If we made it this far, we have nmcli on, ipv6 sinkhole on, firewall on.")
# ============= TESTING =============
+ logger.info("If we made it this far, we have nmcli on, ipv6 sinkhole on, firewall on.")
try:
await_connection(connection_observer=connection_observer)
except ConnectionError:
raise ConnectionError('The connection could not be established.')
+ # ============= UPDATE STATE =============
+ current_state.firewalled = firewall_tracker
+ current_state.dns_set = did_dns_work
+ SystemStateController.update_or_create(current_state)
+
return Result(valid=True)
@@ -339,7 +347,8 @@ def establish_system_connection(
# ================= SETTINGS =================
firewall_setting = get_firewall_setting()
- dns_setting = get_dns_setting()
+ dns_setting = True # for testing
+ # dns_setting = get_dns_setting()
# ================= CONNECT WG ================
try:
diff --git a/core/utils/basic_operations/run_generic_command.py b/core/utils/basic_operations/run_generic_command.py
new file mode 100644
index 0000000..5020425
--- /dev/null
+++ b/core/utils/basic_operations/run_generic_command.py
@@ -0,0 +1,66 @@
+from core.models.Result import Result, ResultError
+from core.errors.logger import logger
+
+import subprocess
+import os
+
+_SUDO_KW = {
+ "stdout": subprocess.PIPE,
+ "stderr": subprocess.PIPE,
+ "stdin": subprocess.DEVNULL,
+ "env": {**os.environ, "SUDO_ASKPASS": "/bin/false"},
+ "text": True,
+}
+
+
+def parse_errors(error_output: str) -> ResultError:
+ # Interface missing
+ if "No such device" in error_output:
+ error_enum = ResultError.INTERFACE
+ # Permission error
+ elif "No permissions" in error_output:
+ error_enum = ResultError.PERMISSION
+ elif "sudo: a password is required" or "password is required" in error_output:
+ error_enum = ResultError.PERMISSION
+ elif "command not found" in error_output:
+ error_enum = ResultError.MISSING_DEPENDENCY
+ else:
+ error_enum = ResultError.UNKNOWN
+ return error_enum
+
+
+def run_generic_command(
+ command: list,
+ human_readable_goal: str,
+ timeout: float | None = None
+) -> Result:
+
+ output_data = None # incase command has nothing.
+
+ # Add -n flag to sudo for non-interactive mode (no prompts)
+ if command and command[0] == "sudo":
+ if "-n" not in command:
+ command = [command[0], "-n"] + command[1:]
+
+ try:
+ result = subprocess.run(
+ command,
+ **_SUDO_KW,
+ timeout=timeout
+ )
+ output_data = result.stdout
+ if result.returncode == 0:
+ logger.info(f"{human_readable_goal} was successful")
+ return Result(valid=True, data=output_data)
+ else:
+ error_output = result.stderr.strip()
+ error_enum = parse_errors(error_output)
+ logger.error(f"{human_readable_goal} Failed, error type is {error_enum}.")
+ return Result(valid=False, error_type=error_enum, goal=human_readable_goal, message=error_output)
+ except subprocess.TimeoutExpired:
+ return Result(valid=False, error_type=ResultError.TIMEOUT, goal=human_readable_goal, data=output_data, message="Command timed out")
+ except FileNotFoundError:
+ return Result(valid=False, error_type=ResultError.MISSING_FILE, goal=human_readable_goal, data=output_data, message="Script is not found")
+ except Exception as e:
+ return Result(valid=False, error_type=ResultError.UNKNOWN, goal=human_readable_goal, data=output_data, message=e)
+
diff --git a/core/utils/basic_operations/wrap_with.py b/core/utils/basic_operations/wrap_with.py
new file mode 100644
index 0000000..d2a6794
--- /dev/null
+++ b/core/utils/basic_operations/wrap_with.py
@@ -0,0 +1,40 @@
+from dataclasses import dataclass
+from typing import TypeVar, Callable
+import functools
+
+"""
+Purpose:
+ Wraps one function with another
+
+Steps:
+ 1) Takes the original function it's wrapping around, and does it
+ 2) Passes that into the second function `wrapper_fn`
+ 3) Returns the resulting object type
+
+Assumes:
+ Same object/type for both functions.
+"""
+
+T = TypeVar('T') # Input type (what the original function returns)
+U = TypeVar('U') # Output type (what wrapper_fn returns)
+
+def wrap_with(wrapper_fn: Callable[[T], U]) -> Callable:
+ def decorator(func: Callable[..., T]) -> Callable[..., U]:
+ @functools.wraps(func)
+ def wrapper(*args, **kwargs) -> U:
+ result = func(*args, **kwargs)
+ return wrapper_fn(result)
+ return wrapper
+ return decorator
+
+
+# def wrap_with(wrapper_fn: Callable[[Result], Result]) -> Callable:
+# def decorator(func: Callable[..., Result]) -> Callable[..., Result]:
+# @functools.wraps(func)
+# def wrapper(*args, **kwargs) -> Result:
+# result = func(*args, **kwargs)
+# processed_result = wrapper_fn(result) # ← Any function plugged in
+# return processed_result
+# return wrapper
+# return decorator
+