slawek/linux-cloud-router
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reengineed the code. Generalized the package. Cloud configurators are modules.

Browse Source
This commit is contained in:
slawek committed 2026-08-10 09:02:45 +02:00
1 parent be4bb34bb0
commit 083ad9a596
40 files changed
+2367 -650
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

debian-package/src/etc/sysctl.d/99-cloud-router.conf → debian-package/src/etc/sysctl.d/99-vpn-router.conf
View File
File renamed without changes.
debian-package/src/opt/cloud-router/bin/simple-ca
-208
View File
@@ -1,208 +0,0 @@
#!/usr/bin/env python3
# MIT License
#
# Copyright (c) 2026 Sławomir Koszewski
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""simple-ca — minimal CA for cloud-router IKEv2 PKI.
Directory layout:
/etc/swanctl/x509ca/ca.pem CA certificate (strongSwan trust anchor)
/etc/swanctl/private/ca.key CA private key
/etc/swanctl/x509/server.pem server certificate
/etc/swanctl/private/server.key server private key
/etc/cloud-router/pki/{name}_cert.pem user/road-warrior certificate
/etc/cloud-router/pki/{name}_key.pem user/road-warrior private key
/etc/cloud-router/pki/{name}.pfx PKCS#12 bundle for client distribution
"""
import argparse
import subprocess
import sys
from pathlib import Path
CA_CERT = Path('/etc/swanctl/x509ca/ca.pem')
CA_KEY = Path('/etc/swanctl/private/ca.key')
SERVER_CERT = Path('/etc/swanctl/x509/server.pem')
SERVER_KEY = Path('/etc/swanctl/private/server.key')
PKI_DIR = Path('/etc/cloud-router/pki')
def _run(*args, stdin=None):
result = subprocess.run(list(args), input=stdin, capture_output=True)
if result.returncode != 0:
sys.stderr.buffer.write(result.stderr)
sys.exit(1)
return result.stdout
def _san_prefix(s):
if '@' in s:
return f'email:{s}'
parts = s.split('.')
if len(parts) == 4 and all(p.isdigit() and 0 <= int(p) <= 255 for p in parts):
return f'IP:{s}'
return f'DNS:{s}'
def cmd_make_ca(args):
if CA_CERT.exists() and CA_KEY.exists():
print('Root CA already exists, skipping.')
return
print('Generating root CA certificate and key...')
_run(
'openssl', 'req',
'-x509',
'-newkey', 'rsa:4096',
'-keyout', str(CA_KEY),
'-out', str(CA_CERT),
'-days', str(args.days),
'-noenc',
'-subj', f'/CN={args.name}',
'-text',
'-addext', 'basicConstraints=critical,CA:TRUE,pathlen:1',
'-addext', 'keyUsage=critical,keyCertSign,cRLSign',
)
CA_KEY.chmod(0o600)
print(f'Root CA created.')
print(f' Certificate : {CA_CERT}')
print(f' Private key : {CA_KEY}')
def cmd_make_cert(args):
name = args.subject.split('.')[0]
if args.type == 'server':
cert_path = SERVER_CERT
key_path = SERVER_KEY
else:
cert_path = PKI_DIR / f'{name}_cert.pem'
key_path = PKI_DIR / f'{name}_key.pem'
if cert_path.exists() and key_path.exists():
print(f'Certificate already exists at {cert_path}, skipping.')
return
if not CA_CERT.exists() or not CA_KEY.exists():
print('ERROR: Root CA not found. Run: simple-ca make-ca <name>', file=sys.stderr)
sys.exit(1)
req_args = [
'openssl', 'req',
'-newkey', 'rsa:4096',
'-keyout', str(key_path),
'-noenc',
'-subj', f'/CN={args.subject}',
'-addext', 'basicConstraints=critical,CA:FALSE',
]
if args.type == 'server':
sans = [f'DNS:{args.subject}'] + [_san_prefix(s) for s in args.san]
req_args += [
'-addext', 'keyUsage=critical,digitalSignature,keyEncipherment',
'-addext', 'extendedKeyUsage=serverAuth,clientAuth',
'-addext', f'subjectAltName={",".join(sans)}',
]
else:
sans = [_san_prefix(s) for s in args.san]
req_args += [
'-addext', 'keyUsage=critical,digitalSignature,nonRepudiation',
'-addext', 'extendedKeyUsage=clientAuth,emailProtection',
]
if sans:
req_args += ['-addext', f'subjectAltName={",".join(sans)}']
x509_args = [
'openssl', 'x509',
'-req',
'-CA', str(CA_CERT),
'-CAkey', str(CA_KEY),
'-copy_extensions', 'copyall',
'-days', str(args.days),
'-text',
'-out', str(cert_path),
]
csr = _run(*req_args)
_run(*x509_args, stdin=csr)
key_path.chmod(0o600)
print('Certificate created.')
print(f' Certificate : {cert_path}')
print(f' Private key : {key_path}')
def cmd_make_pfx(args):
name = args.name
cert_path = PKI_DIR / f'{name}_cert.pem'
key_path = PKI_DIR / f'{name}_key.pem'
pfx_path = PKI_DIR / f'{name}.pfx'
if not cert_path.exists() or not key_path.exists():
print(f'ERROR: Certificate or key not found for {name!r} in {PKI_DIR}.', file=sys.stderr)
sys.exit(1)
if pfx_path.exists():
print(f'ERROR: {pfx_path} already exists.', file=sys.stderr)
sys.exit(1)
password = args.password or 'changeit'
_run(
'openssl', 'pkcs12',
'-export',
'-out', str(pfx_path),
'-inkey', str(key_path),
'-in', str(cert_path),
'-certfile', str(CA_CERT),
'-password', f'pass:{password}',
)
print(f'PKCS#12 created: {pfx_path}')
def main():
ap = argparse.ArgumentParser(
prog='simple-ca',
description='Minimal CA for cloud-router IKEv2 PKI.',
)
sub = ap.add_subparsers(dest='command', required=True)
p_ca = sub.add_parser('make-ca', help='Create root CA')
p_ca.add_argument('name', help='CA common name')
p_ca.add_argument('--days', type=int, default=3650, metavar='N')
p_ca.set_defaults(func=cmd_make_ca)
p_cert = sub.add_parser('make-cert', help='Issue a certificate')
p_cert.add_argument('subject', help='Subject CN (FQDN for server, username for user)')
p_cert.add_argument('san', nargs='*', help='Additional SANs: IP, DNS, or email')
p_cert.add_argument('--type', choices=['server', 'user'], default='server',
help='Certificate type (default: server)')
p_cert.add_argument('--days', type=int, default=365, metavar='N')
p_cert.set_defaults(func=cmd_make_cert)
p_pfx = sub.add_parser('make-pfx', help='Export PKCS#12 bundle for a user certificate')
p_pfx.add_argument('name', help='Certificate name (without extension)')
p_pfx.add_argument('--password', metavar='PASS',
help='Export password (default: changeit)')
p_pfx.set_defaults(func=cmd_make_pfx)
args = ap.parse_args()
args.func(args)
if __name__ == '__main__':
main()
debian-package/src/usr/lib/cloud-router/configure
Vendored
-257
View File
@@ -1,257 +0,0 @@
#!/usr/bin/python3
"""Render cloud-router configuration files and configure the system."""
import os
import sys
import socket
import pathlib
import subprocess
import jinja2
TEMPLATE_DIR = pathlib.Path('/usr/share/cloud-router/templates')
def _require(name):
val = os.environ.get(name)
if val is None:
print(f'ERROR: environment variable {name} is not set', file=sys.stderr)
sys.exit(1)
return val
def build_context():
local_addrs = _require('CLOUD_ROUTER_LOCAL_ADDRS')
local_fqdn = _require('CLOUD_ROUTER_LOCAL_FQDN')
local_id_mode = _require('CLOUD_ROUTER_LOCAL_ID_MODE')
local_cidrs = _require('CLOUD_ROUTER_LOCAL_CIDRS')
remote_addrs = _require('CLOUD_ROUTER_REMOTE_ADDRS')
remote_id = _require('CLOUD_ROUTER_REMOTE_ID')
psk = _require('CLOUD_ROUTER_PSK')
remote_cidrs = _require('CLOUD_ROUTER_REMOTE_CIDRS')
router_int_gateway_ip = _require('CLOUD_ROUTER_ROUTER_INT_GATEWAY_IP')
p2s_address_pool = _require('CLOUD_ROUTER_P2S_ADDRESS_POOL')
wg_enabled = _require('CLOUD_ROUTER_WG_ENABLED')
wg_address = _require('CLOUD_ROUTER_WG_ADDRESS')
wg_listen_port = _require('CLOUD_ROUTER_WG_LISTEN_PORT')
local_subnet = local_cidrs.split(',')[0].strip()
p2s_server_name = local_fqdn.split('.')[0]
if local_id_mode == 'fqdn':
local_id = f'@{local_fqdn}'
elif local_id_mode == 'public_ip':
try:
local_id = socket.getaddrinfo(local_fqdn, None, socket.AF_INET)[0][4][0]
except socket.gaierror as exc:
print(f'ERROR: cannot resolve {local_fqdn}: {exc}', file=sys.stderr)
sys.exit(1)
elif local_id_mode == 'internal_ip':
local_id = local_addrs
else:
local_id = f'@{local_fqdn}'
return {
'local_addrs': local_addrs,
'local_fqdn': local_fqdn,
'local_id_mode': local_id_mode,
'local_cidrs': local_cidrs,
'local_subnet': local_subnet,
'remote_addrs': remote_addrs,
'remote_id': remote_id,
'psk': psk,
'remote_cidrs': remote_cidrs,
'router_int_gateway_ip': router_int_gateway_ip,
'p2s_address_pool': p2s_address_pool,
'p2s_server_name': p2s_server_name,
'wg_enabled': wg_enabled,
'wg_address': wg_address,
'wg_listen_port': wg_listen_port,
'local_id': local_id,
}
def render(jinja_env, ctx, template_name, dest, mode):
content = jinja_env.get_template(template_name).render(ctx)
dest = pathlib.Path(dest)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(content, encoding='utf-8')
os.chmod(dest, mode)
os.chown(dest, 0, 0)
def detect_wan_iface():
result = subprocess.run(
['ip', 'route', 'get', '1.1.1.1'],
capture_output=True, text=True,
)
if result.returncode != 0:
print('ERROR: ip route get 1.1.1.1 failed', file=sys.stderr)
sys.exit(1)
tokens = result.stdout.split()
for i, tok in enumerate(tokens):
if tok == 'dev' and i + 1 < len(tokens):
return tokens[i + 1]
print('ERROR: unable to detect WAN interface', file=sys.stderr)
sys.exit(1)
def setup_wireguard(ctx):
if ctx['wg_enabled'] != 'true':
return
wg_dir = pathlib.Path('/etc/wireguard')
wg_dir.mkdir(mode=0o700, exist_ok=True)
key_file = wg_dir / 'wg0.key'
if not key_file.exists() or key_file.stat().st_size == 0:
result = subprocess.run(['wg', 'genkey'], capture_output=True, check=True)
key_file.write_bytes(result.stdout)
os.chmod(key_file, 0o600)
pub_result = subprocess.run(
['wg', 'pubkey'],
input=key_file.read_bytes(),
capture_output=True, check=True,
)
pub_file = wg_dir / 'wg0.pub'
pub_file.write_bytes(pub_result.stdout)
os.chmod(pub_file, 0o644)
def _insert_after(content, marker, block):
"""Insert block after the first line that exactly matches marker."""
lines = content.splitlines(keepends=True)
result = []
for line in lines:
result.append(line)
if line.rstrip('\n') == marker:
result.append(block)
return ''.join(result)
def _insert_after_first_commit(content, block):
"""Insert block after the first COMMIT line (end of *filter table)."""
lines = content.splitlines(keepends=True)
result = []
inserted = False
for line in lines:
result.append(line)
if not inserted and line.rstrip('\n') == 'COMMIT':
result.append(block)
inserted = True
return ''.join(result)
def _wg_block(ctx):
return (
'\n'
'# WIREGUARD RULES START\n'
f'-A ufw-before-input -p udp --dport {ctx["wg_listen_port"]} -j ACCEPT\n'
'# WIREGUARD RULES END\n'
)
def setup_ufw(ctx, wan_iface):
# ── DEFAULT_FORWARD_POLICY ────────────────────────────────────────────────
ufw_defaults = pathlib.Path('/etc/default/ufw')
if ufw_defaults.exists():
lines = ufw_defaults.read_text().splitlines(keepends=True)
lines = [
'DEFAULT_FORWARD_POLICY="ACCEPT"\n'
if ln.startswith('DEFAULT_FORWARD_POLICY=') else ln
for ln in lines
]
ufw_defaults.write_text(''.join(lines))
# ── before.rules ─────────────────────────────────────────────────────────
before_rules = pathlib.Path('/etc/ufw/before.rules')
if not before_rules.exists():
print(f'ERROR: {before_rules} does not exist', file=sys.stderr)
sys.exit(1)
content = before_rules.read_text()
# Idempotency: handle dpkg-reconfigure re-runs
if '# IPSEC RULES START' in content:
if ctx['wg_enabled'] == 'true' and '# WIREGUARD RULES START' not in content:
content = _insert_after(content, '# P2S DNS RULES END', _wg_block(ctx))
before_rules.write_text(content)
return
# Filter table additions (IPSEC, P2S DNS, FORWARD)
filter_block = (
'\n'
'# IPSEC RULES START\n'
'-A ufw-before-input -p udp --dport 500 -j ACCEPT\n'
'-A ufw-before-input -p udp --dport 4500 -j ACCEPT\n'
'-A ufw-before-input -p esp -j ACCEPT\n'
'-A ufw-before-input -m policy --dir in --pol ipsec -j ACCEPT\n'
'-A ufw-before-output -m policy --dir out --pol ipsec -j ACCEPT\n'
'-A ufw-before-forward -m policy --dir in --pol ipsec -j ACCEPT\n'
'-A ufw-before-forward -m policy --dir out --pol ipsec -j ACCEPT\n'
'# IPSEC RULES END\n'
'\n'
'# P2S DNS RULES START\n'
f'-A ufw-before-input -s {ctx["p2s_address_pool"]} -d {ctx["local_addrs"]} -p udp --dport 53 -j ACCEPT\n'
f'-A ufw-before-input -s {ctx["p2s_address_pool"]} -d {ctx["local_addrs"]} -p tcp --dport 53 -j ACCEPT\n'
'# P2S DNS RULES END\n'
'\n'
'# ROUTER FORWARD RULES START\n'
f'-A ufw-before-forward -s {ctx["local_subnet"]} -o {wan_iface} -j ACCEPT\n'
f'-A ufw-before-forward -d {ctx["local_subnet"]} -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT\n'
'# ROUTER FORWARD RULES END\n'
)
content = _insert_after(content, '# End required lines', filter_block)
# NAT table (inserted after the filter table's COMMIT)
nat_lines = [
'\n# ROUTER NAT RULES START\n',
'*nat\n',
':POSTROUTING ACCEPT [0:0]\n',
'-F POSTROUTING\n',
]
for cidr in ctx['remote_cidrs'].split(','):
nat_lines.append(
f'-A POSTROUTING -s {ctx["local_subnet"]} -d {cidr.strip()} -j RETURN\n'
)
nat_lines.append(
f'-A POSTROUTING -s {ctx["local_subnet"]} -o {wan_iface} -j MASQUERADE\n'
)
nat_lines.append('COMMIT\n# ROUTER NAT RULES END\n')
content = _insert_after_first_commit(content, ''.join(nat_lines))
if ctx['wg_enabled'] == 'true':
content = _insert_after(content, '# P2S DNS RULES END', _wg_block(ctx))
before_rules.write_text(content)
def main():
ctx = build_context()
loader = jinja2.FileSystemLoader(str(TEMPLATE_DIR))
jinja_env = jinja2.Environment(
loader=loader,
keep_trailing_newline=True,
undefined=jinja2.StrictUndefined,
autoescape=False,
)
render(jinja_env, ctx, 'cloud-router.default.j2',
'/etc/default/cloud-router', 0o644)
render(jinja_env, ctx, 'remote-site.conf.j2',
'/etc/swanctl/conf.d/remote-site.conf', 0o600)
render(jinja_env, ctx, 'road-warrior.conf.j2',
'/etc/swanctl/conf.d/road-warrior.conf', 0o600)
render(jinja_env, ctx, 'p2s-forwarder.conf.j2',
'/etc/systemd/resolved.conf.d/p2s-forwarder.conf', 0o644)
render(jinja_env, ctx, '90-cloud-router.yaml.j2',
'/etc/netplan/90-cloud-router.yaml', 0o600)
if ctx['wg_enabled'] == 'true':
render(jinja_env, ctx, 'wg0.conf.j2',
'/etc/wireguard/wg0.conf', 0o600)
wan_iface = detect_wan_iface()
setup_wireguard(ctx)
setup_ufw(ctx, wan_iface)
if __name__ == '__main__':
main()
debian-package/src/usr/lib/vpn-router/configure
Vendored Executable
+22
View File
@@ -0,0 +1,22 @@
#!/usr/bin/python3
"""Render configuration files and provision PKI from vpn-router.conf.
Run directly to regenerate the configuration files without touching the
running system. 'setup' calls the same code before applying the rest.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
import vpnrouter
def main():
vpnrouter.configure()
return 0
if __name__ == '__main__':
sys.exit(main())
debian-package/src/usr/lib/vpn-router/seed-config
Executable
+72
View File
@@ -0,0 +1,72 @@
#!/usr/bin/python3
"""Create /etc/vpn-router/vpn-router.conf from the installer's answers.
Runs once, from postinst, and only when the file does not exist. The
configuration file belongs to the operator from that point on and is never
rewritten by the package.
Answers arrive as VPN_ROUTER_* environment variables. The pre-shared key is
stored base64-encoded so that arbitrary characters cannot collide with INI
syntax.
"""
import base64
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
import vpnrouter
# environment variable -> (section, option)
ANSWERS = {
'VPN_ROUTER_PLATFORM': ('general', 'platform'),
'VPN_ROUTER_EXTERNAL_INTERFACE': ('interfaces', 'external'),
'VPN_ROUTER_INTERNAL_INTERFACE': ('interfaces', 'internal'),
'VPN_ROUTER_LOCAL_FQDN': ('wan', 'local_fqdn'),
'VPN_ROUTER_LOCAL_ID_MODE': ('wan', 'local_id_mode'),
'VPN_ROUTER_LOCAL_CIDRS': ('local', 'cidrs'),
'VPN_ROUTER_INT_GATEWAY_IP': ('local', 'int_gateway_ip'),
'VPN_ROUTER_REMOTE_ADDRS': ('remote', 'addrs'),
'VPN_ROUTER_REMOTE_ID': ('remote', 'id'),
'VPN_ROUTER_REMOTE_CIDRS': ('remote', 'cidrs'),
'VPN_ROUTER_P2S_ENABLED': ('p2s', 'enabled'),
'VPN_ROUTER_P2S_ADDRESS_POOL': ('p2s', 'address_pool'),
'VPN_ROUTER_P2S_CA_NAME': ('p2s', 'ca_name'),
'VPN_ROUTER_WG_ENABLED': ('wireguard', 'enabled'),
'VPN_ROUTER_WG_ADDRESS': ('wireguard', 'address'),
'VPN_ROUTER_WG_LISTEN_PORT': ('wireguard', 'listen_port'),
}
def main():
path = vpnrouter.CONFIG_FILE
if path.exists():
return 0
parser = vpnrouter.default_parser()
for variable, (section, option) in ANSWERS.items():
value = os.environ.get(variable)
if value is not None and value != '':
parser[section][option] = value
psk = os.environ.get('VPN_ROUTER_PSK', '')
if psk:
parser['remote']['psk_b64'] = base64.b64encode(
psk.encode('utf-8')).decode('ascii')
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as handle:
parser.write(handle)
os.chmod(path, 0o600)
os.chown(path, 0, 0)
print(f'vpn-router: wrote {path}')
print('vpn-router: edit it and run "systemctl restart vpn-router-setup" '
'to apply changes')
return 0
if __name__ == '__main__':
sys.exit(main())
debian-package/src/usr/lib/vpn-router/setup
Executable
+337
View File
@@ -0,0 +1,337 @@
#!/usr/bin/python3
"""Apply the configuration to the running system.
Regenerates the configuration files first, then does everything that needs
the live system: routes, DNS-derived identity, WireGuard keys, firewall rules
and service reloads. Idempotent - a second run changes nothing.
Interface names come from the configuration and are never guessed. The
firewall and NAT rules are written against them, so choosing the wrong one
would be a silent and security-relevant mistake.
"""
import os
import pathlib
import socket
import subprocess
import sys
sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
import vpnrouter
from vpnrouter import fail
BEFORE_RULES = pathlib.Path('/etc/ufw/before.rules')
UFW_DEFAULTS = pathlib.Path('/etc/default/ufw')
WG_DIR = pathlib.Path('/etc/wireguard')
FILTER_ANCHOR = '# End required lines'
BLOCKS = (
('# IPSEC RULES START', '# IPSEC RULES END'),
('# WIREGUARD RULES START', '# WIREGUARD RULES END'),
('# P2S DNS RULES START', '# P2S DNS RULES END'),
('# ROUTER FORWARD RULES START', '# ROUTER FORWARD RULES END'),
('# ROUTER NAT RULES START', '# ROUTER NAT RULES END'),
('# ROUTER MSS RULES START', '# ROUTER MSS RULES END'),
)
def run(args, check=True):
result = subprocess.run(args, capture_output=True, text=True)
if check and result.returncode != 0:
fail(f'{" ".join(args)} failed: {result.stderr.strip()}')
return result
def apply_routes(ctx):
"""Route the protected subnets via the configured internal interface.
validate() has already confirmed the interface exists.
"""
gateway = ctx['int_gateway_ip']
iface = ctx['int_iface']
for cidr in ctx['local_cidr_list']:
run(['ip', 'route', 'replace', cidr, 'via', gateway, 'dev', iface])
def resolve_public_ip(fqdn):
try:
return socket.getaddrinfo(fqdn, None, socket.AF_INET)[0][4][0]
except socket.gaierror as exc:
fail(f'cannot resolve {fqdn} for local_id_mode=public_ip: {exc}')
def setup_wireguard(ctx):
"""Generate the WireGuard key pair when it does not exist."""
if not ctx['wg_enabled']:
return
WG_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
key_file = WG_DIR / 'wg0.key'
pub_file = WG_DIR / 'wg0.pub'
if not key_file.exists() or key_file.stat().st_size == 0:
result = run(['wg', 'genkey'])
key_file.write_text(result.stdout)
os.chmod(key_file, 0o600)
result = subprocess.run(['wg', 'pubkey'], input=key_file.read_text(),
capture_output=True, text=True)
if result.returncode != 0:
fail(f'wg pubkey failed: {result.stderr.strip()}')
pub_file.write_text(result.stdout)
os.chmod(pub_file, 0o644)
def strip_blocks(text):
"""Remove every block this package owns, and the blank line before it."""
lines = text.splitlines(keepends=True)
starts = {start for start, _ in BLOCKS}
ends = {end for _, end in BLOCKS}
out = []
skipping = False
for line in lines:
stripped = line.strip()
if not skipping and stripped in starts:
while out and out[-1].strip() == '':
out.pop()
skipping = True
continue
if skipping:
if stripped in ends:
skipping = False
continue
out.append(line)
return ''.join(out)
def is_configured(ctx):
"""True once any feature is actually configured.
While nothing is configured the package leaves the firewall alone: an
install that has not been filled in yet must not start blocking traffic on
a host that was working before.
"""
return any((vpnrouter.s2s_ready(ctx), vpnrouter.routing_ready(ctx),
vpnrouter.p2s_ready(ctx), vpnrouter.wg_ready(ctx)))
def filter_block(ctx):
lines = [
'',
'# IPSEC RULES START',
'-A ufw-before-input -p udp --dport 500 -j ACCEPT',
'-A ufw-before-input -p udp --dport 4500 -j ACCEPT',
'-A ufw-before-input -p esp -j ACCEPT',
'-A ufw-before-input -m policy --dir in --pol ipsec -j ACCEPT',
'-A ufw-before-output -m policy --dir out --pol ipsec -j ACCEPT',
'-A ufw-before-forward -m policy --dir in --pol ipsec -j ACCEPT',
'-A ufw-before-forward -m policy --dir out --pol ipsec -j ACCEPT',
'# IPSEC RULES END',
]
if vpnrouter.wg_ready(ctx):
lines += [
'',
'# WIREGUARD RULES START',
f'-A ufw-before-input -p udp --dport {ctx["wg_listen_port"]} -j ACCEPT',
'# WIREGUARD RULES END',
]
if vpnrouter.p2s_ready(ctx) and ctx['local_addrs']:
pool = ctx['p2s_address_pool']
lines += ['', '# P2S DNS RULES START']
for address in [a.strip() for a in ctx['local_addrs'].split(',') if a.strip()]:
lines += [
f'-A ufw-before-input -s {pool} -d {address} -p udp --dport 53 -j ACCEPT',
f'-A ufw-before-input -s {pool} -d {address} -p tcp --dport 53 -j ACCEPT',
]
lines += ['# P2S DNS RULES END']
if vpnrouter.routing_ready(ctx):
subnet = ctx['local_subnet']
lines += [
'',
'# ROUTER FORWARD RULES START',
f'-A ufw-before-forward -s {subnet} -o {ctx["wan_iface"]} -j ACCEPT',
f'-A ufw-before-forward -d {subnet} -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT',
'# ROUTER FORWARD RULES END',
]
return ''.join(line + '\n' for line in lines)
def nat_and_mangle_block(ctx):
lines = []
if vpnrouter.routing_ready(ctx):
subnet = ctx['local_subnet']
lines += [
'',
'# ROUTER NAT RULES START',
'*nat',
':POSTROUTING ACCEPT [0:0]',
'-F POSTROUTING',
]
for cidr in [c.strip() for c in ctx['remote_cidrs'].split(',') if c.strip()]:
lines.append(f'-A POSTROUTING -s {subnet} -d {cidr} -j RETURN')
lines += [
f'-A POSTROUTING -s {subnet} -o {ctx["wan_iface"]} -j MASQUERADE',
'COMMIT',
'# ROUTER NAT RULES END',
]
lines += [
'',
'# ROUTER MSS RULES START',
'*mangle',
':FORWARD ACCEPT [0:0]',
'-A FORWARD -p tcp --tcp-flags SYN,RST SYN -j TCPMSS --clamp-mss-to-pmtu',
'COMMIT',
'# ROUTER MSS RULES END',
]
return ''.join(line + '\n' for line in lines)
def _insert_after_line(text, marker, block):
lines = text.splitlines(keepends=True)
for index, line in enumerate(lines):
if line.strip() == marker:
lines.insert(index + 1, block)
return ''.join(lines)
return None
def setup_ufw(ctx):
"""Rewrite the package's blocks in before.rules. Returns True on change."""
if not BEFORE_RULES.is_file():
fail(f'{BEFORE_RULES} does not exist')
# Nothing configured: remove anything left from an earlier configuration
# and touch nothing else.
if not is_configured(ctx):
original = BEFORE_RULES.read_text(encoding='utf-8')
text = strip_blocks(original)
if text == original:
return False
BEFORE_RULES.write_text(text, encoding='utf-8')
return True
if UFW_DEFAULTS.is_file():
text = UFW_DEFAULTS.read_text(encoding='utf-8')
updated = ''.join(
'DEFAULT_FORWARD_POLICY="ACCEPT"\n'
if line.startswith('DEFAULT_FORWARD_POLICY=') else line
for line in text.splitlines(keepends=True)
)
if updated != text:
UFW_DEFAULTS.write_text(updated, encoding='utf-8')
original = BEFORE_RULES.read_text(encoding='utf-8')
text = strip_blocks(original)
result = _insert_after_line(text, FILTER_ANCHOR, filter_block(ctx))
if result is None:
fail(f'{BEFORE_RULES}: anchor "{FILTER_ANCHOR}" not found')
text = result
result = _insert_after_line(text, 'COMMIT', nat_and_mangle_block(ctx))
if result is None:
fail(f'{BEFORE_RULES}: no COMMIT line found')
text = result
if text == original:
return False
BEFORE_RULES.write_text(text, encoding='utf-8')
return True
def unit_is_active(unit):
return run(['systemctl', 'is-active', '--quiet', unit],
check=False).returncode == 0
def manage_strongswan(ctx, changed):
"""Make sure the IPsec daemon is running and holding the current config.
strongswan.service loads swanctl configuration itself on start
(ExecStartPost) and re-reads it on reload, so starting it is enough on a
cold boot and a reload is the right response to a configuration change.
"""
wanted = vpnrouter.s2s_ready(ctx) or vpnrouter.p2s_ready(ctx)
if not wanted:
return
if not unit_is_active('strongswan'):
run(['systemctl', 'enable', '--now', 'strongswan'], check=False)
elif 'swanctl' in changed:
run(['systemctl', 'reload', 'strongswan'], check=False)
def manage_wireguard(ctx, changed):
"""Bring wg0 up when WireGuard is configured, and down when it is not."""
unit = 'wg-quick@wg0'
if not vpnrouter.wg_ready(ctx):
if unit_is_active(unit):
run(['systemctl', 'disable', '--now', unit], check=False)
return
if not unit_is_active(unit):
run(['systemctl', 'enable', '--now', unit], check=False)
elif 'wireguard' in changed:
run(['systemctl', 'restart', unit], check=False)
def run_platform_setup(cfg, ctx):
"""Let the platform module apply its own state, if it defines any."""
module = vpnrouter.platform_module(cfg)
apply = getattr(module, 'apply', None)
if apply is None:
return
try:
apply(ctx)
except Exception as exc:
fail(f'platform module {module.__name__} failed: {exc}')
def main():
cfg = vpnrouter.load_config()
ctx, changed = vpnrouter.configure(cfg)
if vpnrouter.routing_ready(ctx):
apply_routes(ctx)
if ctx['local_id_mode'] == 'public_ip' and ctx['local_fqdn']:
address = resolve_public_ip(ctx['local_fqdn'])
ctx, more = vpnrouter.configure(cfg, local_id_override=address)
changed |= more
setup_wireguard(ctx)
rules_changed = setup_ufw(ctx)
if is_configured(ctx):
status = run(['ufw', 'status'], check=False)
if not status.stdout.startswith('Status: active'):
# Only ever enable the firewall after making sure SSH survives it.
run(['ufw', 'allow', '22/tcp'], check=False)
run(['ufw', '--force', 'enable'])
elif rules_changed:
run(['ufw', 'reload'])
elif rules_changed:
run(['ufw', 'reload'], check=False)
manage_strongswan(ctx, changed)
manage_wireguard(ctx, changed)
if 'resolved' in changed:
run(['systemctl', 'restart', 'systemd-resolved'], check=False)
run_platform_setup(cfg, ctx)
return 0
if __name__ == '__main__':
sys.exit(main())
debian-package/src/usr/lib/vpn-router/simple-ca
Executable
+611
View File
@@ -0,0 +1,611 @@
#!/usr/bin/env python3
# MIT License
#
# Copyright (c) 2026 Sławomir Koszewski
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# This module requires Python 3.8+ and OpenSSL to be installed on the system.
import argparse
import datetime
import json
import os
import re
import subprocess
import sys
import tempfile
OPENSSL = "openssl"
class Config:
FILE = "simple-ca.json"
def __init__(self, ca_dir: str):
self._path = os.path.join(ca_dir, self.FILE)
self._data: dict = {}
if not os.path.isfile(self._path):
return
try:
with open(self._path, "r") as f:
self._data = json.load(f)
except (json.JSONDecodeError, OSError) as e:
print(f"WARNING: could not read {self._path}: {e}", file=sys.stderr)
def get(self, key, default=None):
return self._data.get(key, default)
def update(self, patch: dict):
file_missing = not os.path.isfile(self._path)
changed = any(self._data.get(k) != v for k, v in patch.items())
if file_missing or changed:
self._data.update(patch)
self._save()
def append_history(self, ca_key: str, entry: dict):
history = self._data.get("history", {})
history.setdefault(ca_key, []).append(entry)
self._data["history"] = history
self._save()
def revoke_in_history(self, ca_key: str, serial: str, revoked_at: str):
"""Mark a certificate as revoked. Returns True if newly revoked,
False if already revoked, None if serial not found."""
history = self._data.get("history", {})
for entry in history.get(ca_key, []):
if entry.get("serial") == serial:
if "revoked" in entry:
return False
entry["revoked"] = revoked_at
self._data["history"] = history
self._save()
return True
return None
def _save(self):
with open(self._path, "w") as f:
json.dump(self._data, f, indent=2)
f.write("\n")
_config: Config
def _now() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _err(msg):
print(f"ERROR: {msg}", file=sys.stderr)
def _rebuild_ca_bundle(ca_dir):
bundle_path = os.path.join(ca_dir, "ca_bundle.pem")
parts = []
root = os.path.join(ca_dir, "ca_cert.pem")
if os.path.isfile(root):
with open(root, "rb") as f:
parts.append(f.read())
for name in sorted(_config.get("subordinates", [])):
sub_cert = os.path.join(ca_dir, name, "ca_cert.pem")
if os.path.isfile(sub_cert):
with open(sub_cert, "rb") as f:
parts.append(f.read())
with open(bundle_path, "wb") as f:
f.write(b"".join(parts))
def _read_serial(cert_path) -> str:
result = subprocess.run(
[OPENSSL, "x509", "-in", cert_path, "-noout", "-serial"],
capture_output=True, text=True,
)
return result.stdout.strip().split("=", 1)[-1]
def _read_expiry(cert_path) -> str:
"""Return the certificate notAfter date as an ISO 8601 UTC string."""
result = subprocess.run(
[OPENSSL, "x509", "-in", cert_path, "-noout", "-enddate"],
capture_output=True, text=True,
)
raw = result.stdout.strip().split("=", 1)[-1]
raw = " ".join(raw.split()) # normalize whitespace ("May 4" -> "May 4")
dt = datetime.datetime.strptime(raw, "%b %d %H:%M:%S %Y GMT")
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
def _iso_to_asn1(iso: str) -> str:
"""Convert "2026-05-24T14:28:10Z" -> "260524142810Z" for OpenSSL index.txt."""
dt = datetime.datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ")
return dt.strftime("%y%m%d%H%M%SZ")
_IP_RE = re.compile(r"^[0-9]{1,3}(\.[0-9]{1,3}){3}$")
_DNS_RE = re.compile(r"^[a-z0-9-]+(\.[a-z0-9-]+)*$")
def _is_ip(value):
return bool(_IP_RE.match(value))
def _is_dns(value):
return bool(_DNS_RE.match(value))
def _pipe(cmd1, cmd2):
p1 = subprocess.Popen(cmd1, stdout=subprocess.PIPE)
p2 = subprocess.Popen(cmd2, stdin=p1.stdout)
p1.stdout.close()
p2.communicate()
p1.wait()
return p1.returncode == 0 and p2.returncode == 0
def make_ca(ca_dir, ca_name, days=3650, issuing_ca=None, ca_publish_base_url=None):
if issuing_ca == "ca":
_err("--issuing-ca cannot be 'ca' as it is reserved for the root CA.")
return False
if not ca_dir or not os.path.isdir(ca_dir):
_err(f"Certificate directory {ca_dir} does not exist.")
return False
if not ca_name:
_err("CA name is required.")
return False
root_ca_cert_path = os.path.join(ca_dir, "ca_cert.pem")
root_ca_key_path = os.path.join(ca_dir, "ca_key.pem")
if not os.path.isfile(root_ca_cert_path) or not os.path.isfile(root_ca_key_path):
if issuing_ca:
_err(
f"Cannot create issuing CA '{ca_name}' without existing root CA "
"certificate and key. Please create the root CA first."
)
return False
print(f"Generating CA certificate '{ca_name}' and key...")
# Path length constraint of 1: allows one level of issuing CAs.
cmd = [
OPENSSL, "req",
"-x509",
"-newkey", "rsa:4096",
"-keyout", root_ca_key_path,
"-out", root_ca_cert_path,
"-days", str(days),
"-noenc",
"-subj", f"/CN={ca_name}",
"-text",
"-addext", "basicConstraints=critical,CA:TRUE,pathlen:1",
"-addext", "keyUsage=critical,keyCertSign,cRLSign",
]
if subprocess.run(cmd).returncode != 0:
_err("Failed to generate CA certificate and key.")
return False
_rebuild_ca_bundle(ca_dir)
patch = {"name": ca_name, "created": _now()}
if ca_publish_base_url:
patch["ca_publish_base_url"] = ca_publish_base_url
_config.update(patch)
return True
issuing_ca_dir = os.path.join(ca_dir, issuing_ca)
issuing_ca_cert = os.path.join(issuing_ca_dir, "ca_cert.pem")
issuing_ca_key = os.path.join(issuing_ca_dir, "ca_key.pem")
if not os.path.isfile(issuing_ca_cert) or not os.path.isfile(issuing_ca_key):
print(f"Generating issuing CA certificate '{ca_name}' and key...")
os.makedirs(issuing_ca_dir, exist_ok=True)
req_cmd = [
OPENSSL, "req",
"-newkey", "rsa:4096",
"-keyout", issuing_ca_key,
"-noenc",
"-subj", f"/CN={ca_name}",
"-addext", "basicConstraints=critical,CA:TRUE,pathlen:0",
"-addext", "keyUsage=critical,keyCertSign,cRLSign",
]
if ca_publish_base_url:
req_cmd += [
"-addext", f"authorityInfoAccess=caIssuers;URI:{ca_publish_base_url}/ca_cert.crt",
"-addext", f"crlDistributionPoints=URI:{ca_publish_base_url}/crl.pem",
]
x509_cmd = [
OPENSSL, "x509",
"-req",
"-CA", root_ca_cert_path,
"-CAkey", root_ca_key_path,
"-copy_extensions", "copyall",
"-days", str(days),
"-text",
"-out", issuing_ca_cert,
]
if not _pipe(req_cmd, x509_cmd):
_err("Failed to generate issuing CA certificate and key.")
return False
patch = {"ca_publish_base_url": ca_publish_base_url} if ca_publish_base_url else {}
subs = _config.get("subordinates", [])
if issuing_ca not in subs:
patch["subordinates"] = subs + [issuing_ca]
_config.update(patch)
_rebuild_ca_bundle(ca_dir)
return True
def make_cert(cert_subject_name, sans=None, ca_dir=None, cert_dir=None,
issuing_ca=None, days=365, ca_publish_base_url=None):
if issuing_ca == "ca":
_err("--issuing-ca cannot be 'ca' as it is reserved for the root CA.")
return False
if not ca_dir or not os.path.isdir(ca_dir):
_err(f"CA directory {ca_dir} does not exist.")
return False
if not cert_subject_name:
_err("Subject name is required.")
return False
if not _is_dns(cert_subject_name):
_err(f"Invalid subject name '{cert_subject_name}'. Must be a valid DNS name.")
return False
signing_dir = os.path.join(ca_dir, issuing_ca) if issuing_ca else ca_dir
cert_dir = cert_dir or signing_dir
if not os.path.isdir(cert_dir):
_err(f"Certificate directory {cert_dir} does not exist.")
return False
ca_cert_path = os.path.join(signing_dir, "ca_cert.pem")
ca_key_path = os.path.join(signing_dir, "ca_key.pem")
if not os.path.isfile(ca_cert_path) or not os.path.isfile(ca_key_path):
_err(
f"Signing CA certificate and key not found in {signing_dir}. "
"Please set up a signing CA first."
)
return False
aia_url = cdp_url = ""
if ca_publish_base_url:
if issuing_ca:
aia_url = f"{ca_publish_base_url}/{issuing_ca}/ca_cert.crt"
cdp_url = f"{ca_publish_base_url}/{issuing_ca}/crl.pem"
else:
aia_url = f"{ca_publish_base_url}/ca_cert.crt"
cdp_url = f"{ca_publish_base_url}/crl.pem"
cert_name = cert_subject_name.split(".", 1)[0]
san_entries = [f"DNS:{cert_subject_name}"]
for entry in sans or []:
if _is_ip(entry):
san_entries.append(f"IP:{entry}")
elif _is_dns(entry):
san_entries.append(f"DNS:{entry}")
else:
_err(f"Invalid SAN entry '{entry}'")
return False
sans_ext = "subjectAltName=" + ",".join(san_entries)
print(f"Generating server certificate for '{cert_subject_name}' with SANs:")
for san in san_entries:
print(f" - {san}")
cert_out = os.path.join(cert_dir, f"{cert_name}_cert.pem")
key_out = os.path.join(cert_dir, f"{cert_name}_key.pem")
if not os.path.isfile(cert_out) or not os.path.isfile(key_out):
print("Generating server certificate and key...")
req_cmd = [
OPENSSL, "req",
"-newkey", "rsa:4096",
"-keyout", key_out,
"-noenc",
"-subj", f"/CN={cert_subject_name}",
"-addext", "basicConstraints=critical,CA:FALSE",
"-addext", "keyUsage=critical,digitalSignature,keyEncipherment",
"-addext", "extendedKeyUsage=serverAuth,clientAuth",
"-addext", sans_ext,
]
if aia_url:
req_cmd += ["-addext", f"authorityInfoAccess=caIssuers;URI:{aia_url}"]
if cdp_url:
req_cmd += ["-addext", f"crlDistributionPoints=URI:{cdp_url}"]
x509_cmd = [
OPENSSL, "x509",
"-req",
"-CA", ca_cert_path,
"-CAkey", ca_key_path,
"-copy_extensions", "copyall",
"-days", str(days),
"-text",
"-out", cert_out,
]
if not _pipe(req_cmd, x509_cmd):
_err("Failed to generate server certificate and key.")
return False
_config.append_history(issuing_ca or "ca", {
"name": cert_subject_name,
"serial": _read_serial(cert_out),
"created": _now(),
"expires": _read_expiry(cert_out),
})
return True
def make_pfx(cert_path, ca_dir, issuing_ca=None, password=None, apple_openssl=False):
if issuing_ca == "ca":
_err("--issuing-ca cannot be 'ca' as it is reserved for the root CA.")
return False
cert_dir = os.path.dirname(cert_path)
cert_basename = os.path.basename(cert_path)
cert_name = cert_basename[:-len("_cert.pem")] if cert_basename.endswith("_cert.pem") else cert_basename
key_path = os.path.join(cert_dir, f"{cert_name}_key.pem")
if not cert_dir or not os.path.isdir(cert_dir):
_err(f"Certificate directory {cert_dir} does not exist.")
return False
if not ca_dir or not os.path.isdir(ca_dir):
_err(f"CA directory {ca_dir} does not exist.")
return False
if not os.path.isfile(cert_path) or not os.path.isfile(key_path):
_err("Server certificate or key not found.")
return False
root_ca_cert_path = os.path.join(ca_dir, "ca_cert.pem")
root_ca_key_path = os.path.join(ca_dir, "ca_key.pem")
if not os.path.isfile(root_ca_cert_path) or not os.path.isfile(root_ca_key_path):
_err(f"CA certificate or key not found in {ca_dir}.")
return False
if issuing_ca:
issuing_ca_cert_path = os.path.join(ca_dir, issuing_ca, "ca_cert.pem")
if not os.path.isfile(issuing_ca_cert_path):
_err(f"Issuing CA certificate not found: {issuing_ca_cert_path}.")
return False
if not password:
password = "changeit"
pfx_path = os.path.join(cert_dir, f"{cert_name}.pfx")
if os.path.isfile(pfx_path):
print("PKCS#12 (PFX) file already exists, aborting generation.")
return False
print("Generating PKCS#12 (PFX) file...", end="")
chain_bytes = b""
with open(root_ca_cert_path, "rb") as f:
chain_bytes += f.read()
if issuing_ca:
with open(issuing_ca_cert_path, "rb") as f:
chain_bytes += f.read()
chain_fd, chain_file = tempfile.mkstemp()
try:
with os.fdopen(chain_fd, "wb") as f:
f.write(chain_bytes)
cmd = [
"/usr/bin/openssl" if apple_openssl else OPENSSL, "pkcs12",
"-export", "-out", pfx_path,
"-inkey", key_path,
"-in", cert_path,
"-certfile", chain_file,
"-password", f"pass:{password}",
]
if subprocess.run(cmd).returncode != 0:
_err("Failed to generate PKCS#12 (PFX) file.")
return False
finally:
if os.path.exists(chain_file):
os.remove(chain_file)
print("done.")
return True
def make_crl(ca_dir, issuing_ca=None, days=30):
signing_dir = os.path.join(ca_dir, issuing_ca) if issuing_ca else ca_dir
ca_cert = os.path.join(signing_dir, "ca_cert.pem")
ca_key = os.path.join(signing_dir, "ca_key.pem")
if not os.path.isfile(ca_cert) or not os.path.isfile(ca_key):
_err(f"CA certificate or key not found in {signing_dir}.")
return False
crl_path = os.path.join(signing_dir, "crl.pem")
history_key = issuing_ca or "ca"
revoked_entries = [
e for e in _config.get("history", {}).get(history_key, [])
if "revoked" in e and "expires" in e
]
with tempfile.TemporaryDirectory() as tmp:
index_txt = os.path.join(tmp, "index.txt")
crlnumber = os.path.join(tmp, "crlnumber")
cnf_path = os.path.join(tmp, "openssl.cnf")
with open(index_txt, "w") as f:
for e in revoked_entries:
expires_asn1 = _iso_to_asn1(e["expires"])
revoked_asn1 = _iso_to_asn1(e["revoked"])
f.write(f"R\t{expires_asn1}\t{revoked_asn1}\t{e['serial']}\tunknown\t/CN={e['name']}\n")
with open(crlnumber, "w") as f:
f.write("01\n")
with open(cnf_path, "w") as f:
f.write(
"[ ca ]\ndefault_ca = CA_default\n\n"
"[ CA_default ]\n"
f"database = {index_txt}\n"
f"crlnumber = {crlnumber}\n"
f"certificate = {ca_cert}\n"
f"private_key = {ca_key}\n"
f"default_crl_days = {days}\n"
"default_md = sha256\n\n"
"[ crl_ext ]\n"
"authorityKeyIdentifier = keyid:always\n"
)
if subprocess.run([OPENSSL, "ca", "-gencrl", "-config", cnf_path, "-out", crl_path]).returncode != 0:
_err("Failed to generate CRL.")
return False
print(f"CRL written to {crl_path}")
return True
def revoke_cert(cert_path, ca_dir, issuing_ca=None):
if not os.path.isfile(cert_path):
_err(f"Certificate not found: {cert_path}")
return False
signing_dir = os.path.join(ca_dir, issuing_ca) if issuing_ca else ca_dir
if not os.path.isdir(signing_dir):
_err(f"CA directory not found: {signing_dir}")
return False
ca_key = issuing_ca or "ca"
serial = _read_serial(cert_path)
result = _config.revoke_in_history(ca_key, serial, _now())
if result is None:
_err(f"Certificate with serial {serial} not found in history for CA '{ca_key}'.")
return False
if result is False:
print(f"Certificate {cert_path} (serial {serial}) is already revoked.")
return True
print(f"Certificate {cert_path} (serial {serial}) marked as revoked.")
return True
def _build_parser():
parser = argparse.ArgumentParser(
description="Simple CA for creating and managing test certificates."
)
sub = parser.add_subparsers(dest="command", required=True)
p_ca = sub.add_parser("make-ca", help="Create a root or issuing CA.")
p_ca.add_argument("--days", type=int, default=None, help="Validity period in days (default: 3650)")
p_ca.add_argument("--issuing-ca", default=None, help="Specify the issuing CA")
p_ca.add_argument("--ca-publish-base-url", default=None,
help="Base URL for AIA and CRL distribution point extensions")
p_ca.add_argument("--ca-dir", help="Directory to store the CA files")
p_ca.add_argument("--openssl", default=None, metavar="PATH",
help=f"Path to the openssl binary (default: {OPENSSL})")
p_ca.add_argument("ca_name", help="Name of the CA")
p_cert = sub.add_parser("make-cert", help="Create a server/client certificate.")
p_cert.add_argument("--ca-dir", help="Directory of the CA")
p_cert.add_argument("--issuing-ca", default=None, help="Specify the issuing CA")
p_cert.add_argument("--days", type=int, default=None, help="Validity period in days (default: 365)")
p_cert.add_argument("--cert-dir", help="Directory to store the certificate files")
p_cert.add_argument("--openssl", default=None, metavar="PATH",
help=f"Path to the openssl binary (default: {OPENSSL})")
p_cert.add_argument("subject_name", help="Subject name for the certificate")
p_cert.add_argument("sans", nargs="*", help="Subject Alternative Names (SANs) for the certificate")
p_pfx = sub.add_parser("make-pfx", help="Create a PKCS#12 (PFX) bundle.")
p_pfx.add_argument("--issuing-ca", default=None, help="Specify the issuing CA")
p_pfx.add_argument("--ca-dir", help="Directory of the CA")
p_pfx.add_argument("--password", help="Password for the PFX file")
p_pfx.add_argument("--apple-openssl", action="store_true", default=False,
help="Use Apple's bundled /usr/bin/openssl for PKCS12 generation")
p_pfx.add_argument("path", help="Path to the certificate file")
p_crl = sub.add_parser("make-crl", help="Generate a CRL for a CA.")
p_crl.add_argument("--ca-dir", help="Directory of the CA")
p_crl.add_argument("--issuing-ca", default=None, help="Generate CRL for this issuing CA")
p_crl.add_argument("--days", type=int, default=None, help="CRL validity in days (default: 30)")
p_rev = sub.add_parser("revoke-cert", help="Revoke a certificate.")
p_rev.add_argument("--ca-dir", help="Directory of the CA")
p_rev.add_argument("--issuing-ca", default=None, help="Issuing CA that signed the certificate")
p_rev.add_argument("cert_path", help="Path to the certificate file to revoke")
return parser
def main(argv=None):
global OPENSSL, _config
parser = _build_parser()
args = parser.parse_args(argv)
ca_dir = args.ca_dir or os.environ.get("SIMPLE_CA_DIR") or os.getcwd()
_config = Config(ca_dir)
OPENSSL = getattr(args, "openssl", None) or _config.get("openssl", OPENSSL)
issuing_ca = args.issuing_ca or _config.get("issuing_ca")
ca_publish_base_url = getattr(args, "ca_publish_base_url", None) or _config.get("ca_publish_base_url")
days_cfg = _config.get("days", {})
if args.command == "make-ca":
days = args.days or days_cfg.get("ca", 3650)
ok = make_ca(
ca_dir, args.ca_name,
days=days,
issuing_ca=issuing_ca,
ca_publish_base_url=ca_publish_base_url,
)
elif args.command == "make-cert":
days = args.days or days_cfg.get("cert", 365)
ok = make_cert(
args.subject_name,
sans=args.sans,
ca_dir=ca_dir,
cert_dir=getattr(args, "cert_dir", None),
issuing_ca=issuing_ca,
days=days,
ca_publish_base_url=ca_publish_base_url,
)
elif args.command == "make-pfx":
ok = make_pfx(
args.path, ca_dir,
issuing_ca=issuing_ca,
password=args.password,
apple_openssl=args.apple_openssl,
)
elif args.command == "make-crl":
days = args.days or days_cfg.get("crl", 30)
ok = make_crl(ca_dir, issuing_ca=issuing_ca, days=days)
elif args.command == "revoke-cert":
ok = revoke_cert(args.cert_path, ca_dir, issuing_ca=issuing_ca)
else:
parser.error(f"Unknown command: {args.command}")
ok = False
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())
debian-package/src/usr/lib/vpn-router/vpnrouter.py
+503
View File
@@ -0,0 +1,503 @@
"""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': 'generic'},
'interfaces': {'external': '', 'internal': ''},
'wan': {'local_fqdn': '', 'local_id_mode': 'fqdn'},
'local': {'cidrs': '', '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')
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=CONFIG_FILE):
"""Read the configuration, filling in defaults for anything absent."""
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."""
platform = _get(cfg, 'general', 'platform')
installed = 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)}')
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_gateway_ip'), 'local.int_gateway_ip')
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
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 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 build_context(cfg, local_id_override=None):
"""Assemble the template context from the configuration."""
extras = vpnrouter_platforms.context(platform_module(cfg))
ext_iface = _get(cfg, 'interfaces', 'external')
int_iface = _get(cfg, 'interfaces', 'internal')
ext_addrs = interface_addresses(ext_iface, 'interfaces.external')
int_addrs = interface_addresses(int_iface, 'interfaces.internal')
if ext_iface and not ext_addrs:
warn(f'{ext_iface} has no IPv4 address yet')
local_addrs = ', '.join(ext_addrs)
int_addr = int_addrs[0] if int_addrs else ''
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 'generic',
'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_subnet': cidr_list[0] if cidr_list else '',
'local_ts': _join(local_cidrs, extras['EXTRA_LOCAL_TS']),
'int_addr': int_addr,
'int_gateway_ip': _get(cfg, 'local', 'int_gateway_ip'),
'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()
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
debian-package/src/usr/lib/vpn-router/vpnrouter_platforms/__init__.py
+53
View File
@@ -0,0 +1,53 @@
"""Platform modules for vpn-router.
A platform module supplies the values and behaviour that only make sense on one
kind of host. The core is complete without any of them; a module only adds
what the core deliberately refuses to know about, such as a provider's DNS
resolver address.
Modules are discovered by listing this package, so adding a platform means
adding one file here. Nothing in the core has to be edited.
A module may define any of:
EXTRA_P2S_DNS str, added to the road-warrior pool 'dns' list
EXTRA_LOCAL_TS str, added to 'local_ts' on both connections
apply(ctx) called once the system has been configured, to apply
platform-specific state. Must be idempotent. ctx is the
template context, carrying wan_iface, int_iface,
local_cidrs and everything else derived from the
configuration.
Anything a module does not define takes its default, so a module that only
needs one constant is one line long.
"""
import importlib
import pkgutil
DEFAULT = 'generic'
#: Context keys a module may contribute, with the value used when it does not.
CONTEXT_DEFAULTS = {
'EXTRA_P2S_DNS': '',
'EXTRA_LOCAL_TS': '',
}
def available():
"""Names of the installed platform modules."""
return sorted(module.name for module in pkgutil.iter_modules(__path__))
def load(name):
"""Import a platform module by name.
Raises ModuleNotFoundError when no such module is installed.
"""
return importlib.import_module(f'{__name__}.{name}')
def context(module):
"""Read a module's context contributions, applying defaults."""
return {key: getattr(module, key, default)
for key, default in CONTEXT_DEFAULTS.items()}
debian-package/src/usr/lib/vpn-router/vpnrouter_platforms/azure.py
+11
View File
@@ -0,0 +1,11 @@
"""Microsoft Azure.
No platform-specific configuration is currently required.
The Azure platform resolver 168.63.129.16 is deliberately *not* advertised to
road-warrior clients and *not* added to the traffic selectors. Clients are
given the router itself as their DNS server, and the router forwards through
its own systemd-resolved via the p2s-forwarder drop-in. That resolves
Azure-internal names without routing the wireserver address through the tunnel
and without a second resolver in the pool.
"""
debian-package/src/usr/lib/vpn-router/vpnrouter_platforms/gcp.py
+8
View File
@@ -0,0 +1,8 @@
"""Google Cloud Platform.
No platform-specific configuration is currently required.
As on Azure, road-warrior clients are given the router itself as their DNS
server and the router forwards through its own systemd-resolved, so the
metadata resolver does not belong in the pool or in the traffic selectors.
"""
debian-package/src/usr/lib/vpn-router/vpnrouter_platforms/generic.py
+6
View File
@@ -0,0 +1,6 @@
"""Generic platform: no additions.
The default, and the correct choice on any host that needs nothing
platform-specific. It defines nothing, which is the point: the core is
complete on its own.
"""
debian-package/src/usr/share/cloud-router/templates/90-cloud-router.yaml.j2
-7
View File
@@ -1,7 +0,0 @@
network:
version: 2
ethernets:
eth1:
routes:
- to: {{ local_subnet }}
via: {{ router_int_gateway_ip }}
debian-package/src/usr/share/cloud-router/templates/cloud-router.default.j2
-14
View File
@@ -1,14 +0,0 @@
LOCAL_ADDRS="{{ local_addrs }}"
LOCAL_FQDN="{{ local_fqdn }}"
LOCAL_ID_MODE="{{ local_id_mode }}"
LOCAL_CIDRS="{{ local_cidrs }}"
LOCAL_SUBNET="{{ local_subnet }}"
REMOTE_ADDRS="{{ remote_addrs }}"
REMOTE_ID="{{ remote_id }}"
REMOTE_CIDRS="{{ remote_cidrs }}"
ROUTER_INT_GATEWAY_IP="{{ router_int_gateway_ip }}"
P2S_ADDRESS_POOL="{{ p2s_address_pool }}"
P2S_SERVER_NAME="{{ p2s_server_name }}"
WG_ENABLED="{{ wg_enabled }}"
WG_ADDRESS="{{ wg_address }}"
WG_LISTEN_PORT="{{ wg_listen_port }}"
debian-package/src/usr/share/cloud-router/templates/p2s-forwarder.conf.j2
-2
View File
@@ -1,2 +0,0 @@
[Resolve]
DNSStubListenerExtra={{ local_addrs }}
debian-package/src/usr/share/doc/vpn-router/vpn-router.conf.example
+69
View File
@@ -0,0 +1,69 @@
# Example vpn-router configuration.
#
# This file is documentation. The live configuration is
# /etc/vpn-router/vpn-router.conf, which is generated on first install and
# owned by the operator afterwards. Copy settings from here as needed.
#
# Format: INI. Lists are comma-separated, booleans are true/false, an empty
# value means unset, and a _b64 suffix means the value is base64-encoded.
#
# Comments start with # or ; and may follow a value on the same line. That
# means # and ; cannot appear inside a value - which is why the pre-shared
# key, the one setting that can hold arbitrary text, is base64-encoded.
#
# Apply any change with: systemctl restart vpn-router-setup
#
# Every feature is optional. A file with nothing filled in configures nothing
# and leaves the host reachable and unchanged.
[general]
# Platform module to load: generic, azure or gcp.
platform = generic
[interfaces]
# The two inputs everything else is built on. Names only: the addresses on
# these interfaces are read from the system, never configured here.
# External faces the untrusted network; internal faces the protected one.
external =
internal =
[wan]
# This router's fully-qualified domain name.
local_fqdn =
# Source of the local IKE identity: fqdn | public_ip | internal_ip
# fqdn - use local_fqdn (must match what the peer expects)
# public_ip - resolve local_fqdn at boot and use the address
# internal_ip - use the address on the internal interface
local_id_mode = fqdn
[local]
# Local subnets advertised into the tunnel, comma-separated.
cidrs =
# Next hop on the internal side for the subnets above.
int_gateway_ip =
[remote]
# Remote gateway address(es) or FQDN, comma-separated.
addrs =
# Remote IKE identity, without a leading '@'.
id =
# Remote subnets reachable through the tunnel, comma-separated.
cidrs =
# Pre-shared key, base64-encoded: printf %s "$PSK" | base64 -w0
psk_b64 =
# Path to a file containing the raw PSK. Takes precedence over psk_b64.
psk_file =
[p2s]
# Road-warrior access.
enabled = false
# Address pool handed to road-warrior clients.
address_pool =
# CN for the CA created when /etc/vpn-router/pki is empty.
ca_name = VPN Router CA
[wireguard]
enabled = false
# Address and prefix length for wg0, for example 192.168.200.1/24.
address =
listen_port = 51820
debian-package/src/usr/share/vpn-router/templates/p2s-forwarder.conf.j2
+4
View File
@@ -0,0 +1,4 @@
[Resolve]
{% for address in local_addr_list -%}
DNSStubListenerExtra={{ address }}
{% endfor -%}
debian-package/src/usr/share/cloud-router/templates/remote-site.conf.j2 → debian-package/src/usr/share/vpn-router/templates/remote-site.conf.j2
+1 -1
View File
@@ -21,7 +21,7 @@ connections {
children {
site2site {
mode = tunnel
local_ts = {{ local_cidrs }}
local_ts = {{ local_ts }}
remote_ts = {{ remote_cidrs }}
esp_proposals = aes256-sha256,aes256-sha1
life_time = 3600
debian-package/src/usr/share/cloud-router/templates/road-warrior.conf.j2 → debian-package/src/usr/share/vpn-router/templates/road-warrior.conf.j2
+2 -2
View File
@@ -21,7 +21,7 @@ connections {
children {
road-warrior {
mode = tunnel
local_ts = {{ local_cidrs }}
local_ts = {{ local_ts }}
remote_ts = dynamic
esp_proposals = aes256-sha256,aes256-sha1
life_time = 3600
@@ -38,6 +38,6 @@ connections {
pools {
rw-pool {
addrs = {{ p2s_address_pool }}
dns = {{ local_addrs }}
dns = {{ p2s_dns }}
}
}
debian-package/src/usr/share/cloud-router/templates/wg0.conf.j2 → debian-package/src/usr/share/vpn-router/templates/wg0.conf.j2
View File
File renamed without changes.