agent-orchestrator/tools/gateway-domain.py

#!/usr/bin/env python3
"""gateway-domain.py — point a public domain at your box, via the testing gateway.

WHAT THIS IS FOR. Your machine is on the tailnet but has no public IP, and you need a real
HTTPS domain for it — an OAuth callback, a webhook receiver, a demo someone else can open.
The testing gateway at gtest.commoninternet.net already holds a wildcard DNS record, so
*.gtest.commoninternet.net resolves to it. Map your name to your tailnet IP and the gateway
streams matching connections to you:

    browser --TLS--> gateway :443 --reads SNI--> your box over the tailnet

The gateway never terminates that TLS. It reads the SNI name and proxies the still-encrypted
bytes, so **your box serves the certificate**, not the gateway. See SKILL.md.

USAGE
    tools/gateway-domain.py list
    tools/gateway-domain.py add myapp                    # -> myapp.gtest.commoninternet.net -> this box's tailscale IP
    tools/gateway-domain.py add myapp 100.64.1.5         # explicit backend
    tools/gateway-domain.py add myapp 100.64.1.5:8443    # backend listening somewhere other than 443
    tools/gateway-domain.py remove myapp

CREDENTIALS. The admin password is in the orchestrator secret store, never here:

    python3 engine/secrets.py get gateway.admin_password

This script reads it from there itself. You should not need to handle the value.
"""
import argparse
import base64
import json
import os
import re
import shutil
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request

# engine/secrets.py lives one directory up.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import secrets as secret_store  # noqa: E402  (engine/secrets.py, not the stdlib module)

TIMEOUT = 30


def _cfg():
    """Gateway coordinates. Only the password is genuinely secret; the rest lives
    alongside it so there is one thing to change if the gateway ever moves."""
    return (
        secret_store.get("gateway.fqdn", "gtest.commoninternet.net"),
        secret_store.get("gateway.admin_user", "admin"),
        secret_store.get("gateway.admin_password"),
    )


def _request(method="GET", form=None):
    fqdn, user, password = _cfg()
    if not password:
        sys.exit(
            "no gateway.admin_password in the secret store.\n"
            "  check: python3 engine/secrets.py list\n"
            "  see:   skills/gateway-domain/SKILL.md"
        )
    url = f"https://{fqdn}/admin/"
    data = urllib.parse.urlencode(form).encode() if form else None
    req = urllib.request.Request(url, data=data, method=method)
    token = base64.b64encode(f"{user}:{password}".encode()).decode()
    req.add_header("Authorization", f"Basic {token}")
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
            return r.status, r.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "replace")
        if e.code == 401:
            sys.exit(
                f"401 from {url} — the stored password is not what the gateway expects.\n"
                "The e2e suite reseeds .htpasswd while it runs and restores it afterwards;\n"
                "if a run was interrupted, /var/lib/tunnel-gateway/.htpasswd.e2e-backup is\n"
                "still on the box. See SKILL.md, 'When the password stops working'."
            )
        sys.exit(f"HTTP {e.code} from {url}\n{body[:600]}")
    except urllib.error.URLError as e:
        sys.exit(f"cannot reach {url}: {e.reason}")


# The admin app answers 200 and renders a red <p> for a rejected value, rather than
# using a status code. Checking only the status would report success on a no-op.
_ERR = re.compile(r'<p style="color:red">(.*?)</p>', re.S)

# Backends are restricted to a literal IPv4 address, optionally with a port.
#
# The gateway would accept a hostname -- its validate_ip is ^[a-zA-Z0-9_.-]+$ -- but its
# put_domain/remove_domain match existing lines with ^<domain>\s+[\d.:]+; which only ever
# matches a numeric backend. A hostname mapping can therefore be written once and then never
# updated or removed through the admin UI: it becomes an orphan only a hand-edit of
# /var/lib/tunnel-gateway/tunnel_map.conf can clear. Refuse to create one.
_BACKEND = re.compile(r"^(\d{1,3}(?:\.\d{1,3}){3})(?::(\d{1,5}))?$")

# The tailnet ACL only lets the gateway open connections to nodes carrying one of these.
# A mapping to an untagged node is accepted by the gateway and then simply never
# connects, which looks like a gateway fault and is not one.
ALLOWED_TAGS = ("tag:notplants-test-server", "tag:orchestrator")


def _validate_backend(backend):
    m = _BACKEND.match(backend)
    if not m:
        sys.exit(
            f"backend must be an IPv4 address, optionally IP:port -- got {backend!r}.\n"
            "Hostnames are rejected on purpose: the gateway can store one but cannot\n"
            "remove it again. Use the tailnet IP (tailscale ip -4 on the target box)."
        )
    ip, port = m.group(1), m.group(2)
    if any(int(o) > 255 for o in ip.split(".")):
        sys.exit(f"not a valid IPv4 address: {ip}")
    if port is not None and not (1 <= int(port) <= 65535):
        sys.exit(f"port out of range: {port}")
    return backend


def _check(body):
    m = _ERR.search(body)
    if m:
        sys.exit(f"gateway rejected the request: {m.group(1).strip()}")


def _parse_domains(body):
    """Pull the Domain Mappings table out of the admin page."""
    section = body.split("Domain Mappings", 1)[-1].split("Port Mappings", 1)[0]
    return re.findall(r"<td>\s*(.*?)\s*</td>\s*<td>\s*(.*?)\s*</td>", section, re.S)


def _self_tags(exe):
    """This node's tailnet tags, or None if they cannot be determined."""
    try:
        out = subprocess.run([exe, "status", "--json"], capture_output=True, text=True, timeout=15)
        return json.loads(out.stdout).get("Self", {}).get("Tags") or []
    except Exception:
        return None


def _warn_untagged(exe):
    tags = _self_tags(exe)
    if tags is None:
        return
    if not any(t in tags for t in ALLOWED_TAGS):
        print(
            f"warning: this node carries none of {' / '.join(ALLOWED_TAGS)} "
            f"(tags: {', '.join(tags) or 'none'}).\n"
            "         The gateway will accept the mapping but the tailnet ACL will not let it\n"
            "         reach this box, so no traffic will flow. Add the tag in the Tailscale\n"
            "         admin, or map a backend that already has it.",
            file=sys.stderr,
        )


def _tailscale_ip():
    exe = shutil.which("tailscale") or "/run/current-system/sw/bin/tailscale"
    try:
        out = subprocess.run([exe, "ip", "-4"], capture_output=True, text=True, timeout=15)
    except (OSError, subprocess.SubprocessError) as e:
        sys.exit(f"could not run tailscale to detect this box's IP ({e}); pass the backend explicitly")
    _warn_untagged(exe)
    ip = out.stdout.strip().splitlines()[0].strip() if out.stdout.strip() else ""
    if not ip:
        sys.exit(
            "tailscale reported no IPv4 address — is this box on the tailnet?\n"
            f"  {exe} status\n"
            "Or pass the backend explicitly: gateway-domain.py add <name> <ip>"
        )
    return ip


def _qualify(name, fqdn):
    """A bare label becomes a subdomain of the gateway; anything with a dot is used as given."""
    return name if "." in name else f"{name}.{fqdn}"


def cmd_list(args):
    _, body = _request()
    rows = _parse_domains(body)
    if not rows:
        print("no domain mappings")
        return
    width = max(len(d) for d, _ in rows)
    for domain, backend in rows:
        print(f"  {domain:<{width}}  ->  {backend}")


def cmd_add(args):
    fqdn, _, _ = _cfg()
    domain = _qualify(args.name, fqdn)
    backend = args.backend or _tailscale_ip()
    _validate_backend(backend)
    _, body = _request("POST", {"domain": domain, "ip": backend})
    _check(body)
    # Re-read rather than trusting the POST body: app.py mutates its in-memory dict and
    # renders that, so the response shows what it meant to do, not what landed on disk.
    _, fresh = _request()
    if not any(d == domain for d, _ in _parse_domains(fresh)):
        sys.exit(f"POST returned 200 but {domain} is not in the mapping table — check `list`")
    print(f"  {domain}  ->  {backend}")
    print(f"\nYour box must now serve TLS for {domain} on the backend port (443 unless you")
    print("set one). The gateway does not terminate TLS; it proxies the encrypted stream.")


def cmd_remove(args):
    fqdn, _, _ = _cfg()
    domain = _qualify(args.name, fqdn)
    # The admin app treats a backend of exactly "0" as delete.
    _, body = _request("POST", {"domain": domain, "ip": "0"})
    _check(body)
    _, fresh = _request()
    if any(d == domain for d, _ in _parse_domains(fresh)):
        sys.exit(
            f"{domain} is still in the mapping table after the delete.\n"
            "If its backend is a hostname rather than an IP, the gateway cannot remove it:\n"
            "edit /var/lib/tunnel-gateway/tunnel_map.conf on the box and reload nginx."
        )
    print(f"  removed {domain}")


def main():
    ap = argparse.ArgumentParser(
        description="Map a public domain to your tailnet box via the testing gateway.",
        epilog="Password comes from the secret store (gateway.admin_password); see skills/gateway-domain/SKILL.md.",
    )
    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("list", help="show current domain mappings").set_defaults(fn=cmd_list)

    a = sub.add_parser("add", help="point a domain at a backend")
    a.add_argument("name", help="bare label (myapp) or a full domain")
    a.add_argument("backend", nargs="?", help="IP or IP:port (default: this box's tailscale IP, port 443)")
    a.set_defaults(fn=cmd_add)

    r = sub.add_parser("remove", help="delete a domain mapping")
    r.add_argument("name", help="bare label (myapp) or a full domain")
    r.set_defaults(fn=cmd_remove)

    args = ap.parse_args()
    args.fn(args)


if __name__ == "__main__":
    main()