furtka-gateway/control_plane/wgeasy.py

"""Thin REST client for wg-easy's admin API.

Verified directly against the actual running container for the pinned
image (`ghcr.io/wg-easy/wg-easy:14`) — `docker exec furtka-gateway-wg-easy
cat /app/lib/Server.js` — rather than trusted from wg-easy's GitHub
`master` branch, which turned out to describe a different, newer,
Nitro/h3-based rewrite this tag does not actually run. Two things that
draft got wrong, corrected here: this build's auth is a single shared
password sent as a plain `Authorization` header on every request (no
login call, no session cookie, no username), and its routes live under
`/api/wireguard/client`, not `/api/client`. Re-verify against the
container directly (not against docs) if this image tag is ever bumped.

Real routes (confirmed by reading the container's own source):

  GET    /api/wireguard/client                    -> [{id, name, enabled,
                                                        address, publicKey,
                                                        createdAt, ...}]
  POST   /api/wireguard/client        {name}      -> {success: true}
                                                       (no id in the
                                                       response — list and
                                                       match by name)
  GET    /api/wireguard/client/{id}/configuration -> plain-text wg-quick
                                                       .conf (the only place
                                                       the private key ever
                                                       appears)
  DELETE /api/wireguard/client/{id}                -> {success: true}

wg-easy always generates the keypair itself; there is no supported way to
hand it an externally-generated public key. See the plan's "Correction
found during implementation" note for the resulting threat-model
consequence (the gateway sees every box's private key at registration
time).
"""

from __future__ import annotations

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


class WgEasyError(Exception):
    pass


class WgEasyClient:
    def __init__(self, base_url: str, admin_password: str, timeout: float = 10) -> None:
        self._base_url = base_url.rstrip("/")
        self._admin_password = admin_password
        self._timeout = timeout

    def create_client(self, name: str) -> dict:
        """Create a new WireGuard peer and return its full connection material.

        The create call itself never returns the new client's id, so we
        immediately list clients and match by name (most recently created,
        in case an old client happens to share the name) to find it, then
        fetch its private key via /configuration — the only endpoint that
        ever exposes it.
        """
        self._request("POST", "/api/wireguard/client", {"name": name})

        clients = json.loads(self._request("GET", "/api/wireguard/client"))
        matches = [c for c in clients if c["name"] == name]
        if not matches:
            raise WgEasyError(
                f"created client {name!r} but it did not appear in the client list"
            )
        client = max(matches, key=lambda c: c["createdAt"])
        client_id = client["id"]

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

        return {
            "id": client_id,
            "public_key": client["publicKey"],
            "private_key": conf["private_key"],
            "address": conf["address"],
            "server_public_key": conf["peer_public_key"],
            "endpoint": conf["endpoint"],
            "allowed_ips": conf["allowed_ips"],
        }

    def delete_client(self, client_id: str) -> None:
        try:
            self._request("DELETE", f"/api/wireguard/client/{client_id}")
        except WgEasyError:
            # 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._admin_password}
        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 WgEasyError(f"{method} {path} -> HTTP {e.code}: {detail}") from e
        except urllib.error.URLError as e:
            raise WgEasyError(f"{method} {path} -> {e}") from e


def _parse_wg_conf(text: str) -> dict:
    def find(pattern: str) -> str:
        m = re.search(pattern, text, re.MULTILINE)
        if not m:
            raise WgEasyError(f"could not find {pattern!r} in wg-easy client configuration")
        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*(.+)$"),
        "endpoint": find(r"^Endpoint\s*=\s*(.+)$"),
        "allowed_ips": find(r"^AllowedIPs\s*=\s*(.+)$"),
    }