karldns.de/tests/test_api.py

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/<karl_secret>" in docs.text
    assert "GET or POST /api/v1/router/update/<karl_secret>" in docs.text
    assert "GET /api/v1/hostnames/<karl_secret>" in docs.text
    assert "POST /api/v1/hostnames/<karl_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"])
    assert dashboard_secret.startswith("karl_")
    assert update_path.rsplit("/", 1)[1].startswith("karl_")
    assert register_path.rsplit("/", 1)[1].startswith("karl_")

    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 new_update_path.rsplit("/", 1)[1].startswith("karl_")
    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/karl_{'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_without_public_source_ip_waits_for_first_router_update(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()
    update_path = _path(body["update_url"])
    registered = client.post(_path(body["registration_url"]))
    status = client.get(f"/api/v1/hostnames/{body['dashboard_url'].rsplit('/', 1)[1]}")
    update = client.get(update_path, params={"myip": "203.0.113.80"})

    assert registered.status_code == 200
    assert registered.text == "good registered"
    assert status.json()["status"] == "active"
    assert status.json()["ipv4"] is None
    assert update.status_code == 200
    assert update.text == "good 203.0.113.80"
    assert len(published) == 1
    assert published[0].ipv4 == "203.0.113.80"


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() == []