#!/usr/bin/env python3 # -*- coding: utf-8 -*- # # Copyright (C) 2014-2019 LEAP # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # """ This is a privileged helper script for safely running certain commands as root. It should only be called by the Bitmask application. Expected paths: When installed by distro path: /usr/sbin/bitmask-root When installed by bundle or from git: /usr/local/sbin/bitmask-root When installed by snap: /snap/bin/riseup-vpn.bitmask-root USAGE: bitmask-root firewall stop bitmask-root firewall start [restart] GATEWAY1 GATEWAY2 ... bitmask-root openvpn stop bitmask-root openvpn start CONFIG1 CONFIG1 ... All actions return exit code 0 for success, non-zero otherwise. The `openvpn start` action is special: it calls exec on openvpn and replaces the current process. If the `restart` parameter is passed, the firewall will not be teared down in the case of an error during launch. """ import ipaddress import os import re import signal import socket import syslog import subprocess import sys import stat import traceback cmdcheck = subprocess.check_output # # CONSTANTS def get_no_group_name(): """ Return the right group name to use for the current OS. Examples: - Ubuntu: nogroup - Arch: nobody :rtype: str or None """ import grp try: grp.getgrnam('nobody') return 'nobody' except KeyError: try: grp.getgrnam('nogroup') return 'nogroup' except KeyError: return None def is_ipv6_disabled(): """ Return True if ipv6 support is disabled by the kernel. """ code = os.system("sysctl -a 2>/dev/null | grep all.disable_ipv6 | grep 1") return code == 0 def tostr(s): return s.decode('utf-8') VERSION = "14" SCRIPT = "bitmask-root" NAMESERVER_TCP = "10.41.0.1" NAMESERVER_UDP = "10.42.0.1" # for the time being, we're hardcoding tcp on connection params. NAMESERVER = NAMESERVER_TCP BITMASK_CHAIN = "bitmask" BITMASK_CHAIN_NAT_OUT = "bitmask" BITMASK_CHAIN_NAT_POST = "bitmask_postrouting" LOCAL_INTERFACE = "lo" def swhich(binary): """ Find the path to binary in sbin :rtype: str """ for folder in ["/bin", "/sbin", "/usr/bin", "/usr/sbin", "/usr/local/sbin"]: path = os.path.join(folder, binary) if os.path.isfile(path): return path raise Exception("Can't find %s" % (binary,)) IP = swhich("ip") IPTABLES = swhich("iptables") IP6TABLES = swhich("ip6tables") SYSCTL = swhich("sysctl") OPENVPN_USER = "nobody" OPENVPN_GROUP = get_no_group_name() LEAPOPENVPN = "LEAPOPENVPN" OPENVPN_SYSTEM_BIN = "/usr/sbin/openvpn" # Debian location OPENVPN_LEAP_BIN = "/usr/local/sbin/leap-openvpn" # installed by bundle OPENVPN_SNAP_BIN = "/snap/bin/riseup-vpn.openvpn" # installed by snap FIXED_FLAGS = [ "--setenv", "LEAPOPENVPN", "1", "--nobind", "--client", "--dev", "tun", "--tls-client", "--remote-cert-tls", "server", "--management-signal", "--script-security", "1", "--user", "nobody", "--persist-key", "--persist-local-ip", "--tls-version-min", "1.0", ] if OPENVPN_GROUP is not None: FIXED_FLAGS.extend(["--group", OPENVPN_GROUP]) if is_ipv6_disabled(): FIXED_FLAGS.extend([ "--pull-filter", "ignore", "ifconfig-ipv6", "--pull-filter", "ignore", "route-ipv6"]) ALLOWED_FLAGS = { "--remote": ["IP", "NUMBER", "PROTO"], "--tls-cipher": ["CIPHER"], "--cipher": ["CIPHER"], "--auth": ["CIPHER"], "--management": ["DIR||IP", "UNIXSOCKET||NUMBER"], "--management-client-user": ["USER"], "--route": ["IP", "IP", "NETGW"], "--cert": ["FILE"], "--key": ["FILE"], "--ca": ["FILE"], "--fragment": ["NUMBER"], "--keepalive": ["NUMBER", "NUMBER"], "--verb": ["NUMBER"], "--management-client": [], "--tun-ipv6": [], } PARAM_FORMATS = { "NUMBER": lambda s: re.match("^\d+$", s), "PROTO": lambda s: re.match("^(tcp|udp|tcp4|udp4)$", s), "IP": lambda s: is_valid_address(s), "CIPHER": lambda s: re.match("^[A-Z0-9-]+$", s), "USER": lambda s: re.match( "^[a-zA-Z0-9_\.\@][a-zA-Z0-9_\-\.\@]*\$?$", s), # IEEE Std 1003.1-2001 "FILE": lambda s: os.path.isfile(s), "DIR": lambda s: os.path.isdir(os.path.split(s)[0]), "UNIXSOCKET": lambda s: s == "unix", "NETGW": lambda s: s == "net_gateway", "UID": lambda s: re.match("^[a-zA-Z0-9]+$", s) } # Determine Qubes OS version, if any QUBES_PROXY = os.path.exists("/var/run/qubes/this-is-proxyvm") if os.path.isdir("/etc/qubes"): QUBES_CFG = "/rw/config/" QUBES_IPHOOK = QUBES_CFG + "qubes-ip-change-hook" QUBES_FW_SCRIPT = QUBES_CFG + "qubes-firewall-user-script" if subprocess.call([IPTABLES, "--list", "QBS-FORWARD"]) == 0: QUBES_VER = 4 else: QUBES_VER = 3 else: # not a Qubes system QUBES_VER = 0 DEBUG = os.getenv("DEBUG") TEST = os.getenv("TEST") if DEBUG: import logging formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s") ch = logging.StreamHandler() ch.setLevel(logging.DEBUG) ch.setFormatter(formatter) logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) logger.addHandler(ch) syslog.openlog(SCRIPT) # # UTILITY # def is_valid_address(value): """ Validate that the passed ip is a valid IP address. :param value: the value to be validated :type value: str :rtype: bool """ try: socket.inet_aton(value) return True except Exception: log("%s: ERROR: MALFORMED IP: %s!" % (SCRIPT, value)) return False def split_list(_list, regex): """ Split a list based on a regex: e.g. split_list(["xx", "yy", "x1", "zz"], "^x") => [["xx", "yy"], ["x1", "zz"]] :param _list: the list to be split. :type _list: list :param regex: the regex expression to filter with. :type regex: str :rtype: list """ if not hasattr(regex, "match"): regex = re.compile(regex) result = [] i = 0 if not _list: return result while True: if regex.match(_list[i]): result.append([]) while True: result[-1].append(_list[i]) i += 1 if i >= len(_list) or regex.match(_list[i]): break else: i += 1 if i >= len(_list): break return result def get_process_list(): """ Get a process list by reading `/proc` filesystem. :return: a list of tuples, each containing pid and command string. :rtype: tuple if lists """ res = [] pids = [pid for pid in os.listdir('/proc') if pid.isdigit()] for pid in pids: try: res.append((pid, open( os.path.join( '/proc', pid, 'cmdline'), 'rb').read())) except IOError: # proc has already terminated continue return filter(None, res) def getIPv4AllowAddresses(): lines = [] try: with open("/etc/bitmask/ipv4.allow", 'r') as f: lines = [l.strip() for l in f.readlines()] except FileNotFoundError: return lines lines = filter(lambda x: ipaddress.ip_address(x).version == 4, lines) return list(filter(lambda x: ipaddress.ip_address(x).is_private, lines)) def getIPv6AllowAddresses(): lines = [] try: with open("/etc/bitmask/ipv6.allow", 'r') as f: lines = [l.strip() for l in f.readlines()] except FileNotFoundError: return lines lines = filter(lambda x: ipaddress.ip_address(x).version == 6, lines) return list(filter(lambda x: ipaddress.ip_address(x).is_private, lines)) def run(command, *args, **options): """ Run an external command. Options: `check`: If True, check the command's output. bail if non-zero. (the default is true unless detach or input is true) `exitcode`: like `check`, but return exitcode instead of bailing. `detach`: If True, run in detached process. `input`: If True, open command for writing stream to, returning the Popen object. `throw`: If True, raise an exception if there is an error instead of bailing. """ parts = [command] parts.extend(args) debug("%s run: %s " % (SCRIPT, " ".join(parts))) _check = options.get("check", True) _detach = options.get("detach", False) _input = options.get("input", False) _exitcode = options.get("exitcode", False) _throw = options.get("throw", False) if not (_check or _throw) or _detach or _input: if _input: return subprocess.Popen(parts, stdin=subprocess.PIPE) else: subprocess.Popen(parts) return None else: try: devnull = open('/dev/null', 'w') subprocess.check_call(parts, stdout=devnull, stderr=devnull) return 0 except subprocess.CalledProcessError as exc: if _exitcode: if exc.returncode != 1: # 0 or 1 is to be expected, but anything else # should be logged. debug("ERROR: Could not run %s: %s" % (exc.cmd, exc.output), exception=exc) return exc.returncode elif _throw: raise exc else: bail("ERROR: Could not run %s: %s" % (exc.cmd, exc.output), exception=exc) def log(msg=None, exception=None, priority=syslog.LOG_INFO): """ print and log warning message or exception. :param msg: optional error message. :type msg: str :param msg: optional exception. :type msg: Exception :param msg: syslog level :type msg: one of LOG_EMERG, LOG_ALERT, LOG_CRIT, LOG_ERR, LOG_WARNING, LOG_NOTICE, LOG_INFO, LOG_DEBUG """ if msg is not None: print("%s: %s" % (SCRIPT, msg)) syslog.syslog(priority, msg) if exception is not None: if TEST or DEBUG: traceback.print_exc() syslog.syslog(priority, traceback.format_exc()) def debug(msg=None, exception=None): """ Just like log, but is skipped unless DEBUG. Use syslog.LOG_INFO even for debug messages (we don't want to miss them). """ if TEST or DEBUG: log(msg, exception) def bail(msg=None, exception=None): """ abnormal exit. like log(), but exits with error status code. """ log(msg, exception) exit(1) # # OPENVPN # def get_openvpn_bin(): """ Return the path for either the system openvpn or the one the bundle has put there. """ if os.environ.get('SNAP') and os.path.isfile(OPENVPN_SNAP_BIN): # the snap option should be removed from the debian package. return OPENVPN_SNAP_BIN if os.path.isfile(OPENVPN_SYSTEM_BIN): return OPENVPN_SYSTEM_BIN # the bundle option should also be removed from the debian. if os.path.isfile(OPENVPN_LEAP_BIN): return OPENVPN_LEAP_BIN def parse_openvpn_flags(args): """ Take argument list from the command line and parse it, only allowing some configuration flags. :type args: list """ result = [] try: for flag in split_list(args, "^--"): flag_name = flag[0] if flag_name in ALLOWED_FLAGS: result.append(flag_name) required_params = ALLOWED_FLAGS[flag_name] if required_params: # flatten if separated by spaces flag_params = [i for sublist in map( lambda s: s.split(), flag[1:]) for i in sublist] if len(flag_params) != len(required_params): log("%s: ERROR: not enough params for %s" % (SCRIPT, flag_name)) return None for param, param_type in zip(flag_params, required_params): for tpe in param_type.split("||"): if PARAM_FORMATS[tpe](param): result.append(param) break else: log("%s: ERROR: Bad argument %s" % (SCRIPT, param)) return None else: log("WARNING: unrecognized openvpn flag %s" % flag_name) return result except Exception as exc: log("%s: ERROR PARSING FLAGS: %s" % (SCRIPT, exc)) if DEBUG: logger.exception(exc) return None def openvpn_start(args): """ Launch openvpn, sanitizing input, and replacing the current process with the openvpn process. :param args: arguments to be passed to openvpn :type args: list """ openvpn_flags = parse_openvpn_flags(args) if openvpn_flags: OPENVPN = get_openvpn_bin() flags = [OPENVPN] + FIXED_FLAGS + openvpn_flags if DEBUG: log("%s: running openvpn with flags:" % (SCRIPT,)) log(flags) # note: first argument to command is ignored, but customarily set to # the command. os.execv(OPENVPN, flags) else: bail('ERROR: could not parse openvpn options') def openvpn_stop(args): """ Stop the openvpn that has likely been launched by bitmask. :param args: arguments to openvpn :type args: list """ plist = get_process_list() for pid, proc in plist: if bytes("openvpn", 'utf-8') in proc and bytes(LEAPOPENVPN, 'utf-8') in proc: os.kill(int(pid), signal.SIGTERM) break # # FIREWALL # def get_gateways(gateways): """ Filter a passed sequence of gateways, returning only the valid ones. :param gateways: a sequence of gateways to filter. :type gateways: iterable :rtype: iterable """ result = filter(is_valid_address, gateways) if not result: bail("ERROR: No valid gateways specified") else: return result def get_default_device(): """ Retrieve the current default network device. :rtype: str """ routes = subprocess.check_output([IP, "route", "show"]) match = re.search(rb"^default .*dev ([^\s]*) .*$", routes, flags=re.M) if match and match.groups(): return tostr(match.group(1)) else: bail("Could not find default device") def get_local_network_ipv4(device): """ Get the local ipv4 addres for a given device. :param device: :type device: str """ addresses = cmdcheck([IP, "-o", "address", "show", "dev", device]) match = re.search(rb"^.*inet ([^ ]*) .*$", addresses, flags=re.M) if match and match.groups(): return tostr(match.group(1)) else: return None def get_local_network_ipv6(device): """ Get the local ipv6 addres for a given device. :param device: :type device: str """ addresses = cmdcheck([IP, "-o", "address", "show", "dev", device]) match = re.search(rb"^.*inet6 ([^ ]*) .*$", addresses, flags=re.M) if match and match.groups(): return tostr(match.group(1)) else: return None def run_iptable_with_check(cmd, *args, **options): """ Run an iptables command checking to see if it should: for --append: run only if rule does not already exist. for --insert: run only if rule does not already exist. for --delete: run only if rule does exist. other commands are run normally. """ if "--insert" in args: check_args = [arg.replace("--insert", "--check") for arg in args] check_code = run(cmd, *check_args, exitcode=True) if check_code != 0: run(cmd, *args, **options) elif "--append" in args: check_args = [arg.replace("--append", "--check") for arg in args] check_code = run(cmd, *check_args, exitcode=True) if check_code != 0: run(cmd, *args, **options) elif "--delete" in args: check_args = [arg.replace("--delete", "--check") for arg in args] check_code = run(cmd, *check_args, exitcode=True) if check_code == 0: run(cmd, *args, **options) else: run(cmd, *args, **options) def iptables(*args, **options): """ Run iptables4 and iptables6. """ ip4tables(*args, **options) ip6tables(*args, **options) def ip4tables(*args, **options): """ Run iptables4 with checks. """ run_iptable_with_check(IPTABLES, *args, **options) def ip6tables(*args, **options): """ Run iptables6 with checks. """ run_iptable_with_check(IP6TABLES, *args, **options) def toggle_ipv6(status='disable'): if status == 'disable': arg = 1 elif status == 'enable': arg = 0 else: return cmdcheck([SYSCTL, '-w', 'net.ipv6.conf.all.disable_ipv6=%s' % arg]) # # NOTE: these tests to see if a chain exists might incorrectly return false. # This happens when there is an error in calling `iptables --list bitmask`. # # For this reason, when stopping the firewall, we do not trust the # output of ipvx_chain_exists() but instead always attempt to delete # the chain. # def ipv4_chain_exists(chain, table=None): """ Check if a given chain exists. Only returns true if it actually exists, but might return false if it exists and iptables failed to run. :param chain: the chain to check against :type chain: str :rtype: bool """ if table is not None: code = run(IPTABLES, "-t", table, "--list", chain, "--numeric", exitcode=True) else: code = run(IPTABLES, "--list", chain, "--numeric", exitcode=True) if code == 0: return True elif code == 1: return False else: log("ERROR: Could not determine state of iptable chain") return False def ipv6_chain_exists(chain): """ see ipv4_chain_exists() :param chain: the chain to check against :type chain: str :rtype: bool """ code = run(IP6TABLES, "--list", chain, "--numeric", exitcode=True) if code == 0: return True elif code == 1: return False else: log("ERROR: Could not determine state of iptable chain") return False def enable_ip_forwarding(): """ ip_fowarding must be enabled for the firewall to work. """ with open('/proc/sys/net/ipv4/ip_forward', 'w') as f: f.write('1\n') def firewall_start(args): """ Bring up the firewall. :param args: list of gateways, to be sanitized. :type args: list """ default_device = get_default_device() local_network_ipv4 = get_local_network_ipv4(default_device) local_network_ipv6 = get_local_network_ipv6(default_device) gateways = get_gateways(args) # allow local address in listed exception list # this will allow all ports and both tcp and udp. def allow4(ip): ip4tables("--append", BITMASK_CHAIN, "--destination", ip, "-o", default_device, "--jump", "ACCEPT") def allow6(ip): ip6tables("--append", BITMASK_CHAIN, "--destination", ip, "-o", default_device, "--jump", "ACCEPT") # add custom chain "bitmask" to front of OUTPUT chain for both # the 'filter' and the 'nat' tables. if not ipv4_chain_exists(BITMASK_CHAIN): ip4tables("--new-chain", BITMASK_CHAIN) if not ipv4_chain_exists(BITMASK_CHAIN_NAT_OUT, 'nat'): ip4tables("--table", "nat", "--new-chain", BITMASK_CHAIN_NAT_OUT) if not ipv4_chain_exists(BITMASK_CHAIN_NAT_POST, 'nat'): ip4tables("--table", "nat", "--new-chain", BITMASK_CHAIN_NAT_POST) if not ipv6_chain_exists(BITMASK_CHAIN): ip6tables("--new-chain", BITMASK_CHAIN) ip4tables("--table", "nat", "--insert", "OUTPUT", "--jump", BITMASK_CHAIN_NAT_OUT) ip4tables("--table", "nat", "--insert", "POSTROUTING", "--jump", BITMASK_CHAIN_NAT_POST) iptables("--insert", "OUTPUT", "--jump", BITMASK_CHAIN) # route all ipv4 DNS over VPN # (note: NAT does not work with ipv6 until kernel 3.7) enable_ip_forwarding() if QUBES_PROXY and QUBES_VER >= 3: # rewrite DNS packets for VPN DNS; Qubes preconfigures masquerade ip4tables("-t", "nat", "--flush", "PR-QBS") ip4tables("-t", "nat", "--append", "PR-QBS", "-p", "udp", "--dport", "53", "--jump", "DNAT", "--to", NAMESERVER + ":53") ip4tables("-t", "nat", "--append", "PR-QBS", "-p", "tcp", "--dport", "53", "--jump", "DNAT", "--to", NAMESERVER + ":53") else: # allow dns to localhost ip4tables("-t", "nat", "--append", BITMASK_CHAIN, "--protocol", "udp", "--dest", "127.0.1.1,127.0.0.1,127.0.0.53", "--dport", "53", "--jump", "ACCEPT") # rewrite all outgoing packets to use VPN DNS server # (DNS does sometimes use TCP!) ip4tables("-t", "nat", "--append", BITMASK_CHAIN_NAT_OUT, "-p", "udp", "--dport", "53", "--jump", "DNAT", "--to", NAMESERVER + ":53") ip4tables("-t", "nat", "--append", BITMASK_CHAIN_NAT_OUT, "-p", "tcp", "--dport", "53", "--jump", "DNAT", "--to", NAMESERVER + ":53") # enable masquerading, so that DNS packets rewritten by DNAT will # have the correct source IPs. Apply masquerade only to the NAMESERVER, # we don't want to apply it to the localhost dns resolver. ip4tables("-t", "nat", "--append", BITMASK_CHAIN_NAT_POST, "--dest", NAMESERVER, "--protocol", "udp", "--dport", "53", "--jump", "MASQUERADE") ip4tables("-t", "nat", "--append", BITMASK_CHAIN_NAT_POST, "--dest", NAMESERVER, "--protocol", "tcp", "--dport", "53", "--jump", "MASQUERADE") # allow local network traffic ipv4_exceptions = getIPv4AllowAddresses() if local_network_ipv4: if len(ipv4_exceptions) == 0: # allow all local network destinations if no explicit allow rules defined ip4tables("--append", BITMASK_CHAIN, "--destination", local_network_ipv4, "-o", default_device, "--jump", "ACCEPT") # allow local network sources for DNS # (required to allow local network DNS that gets rewritten by NAT # to get passed through so that MASQUERADE can set correct source IP) ip4tables("--append", BITMASK_CHAIN, "--source", local_network_ipv4, "-o", default_device, "-p", "udp", "--dport", "53", "--jump", "ACCEPT") ip4tables("--append", BITMASK_CHAIN, "--source", local_network_ipv4, "-o", default_device, "-p", "tcp", "--dport", "53", "--jump", "ACCEPT") # allow multicast Simple Service Discovery Protocol ip4tables("--append", BITMASK_CHAIN, "--protocol", "udp", "--destination", "239.255.255.250", "--dport", "1900", "-o", default_device, "--jump", "RETURN") # allow multicast Bonjour/mDNS ip4tables("--append", BITMASK_CHAIN, "--protocol", "udp", "--destination", "224.0.0.251", "--dport", "5353", "-o", default_device, "--jump", "RETURN") ipv6_exceptions = getIPv6AllowAddresses() if local_network_ipv6: if len(ipv6_exceptions) == 0: # allow all local network destinations if no explicit allow rules defined ip6tables("--append", BITMASK_CHAIN, "--destination", local_network_ipv6, "-o", default_device, "--jump", "ACCEPT") # allow multicast Simple Service Discovery Protocol ip6tables("--append", BITMASK_CHAIN, "--protocol", "udp", "--destination", "FF05::C", "--dport", "1900", "-o", default_device, "--jump", "RETURN") # allow multicast Bonjour/mDNS ip6tables("--append", BITMASK_CHAIN, "--protocol", "udp", "--destination", "FF02::FB", "--dport", "5353", "-o", default_device, "--jump", "RETURN") # allow ipv4 traffic to gateways for gateway in gateways: ip4tables("--append", BITMASK_CHAIN, "--destination", gateway, "-o", default_device, "--jump", "ACCEPT") # TODO allow ipv6 traffic to gws too # log rejected packets to syslog if DEBUG: iptables("--append", BITMASK_CHAIN, "-o", default_device, "--jump", "LOG", "--log-prefix", "iptables denied: ", "--log-level", "7") # allow explicit private exceptions if len(ipv4_exceptions) != 0: for ip in ipv4_exceptions: allow4(ip) ip4tables("--append", BITMASK_CHAIN, "--destination", local_network_ipv4, "-o", default_device, "--jump", "REJECT") if len(ipv6_exceptions) != 0: for ip in ipv6_exceptions: allow6(ip) ip6tables("--append", BITMASK_CHAIN, "--destination", local_network_ipv6, "-o", default_device, "--jump", "REJECT") # for now, ensure all other ipv6 packets get rejected (regardless of # device). not sure why, but "-p any" doesn't work. ip6tables("--append", BITMASK_CHAIN, "-p", "tcp", "--jump", "REJECT") ip6tables("--append", BITMASK_CHAIN, "-p", "udp", "--jump", "REJECT") # reject all other ipv4 sent over the default device ip4tables("--append", BITMASK_CHAIN, "-o", default_device, "--jump", "REJECT") # On Qubes OS, add anti-leak rules for proxyVM qubes-firewall.service # Must stay on 'top' of chain! if QUBES_PROXY and QUBES_VER >= 3 and run("grep", "installed\ by\ " + SCRIPT, QUBES_FW_SCRIPT, exitcode=True) != 0: with open(QUBES_FW_SCRIPT, mode="w") as qfile: qfile.write("#!/bin/sh\n") qfile.write("# Anti-leak rules installed by " + SCRIPT + " " + VERSION + "\n") qfile.write("iptables --insert FORWARD -i eth0 -j DROP\n") qfile.write("iptables --insert FORWARD -o eth0 -j DROP\n") qfile.write("ip6tables --insert FORWARD -i eth0 -j DROP\n") qfile.write("ip6tables --insert FORWARD -o eth0 -j DROP\n") qfile.write("iptables --insert INPUT -i tun+ -j DROP\n") qfile.write("ip6tables --insert INPUT -i tun+ -j DROP\n") os.chmod(QUBES_FW_SCRIPT, stat.S_IRWXU) if not os.path.exists(QUBES_IPHOOK): os.symlink(QUBES_FW_SCRIPT, QUBES_IPHOOK) if QUBES_VER == 4: run(QUBES_FW_SCRIPT) elif QUBES_VER == 3: run("systemctl", "restart", "qubes-firewall.service") # toggle_ipv6('disable') def firewall_stop(): """ Stop the firewall. Because we really really always want the firewall to be stopped if at all possible, this function is cautious and contains a lot of trys and excepts. If there were any problems, we raise an exception at the end. This allows the calling code to retry stopping the firewall. Stopping the firewall can fail if iptables is being run by another process (only one iptables command can be run at a time). """ ok = True # -t filter -D OUTPUT -j bitmask try: iptables("--delete", "OUTPUT", "--jump", BITMASK_CHAIN, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to remove bitmask firewall from OUTPUT chain " "(maybe it is already removed?)", exc) ok = False # -t nat -D OUTPUT -j bitmask try: ip4tables("-t", "nat", "--delete", "OUTPUT", "--jump", BITMASK_CHAIN_NAT_OUT, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to remove bitmask firewall from OUTPUT chain " "in 'nat' table (maybe it is already removed?)", exc) ok = False # -t nat -D POSTROUTING -j bitmask_postrouting try: ip4tables("-t", "nat", "--delete", "POSTROUTING", "--jump", BITMASK_CHAIN_NAT_POST, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to remove bitmask firewall from POSTROUTING " "chain in 'nat' table (maybe it is already removed?)", exc) ok = False # -t filter --delete-chain bitmask try: ip4tables("--flush", BITMASK_CHAIN, throw=True) ip4tables("--delete-chain", BITMASK_CHAIN, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to flush and delete bitmask ipv4 firewall " "chain (maybe it is already destroyed?)", exc) ok = False # -t nat --delete-chain bitmask try: ip4tables("-t", "nat", "--flush", BITMASK_CHAIN_NAT_OUT, throw=True) ip4tables("-t", "nat", "--delete-chain", BITMASK_CHAIN_NAT_OUT, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to flush and delete bitmask ipv4 firewall " "chain in 'nat' table (maybe it is already destroyed?)", exc) ok = False # -t nat --delete-chain bitmask_postrouting try: ip4tables("-t", "nat", "--flush", BITMASK_CHAIN_NAT_POST, throw=True) ip4tables("-t", "nat", "--delete-chain", BITMASK_CHAIN_NAT_POST, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to flush and delete bitmask ipv4 firewall " "chain in 'nat' table (maybe it is already destroyed?)", exc) ok = False # -t filter --delete-chain bitmask (ipv6) try: ip6tables("--flush", BITMASK_CHAIN, throw=True) ip6tables("--delete-chain", BITMASK_CHAIN, throw=True) except subprocess.CalledProcessError as exc: debug("INFO: not able to flush and delete bitmask ipv6 firewall " "chain (maybe it is already destroyed?)", exc) ok = False # toggle_ipv6('enable') if not (ok or ipv4_chain_exists or ipv6_chain_exists): raise Exception("firewall might still be left up. " "Please try `firewall stop` again.") # # MAIN # USAGE = """ This is bitmask-root version {VERSION} This program manipulates the Bitmask firewall. It is *not* intented to be used manually. Commands: {SCRIPT} version {SCRIPT} restart {SCRIPT} openvpn start {SCRIPT} openvpn stop {SCRIPT} firewall start {SCRIPT} firewall stop {SCRIPT} firewall isup """.format(SCRIPT=SCRIPT, VERSION=VERSION) def main(): """ Entry point for cmdline execution. """ # TODO use argparse instead please. if len(sys.argv) >= 2: command = "_".join(sys.argv[1:3]) args = sys.argv[3:] is_restart = False if command == 'help' or command == '-h': print(USAGE) exit(0) if args and args[0] == 'restart': is_restart = True args.remove('restart') if command == "version": print(VERSION) exit(0) if os.getuid() != 0: bail("ERROR: must be run as root") if command == "openvpn_start": openvpn_start(args) elif command == "openvpn_stop": openvpn_stop(args) elif command == "firewall_start": try: firewall_start(args) except Exception as ex: if not is_restart: firewall_stop() bail("ERROR: could not start firewall", ex) elif command == "firewall_stop": try: firewall_stop() except Exception as ex: bail("ERROR: could not stop firewall", ex) elif command == "firewall_isup": if ipv4_chain_exists(BITMASK_CHAIN): # too verbose since bitmask polls this pass else: bail("INFO: bitmask firewall is down") else: bail("ERROR: No such command. Try bitmask-root help") else: bail("ERROR: No such command. Try bitmask-root help") if __name__ == "__main__": debug(" ".join(sys.argv)) main() exit(0)