#!/usr/bin/env python3 """Tailnet-only transport edge for the loopback Subrouter authority. The edge never reads provider credentials. It resolves the workstation and approved peer Tailscale IPv4 addresses from local tailscaled state, binds only the workstation's CGNAT address, rejects unapproved peers before forwarding any bytes, and proxies approved connections byte-for-byte to the loopback authority. """ from __future__ import annotations import argparse import ipaddress import json import os import signal import socket import subprocess import sys import threading from pathlib import Path from typing import Any SCHEMA = "overdeck-subrouter-tailnet-edge/v1" TAILNET_V4 = ipaddress.ip_network("100.64.0.0/10") MAX_CONFIG = 16 * 1024 FORBIDDEN_RESPONSE = ( b"HTTP/1.1 403 Forbidden\r\n" b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Cache-Control: no-store\r\n\r\n" ) def _tailnet_ipv4(values: object, *, what: str) -> str: if not isinstance(values, list): raise ValueError(f"{what} Tailscale addresses are missing") matches: list[str] = [] for raw in values: if not isinstance(raw, str): continue try: ip = ipaddress.ip_address(raw) except ValueError: continue if isinstance(ip, ipaddress.IPv4Address) and ip in TAILNET_V4: matches.append(str(ip)) if len(matches) != 1: raise ValueError(f"{what} must have exactly one tailnet IPv4 address") return matches[0] def resolve_tailnet(status: dict[str, Any], allowed_hosts: list[str]) -> tuple[str, set[str]]: if status.get("BackendState") != "Running": raise ValueError("tailscale backend is not running") self_info = status.get("Self") if not isinstance(self_info, dict): raise ValueError("tailscale self identity is missing") listen_host = _tailnet_ipv4(self_info.get("TailscaleIPs"), what="self") peers = status.get("Peer") if not isinstance(peers, dict): raise ValueError("tailscale peer inventory is missing") by_name: dict[str, list[dict[str, Any]]] = {} for peer in peers.values(): if not isinstance(peer, dict): continue name = peer.get("HostName") if isinstance(name, str) and name: by_name.setdefault(name, []).append(peer) allowed_ips: set[str] = set() for host in allowed_hosts: matches = by_name.get(host, []) if len(matches) != 1: raise ValueError(f"approved tailnet host {host!r} is not unique/present") peer = matches[0] if peer.get("Online") is not True: raise ValueError(f"approved tailnet host {host!r} is offline") allowed_ips.add(_tailnet_ipv4(peer.get("TailscaleIPs"), what=f"peer {host}")) if not allowed_ips: raise ValueError("tailnet edge has no approved peers") return listen_host, allowed_ips def load_config(path: Path) -> dict[str, Any]: st = path.stat() if not path.is_file() or st.st_size <= 0 or st.st_size > MAX_CONFIG: raise ValueError("tailnet edge config is not a bounded regular file") data = json.loads(path.read_text(encoding="utf-8")) expected = {"schema", "listen_port", "upstream_host", "upstream_port", "allowed_hosts"} if not isinstance(data, dict) or set(data) != expected or data.get("schema") != SCHEMA: raise ValueError("tailnet edge config schema is invalid") if not isinstance(data["listen_port"], int) or not 1024 <= data["listen_port"] <= 65535: raise ValueError("tailnet edge listen port is invalid") if data["upstream_host"] != "127.0.0.1": raise ValueError("tailnet edge upstream must be loopback") if not isinstance(data["upstream_port"], int) or not 1024 <= data["upstream_port"] <= 65535: raise ValueError("tailnet edge upstream port is invalid") hosts = data["allowed_hosts"] if not isinstance(hosts, list) or not hosts or any( not isinstance(host, str) or not host or len(host) > 63 or any(ch not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-." for ch in host) for host in hosts ): raise ValueError("tailnet edge approved hosts are invalid") if len(hosts) != len(set(hosts)): raise ValueError("tailnet edge approved hosts contain duplicates") return data def tailscale_status() -> dict[str, Any]: proc = subprocess.run( ["/usr/bin/tailscale", "status", "--json"], check=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, timeout=10, ) data = json.loads(proc.stdout) if not isinstance(data, dict): raise ValueError("tailscale status is not an object") return data def _pump(source: socket.socket, target: socket.socket) -> None: try: while True: chunk = source.recv(65536) if not chunk: break target.sendall(chunk) except (ConnectionResetError, BrokenPipeError, OSError): pass finally: try: target.shutdown(socket.SHUT_WR) except OSError: pass def _proxy(client: socket.socket, upstream: tuple[str, int]) -> None: try: server = socket.create_connection(upstream, timeout=5) except OSError: client.close() return client.settimeout(None) server.settimeout(None) outbound = threading.Thread(target=_pump, args=(client, server), daemon=True) outbound.start() try: _pump(server, client) outbound.join(timeout=1) finally: for sock in (client, server): try: sock.close() except OSError: pass def handle_client(client: socket.socket, peer_ip: str, allowed_ips: set[str], upstream: tuple[str, int]) -> None: if peer_ip not in allowed_ips: try: client.sendall(FORBIDDEN_RESPONSE) except OSError: pass client.close() return _proxy(client, upstream) def serve(listen_host: str, listen_port: int, upstream_host: str, upstream_port: int, allowed_ips: set[str]) -> None: if ipaddress.ip_address(listen_host) not in TAILNET_V4: raise ValueError("tailnet edge refuses non-tailnet listen address") if any(ipaddress.ip_address(ip) not in TAILNET_V4 for ip in allowed_ips): raise ValueError("tailnet edge refuses non-tailnet approved peer") stop = threading.Event() signal.signal(signal.SIGTERM, lambda *_: stop.set()) signal.signal(signal.SIGINT, lambda *_: stop.set()) listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.bind((listen_host, listen_port)) listener.listen(64) listener.settimeout(1) print( f"subrouter-tailnet-edge: listening tailnet={listen_host}:{listen_port} approved_peers={len(allowed_ips)}", file=sys.stderr, flush=True, ) try: while not stop.is_set(): try: client, peer = listener.accept() except socket.timeout: continue except OSError: if stop.is_set(): break raise peer_ip = peer[0] threading.Thread( target=handle_client, args=(client, peer_ip, allowed_ips, (upstream_host, upstream_port)), daemon=True, name="subrouter-tailnet-proxy", ).start() finally: listener.close() def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(add_help=True) parser.add_argument("--config", required=True) args = parser.parse_args(argv) try: config = load_config(Path(args.config)) listen_host, allowed_ips = resolve_tailnet(tailscale_status(), config["allowed_hosts"]) serve( listen_host, config["listen_port"], config["upstream_host"], config["upstream_port"], allowed_ips, ) return 0 except Exception as exc: print(f"subrouter-tailnet-edge: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())