karldns.de/tests/test_admin.py

from pathlib import Path

from fastapi.testclient import TestClient

from karldns import DdnsSettings, DdnsStore, make_app
from karldns.security import admin_csrf_token


def test_admin_page_shows_operator_views_without_create_form(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get("/admin", auth=("admin", "admin"))

    assert response.status_code == 200
    assert response.headers["cache-control"] == "no-store"
    assert '<link href="/admin.css?v=4" rel="stylesheet">' in response.text
    assert '<script src="/admin-tables.js?v=2" defer></script>' in response.text
    assert "Operator console" in response.text
    assert "Generate credentials" not in response.text
    assert 'action="/admin/accounts"' not in response.text
    assert response.text.count("data-admin-table") == 2
    assert response.text.count("data-table-search") == 2
    assert response.text.count("data-table-page-size") == 2
    assert response.text.count("data-table-pagination") == 2
    assert "admin-table-scroll" in response.text
    assert "Cleanup" in response.text
    assert "Run now" in response.text
    assert "Next run" in response.text
    assert "Service metrics" in response.text
    assert "Blocked requests" in response.text
    assert "Hostnames" in response.text
    assert "Update events" in response.text


def test_custom_admin_path_moves_operator_routes(tmp_path: Path) -> None:
    settings = DdnsSettings(
        database_path=tmp_path / "ddns.sqlite3",
        admin_password="admin",
        admin_path="/ops",
    )
    store = DdnsStore(settings.database_path)
    store.create_account("home.example.net", "router")
    app = make_app(settings)
    client = TestClient(app)

    response = client.get("/ops", auth=("admin", "admin"))

    assert response.status_code == 200
    assert 'href="/ops/records"' not in response.text
    assert 'href="/ops/events"' not in response.text
    assert 'action="/ops/cleanup"' in response.text
    assert 'action="/ops/hostnames/rotate"' in response.text
    assert 'action="/admin/cleanup"' not in response.text
    assert client.get("/ops/records", auth=("admin", "admin")).status_code == 404
    assert client.get("/records", auth=("admin", "admin")).status_code == 404
    assert client.get("/admin", auth=("admin", "admin")).status_code == 404


def test_admin_records_and_events_are_paginated(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    first = store.create_account("first.example.net", "router")
    second = store.create_account("second.example.net", "router")
    store.upsert(first["hostname"], "203.0.113.1", None)
    store.upsert(second["hostname"], "203.0.113.2", None)
    store.log_update_event(first["hostname"], "203.0.113.1", None, "updated", "ok", "198.51.100.1")
    store.log_update_event(second["hostname"], "203.0.113.2", None, "updated", "ok", "198.51.100.2")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    records = client.get("/api/v1/admin/records?limit=1&offset=1", auth=("admin", "admin")).json()
    events = client.get("/api/v1/admin/events?limit=1&offset=1", auth=("admin", "admin")).json()

    assert records["limit"] == 1
    assert records["offset"] == 1
    assert records["total"] == 2
    assert len(records["items"]) == 1
    assert events["limit"] == 1
    assert events["offset"] == 1
    assert "next_before" in events
    assert events["total"] == 2
    assert len(events["items"]) == 1


def test_admin_api_exposes_dashboard_data_and_history(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "router")
    store.upsert(account["hostname"], "203.0.113.10", None)
    store.log_update_event(account["hostname"], "203.0.113.10", None, "updated", "ok", "198.51.100.1")
    store.increment_metric("update_requests_total")
    settings = DdnsSettings(database_path=database_path, admin_password="admin")
    app = make_app(settings)
    client = TestClient(app)

    assert client.get("/api/v1/admin/dashboard").status_code == 401

    dashboard = client.get("/api/v1/admin/dashboard?limit=10", auth=("admin", "admin"))
    hostnames = client.get("/api/v1/admin/hostnames?limit=1&offset=0", auth=("admin", "admin"))
    records = client.get("/api/v1/admin/records?limit=1&offset=0", auth=("admin", "admin"))
    events = client.get("/api/v1/admin/events?limit=1&offset=0", auth=("admin", "admin"))

    assert dashboard.status_code == 200
    assert dashboard.headers["cache-control"] == "no-store"
    dashboard_json = dashboard.json()
    assert dashboard_json["totals"]["hostnames"] == 1
    assert dashboard_json["totals"]["records"] == 1
    assert dashboard_json["totals"]["update_events"] == 1
    assert dashboard_json["metrics"]["values"]["update_requests_total"] == 1
    assert dashboard_json["hostnames"][0]["hostname"] == "home.example.net"
    assert dashboard_json["hostnames"][0]["status"] == "active"
    assert dashboard_json["hostnames"][0]["ipv4"] == "203.0.113.10"
    assert dashboard_json["update_events"][0]["status"] == "updated"
    assert dashboard_json["scheduled_tasks"][0]["name"] == "cleanup"

    assert hostnames.json()["items"][0]["hostname"] == "home.example.net"
    assert records.json()["items"][0]["ipv4"] == "203.0.113.10"
    assert events.json()["items"][0]["hostname"] == "home.example.net"


def test_admin_records_and_events_reject_excessive_offsets(tmp_path: Path) -> None:
    app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
    client = TestClient(app)

    records = client.get("/api/v1/admin/records?offset=10001", auth=("admin", "admin"))
    events = client.get("/api/v1/admin/events?offset=10001", auth=("admin", "admin"))

    assert records.status_code == 422
    assert events.status_code == 422


def test_update_events_page_orders_raw_and_rollup_events_in_sql(tmp_path: Path) -> None:
    store = DdnsStore(tmp_path / "ddns.sqlite3")
    store.log_update_event("raw-old.example.net", "203.0.113.1", None, "updated", "ok", "198.51.100.1")
    store.log_update_event("raw-new.example.net", "203.0.113.2", None, "updated", "ok", "198.51.100.2")
    store.log_update_rollup("rollup-mid.example.net", "203.0.113.3", None, "unchanged", "ok", "198.51.100.3")

    with store.connect() as conn:
        conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-01T00:00:00+00:00", "raw-old.example.net"))
        conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-03T00:00:00+00:00", "raw-new.example.net"))
        conn.execute(
            "UPDATE update_event_rollups SET last_seen_at = ? WHERE hostname = ?",
            ("2026-01-02T00:00:00+00:00", "rollup-mid.example.net"),
        )

    page = store.list_update_events_page(limit=2, offset=1)

    assert [event["hostname"] for event in page] == ["rollup-mid.example.net", "raw-old.example.net"]


def test_update_events_support_keyset_pagination(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    store.log_update_event("raw-old.example.net", "203.0.113.1", None, "updated", "ok", "198.51.100.1")
    store.log_update_event("raw-new.example.net", "203.0.113.2", None, "updated", "ok", "198.51.100.2")
    with store.connect() as conn:
        conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-01T00:00:00+00:00", "raw-old.example.net"))
        conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-02T00:00:00+00:00", "raw-new.example.net"))
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    first = client.get("/api/v1/admin/events?limit=1", auth=("admin", "admin")).json()
    second = client.get(f"/api/v1/admin/events?limit=1&before={first['next_before']}", auth=("admin", "admin")).json()

    assert first["items"][0]["hostname"] == "raw-new.example.net"
    assert second["items"][0]["hostname"] == "raw-old.example.net"


def test_admin_posts_require_csrf(tmp_path: Path) -> None:
    app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
    client = TestClient(app)

    response = client.post(
        "/admin/cleanup",
        auth=("admin", "admin"),
        data={},
    )

    assert response.status_code == 403


def test_admin_cleanup_uses_post_redirect_get(tmp_path: Path) -> None:
    settings = DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")
    app = make_app(settings)
    client = TestClient(app)

    response = client.post(
        "/admin/cleanup",
        auth=("admin", "admin"),
        data={"csrf": admin_csrf_token(settings)},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert response.headers["location"].startswith("/admin?cleanup_done=1")

    redirected = client.get(response.headers["location"], auth=("admin", "admin"))
    assert redirected.status_code == 200
    assert "Cleanup finished." in redirected.text
    assert "Copy these values now" not in redirected.text


def test_admin_password_query_string_is_rejected(tmp_path: Path) -> None:
    app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
    client = TestClient(app)

    response = client.get("/admin", params={"password": "admin"})

    assert response.status_code == 401


def test_admin_routes_are_rate_limited(tmp_path: Path) -> None:
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="admin",
            admin_rate_limit_per_minute=1,
        )
    )
    client = TestClient(app)

    first = client.get("/admin", auth=("admin", "wrong"))
    second = client.get("/admin", auth=("admin", "wrong"))

    assert first.status_code == 401
    assert second.status_code == 429


def test_admin_can_rotate_links_and_password(tmp_path: Path) -> None:
    settings = DdnsSettings(
        database_path=tmp_path / "ddns.sqlite3",
        admin_password="admin",
        public_base_url="http://ddns.example.net",
    )
    store = DdnsStore(settings.database_path)
    account = store.create_account("home.example.net", "router")
    app = make_app(settings)
    client = TestClient(app)
    csrf = admin_csrf_token(settings)

    update = client.post(
        "/admin/hostnames/rotate",
        auth=("admin", "admin"),
        data={"csrf": csrf, "hostname": "home.example.net", "action": "update"},
    )
    assert update.status_code == 200
    assert "Update URL changed" in update.text
    assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404

    management = client.post(
        "/admin/hostnames/rotate",
        auth=("admin", "admin"),
        data={"csrf": csrf, "hostname": "home.example.net", "action": "management"},
    )
    assert management.status_code == 200
    assert "Dashboard link changed" in management.text
    assert "/m/" in management.text

    statuses = {event["status"] for event in store.list_update_events_for_hostname("home.example.net", 10)}
    assert "admin_update_url_rotated" in statuses
    assert "admin_management_url_rotated" in statuses
    assert "admin_router_credentials_rotated" not in statuses


def test_admin_delete_normalizes_hostname_and_logs_audit_event(tmp_path: Path) -> None:
    settings = DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")
    store = DdnsStore(settings.database_path)
    store.create_account("home.example.net", "router")
    app = make_app(settings)
    client = TestClient(app)

    response = client.post(
        "/admin/hostnames/delete",
        auth=("admin", "admin"),
        data={"csrf": admin_csrf_token(settings), "hostname": " HOME.EXAMPLE.NET. "},
        follow_redirects=False,
    )

    assert response.status_code == 303
    assert store.get_account_by_hostname("home.example.net") is None
    events = store.list_update_events_for_hostname("home.example.net", 10)
    assert events[0]["status"] == "admin_hostname_deleted"
    assert "secret" not in str(events[0]["detail"]).lower()


def test_admin_template_escapes_update_event_html(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    store.create_account("home.example.net", "router")
    store.log_update_event(
        "home.example.net",
        "203.0.113.10",
        None,
        "dns_failed",
        '<script>alert("xss")</script>',
        "198.51.100.20",
    )
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get("/admin", auth=("admin", "admin"))

    assert response.status_code == 200
    assert '<script>alert("xss")</script>' not in response.text
    assert "&lt;script&gt;alert(&quot;xss&quot;)&lt;/script&gt;" in response.text