from pathlib import Path from urllib.parse import urlparse from fastapi.testclient import TestClient import karldns.services.ddns as ddns_module from karldns import DdnsSettings, DdnsStore, DnsProviderError, make_app def _path(url: str) -> str: parsed = urlparse(url) return parsed.path def test_custom_api_docs_include_only_current_api_surfaces(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", public_base_url="https://karldns.de", ) ) client = TestClient(app) docs = client.get("/docs") redoc = client.get("/redoc") schema = client.get("/openapi.json").json() assert docs.status_code == 200 assert redoc.status_code == 404 assert "Small API surface." in docs.text assert "POST /api/v1/hostnames" in docs.text assert "GET or POST /api/v1/router/register/<secret>" in docs.text assert "GET or POST /api/v1/router/update/<secret>" in docs.text assert "GET /api/v1/hostnames/<secret>" in docs.text assert "POST /api/v1/hostnames/<secret>/rotate-update-url" in docs.text assert "Rotate update URL" in docs.text assert "Operator API" not in docs.text assert "/api/v1/admin" not in docs.text assert "admin:<admin-password>" not in docs.text assert "https://karldns.de/api/v1/hostnames" in docs.text assert "http://localhost:8080/api/v1/hostnames" not in docs.text assert "registration_url" in docs.text assert "/api/v1/management" not in docs.text assert "/api/v1/hostnames/magic" not in docs.text assert "/nic/update" not in docs.text assert "/u/<secret>" not in docs.text assert "swagger-ui" not in docs.text.lower() assert "redoc" not in docs.text.lower() assert "x-robots-tag" not in docs.headers assert client.get("/openapi.json").headers["x-robots-tag"] == "noindex, nofollow" assert "cdn.jsdelivr.net" not in docs.headers["content-security-policy"] assert "fonts.googleapis.com" not in docs.headers["content-security-policy"] expected_paths = { "/api/v1/hostnames", "/api/v1/hostnames/{dashboard_secret}", "/api/v1/hostnames/{dashboard_secret}/rotate-update-url", "/api/v1/router/register/{dashboard_secret}", "/api/v1/router/update/{update_slug}", } assert set(schema["paths"]) == expected_paths assert "/" not in schema["paths"] assert "/admin" not in schema["paths"] assert "/healthz" not in schema["paths"] assert "/records" not in schema["paths"] assert "/events" not in schema["paths"] assert any(tag["name"] == "hostnames" for tag in schema["tags"]) assert any(tag["name"] == "router" for tag in schema["tags"]) assert all(tag["name"] != "admin" for tag in schema["tags"]) schema_names = set(schema["components"]["schemas"]) assert "AdminDashboardResponse" not in schema_names assert "AdminHostnamesResponse" not in schema_names assert "AdminEventsResponse" not in schema_names assert "AdminRecordsResponse" not in schema_names def test_custom_api_docs_use_visitor_public_ip(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", public_base_url="https://karldns.de", ) ) client = TestClient(app, client=("8.8.4.4", 12345)) response = client.get("/docs") assert response.status_code == 200 assert "myip=8.8.4.4" in response.text def test_openapi_hides_custom_operator_routes(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", admin_path="/ops", ) ) client = TestClient(app) schema = client.get("/openapi.json").json() forbidden_paths = { "/ops", "/ops/hostnames/delete", "/ops/hostnames/rotate", "/ops/cleanup", "/ops/events", "/ops/records", "/records", "/events", "/api/v1/admin/dashboard", "/api/v1/admin/events", "/api/v1/admin/hostnames", "/api/v1/admin/records", } assert client.get("/redoc").status_code == 404 assert forbidden_paths.isdisjoint(schema["paths"]) assert not any(path.startswith("/api/v1/admin") for path in schema["paths"]) def test_openapi_marks_generated_links_as_secret(tmp_path: Path) -> None: app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) client = TestClient(app) schema = client.get("/openapi.json").json() created = schema["components"]["schemas"]["HostnameCreateResponse"]["properties"] status = schema["components"]["schemas"]["HostnameStatusResponse"]["properties"] for name in ["update_url", "dashboard_url", "registration_url"]: assert "Secret" in created[name]["description"] for name in ["update_url", "dashboard_url"]: assert "Secret" in status[name]["description"] def test_hostname_api_lifecycle_and_history(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", hostname_suffix="ddns.example.net", public_base_url="http://ddns.example.net", ) ) client = TestClient(app) created = client.post("/api/v1/hostnames") assert created.status_code == 201 assert created.headers["cache-control"] == "no-store" body = created.json() assert body["hostname"].endswith(".ddns.example.net") assert body["status"] == "pending" assert body["update_url"].startswith("http://ddns.example.net/api/v1/router/update/") assert body["dashboard_url"].startswith("http://ddns.example.net/m/") assert body["registration_url"].startswith("http://ddns.example.net/api/v1/router/register/") dashboard_secret = body["dashboard_url"].rsplit("/", 1)[1] update_path = _path(body["update_url"]) register_path = _path(body["registration_url"]) pending = client.get(f"/api/v1/hostnames/{dashboard_secret}") assert pending.status_code == 200 assert pending.json()["status"] == "pending" assert pending.json()["history"] == [] assert client.get(update_path, params={"myip": "203.0.113.11"}).status_code == 404 registered = client.post(register_path, params={"myip": "203.0.113.10"}) assert registered.status_code == 200 assert registered.text == "good registered" update = client.get(update_path, params={"myip": "203.0.113.11"}) assert update.status_code == 200 assert update.text == "good 203.0.113.11" active = client.get(f"/api/v1/hostnames/{dashboard_secret}", params={"history_limit": 10}) assert active.status_code == 200 payload = active.json() assert payload["status"] == "active" assert payload["ipv4"] == "203.0.113.11" assert payload["history"][0]["hostname"] == body["hostname"] assert "source_ip" not in payload["history"][0] deleted = client.delete(f"/api/v1/hostnames/{dashboard_secret}") assert deleted.status_code == 204 assert client.get(f"/api/v1/hostnames/{dashboard_secret}").status_code == 404 def test_hostname_api_can_rotate_update_url(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", hostname_suffix="ddns.example.net", public_base_url="http://ddns.example.net", ) ) client = TestClient(app) body = client.post("/api/v1/hostnames").json() dashboard_secret = body["dashboard_url"].rsplit("/", 1)[1] old_update_path = _path(body["update_url"]) pending_rotate = client.post(f"/api/v1/hostnames/{dashboard_secret}/rotate-update-url") registered = client.post(_path(body["registration_url"]), params={"myip": "203.0.113.10"}) rotated = client.post(f"/api/v1/hostnames/{dashboard_secret}/rotate-update-url") new_update_path = _path(rotated.json()["update_url"]) assert pending_rotate.status_code == 400 assert registered.status_code == 200 assert rotated.status_code == 200 assert rotated.json()["hostname"] == body["hostname"] assert rotated.json()["update_url"] != body["update_url"] assert client.get(old_update_path, params={"myip": "203.0.113.11"}).status_code == 404 assert client.get(new_update_path, params={"myip": "203.0.113.12"}).status_code == 200 def test_hostname_api_history_is_scoped_to_dashboard_secret(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) own = store.create_account("own.ddns.example.net", "own-router") other = store.create_account("other.ddns.example.net", "other-router") store.log_update_event("own.ddns.example.net", "203.0.113.21", None, "good", "updated", "198.51.100.10") store.log_update_event("other.ddns.example.net", "203.0.113.22", None, "good", "updated", "198.51.100.11") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.get(f"/api/v1/hostnames/{own['management_slug']}", params={"history_limit": 10}) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" payload = response.json() assert len(payload["history"]) == 1 assert payload["history"][0]["hostname"] == "own.ddns.example.net" assert "source_ip" not in payload["history"][0] assert "other.ddns.example.net" not in response.text assert client.get(f"/api/v1/hostnames/{other['management_slug']}", params={"history_limit": 101}).status_code == 422 assert client.get(f"/api/v1/hostnames/{'a' * 32}").status_code == 404 def test_registration_publishes_initial_dns_when_provider_is_configured(tmp_path: Path, monkeypatch) -> None: published = [] monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result)) app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", hostname_suffix="ddns.example.net", public_base_url="http://ddns.example.net", dns_provider="cloudflare", cloudflare_api_token="token", cloudflare_zone_id="zone", ) ) client = TestClient(app) body = client.post("/api/v1/hostnames").json() registered = client.post(_path(body["registration_url"]), params={"myip": "203.0.113.77"}) status = client.get(f"/api/v1/hostnames/{body['dashboard_url'].rsplit('/', 1)[1]}") assert registered.status_code == 200 assert status.json()["status"] == "active" assert status.json()["ipv4"] == "203.0.113.77" assert len(published) == 1 assert published[0].hostname == body["hostname"] assert published[0].ipv4 == "203.0.113.77" def test_registration_requires_ip_when_provider_source_ip_is_not_public(tmp_path: Path, monkeypatch) -> None: monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: None) app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", hostname_suffix="ddns.example.net", public_base_url="http://ddns.example.net", dns_provider="cloudflare", cloudflare_api_token="token", cloudflare_zone_id="zone", ) ) client = TestClient(app) body = client.post("/api/v1/hostnames").json() registered = client.post(_path(body["registration_url"])) status = client.get(f"/api/v1/hostnames/{body['dashboard_url'].rsplit('/', 1)[1]}") assert registered.status_code == 400 assert registered.text.startswith("badip registration needs a public source IP") assert status.json()["status"] == "pending" def test_api_update_publishes_before_recording_ip(tmp_path: Path, monkeypatch) -> None: published = [] monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result)) database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "router") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.11"}) assert response.status_code == 200 assert response.text == "good 203.0.113.11" assert len(published) == 1 assert store.list_records()[0]["ipv4"] == "203.0.113.11" def test_api_dns_publish_failure_keeps_record_unchanged(tmp_path: Path, monkeypatch) -> None: def fail_publish(result, settings): raise DnsProviderError("dns down") monkeypatch.setattr(ddns_module, "publish_dns", fail_publish) database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "router") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.11"}) assert response.status_code == 500 assert store.list_records() == [] assert store.list_update_events()[0]["status"] == "dns_failed" def test_api_creation_is_rate_limited(tmp_path: Path) -> None: app = make_app( DdnsSettings( database_path=tmp_path / "ddns.sqlite3", admin_password="admin", hostname_suffix="ddns.example.net", rate_limit_per_minute=1, ) ) client = TestClient(app) first = client.post("/api/v1/hostnames") second = client.post("/api/v1/hostnames") assert first.status_code == 201 assert second.status_code == 429 def test_api_invalid_ip_returns_clear_400(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.example.net", "alice") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myipv6": "203.0.113.9"}) assert response.status_code == 400 assert response.text == "badip" assert store.list_records() == []