"""Shared library for the vpn-router package. Everything derivable from the configuration file alone, with no running network required. Used by both 'configure' and 'setup'. """ import base64 import binascii import configparser import ipaddress import os import pathlib import shutil import subprocess import sys import jinja2 LIB_DIR = pathlib.Path(__file__).resolve().parent if str(LIB_DIR) not in sys.path: sys.path.insert(0, str(LIB_DIR)) import vpnrouter_platforms CONFIG_FILE = pathlib.Path('/etc/vpn-router/vpn-router.conf') TEMPLATE_DIR = pathlib.Path('/usr/share/vpn-router/templates') PKI_DIR = pathlib.Path('/etc/vpn-router/pki') SIMPLE_CA = LIB_DIR / 'simple-ca' SWANCTL_CONF_D = pathlib.Path('/etc/swanctl/conf.d') REMOTE_SITE = SWANCTL_CONF_D / 'remote-site.conf' ROAD_WARRIOR = SWANCTL_CONF_D / 'road-warrior.conf' P2S_FORWARDER = pathlib.Path('/etc/systemd/resolved.conf.d/p2s-forwarder.conf') WG_CONF = pathlib.Path('/etc/wireguard/wg0.conf') CA_CERT_DEST = pathlib.Path('/etc/swanctl/x509ca/ca.pem') SERVER_CERT = pathlib.Path('/etc/swanctl/x509/server.pem') SERVER_KEY = pathlib.Path('/etc/swanctl/private/server.key') CRL_DEST = pathlib.Path('/etc/swanctl/x509crl/crl.pem') TRUST_ANCHOR = pathlib.Path('/usr/local/share/ca-certificates/vpn-router-ca.crt') # section -> option -> default SCHEMA = { 'general': {'platform': 'none', 'mode': 'manual'}, 'interfaces': {'external': '', 'internal': ''}, 'wan': {'local_fqdn': '', 'local_id_mode': 'fqdn'}, 'local': {'cidrs': '', 'int_addr': '', 'int_gateway_ip': ''}, 'remote': {'addrs': '', 'id': '', 'cidrs': '', 'psk_b64': '', 'psk_file': ''}, 'p2s': {'enabled': 'false', 'address_pool': '', 'ca_name': 'VPN Router CA'}, 'wireguard': {'enabled': 'false', 'address': '', 'listen_port': '51820'}, } ID_MODES = ('fqdn', 'public_ip', 'internal_ip') #: How much the operator supplies, and therefore how much may be detected. #: manual - names and addresses given; nothing is detected #: interfaces - names given; addresses and gateway derived from them #: auto - nothing given; the interfaces are worked out too, accepting #: that the result may be wrong MODES = ('manual', 'interfaces', 'auto') #: platform = none means configuration is deferred: files are installed and #: nothing else happens until the operator says otherwise. PLATFORM_NONE = 'none' class ConfigError(Exception): """A setting is present but malformed.""" def warn(message): print(f'vpn-router: warning: {message}', file=sys.stderr) def fail(message): print(f'vpn-router: error: {message}', file=sys.stderr) sys.exit(1) def new_parser(): """A parser that tolerates operator comments and never interpolates. interpolation=None matters: the default BasicInterpolation treats '%' as syntax, which would break any value containing it. Inline comments are enabled, which configparser leaves off by default because it makes '#' and ';' unusable inside values. That is an acceptable trade here precisely because the one setting that can hold arbitrary characters, the pre-shared key, is base64-encoded, and base64 uses none of those characters. """ return configparser.ConfigParser( interpolation=None, comment_prefixes=('#', ';'), inline_comment_prefixes=('#', ';'), ) def default_parser(): parser = new_parser() for section, options in SCHEMA.items(): parser[section] = dict(options) return parser def load_config(path=None): """Read the configuration, filling in defaults for anything absent. The path is resolved when called, not when this function is defined, so CONFIG_FILE stays overridable. """ path = pathlib.Path(path) if path else CONFIG_FILE parser = default_parser() if path.exists(): try: parser.read(path, encoding='utf-8') except configparser.Error as exc: fail(f'cannot parse {path}: {exc}') return parser def _split_list(value): return [item.strip() for item in value.split(',') if item.strip()] def _get(cfg, section, option): return cfg.get(section, option, fallback=SCHEMA[section][option]).strip() def _get_bool(cfg, section, option): try: return cfg.getboolean(section, option, fallback=SCHEMA[section][option] == 'true') except ValueError: raise ConfigError(f'{section}.{option} is not a boolean') def _check_networks(value, name): for item in _split_list(value): try: ipaddress.ip_network(item, strict=False) except ValueError: raise ConfigError(f'{name}: {item!r} is not a valid CIDR') def _check_address(value, name): if not value: return try: ipaddress.ip_address(value) except ValueError: raise ConfigError(f'{name}: {value!r} is not a valid IP address') def validate(cfg): """Reject values that are present but malformed. Absent is not an error. The exception is general.mode, which states what the operator promised to supply, so the values that mode requires are checked for presence too. """ platform = _get(cfg, 'general', 'platform') installed = [PLATFORM_NONE] + vpnrouter_platforms.available() if platform and platform not in installed: raise ConfigError( f'general.platform: {platform!r} is not installed, expected one of ' f'{", ".join(installed)}') mode = _get(cfg, 'general', 'mode') if mode and mode not in MODES: raise ConfigError( f'general.mode: {mode!r} is not one of {", ".join(MODES)}') if not deferred(cfg): required = { 'manual': (('interfaces', 'external'), ('interfaces', 'internal'), ('local', 'int_addr'), ('local', 'int_gateway_ip')), 'interfaces': (('interfaces', 'external'), ('interfaces', 'internal')), 'auto': (), }[mode or 'manual'] for section, option in required: if not _get(cfg, section, option): raise ConfigError( f'{section}.{option} is required in mode {mode or "manual"}') id_mode = _get(cfg, 'wan', 'local_id_mode') if id_mode and id_mode not in ID_MODES: raise ConfigError( f'wan.local_id_mode: {id_mode!r} is not one of {", ".join(ID_MODES)}') _check_networks(_get(cfg, 'local', 'cidrs'), 'local.cidrs') _check_networks(_get(cfg, 'remote', 'cidrs'), 'remote.cidrs') _check_networks(_get(cfg, 'p2s', 'address_pool'), 'p2s.address_pool') _check_address(_get(cfg, 'local', 'int_addr'), 'local.int_addr') _check_address(_get(cfg, 'local', 'int_gateway_ip'), 'local.int_gateway_ip') # In auto mode the names are worked out later, so there is nothing to check. if mode != 'auto': for option in ('external', 'internal'): name = _get(cfg, 'interfaces', option) if name and not (pathlib.Path('/sys/class/net') / name).exists(): raise ConfigError( f'interfaces.{option}: interface {name!r} does not exist') wg_address = _get(cfg, 'wireguard', 'address') if wg_address: try: ipaddress.ip_interface(wg_address) except ValueError: raise ConfigError( f'wireguard.address: {wg_address!r} is not an address with prefix') port = _get(cfg, 'wireguard', 'listen_port') if port: if not port.isdigit() or not 1 <= int(port) <= 65535: raise ConfigError(f'wireguard.listen_port: {port!r} is not a valid port') for section in ('p2s', 'wireguard'): _get_bool(cfg, section, 'enabled') def resolve_psk(cfg): """Return the pre-shared key, or '' when none is configured.""" psk_file = _get(cfg, 'remote', 'psk_file') if psk_file: path = pathlib.Path(psk_file) if not path.is_file(): raise ConfigError(f'remote.psk_file: {psk_file} does not exist') return path.read_text(encoding='utf-8').strip() psk_b64 = _get(cfg, 'remote', 'psk_b64') if not psk_b64: return '' try: return base64.b64decode(psk_b64, validate=True).decode('utf-8') except (binascii.Error, UnicodeDecodeError) as exc: raise ConfigError(f'remote.psk_b64 is not valid base64 text: {exc}') def platform_module(cfg): """Import the configured platform module, falling back to the default.""" name = _get(cfg, 'general', 'platform') or vpnrouter_platforms.DEFAULT if name == PLATFORM_NONE: # Not normally reached: configure() returns before this when deferred. name = vpnrouter_platforms.DEFAULT try: return vpnrouter_platforms.load(name) except ModuleNotFoundError: warn(f'platform {name!r} is not installed, using ' f'{vpnrouter_platforms.DEFAULT}') return vpnrouter_platforms.load(vpnrouter_platforms.DEFAULT) def _join(*groups): items = [] for group in groups: for item in _split_list(group): if item not in items: items.append(item) return ', '.join(items) def deferred(cfg): """True when platform is none, meaning nothing is to be configured yet.""" return (_get(cfg, 'general', 'platform') or PLATFORM_NONE) == PLATFORM_NONE def _ip_route(*args): result = subprocess.run(['ip', *args], capture_output=True, text=True) return result.stdout if result.returncode == 0 else '' def detect_interfaces(): """Work out the external and internal interface names. Only reached in auto mode, where the operator has accepted that a wrong guess is possible. The external interface is whichever one carries the default route; the internal one is the only other addressed interface. """ tokens = _ip_route('route', 'get', '1.1.1.1').split() external = '' for index, token in enumerate(tokens): if token == 'dev' and index + 1 < len(tokens): external = tokens[index + 1] break if not external: raise ConfigError('auto mode: cannot determine the external interface') candidates = [name for name in _addressed_interfaces() if name != external] if len(candidates) != 1: raise ConfigError( f'auto mode: expected one internal interface, found {len(candidates)}; ' 'use mode = interfaces and name them') return external, candidates[0] def _addressed_interfaces(): """Interface names holding at least one IPv4 address, excluding loopback.""" names = [] for line in _ip_route('-o', '-4', 'addr', 'show').splitlines(): fields = line.split() if len(fields) >= 4 and fields[1] != 'lo' and fields[1] not in names: names.append(fields[1]) return names def detect_gateway(iface, address): """Derive the next hop on the internal side. The routing table is the honest source. When it has nothing to say, fall back to the first host address of the connected subnet, which is a convention rather than a fact - so say so. """ for line in _ip_route('-o', 'route', 'show', 'dev', iface).splitlines(): fields = line.split() if 'via' in fields: return fields[fields.index('via') + 1] for line in _ip_route('-o', '-4', 'addr', 'show', 'dev', iface).splitlines(): fields = line.split() if len(fields) >= 4 and fields[3].split('/')[0] == address: network = ipaddress.ip_interface(fields[3]).network gateway = str(next(network.hosts())) warn(f'{iface}: no gateway in the routing table, assuming {gateway} ' f'as the first host of {network}') return gateway return '' def interface_addresses(name, setting): """IPv4 addresses configured on an interface. The interface name is configuration; its addresses are read from the system. Looking up the addresses of a named interface is deterministic - unlike working out which interface to use, which is a guess the package deliberately does not make. """ if not name: return [] if not (pathlib.Path('/sys/class/net') / name).exists(): raise ConfigError(f'{setting}: interface {name!r} does not exist') result = subprocess.run( ['ip', '-o', '-4', 'addr', 'show', 'dev', name], capture_output=True, text=True, ) if result.returncode != 0: raise ConfigError(f'{setting}: cannot read addresses of {name!r}') addresses = [] for line in result.stdout.splitlines(): fields = line.split() if len(fields) >= 4: addresses.append(fields[3].split('/')[0]) return addresses def resolve_interfaces(cfg): """Return (ext_iface, int_iface, local_addrs, int_addr, int_gateway_ip). What is read from the system and what is taken verbatim depends entirely on general.mode. In manual mode nothing below touches the system at all. """ mode = _get(cfg, 'general', 'mode') or 'manual' ext_iface = _get(cfg, 'interfaces', 'external') int_iface = _get(cfg, 'interfaces', 'internal') int_addr = _get(cfg, 'local', 'int_addr') gateway = _get(cfg, 'local', 'int_gateway_ip') if mode == 'auto': ext_iface, int_iface = detect_interfaces() ext_addrs = interface_addresses(ext_iface, 'interfaces.external') local_addrs = ', '.join(ext_addrs) if ext_iface and not ext_addrs: warn(f'{ext_iface} has no IPv4 address yet') if mode == 'manual': return ext_iface, int_iface, local_addrs, int_addr, gateway int_addrs = interface_addresses(int_iface, 'interfaces.internal') int_addr = int_addrs[0] if int_addrs else '' if int_iface and not int_addr: warn(f'{int_iface} has no IPv4 address yet') gateway = detect_gateway(int_iface, int_addr) if int_addr else '' return ext_iface, int_iface, local_addrs, int_addr, gateway def build_context(cfg, local_id_override=None): """Assemble the template context from the configuration.""" extras = vpnrouter_platforms.context(platform_module(cfg)) ext_iface, int_iface, local_addrs, int_addr, gateway = resolve_interfaces(cfg) local_fqdn = _get(cfg, 'wan', 'local_fqdn') id_mode = _get(cfg, 'wan', 'local_id_mode') or 'fqdn' local_cidrs = _get(cfg, 'local', 'cidrs') if local_id_override: local_id = local_id_override elif id_mode == 'fqdn': local_id = f'@{local_fqdn}' if local_fqdn else '' elif id_mode == 'internal_ip': local_id = int_addr else: # public_ip is resolved once the network is up, and the caller then # re-renders with local_id_override set. local_id = '' cidr_list = _split_list(local_cidrs) return { 'platform': _get(cfg, 'general', 'platform') or PLATFORM_NONE, 'mode': _get(cfg, 'general', 'mode') or 'manual', 'wan_iface': ext_iface, 'int_iface': int_iface, 'local_addrs': local_addrs, 'local_addr_list': _split_list(local_addrs), 'local_fqdn': local_fqdn, 'local_id_mode': id_mode, 'local_id': local_id, 'local_cidrs': local_cidrs, 'local_cidr_list': cidr_list, 'local_ts': _join(local_cidrs, extras['EXTRA_LOCAL_TS']), 'int_addr': int_addr, 'int_gateway_ip': gateway, 'remote_addrs': _get(cfg, 'remote', 'addrs'), 'remote_id': _get(cfg, 'remote', 'id'), 'remote_cidrs': _get(cfg, 'remote', 'cidrs'), 'psk': resolve_psk(cfg), 'p2s_enabled': _get_bool(cfg, 'p2s', 'enabled'), 'p2s_address_pool': _get(cfg, 'p2s', 'address_pool'), 'p2s_ca_name': _get(cfg, 'p2s', 'ca_name'), 'p2s_dns': _join(local_addrs, extras['EXTRA_P2S_DNS']), 'wg_enabled': _get_bool(cfg, 'wireguard', 'enabled'), 'wg_address': _get(cfg, 'wireguard', 'address'), 'wg_listen_port': _get(cfg, 'wireguard', 'listen_port'), } def s2s_ready(ctx): return all([ ctx['local_addrs'], ctx['local_fqdn'], ctx['local_cidrs'], ctx['remote_addrs'], ctx['remote_id'], ctx['remote_cidrs'], ctx['psk'], ctx['local_id'], ]) def routing_ready(ctx): """Routing and NAT need both interfaces named, plus the subnets involved.""" return all([ctx['local_cidrs'], ctx['int_gateway_ip'], ctx['int_iface'], ctx['wan_iface']]) def p2s_ready(ctx): return bool(ctx['p2s_enabled'] and ctx['p2s_address_pool'] and ctx['local_fqdn']) def wg_ready(ctx): return bool(ctx['wg_enabled'] and ctx['wg_address']) def _jinja_env(): return jinja2.Environment( loader=jinja2.FileSystemLoader(str(TEMPLATE_DIR)), keep_trailing_newline=True, undefined=jinja2.StrictUndefined, autoescape=False, ) def render(env, ctx, template_name, dest, mode): """Render a template, writing only when the content changes.""" content = env.get_template(template_name).render(ctx) dest = pathlib.Path(dest) if dest.exists() and dest.read_text(encoding='utf-8') == content: os.chmod(dest, mode) return False dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(content, encoding='utf-8') os.chmod(dest, mode) return True def existing_peers(path): """Return the [Peer] sections of an existing WireGuard config. Peers are added by the operator by hand, so re-rendering the [Interface] section must not discard them. """ path = pathlib.Path(path) if not path.is_file(): return '' lines = path.read_text(encoding='utf-8').splitlines(keepends=True) for index, line in enumerate(lines): if line.strip().lower() == '[peer]': return ''.join(lines[index:]) return '' def render_wg_conf(env, ctx, dest=WG_CONF, mode=0o600): """Render wg0.conf, carrying any operator-defined peers across.""" content = env.get_template('wg0.conf.j2').render(ctx) peers = existing_peers(dest) if peers: content = content.rstrip('\n') + '\n\n' + peers dest = pathlib.Path(dest) if dest.exists() and dest.read_text(encoding='utf-8') == content: os.chmod(dest, mode) return False dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(content, encoding='utf-8') os.chmod(dest, mode) return True def remove(path): """Remove a generated file. Returns True when something was removed.""" path = pathlib.Path(path) if path.exists(): path.unlink() return True return False def install_file(src, dest, mode): """Copy src to dest when the content differs. Returns True on change.""" src, dest = pathlib.Path(src), pathlib.Path(dest) if not src.is_file(): return False if dest.exists() and dest.read_bytes() == src.read_bytes(): os.chmod(dest, mode) return False dest.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(src, dest) os.chmod(dest, mode) return True def _simple_ca(*args): result = subprocess.run( [sys.executable, str(SIMPLE_CA), *args], capture_output=True, text=True, ) if result.returncode != 0: fail(f'simple-ca {" ".join(args)} failed: {result.stderr.strip()}') return result.stdout def provision_pki(ctx): """Create the CA and server certificate when the PKI directory is empty.""" PKI_DIR.mkdir(parents=True, exist_ok=True) os.chmod(PKI_DIR, 0o700) if any(PKI_DIR.iterdir()): return if not ctx['p2s_ca_name']: warn('p2s.ca_name is empty, not creating a local CA') return if not SIMPLE_CA.is_file(): fail(f'{SIMPLE_CA} is missing') _simple_ca('make-ca', '--ca-dir', str(PKI_DIR), ctx['p2s_ca_name']) _simple_ca('make-cert', '--ca-dir', str(PKI_DIR), ctx['local_fqdn'], ctx['local_fqdn']) def distribute_pki(ctx): """Copy PKI material to each consumer. Returns True when anything changed.""" label = ctx['local_fqdn'].split('.', 1)[0] if ctx['local_fqdn'] else '' bundle = PKI_DIR / 'ca_bundle.pem' ca_src = bundle if bundle.is_file() else PKI_DIR / 'ca_cert.pem' changed = False changed |= install_file(ca_src, CA_CERT_DEST, 0o644) if label: changed |= install_file(PKI_DIR / f'{label}_cert.pem', SERVER_CERT, 0o644) changed |= install_file(PKI_DIR / f'{label}_key.pem', SERVER_KEY, 0o600) changed |= install_file(PKI_DIR / 'crl.pem', CRL_DEST, 0o644) if install_file(PKI_DIR / 'ca_cert.pem', TRUST_ANCHOR, 0o644): changed = True if shutil.which('update-ca-certificates'): subprocess.run(['update-ca-certificates'], capture_output=True, text=True) return changed def configure(cfg=None, local_id_override=None): """Render configuration files and provision PKI. Returns (ctx, changed) where changed is a set naming what needs reloading: 'swanctl', 'resolved', 'wireguard'. """ if cfg is None: cfg = load_config() # platform = none: configuration is deferred. Files are installed and # nothing is rendered, provisioned or removed until a platform is chosen. if deferred(cfg): return None, set() try: validate(cfg) ctx = build_context(cfg, local_id_override=local_id_override) except ConfigError as exc: fail(str(exc)) changed = set() if p2s_ready(ctx): provision_pki(ctx) if distribute_pki(ctx): changed.add('swanctl') env = _jinja_env() if s2s_ready(ctx): if render(env, ctx, 'remote-site.conf.j2', REMOTE_SITE, 0o600): changed.add('swanctl') elif remove(REMOTE_SITE): changed.add('swanctl') if p2s_ready(ctx): if render(env, ctx, 'road-warrior.conf.j2', ROAD_WARRIOR, 0o600): changed.add('swanctl') if render(env, ctx, 'p2s-forwarder.conf.j2', P2S_FORWARDER, 0o644): changed.add('resolved') else: if remove(ROAD_WARRIOR): changed.add('swanctl') if remove(P2S_FORWARDER): changed.add('resolved') if wg_ready(ctx): if render_wg_conf(env, ctx): changed.add('wireguard') elif remove(WG_CONF): changed.add('wireguard') return ctx, changed