Robert Syrnicki
3a9d18fcd5
wg-easy + Traefik docker-compose stack (Phase 1) plus a stdlib-only control-plane API for box registration, WireGuard peer provisioning via wg-easy, and per-box route publish/unpublish backed by Traefik's file provider (Phase 2, single-tenant mode). SQLite holds accounts/boxes/routes so a later multi-tenant shared instance is the same schema with more rows, not a reshape. wg-easy's actual REST API was verified against its source rather than assumed: it has no bearer-token auth (session-cookie login via POST /api/auth/password) and no way to accept an externally-generated public key (it always mints the keypair itself, private key included) — both corrected from the original plan during implementation.
130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
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
|