diff --git a/core/controllers/ConfigurationController.py b/core/controllers/ConfigurationController.py index 2488396..78499d7 100644 --- a/core/controllers/ConfigurationController.py +++ b/core/controllers/ConfigurationController.py @@ -103,4 +103,3 @@ class ConfigurationController: configuration = ConfigurationController.get_or_new() configuration.dns = new_value configuration.save() - diff --git a/core/services/networking/systemwide/systemwide_wireguard.py b/core/services/networking/systemwide/systemwide_wireguard.py index e909673..1f902ee 100644 --- a/core/services/networking/systemwide/systemwide_wireguard.py +++ b/core/services/networking/systemwide/systemwide_wireguard.py @@ -6,6 +6,7 @@ from core.services.networking.general_connection_tools.config_tools import extra 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.errors.logger import logger from core.Constants import Constants @@ -161,40 +162,6 @@ def terminate_system_connection( raise ConnectionTerminationError('The connection could not be terminated.') -def setup_nmcli_connection(wg_config_path: str) -> Result: - try: - process_output = subprocess.check_output(('nmcli', 'connection', 'import', '--temporary', 'type', 'wireguard', 'file', wg_config_path), text=True) - return Result(valid=True, data=process_output) - - except CalledProcessError as e: - return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error with nmcli: {e}") - # raise ConnectionError('The connection could not be established.') - - -def setup_ipv6_sinkhole(process_output: str) -> Result: - error_info = "with nmcli setting up the IPv6 sinkhole, error is " - try: - connection_id = (m := re.search(r'(?<=\()([a-f0-9-]+?)(?=\))', process_output)) and m.group(1) - ipv6_method = subprocess.check_output(('nmcli', '-g', 'ipv6.method', 'connection', 'show', connection_id), text=True).strip() - except CalledProcessError as e: - return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}: {e}") - # raise ConnectionError('The connection could not be established.') - - if ipv6_method in ('disabled', 'ignore'): - try: - subprocess.run(('dbus-send', '--system', '--print-reply', '--dest=org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager', 'org.freedesktop.DBus.Properties.Set', 'string:org.freedesktop.NetworkManager', 'string:ConnectivityCheckEnabled', 'variant:boolean:false'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) - except CalledProcessError as e: - return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}: {e}") - # raise ConnectionError('The connection could not be established.') - - try: - subprocess.run(('nmcli', 'connection', 'add', 'type', 'dummy', 'save', 'no', 'con-name', 'hv-ipv6-sink', 'ifname', 'hvipv6sink0', 'ipv6.method', 'manual', 'ipv6.addresses', 'fd7a:fd4b:54e3:077c::/64', 'ipv6.gateway', 'fd7a:fd4b:54e3:077c::1', 'ipv6.dns', '::1', 'ipv6.route-metric', '72'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) - return Result(valid=True) - - except CalledProcessError: - return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}") - # raise ConnectionError('The connection could not be established.') - def __establish_system_connection( profile: SystemProfile, @@ -347,8 +314,8 @@ def establish_system_connection( # ================= SETTINGS ================= firewall_setting = get_firewall_setting() - dns_setting = True # for testing - # dns_setting = get_dns_setting() + # dns_setting = True # for testing + dns_setting = get_dns_setting() # ================= CONNECT WG ================ try: diff --git a/core/services/networking/systemwide/wireguard/nmcli_tools.py b/core/services/networking/systemwide/wireguard/nmcli_tools.py new file mode 100644 index 0000000..9e3f263 --- /dev/null +++ b/core/services/networking/systemwide/wireguard/nmcli_tools.py @@ -0,0 +1,42 @@ +from core.models.Result import Result, ResultError +from subprocess import CalledProcessError +import os +import re +import shutil +import subprocess + + +def setup_nmcli_connection(wg_config_path: str) -> Result: + try: + process_output = subprocess.check_output(('nmcli', 'connection', 'import', '--temporary', 'type', 'wireguard', 'file', wg_config_path), text=True) + return Result(valid=True, data=process_output) + + except CalledProcessError as e: + return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error with nmcli: {e}") + # raise ConnectionError('The connection could not be established.') + + +def setup_ipv6_sinkhole(process_output: str) -> Result: + error_info = "with nmcli setting up the IPv6 sinkhole, error is " + try: + connection_id = (m := re.search(r'(?<=\()([a-f0-9-]+?)(?=\))', process_output)) and m.group(1) + ipv6_method = subprocess.check_output(('nmcli', '-g', 'ipv6.method', 'connection', 'show', connection_id), text=True).strip() + except CalledProcessError as e: + return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}: {e}") + # raise ConnectionError('The connection could not be established.') + + if ipv6_method in ('disabled', 'ignore'): + try: + subprocess.run(('dbus-send', '--system', '--print-reply', '--dest=org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager', 'org.freedesktop.DBus.Properties.Set', 'string:org.freedesktop.NetworkManager', 'string:ConnectivityCheckEnabled', 'variant:boolean:false'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) + except CalledProcessError as e: + return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}: {e}") + # raise ConnectionError('The connection could not be established.') + + try: + subprocess.run(('nmcli', 'connection', 'add', 'type', 'dummy', 'save', 'no', 'con-name', 'hv-ipv6-sink', 'ifname', 'hvipv6sink0', 'ipv6.method', 'manual', 'ipv6.addresses', 'fd7a:fd4b:54e3:077c::/64', 'ipv6.gateway', 'fd7a:fd4b:54e3:077c::1', 'ipv6.dns', '::1', 'ipv6.route-metric', '72'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) + return Result(valid=True) + + except CalledProcessError: + return Result(valid=False, error_type=ResultError.NMCLI, message=f"Called Process error {error_info}") + # raise ConnectionError('The connection could not be established.') +