furtka/furtka/gateway.py

"""Client for a Furtka box registering with (and talking to) a
furtka-gateway.

Talks to the gateway's `/v1/boxes/*` API (see the furtka-gateway repo's
`control_plane/api.py` for the server side) using the same stdlib-only
`urllib.request` convention as `_release_common.py`/`updater.py` — no new
dependency for this.

`register()` persists the result to `paths.gateway_config_file()` (0600,
atomic write — the same tmp-file-then-replace pattern `auth.save_users`
uses) so later calls don't need the box to re-register, and so
`/api/gateway/status` has something to read without holding a live
connection open.

`run_connect()` plus the state/lock helpers below follow the exact shape
of furtka.catalog's own state/lock functions — this module is the
"runner" `furtka gateway connect-bg` calls after `api.py`'s
`POST /api/gateway/connect` dispatches it via systemd-run, the same
async-with-state-file-polling idiom used for install/update/catalog-sync.
"""

from __future__ import annotations

import fcntl
import json
import os
import time
import urllib.error
import urllib.request
from pathlib import Path

from furtka import wireguard
from furtka.paths import gateway_config_file

_CONNECT_STATE = Path(
    os.environ.get("FURTKA_GATEWAY_CONNECT_STATE", "/var/lib/furtka/gateway-connect-state.json")
)
_CONNECT_LOCK = Path(
    os.environ.get("FURTKA_GATEWAY_CONNECT_LOCK", "/run/furtka/gateway-connect.lock")
)


class GatewayError(Exception):
    """Recoverable failure talking to a furtka-gateway — the caller should 5xx."""


def _request(method: str, url: str, body: dict | None = None, *, token: str | None = None) -> dict:
    data = json.dumps(body).encode() if body is not None else None
    headers = {}
    if data is not None:
        headers["Content-Type"] = "application/json"
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")
        raise GatewayError(f"{method} {url} -> HTTP {e.code}: {detail}") from e
    except urllib.error.URLError as e:
        raise GatewayError(f"{method} {url}: {e}") from e


def _post(url: str, body: dict, *, token: str | None = None) -> dict:
    return _request("POST", url, body, token=token)


def _delete(url: str, *, token: str | None = None) -> dict:
    return _request("DELETE", url, token=token)


def _save_config(config: dict) -> None:
    path = gateway_config_file()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(config, indent=2) + "\n")
    tmp.chmod(0o600)
    tmp.replace(path)


def load_config() -> dict | None:
    """Return the persisted {gateway_url, box_id, box_token, wg} config, or
    None if this box has never registered with a gateway."""
    path = gateway_config_file()
    if not path.is_file():
        return None
    return json.loads(path.read_text())


def register(gateway_url: str, registration_token: str, box_name: str) -> dict:
    """Register this box with a furtka-gateway.

    No local keypair is generated here — the gateway's response carries the
    full WireGuard interface config, private key included (see
    furtka-gateway's `wgeasy.py` module docstring for why stock wg-easy
    makes that unavoidable). Persists and returns
    {gateway_url, box_id, box_token, wg}.
    """
    gateway_url = gateway_url.rstrip("/")
    result = _post(
        f"{gateway_url}/v1/boxes/register",
        {"registration_token": registration_token, "box_name": box_name},
    )
    config = {
        "gateway_url": gateway_url,
        "box_id": result["box_id"],
        "box_token": result["box_token"],
        "wg": result["wg"],
    }
    _save_config(config)
    return config


def rotate_token() -> str:
    """Ask the gateway for a new box token, replacing the stored one."""
    config = load_config()
    if config is None:
        raise GatewayError("not connected to a gateway")
    result = _post(
        f"{config['gateway_url']}/v1/boxes/{config['box_id']}/rotate-token",
        {},
        token=config["box_token"],
    )
    config["box_token"] = result["box_token"]
    _save_config(config)
    return config["box_token"]


def deregister() -> None:
    """Tell the gateway to forget this box, then drop the local config.

    Raises on a failed gateway call rather than silently clearing local
    state — a box that can't reach the gateway should stay "connected"
    locally so the operation can be retried, instead of losing its
    box_token while the gateway's own row (and the wg-easy peer / any
    published routes) linger orphaned.
    """
    config = load_config()
    if config is None:
        return
    _post(
        f"{config['gateway_url']}/v1/boxes/{config['box_id']}/deregister",
        {},
        token=config["box_token"],
    )
    gateway_config_file().unlink(missing_ok=True)


def publish_route(app_name: str, subdomain: str, port: int) -> dict:
    """Publish `app_name` at `subdomain` through the connected gateway.

    Returns {route_id, public_url}. The gateway independently re-validates
    subdomain/port server-side regardless of what's sent here — see
    furtka-gateway's routes.py — this call just carries the box owner's
    request, it isn't itself the security boundary.
    """
    config = load_config()
    if config is None:
        raise GatewayError("not connected to a gateway")
    return _post(
        f"{config['gateway_url']}/v1/routes",
        {"app_name": app_name, "subdomain": subdomain, "port": port},
        token=config["box_token"],
    )


def unpublish_route(route_id: str) -> None:
    config = load_config()
    if config is None:
        raise GatewayError("not connected to a gateway")
    _delete(f"{config['gateway_url']}/v1/routes/{route_id}", token=config["box_token"])


def state_path() -> Path:
    return _CONNECT_STATE


def lock_path() -> Path:
    return _CONNECT_LOCK


def write_state(stage: str, **extra) -> None:
    """Atomic JSON state write — same shape as catalog/update-state."""
    state_path().parent.mkdir(parents=True, exist_ok=True)
    tmp = state_path().with_suffix(".tmp")
    payload = {"stage": stage, "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), **extra}
    tmp.write_text(json.dumps(payload, indent=2))
    tmp.replace(state_path())


def read_state() -> dict:
    try:
        return json.loads(state_path().read_text())
    except (FileNotFoundError, json.JSONDecodeError):
        return {}


def acquire_lock():
    path = lock_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    fh = path.open("w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as e:
        fh.close()
        raise GatewayError("another gateway connect is already in progress") from e
    return fh


def run_connect(gateway_url: str, registration_token: str, box_name: str) -> None:
    """Register with the gateway, then bring the tunnel up — the slow part
    `furtka gateway connect-bg` runs in the background (dispatched via
    systemd-run from `POST /api/gateway/connect`), writing state
    transitions as it goes so the UI can poll instead of the request
    thread blocking on a slow/unreachable gateway.

    A registration saved by an earlier attempt whose tunnel failed to come
    up is reused rather than registering again — re-registering would leave
    the first box (and its wg-easy peer) orphaned on the gateway.
    """
    existing = load_config()
    if existing is not None:
        if existing["gateway_url"] != gateway_url.rstrip("/"):
            write_state(
                "error",
                error=f"already registered with {existing['gateway_url']} — disconnect first",
            )
            return
        config = existing
    else:
        write_state("registering")
        try:
            config = register(gateway_url, registration_token, box_name)
        except GatewayError as e:
            write_state("error", error=str(e))
            return

    write_state("starting_tunnel")
    try:
        wireguard.set_tunnel_enabled(True, wg=config["wg"])
    except wireguard.WireguardError as e:
        write_state("error", error=str(e))
        return

    write_state("done")