karldns.de/tests/test_security.py

import hashlib
import hmac
import logging
from pathlib import Path

import pytest
from fastapi import Request
from fastapi.testclient import TestClient

import karldns.services.ddns as ddns_module
from karldns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, UpdateResult, make_app
from karldns.dns import validate_dns_provider_settings
from karldns.security import DASHBOARD_UNLOCK_COOKIE, client_ip


def test_generated_hostname_is_not_an_update_secret(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    store.create_account("public-target.ddns.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/karl_{'a' * 43}",
        params={"hostname": "public-target.ddns.example.net", "myip": "203.0.113.9"},
    )

    assert response.status_code == 404
    assert store.list_records() == []


def test_generated_account_cannot_update_other_hostname(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={"hostname": "other.example.net", "myip": "203.0.113.9"},
    )

    assert response.status_code == 200
    assert store.list_records()[0]["hostname"] == "home.example.net"
    assert store.get_account_by_hostname("other.example.net") is None


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

    first = client.get(f"/m/karl_{'a' * 43}")
    second = client.get(f"/m/karl_{'b' * 43}")

    assert first.status_code == 404
    assert second.status_code == 429
    assert app.state.store.get_service_metrics()["rate_limit_hits_total"] == 1


def test_redacted_access_log_masks_bearer_paths_and_query_strings(tmp_path: Path, caplog) -> None:
    caplog.set_level(logging.INFO, logger="karldns.access")
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="admin",
            redacted_access_log=True,
        )
    )
    account = app.state.store.create_account("home.example.net", "router")
    client = TestClient(app)

    client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.1"})
    client.get(f"/api/v1/hostnames/{account['management_slug']}")

    assert "/api/v1/router/update/<secret>" in caplog.text
    assert "/api/v1/hostnames/<secret>" in caplog.text
    assert account["update_slug"] not in caplog.text
    assert account["management_slug"] not in caplog.text


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/router/update/short",
        "/api/v1/router/register/short",
        "/api/v1/hostnames/short",
        "/m/short",
        "/api/v1/router/update/not_karl_but_long_enough_1234567890",
        "/api/v1/router/register/not_karl_but_long_enough_1234567890",
        "/api/v1/hostnames/not_karl_but_long_enough_1234567890",
        "/m/not_karl_but_long_enough_1234567890",
    ],
)
def test_secret_path_tokens_are_constrained(tmp_path: Path, path: str) -> None:
    app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
    client = TestClient(app)

    response = client.get(path)

    assert response.status_code == 422


def test_dashboard_unlock_cookie_rejects_password_hash_signed_forgery(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "router")
    store.set_dashboard_password("home.example.net", "correct horse")
    settings = DdnsSettings(
        database_path=database_path,
        admin_password="admin",
        dashboard_cookie_secret="s" * 32,
    )
    app = make_app(settings)
    client = TestClient(app)
    stored = store.get_account_by_management_slug(account["management_slug"])
    password_hash = str(stored["dashboard_password_hash"])
    timestamp = 1_900_000_000
    max_age = settings.dashboard_unlock_seconds
    payload = f"{account['hostname']}\0{account['management_slug']}\0{timestamp}\0{max_age}".encode()
    forged = f"v2:{timestamp}:{max_age}:{hmac.new(password_hash.encode(), payload, hashlib.sha256).hexdigest()}"
    client.cookies.set(DASHBOARD_UNLOCK_COOKIE, forged, path=f"/m/{account['management_slug']}")

    response = client.get(f"/m/{account['management_slug']}")

    assert response.status_code == 200
    assert "Dashboard locked" in response.text
    assert "Update-URL:" not in response.text


def test_settings_reject_unknown_constructor_fields() -> None:
    with pytest.raises(ValueError, match="Extra inputs are not permitted"):
        DdnsSettings(**{"admn_password": "typo"})


def test_trusted_host_middleware_rejects_unlisted_host(tmp_path: Path) -> None:
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="admin",
            public_base_url="http://ddns.example.net",
            trusted_hosts={"ddns.example.net"},
        )
    )
    client = TestClient(app)

    response = client.get("/", headers={"host": "evil.example.net"})

    assert response.status_code == 400


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

    created = client.post("/")

    assert created.status_code == 200
    assert created.headers["cache-control"] == "no-store"
    assert created.headers["pragma"] == "no-cache"


def test_oversized_request_body_is_rejected(tmp_path: Path) -> None:
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="admin",
            hostname_suffix="ddns.example.net",
            max_request_body_bytes=1024,
        )
    )
    client = TestClient(app)

    response = client.post("/", data={"username": "x" * 2000})

    assert response.status_code == 413


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

    response = client.post(
        "/",
        content=b"username=\xff",
        headers={"content-type": "application/x-www-form-urlencoded"},
    )

    assert response.status_code == 400
    assert response.json()["detail"] == "form body must be valid UTF-8"


def test_too_many_form_fields_are_rejected(tmp_path: Path) -> None:
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="admin",
            hostname_suffix="ddns.example.net",
        )
    )
    client = TestClient(app)
    body = "&".join(f"k{index}=v" for index in range(101))

    response = client.post(
        "/",
        content=body.encode("ascii"),
        headers={"content-type": "application/x-www-form-urlencoded"},
    )

    assert response.status_code == 400
    assert response.json()["detail"] == "form must not contain more than 100 fields"


def test_untrusted_forwarded_for_does_not_bypass_rate_limit(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("/", headers={"x-forwarded-for": "198.51.100.10"})
    second = client.post("/", headers={"x-forwarded-for": "198.51.100.11"})

    assert first.status_code == 200
    assert second.status_code == 429


def test_trusted_proxy_uses_cloudflare_connecting_ip(tmp_path: Path) -> None:
    settings = DdnsSettings(
        database_path=tmp_path / "ddns.sqlite3",
        admin_password="admin",
        trusted_proxy_ips={"10.0.0.0/8"},
    )
    app = make_app(settings)

    @app.get("/debug-ip", include_in_schema=False)
    def debug_ip(request: Request) -> dict[str, str]:
        return {"ip": client_ip(request, settings)}

    client = TestClient(app, client=("10.1.2.3", 12345))
    response = client.get(
        "/debug-ip",
        headers={
            "cf-connecting-ip": "198.51.100.20",
            "x-forwarded-for": "203.0.113.5",
        },
    )

    assert response.status_code == 200
    assert response.json()["ip"] == "198.51.100.20"


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

    @app.get("/debug-ip", include_in_schema=False)
    def debug_ip(request: Request) -> dict[str, str]:
        return {"ip": client_ip(request, settings)}

    client = TestClient(app, client=("172.18.0.5", 12345))
    response = client.get(
        "/debug-ip",
        headers={
            "cf-connecting-ip": "8.8.8.8",
            "x-forwarded-for": "1.1.1.1",
        },
    )

    assert response.status_code == 200
    assert response.json()["ip"] == "8.8.8.8"


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

    @app.get("/debug-ip", include_in_schema=False)
    def debug_ip(request: Request) -> dict[str, str]:
        return {"ip": client_ip(request, settings)}

    client = TestClient(app, client=("172.18.0.5", 12345))
    response = client.get("/debug-ip", headers={"x-forwarded-for": "8.8.4.4, 172.18.0.5"})

    assert response.status_code == 200
    assert response.json()["ip"] == "8.8.4.4"


def test_private_reverse_proxy_trust_can_be_disabled(tmp_path: Path) -> None:
    settings = DdnsSettings(
        database_path=tmp_path / "ddns.sqlite3",
        admin_password="admin",
        trust_private_proxy_headers=False,
    )
    app = make_app(settings)

    @app.get("/debug-ip", include_in_schema=False)
    def debug_ip(request: Request) -> dict[str, str]:
        return {"ip": client_ip(request, settings)}

    client = TestClient(app, client=("172.18.0.5", 12345))
    response = client.get("/debug-ip", headers={"cf-connecting-ip": "8.8.8.8"})

    assert response.status_code == 200
    assert response.json()["ip"] == "172.18.0.5"


def test_untrusted_proxy_headers_are_ignored(tmp_path: Path) -> None:
    settings = DdnsSettings(
        database_path=tmp_path / "ddns.sqlite3",
        admin_password="admin",
        trusted_proxy_ips={"10.0.0.0/8"},
    )
    app = make_app(settings)

    @app.get("/debug-ip", include_in_schema=False)
    def debug_ip(request: Request) -> dict[str, str]:
        return {"ip": client_ip(request, settings)}

    client = TestClient(app, client=("198.51.100.1", 12345))
    response = client.get("/debug-ip", headers={"cf-connecting-ip": "198.51.100.20"})

    assert response.status_code == 200
    assert response.json()["ip"] == "198.51.100.1"


def test_dns_delete_failure_is_logged_without_secret_management_slug(tmp_path: Path, monkeypatch, caplog) -> None:
    def fail_delete(hostname, settings):
        raise DnsProviderError("dns down")

    caplog.set_level(logging.WARNING, logger="karldns.services.ddns")
    monkeypatch.setattr(ddns_module, "delete_dns_records", fail_delete)
    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"))

    with pytest.raises(DnsProviderError):
        app.state.service.delete_account("home.example.net", "unit-test")

    assert store.get_account_by_slug(account["update_slug"]) is not None
    assert store.list_update_events()[0]["status"] == "dns_delete_failed"
    assert "DNS delete failed for hostname=home.example.net source=unit-test" in caplog.text
    assert account["management_slug"] not in caplog.text
    assert "dns down" not in caplog.text


def test_require_dns_provider_fails_startup_without_backend(tmp_path: Path) -> None:
    try:
        make_app(
            DdnsSettings(
                database_path=tmp_path / "ddns.sqlite3",
                admin_password="admin",
                require_dns_provider=True,
            )
        )
    except RuntimeError as exc:
        assert "no DNS backend" in str(exc)
    else:
        raise AssertionError("expected startup failure")


def test_invalid_dns_provider_is_rejected() -> None:
    settings = DdnsSettings(dns_provider="route53")

    with pytest.raises(RuntimeError, match="Unsupported DDNS_DNS_PROVIDER"):
        validate_dns_provider_settings(settings)


def test_required_cloudflare_provider_requires_token_and_zone(tmp_path: Path) -> None:
    with pytest.raises(RuntimeError, match="DDNS_CLOUDFLARE_API_TOKEN"):
        make_app(
            DdnsSettings(
                database_path=tmp_path / "ddns.sqlite3",
                admin_password="a" * 16,
                public_base_url="http://ddns.example.net",
                hostname_suffix="ddns.example.net",
                dns_provider="cloudflare",
                require_dns_provider=True,
            )
        )


def test_required_dns_provider_requires_strong_dashboard_cookie_signing_source(tmp_path: Path) -> None:
    with pytest.raises(RuntimeError, match="DDNS_DASHBOARD_COOKIE_SECRET"):
        make_app(
            DdnsSettings(
                database_path=tmp_path / "ddns.sqlite3",
                admin_password="a" * 16,
                public_base_url="http://ddns.example.net",
                hostname_suffix="ddns.example.net",
                dns_provider="rfc2136",
                rfc2136_server="192.0.2.53",
                rfc2136_zone="example.net",
                require_dns_provider=True,
            )
        )


def test_required_dns_provider_accepts_explicit_dashboard_cookie_secret(tmp_path: Path) -> None:
    app = make_app(
        DdnsSettings(
            database_path=tmp_path / "ddns.sqlite3",
            admin_password="a" * 16,
            dashboard_cookie_secret="s" * 32,
            public_base_url="http://ddns.example.net",
            hostname_suffix="ddns.example.net",
            dns_provider="rfc2136",
            rfc2136_server="192.0.2.53",
            rfc2136_zone="example.net",
            require_dns_provider=True,
        )
    )

    assert app.state.settings.dashboard_cookie_secret == "s" * 32


def test_required_rfc2136_provider_requires_server_and_zone(tmp_path: Path) -> None:
    with pytest.raises(RuntimeError, match="DDNS_RFC2136_SERVER and DDNS_RFC2136_ZONE"):
        make_app(
            DdnsSettings(
                database_path=tmp_path / "ddns.sqlite3",
                admin_password="a" * 16,
                public_base_url="http://ddns.example.net",
                hostname_suffix="ddns.example.net",
                dns_provider="rfc2136",
                rfc2136_server="192.0.2.53",
                require_dns_provider=True,
            )
        )


def test_cloudflare_publish_reconciles_provider_state(monkeypatch) -> None:
    import karldns.dns.cloudflare as cloudflare

    calls = {"find": 0}

    def fake_find(settings, hostname, record_type):
        calls["find"] += 1
        if calls["find"] == 1:
            return None
        return {"id": "record-id", "content": "203.0.113.99"}

    monkeypatch.setattr(cloudflare, "_cloudflare_find_record", fake_find)
    monkeypatch.setattr(cloudflare, "_cloudflare_request", lambda *args, **kwargs: {"success": True, "result": []})

    with pytest.raises(DnsReconciliationError, match="did not match"):
        cloudflare.publish_cloudflare(
            UpdateResult(
                hostname="home.example.net",
                ipv4="203.0.113.10",
                ipv6=None,
                changed=True,
                update_ipv4=True,
                update_ipv6=False,
            ),
            DdnsSettings(cloudflare_api_token="token", cloudflare_zone_id="zone"),
        )


def test_rfc2136_delete_checks_response_code(monkeypatch) -> None:
    import dns.query
    import dns.rcode

    from karldns.dns.rfc2136 import delete_rfc2136_records

    class RefusedResponse:
        def rcode(self) -> int:
            return dns.rcode.REFUSED

    monkeypatch.setattr(dns.query, "tcp", lambda *args, **kwargs: RefusedResponse())

    with pytest.raises(DnsProviderError, match="RFC2136 delete failed"):
        delete_rfc2136_records(
            "home.example.net",
            DdnsSettings(rfc2136_server="192.0.2.53", rfc2136_zone="example.net"),
        )


def test_invalid_integer_env_value_fails_startup(monkeypatch) -> None:
    monkeypatch.setenv("DDNS_RATE_LIMIT_PER_MINUTE", "not-an-int")

    with pytest.raises(ValueError, match="DDNS_RATE_LIMIT_PER_MINUTE must be an integer"):
        DdnsSettings.from_env()


def test_invalid_admin_path_fails_startup() -> None:
    with pytest.raises(ValueError, match="admin_path"):
        DdnsSettings(admin_path="admin")

    with pytest.raises(ValueError, match="admin_path"):
        DdnsSettings(admin_path="/")

    with pytest.raises(ValueError, match="admin_path"):
        DdnsSettings(admin_path="/admin?")

    with pytest.raises(ValueError, match="public route"):
        DdnsSettings(admin_path="/api")


def test_launch_mode_requires_strong_admin_password(tmp_path: Path) -> None:
    try:
        make_app(
            DdnsSettings(
                database_path=tmp_path / "ddns.sqlite3",
                admin_password="admin",
                public_base_url="http://ddns.example.net",
                hostname_suffix="ddns.example.net",
                dns_provider="cloudflare",
                cloudflare_api_token="token",
                cloudflare_zone_id="zone",
                require_dns_provider=True,
            )
        )
    except RuntimeError as exc:
        assert "DDNS_ADMIN_PASSWORD" in str(exc)
    else:
        raise AssertionError("expected startup failure")