furtka/furtka/wireguard.py

"""WireGuard tunnel to a connected furtka-gateway.

Structural mirror of furtka.https's snippet-write-plus-reload pattern:
write/remove /etc/wireguard/furtka-gw.conf, then flip the wg-quick@furtka-gw
systemd unit, reverting the conf file if that call fails so a bad config
can't leave the tunnel half-configured. One deliberate difference from
https.py: enabling and disabling are genuinely different systemctl
subcommands here (Caddy only ever needs a single `reload` either way), so
the injectable `reload` callable takes the resolved action ("enable" or
"disable") rather than being a fixed zero-argument call.

This module never generates its own keypair or contacts furtka-gateway
directly — furtka.gateway.register() gets the full WireGuard interface
config (including the private key; see furtka-gateway's wgeasy.py for why
that's unavoidable with stock wg-easy) and hands it to
set_tunnel_enabled() here.
"""

import os
import subprocess
from pathlib import Path

DEFAULT_CONF_PATH = Path("/etc/wireguard/furtka-gw.conf")
INTERFACE = "furtka-gw"


def default_conf_path() -> Path:
    return Path(os.environ.get("FURTKA_WIREGUARD_CONF", DEFAULT_CONF_PATH))


class WireguardError(Exception):
    """Recoverable failure from set_tunnel_enabled — the caller should 5xx."""


def _conf_content(wg: dict) -> str:
    """Render a wg-quick .conf from a furtka-gateway registration response's
    `wg` block. AllowedIPs here governs what THIS box routes *through* the
    tunnel — furtka-gateway hands back a split-tunnel value (its own peer
    subnet only), which this module passes through verbatim rather than
    widening it, so a compromised gateway can't pull the box's other
    traffic through it.
    """
    return (
        "[Interface]\n"
        f"PrivateKey = {wg['private_key']}\n"
        f"Address = {wg['address']}\n"
        "\n"
        "[Peer]\n"
        f"PublicKey = {wg['server_public_key']}\n"
        f"Endpoint = {wg['endpoint']}\n"
        f"AllowedIPs = {wg['allowed_ips']}\n"
        "PersistentKeepalive = 25\n"
    )


def _default_reload(action: str) -> None:
    subprocess.run(
        ["systemctl", action, "--now", f"wg-quick@{INTERFACE}"],
        check=True,
        capture_output=True,
        text=True,
    )


def set_tunnel_enabled(
    enabled: bool,
    wg: dict | None = None,
    conf_path: Path | None = None,
    reload=_default_reload,
) -> bool:
    """Bring the tunnel up (writing conf_path from `wg`) or down.

    `wg` is required when enabling ({private_key, address, server_public_key,
    endpoint, allowed_ips} — furtka.gateway.register()'s return shape),
    ignored when disabling. Reverts conf_path to its pre-call state if the
    systemctl action fails, so a bad config can't leave the tunnel wedged.
    """
    if enabled and wg is None:
        raise ValueError("wg config is required to enable the tunnel")
    if conf_path is None:
        conf_path = default_conf_path()

    try:
        conf_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        had_conf = conf_path.is_file()
        previous = conf_path.read_text() if had_conf else None

        if enabled:
            conf_path.write_text(_conf_content(wg))
            conf_path.chmod(0o600)
        elif had_conf:
            conf_path.unlink()
    except OSError as e:
        raise WireguardError(f"could not access {conf_path}: {e}") from e

    try:
        reload("enable" if enabled else "disable")
    except subprocess.CalledProcessError as e:
        _revert(conf_path, previous)
        msg = (e.stderr or e.stdout or "").strip() or f"exit {e.returncode}"
        raise WireguardError(f"systemctl failed: {msg}") from e
    except FileNotFoundError as e:
        _revert(conf_path, previous)
        raise WireguardError(f"systemctl not available: {e}") from e
    return enabled


def _revert(conf_path: Path, previous: str | None) -> None:
    if previous is None:
        try:
            conf_path.unlink()
        except FileNotFoundError:
            pass
    else:
        conf_path.write_text(previous)


def _default_wg_show() -> str:
    result = subprocess.run(
        ["wg", "show", INTERFACE, "latest-handshakes"],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def status(conf_path: Path | None = None, wg_show=_default_wg_show) -> dict:
    """Whether the tunnel is configured, and its latest handshake if `wg`
    tooling is available (None if not — e.g. wg-tools missing, or the
    interface isn't actually up yet right after enabling).

    Read-only and best-effort: any OSError stat-ing conf_path (missing
    parent dir, an unexpected permission situation) is treated the same as
    "not configured" rather than raised — a status check must never 500
    the whole request over something this innocuous.
    """
    if conf_path is None:
        conf_path = default_conf_path()
    try:
        configured = conf_path.is_file()
    except OSError:
        configured = False
    latest_handshake = None
    if configured:
        try:
            latest_handshake = wg_show()
        except (subprocess.CalledProcessError, FileNotFoundError):
            latest_handshake = None
    return {"configured": configured, "latest_handshake": latest_handshake}