import base64 import json import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import pytest from control_plane.wgeasy_v15 import WgEasyV15Client, WgEasyV15Error USERNAME = "admin" PASSWORD = "hunter2" _AUTH_HEADER = "Basic " + base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode() FAKE_CONF = """[Interface] PrivateKey = client-private-key== Address = 10.8.0.2/32 MTU = 1420 DNS = 1.1.1.1 [Peer] PublicKey = server-public-key== PresharedKey = preshared-key== AllowedIPs = 10.8.0.0/24 PersistentKeepalive = 0 Endpoint = gateway.example.com:51820 """ CLIENT_RECORD = { "id": 1, "name": "box-1", "enabled": True, "publicKey": "client-public-key==", } USERCONFIG = { "id": "wg0", "defaultMtu": 1420, "defaultAllowedIps": ["0.0.0.0/0", "::/0"], "host": "gateway.example.com", "port": 51820, "createdAt": "2026-01-01 00:00:00", "updatedAt": "2026-01-01 00:00:00", } class FakeWgEasyV15Handler(BaseHTTPRequestHandler): def log_message(self, format, *args): pass def _json(self, status, payload): body = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _check_auth(self) -> bool: ok = self.headers.get("Authorization") == _AUTH_HEADER if not ok: self._json(401, {"error": True, "statusMessage": "Not authenticated"}) return ok def _body(self) -> dict: length = int(self.headers.get("Content-Length", 0)) return json.loads(self.rfile.read(length)) if length else {} def do_POST(self): self.server.requests.append(("POST", self.path)) if not self._check_auth(): return if self.path == "/api/client": body = self._body() assert body["expiresAt"] is None self._json(200, {"success": True, "clientId": CLIENT_RECORD["id"]}) return if self.path == "/api/admin/userconfig": self.server.last_userconfig_post = self._body() self._json(200, {"success": True}) return self._json(404, {"error": True, "statusMessage": "not found"}) def do_GET(self): self.server.requests.append(("GET", self.path)) if not self._check_auth(): return if self.path == "/api/client": self._json(200, [CLIENT_RECORD]) return if self.path == f"/api/client/{CLIENT_RECORD['id']}/configuration": body = FAKE_CONF.encode() self.send_response(200) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if self.path == "/api/admin/userconfig": self._json(200, USERCONFIG) return self._json(404, {"error": True, "statusMessage": "not found"}) def do_DELETE(self): self.server.requests.append(("DELETE", self.path)) if not self._check_auth(): return if self.path == f"/api/client/{CLIENT_RECORD['id']}": self._json(200, {"success": True}) return self._json(404, {"error": True, "statusMessage": "not found"}) @pytest.fixture def fake_wgeasy(): server = ThreadingHTTPServer(("127.0.0.1", 0), FakeWgEasyV15Handler) server.requests = [] server.last_userconfig_post = None thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: yield server finally: server.shutdown() thread.join() def _client(server, username=USERNAME, password=PASSWORD): port = server.server_address[1] return WgEasyV15Client(f"http://127.0.0.1:{port}", username, password) def test_create_client_returns_full_connection_material(fake_wgeasy): client = _client(fake_wgeasy) peer = client.create_client("box-1") assert peer["id"] == CLIENT_RECORD["id"] assert peer["public_key"] == "client-public-key==" assert peer["private_key"] == "client-private-key==" assert peer["address"] == "10.8.0.2/32" assert peer["server_public_key"] == "server-public-key==" assert peer["preshared_key"] == "preshared-key==" assert peer["endpoint"] == "gateway.example.com:51820" assert peer["allowed_ips"] == "10.8.0.0/24" paths_hit = [(m, p) for m, p in fake_wgeasy.requests] assert ("POST", "/api/client") in paths_hit assert ("GET", "/api/client") in paths_hit assert ("GET", f"/api/client/{CLIENT_RECORD['id']}/configuration") in paths_hit def test_every_request_uses_basic_auth_and_wrong_password_fails(fake_wgeasy): client = _client(fake_wgeasy, password="wrong-password") with pytest.raises(WgEasyV15Error): client.create_client("box-1") def test_delete_client_calls_endpoint(fake_wgeasy): client = _client(fake_wgeasy) client.delete_client(str(CLIENT_RECORD["id"])) assert ("DELETE", f"/api/client/{CLIENT_RECORD['id']}") in [ (m, p) for m, p in fake_wgeasy.requests ] def test_delete_client_swallows_not_found(fake_wgeasy): client = _client(fake_wgeasy) client.delete_client("999") # 404 from fake server — must not raise def test_restrict_default_allowed_ips_sends_full_config_back(fake_wgeasy): client = _client(fake_wgeasy) client.restrict_default_allowed_ips(["10.8.0.0/24"]) sent = fake_wgeasy.last_userconfig_post assert sent["defaultAllowedIps"] == ["10.8.0.0/24"] # The full current config must be echoed back, not a partial patch — # confirmed live against v15 that a partial body 400s. assert sent["defaultMtu"] == USERCONFIG["defaultMtu"] assert sent["host"] == USERCONFIG["host"] # Server-managed fields must not be sent back. assert "id" not in sent assert "createdAt" not in sent assert "updatedAt" not in sent def test_bad_endpoint_raises_wgeasy_v15_error(fake_wgeasy): client = _client(fake_wgeasy) with pytest.raises(WgEasyV15Error): client._request("GET", "/api/does-not-exist")