furtka-gateway/tests/test_api.py

import json
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer

import pytest

from control_plane.api import Context, Handler
from control_plane.db import Database


class FakeWgEasy:
    def __init__(self):
        self._counter = 0
        self.deleted = []

    def create_client(self, name):
        self._counter += 1
        return {
            "id": f"peer-{self._counter}",
            "public_key": f"pubkey-{self._counter}==",
            "private_key": f"privkey-{self._counter}==",
            "address": f"10.8.0.{self._counter}/32",
            "server_public_key": "server-pubkey==",
            "endpoint": "gateway.example.com:51820",
            "allowed_ips": "10.8.0.0/24",
        }

    def delete_client(self, client_id):
        self.deleted.append(client_id)


@pytest.fixture
def server(tmp_path, gateway_paths, monkeypatch):
    monkeypatch.setenv("GATEWAY_BOX_TOKEN", "test-box-token")
    db = Database(db_path=tmp_path / "gateway.db")
    ctx = Context(
        db=db,
        wgeasy=FakeWgEasy(),
        mode="single",
        cert_resolver="le",
        single_account_id="default",
        box_limit=5,
        route_limit_per_box=5,
    )
    db.execute(
        "INSERT INTO accounts (id, email, created_at, registration_token_hash, "
        "account_token_hash, box_limit, route_limit_per_box) "
        "VALUES ('default', NULL, '2026-01-01T00:00:00', '', '', 5, 5)"
    )
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    httpd.ctx = ctx
    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
    thread.start()
    try:
        yield httpd
    finally:
        httpd.shutdown()
        thread.join()


def _url(server, path):
    port = server.server_address[1]
    return f"http://127.0.0.1:{port}{path}"


def _request(server, method, path, body=None, token=None):
    data = json.dumps(body).encode() if body is not None else None
    headers = {"Content-Type": "application/json"} if data else {}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(_url(server, path), data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())


def test_healthz(server):
    status, body = _request(server, "GET", "/healthz")
    assert status == 200
    assert body == {"status": "ok"}


def _register(server, box_name="my-box"):
    return _request(
        server,
        "POST",
        "/v1/boxes/register",
        {"registration_token": "test-box-token", "box_name": box_name},
    )


def test_register_rejects_wrong_token(server):
    status, _ = _request(
        server,
        "POST",
        "/v1/boxes/register",
        {"registration_token": "wrong", "box_name": "my-box"},
    )
    assert status == 401


def test_register_rejects_missing_fields(server):
    status, _ = _request(server, "POST", "/v1/boxes/register", {"box_name": "my-box"})
    assert status == 400


def test_full_box_and_route_lifecycle(server):
    status, body = _register(server)
    assert status == 201
    box_token = body["box_token"]
    assert body["wg"]["private_key"] == "privkey-1=="

    # Unauthenticated calls are rejected.
    status, _ = _request(server, "GET", "/v1/routes")
    assert status == 401

    status, _ = _request(server, "GET", "/v1/routes", token=box_token)
    assert status == 200

    status, body = _request(
        server,
        "POST",
        "/v1/routes",
        {"app_name": "vaultwarden", "subdomain": "vault.example.com", "port": 8081},
        token=box_token,
    )
    assert status == 201
    route_id = body["route_id"]
    assert body["public_url"] == "https://vault.example.com/"

    status, body = _request(server, "GET", "/v1/routes", token=box_token)
    assert status == 200
    assert len(body["routes"]) == 1
    assert body["routes"][0]["subdomain"] == "vault.example.com"

    status, _ = _request(server, "POST", f"/v1/boxes/{_box_id(box_token)}/heartbeat", token=box_token)
    assert status == 200

    status, _ = _request(server, "DELETE", f"/v1/routes/{route_id}", token=box_token)
    assert status == 200

    status, body = _request(server, "GET", "/v1/routes", token=box_token)
    assert body["routes"] == []


def test_route_ownership_is_enforced_across_boxes(server):
    _, box_a = _register(server, "box-a")
    _, box_b = _register(server, "box-b")

    _, route = _request(
        server,
        "POST",
        "/v1/routes",
        {"app_name": "vaultwarden", "subdomain": "vault.example.com", "port": 8081},
        token=box_a["box_token"],
    )

    # box-b's token must not be able to delete box-a's route.
    status, _ = _request(
        server, "DELETE", f"/v1/routes/{route['route_id']}", token=box_b["box_token"]
    )
    assert status == 404


def test_box_action_rejects_mismatched_box_id(server):
    _, box_a = _register(server, "box-a")
    _, box_b = _register(server, "box-b")

    status, _ = _request(
        server,
        "POST",
        f"/v1/boxes/{_box_id(box_b['box_token'])}/heartbeat",
        token=box_a["box_token"],
    )
    assert status == 403


def test_rotate_token_invalidates_old_token(server):
    _, box = _register(server)
    old_token = box["box_token"]
    box_id = _box_id(old_token)

    status, body = _request(server, "POST", f"/v1/boxes/{box_id}/rotate-token", token=old_token)
    assert status == 200
    new_token = body["box_token"]

    status, _ = _request(server, "GET", "/v1/routes", token=old_token)
    assert status == 401
    status, _ = _request(server, "GET", "/v1/routes", token=new_token)
    assert status == 200


def test_deregister_removes_routes_and_wgeasy_peer(server):
    _, box = _register(server)
    box_token = box["box_token"]
    box_id = _box_id(box_token)
    _request(
        server,
        "POST",
        "/v1/routes",
        {"app_name": "vaultwarden", "subdomain": "vault.example.com", "port": 8081},
        token=box_token,
    )

    status, _ = _request(server, "POST", f"/v1/boxes/{box_id}/deregister", token=box_token)
    assert status == 200

    # The box token no longer authenticates anything after deregistration.
    status, _ = _request(server, "GET", "/v1/routes", token=box_token)
    assert status == 401
    assert server.ctx.wgeasy.deleted == ["peer-1"]


def _box_id(box_token: str) -> str:
    return box_token.split(".", 1)[0]