Isolate DNS & Firewall functions for wireguard
This commit is contained in:
parent
3afd4a8f9b
commit
c67f12a306
2 changed files with 106 additions and 98 deletions
|
|
@ -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()
|
||||
|
|
|
|||
105
core/services/networking/systemwide/wireguard/wg_firewall_dns.py
Normal file
105
core/services/networking/systemwide/wireguard/wg_firewall_dns.py
Normal file
|
|
@ -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
|
||||
|
||||
Loading…
Reference in a new issue