Isolating the nmcli segments to a separate module
This commit is contained in:
parent
1baaa4042c
commit
3afd4a8f9b
3 changed files with 45 additions and 37 deletions
|
|
@ -103,4 +103,3 @@ class ConfigurationController:
|
||||||
configuration = ConfigurationController.get_or_new()
|
configuration = ConfigurationController.get_or_new()
|
||||||
configuration.dns = new_value
|
configuration.dns = new_value
|
||||||
configuration.save()
|
configuration.save()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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 import killswitch, dns
|
||||||
from core.services.networking.systemwide.dns_tools.process_dns_result import orchestrate_dns_check
|
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.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.errors.logger import logger
|
||||||
|
|
||||||
from core.Constants import Constants
|
from core.Constants import Constants
|
||||||
|
|
@ -161,40 +162,6 @@ def terminate_system_connection(
|
||||||
raise ConnectionTerminationError('The connection could not be terminated.')
|
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(
|
def __establish_system_connection(
|
||||||
profile: SystemProfile,
|
profile: SystemProfile,
|
||||||
|
|
@ -347,8 +314,8 @@ def establish_system_connection(
|
||||||
|
|
||||||
# ================= SETTINGS =================
|
# ================= SETTINGS =================
|
||||||
firewall_setting = get_firewall_setting()
|
firewall_setting = get_firewall_setting()
|
||||||
dns_setting = True # for testing
|
# dns_setting = True # for testing
|
||||||
# dns_setting = get_dns_setting()
|
dns_setting = get_dns_setting()
|
||||||
|
|
||||||
# ================= CONNECT WG ================
|
# ================= CONNECT WG ================
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
42
core/services/networking/systemwide/wireguard/nmcli_tools.py
Normal file
42
core/services/networking/systemwide/wireguard/nmcli_tools.py
Normal file
|
|
@ -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.')
|
||||||
|
|
||||||
Loading…
Reference in a new issue