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.
105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
import pytest
|
|
|
|
from control_plane import paths, routes
|
|
from control_plane.db import Database
|
|
|
|
|
|
@pytest.fixture
|
|
def db_with_box(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)"
|
|
)
|
|
db.execute(
|
|
"INSERT INTO boxes (id, account_id, name, wg_public_key, wg_peer_id, "
|
|
"wg_allowed_ip, box_token_hash, registered_at) "
|
|
"VALUES ('box1', 'acc1', 'my-box', 'pk==', 'peer-1', '10.8.0.5/32', 'h', "
|
|
"'2026-01-01T00:00:00')"
|
|
)
|
|
return db
|
|
|
|
|
|
def test_publish_route_success(db_with_box):
|
|
result = routes.publish_route(
|
|
db_with_box, "box1", "acc1", "vaultwarden", "vault.example.com", 8081, 5, "le"
|
|
)
|
|
|
|
assert result["public_url"] == "https://vault.example.com/"
|
|
row = db_with_box.query_one("SELECT * FROM routes WHERE id = ?", (result["route_id"],))
|
|
assert row["subdomain"] == "vault.example.com"
|
|
assert (paths.dynamic_dir() / f"route-{result['route_id']}.yml").exists()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"app_name,subdomain,port",
|
|
[
|
|
("Bad Name", "vault.example.com", 8081),
|
|
("vaultwarden", "not a domain", 8081),
|
|
("vaultwarden", "www.example.com", 8081),
|
|
("vaultwarden", "vault.example.com", 0),
|
|
("vaultwarden", "vault.example.com", 70000),
|
|
("vaultwarden", "vault.example.com", "8081"),
|
|
],
|
|
)
|
|
def test_publish_route_rejects_invalid_input(db_with_box, app_name, subdomain, port):
|
|
with pytest.raises(routes.InvalidRoute):
|
|
routes.publish_route(db_with_box, "box1", "acc1", app_name, subdomain, port, 5, "le")
|
|
|
|
|
|
def test_publish_route_rejects_duplicate_subdomain(db_with_box):
|
|
routes.publish_route(
|
|
db_with_box, "box1", "acc1", "vaultwarden", "vault.example.com", 8081, 5, "le"
|
|
)
|
|
with pytest.raises(routes.SubdomainTaken):
|
|
routes.publish_route(
|
|
db_with_box, "box1", "acc1", "jellyfin", "vault.example.com", 8096, 5, "le"
|
|
)
|
|
|
|
|
|
def test_publish_route_enforces_route_limit(db_with_box):
|
|
routes.publish_route(db_with_box, "box1", "acc1", "app1", "app1.example.com", 80, 1, "le")
|
|
with pytest.raises(routes.RouteLimitExceeded):
|
|
routes.publish_route(db_with_box, "box1", "acc1", "app2", "app2.example.com", 80, 1, "le")
|
|
|
|
|
|
def test_publish_route_unknown_box(db_with_box):
|
|
with pytest.raises(routes.InvalidRoute):
|
|
routes.publish_route(
|
|
db_with_box, "nope", "acc1", "vaultwarden", "vault.example.com", 8081, 5, "le"
|
|
)
|
|
|
|
|
|
def test_unpublish_route_removes_row_and_file(db_with_box):
|
|
result = routes.publish_route(
|
|
db_with_box, "box1", "acc1", "vaultwarden", "vault.example.com", 8081, 5, "le"
|
|
)
|
|
route_id = result["route_id"]
|
|
|
|
assert routes.unpublish_route(db_with_box, "box1", route_id) is True
|
|
assert db_with_box.query_one("SELECT id FROM routes WHERE id = ?", (route_id,)) is None
|
|
assert not (paths.dynamic_dir() / f"route-{route_id}.yml").exists()
|
|
|
|
|
|
def test_unpublish_route_ownership_check(db_with_box):
|
|
result = routes.publish_route(
|
|
db_with_box, "box1", "acc1", "vaultwarden", "vault.example.com", 8081, 5, "le"
|
|
)
|
|
assert routes.unpublish_route(db_with_box, "some-other-box", result["route_id"]) is False
|
|
|
|
|
|
def test_unpublish_all_for_box(db_with_box):
|
|
routes.publish_route(db_with_box, "box1", "acc1", "app1", "app1.example.com", 80, 5, "le")
|
|
routes.publish_route(db_with_box, "box1", "acc1", "app2", "app2.example.com", 80, 5, "le")
|
|
|
|
routes.unpublish_all_for_box(db_with_box, "box1")
|
|
|
|
assert db_with_box.query_all("SELECT id FROM routes WHERE box_id = 'box1'") == []
|
|
assert traefik_dir_empty(db_with_box)
|
|
|
|
|
|
def traefik_dir_empty(db) -> bool:
|
|
from control_plane import traefikconf
|
|
|
|
return traefikconf.existing_route_ids() == set()
|