furtka-gateway/tests/test_boxes.py

import pytest

from control_plane import boxes, routes
from control_plane.db import Database


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

    def create_client(self, name):
        self._counter += 1
        peer_id = f"peer-{self._counter}"
        self.created.append(name)
        return {
            "id": peer_id,
            "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 db_with_account(tmp_path, gateway_paths):
    db = Database(db_path=tmp_path / "gateway.db")
    db.execute(
        "INSERT INTO accounts (id, email, created_at, registration_token_hash, "
        "account_token_hash, box_limit, route_limit_per_box) "
        "VALUES ('acc1', NULL, '2026-01-01T00:00:00', '', '', 5, 5)"
    )
    return db


def test_register_box_success(db_with_account):
    wgeasy = FakeWgEasy()

    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)

    assert "box_id" in result and "box_token" in result
    assert result["wg"]["private_key"] == "privkey-1=="
    assert wgeasy.created == ["my-box"]

    row = db_with_account.query_one("SELECT * FROM boxes WHERE id = ?", (result["box_id"],))
    assert row["name"] == "my-box"
    assert row["wg_public_key"] == "pubkey-1=="


def test_register_box_enforces_box_limit(db_with_account):
    wgeasy = FakeWgEasy()
    boxes.register_box(db_with_account, wgeasy, "acc1", "box-a", box_limit=1)
    with pytest.raises(boxes.BoxLimitExceeded):
        boxes.register_box(db_with_account, wgeasy, "acc1", "box-b", box_limit=1)


def test_authenticate_box_token_roundtrip(db_with_account):
    wgeasy = FakeWgEasy()
    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)

    row = boxes.authenticate_box_token(db_with_account, result["box_token"])

    assert row is not None
    assert row["id"] == result["box_id"]


@pytest.mark.parametrize(
    "bad_token",
    ["not-a-real-token", "unknown-box-id.somesecret", ""],
)
def test_authenticate_box_token_rejects_bad_tokens(db_with_account, bad_token):
    wgeasy = FakeWgEasy()
    boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)

    assert boxes.authenticate_box_token(db_with_account, bad_token) is None


def test_authenticate_box_token_rejects_wrong_secret(db_with_account):
    wgeasy = FakeWgEasy()
    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)
    box_id = result["box_id"]

    tampered = f"{box_id}.wrong-secret"

    assert boxes.authenticate_box_token(db_with_account, tampered) is None


def test_rotate_box_token_invalidates_old_one(db_with_account):
    wgeasy = FakeWgEasy()
    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)
    box_id = result["box_id"]

    new_token = boxes.rotate_box_token(db_with_account, box_id)

    assert boxes.authenticate_box_token(db_with_account, result["box_token"]) is None
    assert boxes.authenticate_box_token(db_with_account, new_token)["id"] == box_id


def test_touch_last_seen(db_with_account):
    wgeasy = FakeWgEasy()
    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)

    boxes.touch_last_seen(db_with_account, result["box_id"])

    row = db_with_account.query_one("SELECT last_seen_at FROM boxes WHERE id = ?", (result["box_id"],))
    assert row["last_seen_at"] is not None


def test_deregister_box_removes_peer_and_routes(db_with_account):
    wgeasy = FakeWgEasy()
    result = boxes.register_box(db_with_account, wgeasy, "acc1", "my-box", box_limit=5)
    box_id = result["box_id"]
    routes.publish_route(db_with_account, box_id, "acc1", "app1", "app1.example.com", 80, 5, "le")

    assert boxes.deregister_box(db_with_account, wgeasy, box_id) is True

    assert db_with_account.query_one("SELECT id FROM boxes WHERE id = ?", (box_id,)) is None
    assert db_with_account.query_all("SELECT id FROM routes WHERE box_id = ?", (box_id,)) == []
    assert wgeasy.deleted == ["peer-1"]


def test_deregister_box_unknown_id(db_with_account):
    wgeasy = FakeWgEasy()
    assert boxes.deregister_box(db_with_account, wgeasy, "nope") is False