furtka-gateway/control_plane/wgeasy_v15.py

"""REST client for wg-easy v15's admin API — the Nitro/h3-based rewrite
wgeasy.py's module docstring warned about, confirmed here to be a real,
materially different API rather than a compatible evolution of v14's.

Verified directly against a real running container
(`ghcr.io/wg-easy/wg-easy:15`, booted headlessly via its `INIT_ENABLED=true`
bootstrap env vars, every endpoint below hit with live `curl` calls) rather
than trusted from reading the bundled source alone — the compiled Nitro
output's chunk filenames turned out to be a false trail (e.g.
`routes/api/client/index.post.mjs`'s handler body reads a router-param
`clientId` that doesn't exist on that literal path — it's not actually the
client-creation endpoint despite the name), so treat only the behavior
below as trustworthy, not the source layout.

What's different from v14 (see wgeasy.py):

  - Auth: HTTP Basic (`Authorization: Basic base64(username:password)`)
    against a real user account, not a bare plaintext-password header. An
    admin user must already exist — create one headlessly by setting
    `INIT_ENABLED=true`/`INIT_USERNAME`/`INIT_PASSWORD`/`INIT_HOST`/
    `INIT_PORT` on the wg-easy container itself (confirmed live: this
    fully skips the interactive setup wizard). Basic auth is rejected for
    a user with TOTP (2FA) enabled — keep the automation account's 2FA off.
  - `POST /api/client` requires `{"name": ..., "expiresAt": null}` (v15
    added per-client expiry; `null` means never) and its response DOES
    carry `{"clientId": ...}` directly — no more list-and-match-by-name.
    The public key isn't in that response though, so `create_client()`
    still does one follow-up `GET /api/client` list call, matched by the
    now-unique numeric id (far less fragile than v14's match-by-name).
  - `GET /api/client/{id}/configuration` returns a wg-quick .conf that now
    also includes a `PresharedKey` line (v15 generates one per client by
    default) — callers must thread this through to the box's own
    wg-quick config, or the handshake fails silently (the server enforces
    the PSK even if the client's own conf omits it).
  - `DELETE /api/client/{id}` (id in the URL path — confirmed live; DELETE
    on the bare `/api/client` collection 404s).
  - New clients default to full-tunnel AllowedIPs (`0.0.0.0/0, ::/0`)
    unless the instance's interface-wide default is restricted first via
    `POST /api/admin/userconfig` — confirmed live that endpoint requires
    the FULL current config object back (a partial/PATCH-style body
    400s), so `restrict_default_allowed_ips()` GETs the current config
    before POSTing the change.

Re-verify against the running container directly (not upstream docs) if
this image tag is ever bumped — the same lesson wgeasy.py's docstring
already learned once for v14.
"""

from __future__ import annotations

import base64
import json
import re
import urllib.error
import urllib.request


class WgEasyV15Error(Exception):
    pass


class WgEasyV15Client:
    def __init__(self, base_url: str, username: str, password: str, timeout: float = 10) -> None:
        self._base_url = base_url.rstrip("/")
        self._auth_header = "Basic " + base64.b64encode(
            f"{username}:{password}".encode()
        ).decode()
        self._timeout = timeout

    def restrict_default_allowed_ips(self, allowed_ips: list[str]) -> None:
        """Set the interface-wide default AllowedIPs new clients get (e.g.
        split-tunnel `["10.8.0.0/24"]`) instead of v15's full-tunnel
        default — the same concern v14's WG_ALLOWED_IPS container env var
        addresses, but v15 has no equivalent env var that applies to an
        already-initialized instance, so this calls the admin API
        directly. Idempotent — safe to call on every control-plane
        startup.
        """
        current = json.loads(self._request("GET", "/api/admin/userconfig"))
        current.pop("id", None)
        current.pop("createdAt", None)
        current.pop("updatedAt", None)
        current["defaultAllowedIps"] = allowed_ips
        self._request("POST", "/api/admin/userconfig", current)

    def create_client(self, name: str) -> dict:
        """Create a new WireGuard peer and return its full connection material."""
        created = json.loads(
            self._request("POST", "/api/client", {"name": name, "expiresAt": None})
        )
        client_id = created["clientId"]

        clients = json.loads(self._request("GET", "/api/client"))
        matches = [c for c in clients if c["id"] == client_id]
        if not matches:
            raise WgEasyV15Error(
                f"created client id {client_id!r} but it did not appear in the client list"
            )
        public_key = matches[0]["publicKey"]

        conf_text = self._request(
            "GET", f"/api/client/{client_id}/configuration"
        ).decode()
        conf = _parse_wg_conf(conf_text)

        return {
            "id": client_id,
            "public_key": public_key,
            "private_key": conf["private_key"],
            "address": conf["address"],
            "server_public_key": conf["peer_public_key"],
            "endpoint": conf["endpoint"],
            "allowed_ips": conf["allowed_ips"],
            "preshared_key": conf.get("preshared_key"),
        }

    def delete_client(self, client_id: str) -> None:
        try:
            self._request("DELETE", f"/api/client/{client_id}")
        except WgEasyV15Error:
            # Already gone on wg-easy's side shouldn't block us cleaning up
            # our own box/route rows — deregistration must still succeed.
            pass

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


def _parse_wg_conf(text: str) -> dict:
    def find(pattern: str, required: bool = True) -> str | None:
        m = re.search(pattern, text, re.MULTILINE)
        if not m:
            if required:
                raise WgEasyV15Error(f"could not find {pattern!r} in wg-easy client configuration")
            return None
        return m.group(1).strip()

    return {
        "private_key": find(r"^PrivateKey\s*=\s*(.+)$"),
        "address": find(r"^Address\s*=\s*(.+)$"),
        "peer_public_key": find(r"^PublicKey\s*=\s*(.+)$"),
        "preshared_key": find(r"^PresharedKey\s*=\s*(.+)$", required=False),
        "endpoint": find(r"^Endpoint\s*=\s*(.+)$"),
        "allowed_ips": find(r"^AllowedIPs\s*=\s*(.+)$"),
    }