#!/usr/bin/env python3 import sys import os import logging import subprocess def run_cmd_safe(cmd, expire_cmd=None, timeout=6, retry=10): """Safely executes a command with timeout. Logs stdout and stderr. Captures TimeOutException. Args: cmd (list): Command to be executed """ result = None retry_count = 0 while retry_count < retry: if retry_count > 0 and expire_cmd: logging.warn(f"Running expire command {expire_cmd}") run_cmd_safe(expire_cmd) try: logging.debug(f"Executing {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, timeout=timeout) logging.info(f"{' '.join(cmd)}: {result.stdout}") logging.error(f"{' '.join(cmd)}: {result.stderr}") break except subprocess.TimeoutExpired: logging.error(f"Attempt {retry_count}") logging.error(f"Command expired: {cmd}") retry_count += 1 return result def get_network_state(conn_uuid): """Using nmcli, retreive the state of the given network Args: conn_uuid (str): The connection UUID provided by `nmcli connection show` Returns: str: The state of the connection provided by nmcli """ UUID = 0 STATE = 1 CMD = ['nmcli', '-t', '-f', 'con-uuid,state', 'device', 'status'] result = run_cmd_safe(CMD) decoded_result = result.stdout.decode() network_connections = decoded_result.split("\n") valid_connections = list(filter( lambda item: item[0] != "", [conn.split(':') for conn in network_connections])) selected_network = list(filter(lambda item: item[UUID] == conn_uuid, valid_connections)) if len(selected_network) > 0: return selected_network[0][STATE] else: return '' def network_connected(conn_uuid): """Returns True if the given connection UUID is connected Args: conn_uuid (str): The connection UUID provided by `nmcli connection show` Returns: bool: True if connected, False otherwise """ CONNECTED = "connected" current_state = get_network_state(conn_uuid) return current_state == CONNECTED def one_up(conn_uuids): """Returns True if at least one of the provided network connections is up Args: conn_uuids (list): List of connections to check """ for conn_uuid in conn_uuids: if network_connected(conn_uuid): return True return False def set_wifi_state(on=True): """Turns the wifi on and off Args: on (bool, optional): Set to False to turn wifi off. Defaults to True. Returns: bool: True if command successful, False if otherwise """ desired_state = "on" if on else "off" cmd = ["nmcli", "radio", "wifi", desired_state] result = run_cmd_safe(cmd) return result.returncode == 0 def is_mountpoint(path): cmd = ["mountpoint", path] result = run_cmd_safe(cmd) return result.returncode == 0 if __name__ == "__main__": logging.basicConfig( filename='/var/log/nmd.log', encoding='utf-8', level=logging.DEBUG, format='%(asctime)s %(levelname)s: %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p') logging.debug("----------Start----------") # List of connections relevant to this script # Use tags to denote ethernet or wifi CONNECTIONS = { "home": { "029a0daa-9dcd-36c2-9f3f-8c8a4da10da0": { "tags": ["ethernet"] }, "991b3332-3b25-467d-b49d-daecb968b4f8": { "tags": ["wifi"] } } } # List of valid states for NetworkManager # Taken from https://developer-old.gnome.org/NetworkManager/unstable/NetworkManager-dispatcher.html STATES = { "pre-up": "pre-up", "up": "up", "pre-down": "pre-down", "down": "down", "vpn-pre-up": "vpn-pre-up", "vpn-up": "vpn-up", "vpn-pre-down": "vpn-pre-down", "vpn-down": "vpn-down", "hostname": "hostname", "dhcp4-change": "dhcp4-change", "dhcp6-change": "dhcp6-change", "connectivity-change": "connectivity-change", } # List of available environment variables given by NetworkManager # Taken from https://developer-old.gnome.org/NetworkManager/unstable/NetworkManager-dispatcher.html # Note: omits DHCP4_ and IP6_ for simplicity's sake ENV_VARS = { "NM_DISPATCHER_ACTION": "NM_DISPATCHER_ACTION", "CONNECTION_UUID": "CONNECTION_UUID", "CONNECTION_ID": "CONNECTION_ID", "CONNECTION_DBUS_PATH": "CONNECTION_DBUS_PATH", "CONNECTION_FILENAME": "CONNECTION_FILENAME", "CONNECTION_EXTERNAL": "CONNECTION_EXTERNAL", "DEVICE_IFACE": "DEVICE_IFACE", "DEVICE_IP_IFACE": "DEVICE_IP_IFACE", "IP4_ADDRESS_N": "IP4_ADDRESS_N", "IP4_NUM_ADDRESSES": "IP4_NUM_ADDRESSES", "IP4_GATEWAY": "IP4_GATEWAY", "IP4_ROUTE_N": "IP4_ROUTE_N", "IP4_NUM_ROUTES": "IP4_NUM_ROUTES", "IP4_NAMESERVERS": "IP4_NAMESERVERS", "IP4_DOMAINS": "IP4_DOMAINS", "CONNECTIVITY_STATE": "CONNECTIVITY_STATE", } # Used to retrive values from dictionaries after they've been turned into .items() KEY = 0 VALUE = 1 # Filter out all home connections home_connections = CONNECTIONS.get("home").keys() logging.debug(f"Home connections: {home_connections}") # Filter out our ethernet connections per their tags and save to a list ethernets = list(map( lambda conn: conn[KEY], filter( lambda conn: "ethernet" in conn[VALUE].get("tags") or [], CONNECTIONS["home"].items()))) logging.debug(f"Ethernet connections: {ethernets}") # The interface and state are always passed as positional arguments logging.debug(f"arguments: {sys.argv}") interface, state = sys.argv[1:3] logging.debug(f"interface: {interface}") logging.debug(f"state: {state}") # Get the environment variables from our dictionary above environment = {var[KEY]: os.getenv(var[VALUE]) for var in ENV_VARS.items()} logging.debug(f"enviroment: {environment}") # Get our conn_uuid from the dictionary of environment variables conn_uuid = environment.get(ENV_VARS["CONNECTION_UUID"]) logging.debug(f"Connection UUID: {conn_uuid}") # check if we need to turn the wifi on or off if conn_uuid in ethernets: # If the state of our home ethernet connection is "up" (we've just connected to ethernet), # turn wifi off. if state == STATES["up"]: set_wifi_state(on=False) # If the state of our home ethernet connection is "down" (we've just disconnected from # ethernet), turn wifi back on. elif state == STATES["down"]: set_wifi_state(on=True) # When we connect to a home network, mount our shares # When we disconnect from all home networks, unmount our shares # one_home_connection_up = one_up(home_connections) # logging.debug(f"One Home Connection Up: {one_home_connection_up}") # umount_cmd = ["umount", "-a", "-l", "-t", "cifs"] # mount_cmd = ["mount", "/mnt/truenas"] # if one_home_connection_up: # run_cmd_safe(mount_cmd, expire_cmd=umount_cmd) # else: # run_cmd_safe(umount_cmd) # Log Done logging.debug("----------Done----------")