From c67f12a30624bba0df31f9af6642f685387447d5 Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Sat, 25 Jul 2026 14:15:48 -0400 Subject: [PATCH] Isolate DNS & Firewall functions for wireguard --- .../systemwide/systemwide_wireguard.py | 99 +---------------- .../systemwide/wireguard/wg_firewall_dns.py | 105 ++++++++++++++++++ 2 files changed, 106 insertions(+), 98 deletions(-) create mode 100644 core/services/networking/systemwide/wireguard/wg_firewall_dns.py diff --git a/core/services/networking/systemwide/systemwide_wireguard.py b/core/services/networking/systemwide/systemwide_wireguard.py index 1f902ee..0f1b20c 100644 --- a/core/services/networking/systemwide/systemwide_wireguard.py +++ b/core/services/networking/systemwide/systemwide_wireguard.py @@ -7,6 +7,7 @@ from core.services.networking.systemwide import killswitch, dns from core.services.networking.systemwide.dns_tools.process_dns_result import orchestrate_dns_check from core.services.networking.systemwide.dns_tools.SearchResult import SearchResult, SearchError from core.services.networking.systemwide.wireguard.nmcli_tools import setup_nmcli_connection, setup_ipv6_sinkhole +from core.services.networking.systemwide.wireguard.wg_firewall_dns import set_dns, revert_dns, turn_on_firewall, check_and_kill_firewall from core.errors.logger import logger from core.Constants import Constants @@ -27,104 +28,6 @@ import subprocess import time -def set_dns(profile_id: str, network_interface: str) -> Result: - do_they_even_use_systemd = dns.is_systemd_enabled() - - if not do_they_even_use_systemd.valid: - error_msg = "You don't use SystemD, so we are skipping setting your DNS." - logger.info(error_msg) - return Result(valid=True, message=error_msg) # true to not prompt errors - - dns_should_be = extract_dns(profile_id) - - logger.info(f"[CONNECT] About to set the DNS for {network_interface}..") - return dns.set_on(network_interface=network_interface, dns_should_be=dns_should_be) - - - -def revert_dns() -> Result: - do_they_even_use_systemd = dns.is_systemd_enabled() - - if not do_they_even_use_systemd.valid: - error_msg = "You don't use SystemD, so we are skipping setting your DNS." - logger.info(error_msg) - return Result(valid=True, message=error_msg) # true to not prompt errors - - logger.info("Reverting DNS Settings..") - 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. Error type: {turn_off.error_type} with {turn_off.message}") - return turn_off.valid - - - -def turn_on_firewall(profile_id: str, wg_interface_name: str) -> Result: - """ - Purpose: - This is a wireguard specific function, using the WG interface, - but the underlying killswitch.arm function it uses is protocol neutral. - - Steps: - 1) Get the WG interface from the process output - 2) Get the server IP from the backup WG config - 3) Use the killswitch's arm - """ - logger.info(f"[SYSTEMWIDE WG] Turning on Firewall..") - - server_ip_address = extract_endpoint_ip(profile_id) - if wg_interface_name != 'wg': - logger.info(f"Note: The WG interface is NOT WG. It's {wg_interface_name}. Be careful about what we kill switch compared to what goes up. Although we kill the nftable rule regardless on down.") - - return killswitch.arm( - server_ip=server_ip_address, - tunnel_if=wg_interface_name, - internal_subnet=None - ) - - -# This raises errors on failure. -def check_and_kill_firewall(): - firewall_on = killswitch.status() # produces a boolean - logger.info(f"The Firewall is currently: {firewall_on}") - - # is it even armed? if not, get lost, - if firewall_on: - # if it's armed, kill it, - logger.info(f"Killing the firewall..") - disable_result_object = killswitch.disarm() # produces a Result Object - - # did that work? - if disable_result_object.valid: - logger.info(f"Success! We disabled the firewall correctly.") - return - 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.") - - -# This is dead code right now -def confirm_dns_is_reverted(wg_interface_name: str = 'wg'): - do_they_even_use_systemd = dns.is_systemd_enabled() - - if not do_they_even_use_systemd: - return True - - check_on_dns = orchestrate_dns_check(wg_interface_name) - - if check_on_dns.valid: - logger.error("CRITICAL DNS PROBLEM! We turned WG off, but the DNS check shows it's still on. Trying again..") - return False - else: - logger.info("We turned WG off, and the DNS check shows it reverted also. This is great.") - return True - - def terminate_system_connection( firewall_setting: Optional[bool] = get_firewall_setting(), dns_setting: Optional[bool] = get_dns_setting() diff --git a/core/services/networking/systemwide/wireguard/wg_firewall_dns.py b/core/services/networking/systemwide/wireguard/wg_firewall_dns.py new file mode 100644 index 0000000..17122ab --- /dev/null +++ b/core/services/networking/systemwide/wireguard/wg_firewall_dns.py @@ -0,0 +1,105 @@ +from core.models.Result import Result, ResultError +from core.services.networking.systemwide import killswitch, dns +from core.services.networking.general_connection_tools.config_tools import extract_endpoint_ip, extract_dns +from core.errors.logger import logger + +def set_dns(profile_id: str, network_interface: str) -> Result: + do_they_even_use_systemd = dns.is_systemd_enabled() + + if not do_they_even_use_systemd.valid: + error_msg = "You don't use SystemD, so we are skipping setting your DNS." + logger.info(error_msg) + return Result(valid=True, message=error_msg) # true to not prompt errors + + dns_should_be = extract_dns(profile_id) + + logger.info(f"[CONNECT] About to set the DNS for {network_interface}..") + return dns.set_on(network_interface=network_interface, dns_should_be=dns_should_be) + + + +def revert_dns() -> Result: + do_they_even_use_systemd = dns.is_systemd_enabled() + + if not do_they_even_use_systemd.valid: + error_msg = "You don't use SystemD, so we are skipping setting your DNS." + logger.info(error_msg) + return Result(valid=True, message=error_msg) # true to not prompt errors + + logger.info("Reverting DNS Settings..") + 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. Error type: {turn_off.error_type} with {turn_off.message}") + return turn_off.valid + + + +def turn_on_firewall(profile_id: str, wg_interface_name: str) -> Result: + """ + Purpose: + This is a wireguard specific function, using the WG interface, + but the underlying killswitch.arm function it uses is protocol neutral. + + Steps: + 1) Get the WG interface from the process output + 2) Get the server IP from the backup WG config + 3) Use the killswitch's arm + """ + logger.info(f"[SYSTEMWIDE WG] Turning on Firewall..") + + server_ip_address = extract_endpoint_ip(profile_id) + if wg_interface_name != 'wg': + logger.info(f"Note: The WG interface is NOT WG. It's {wg_interface_name}. Be careful about what we kill switch compared to what goes up. Although we kill the nftable rule regardless on down.") + + return killswitch.arm( + server_ip=server_ip_address, + tunnel_if=wg_interface_name, + internal_subnet=None + ) + + +# This raises errors on failure. +def check_and_kill_firewall(): + firewall_on = killswitch.status() # produces a boolean + logger.info(f"The Firewall is currently: {firewall_on}") + + # is it even armed? if not, get lost, + if firewall_on: + # if it's armed, kill it, + logger.info(f"Killing the firewall..") + disable_result_object = killswitch.disarm() # produces a Result Object + + # did that work? + if disable_result_object.valid: + logger.info(f"Success! We disabled the firewall correctly.") + return + 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.") + + +# used in dead code at bottom: +# from core.services.networking.systemwide.dns_tools.process_dns_result import orchestrate_dns_check + +# This is dead code right now +# def confirm_dns_is_reverted(wg_interface_name: str = 'wg'): +# do_they_even_use_systemd = dns.is_systemd_enabled() + +# if not do_they_even_use_systemd: +# return True + +# check_on_dns = orchestrate_dns_check(wg_interface_name) + +# if check_on_dns.valid: +# logger.error("CRITICAL DNS PROBLEM! We turned WG off, but the DNS check shows it's still on. Trying again..") +# return False +# else: +# logger.info("We turned WG off, and the DNS check shows it reverted also. This is great.") +# return True +