karl/karldns.de
Archived
1
0 Fork 0
Code Issues Packages Projects Releases Wiki Activity

Prepare KarlDNS production deployment

Browse Source
This commit is contained in:
StasonJatham committed 2026-05-21 17:23:08 +02:00
1 parent 5a90bd9d87
commit ce5b4e0ff2
101 files changed
+3364 -2018
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_admin.py
+53 -27
View File
@@ -2,8 +2,8 @@ from pathlib import Path
from fastapi.testclient import TestClient
from router_dyndns.ddns import DdnsSettings, DdnsStore, make_app
from router_dyndns.ddns_security import admin_csrf_token
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:
@@ -48,12 +48,12 @@ def test_custom_admin_path_moves_operator_routes(tmp_path: Path) -> None:
response = client.get("/ops", auth=("admin", "admin"))
assert response.status_code == 200
assert 'href="/ops/records"' in response.text
assert 'href="/ops/events"' in response.text
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/accounts/rotate"' 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 == 200
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
@@ -70,8 +70,8 @@ def test_admin_records_and_events_are_paginated(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
records = client.get("/records?limit=1&offset=1", auth=("admin", "admin")).json()
events = client.get("/events?limit=1&offset=1", auth=("admin", "admin")).json()
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
@@ -84,12 +84,48 @@ def test_admin_records_and_events_are_paginated(tmp_path: Path) -> None:
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("/records?offset=10001", auth=("admin", "admin"))
events = client.get("/events?offset=10001", auth=("admin", "admin"))
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
@@ -125,8 +161,8 @@ def test_update_events_support_keyset_pagination(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
first = client.get("/events?limit=1", auth=("admin", "admin")).json()
second = client.get(f"/events?limit=1&before={first['next_before']}", auth=("admin", "admin")).json()
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"
@@ -184,16 +220,16 @@ def test_admin_can_rotate_links_and_password(tmp_path: Path) -> None:
csrf = admin_csrf_token(settings)
update = client.post(
"/admin/accounts/rotate",
"/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"/u/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404
assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404
management = client.post(
"/admin/accounts/rotate",
"/admin/hostnames/rotate",
auth=("admin", "admin"),
data={"csrf": csrf, "hostname": "home.example.net", "action": "management"},
)
@@ -201,20 +237,10 @@ def test_admin_can_rotate_links_and_password(tmp_path: Path) -> None:
assert "Dashboard link changed" in management.text
assert "/m/" in management.text
password = client.post(
"/admin/accounts/rotate",
auth=("admin", "admin"),
data={"csrf": csrf, "hostname": "home.example.net", "action": "password"},
)
assert password.status_code == 200
assert "Router password changed" in password.text
assert "Full update URL:" in password.text
assert "/nic/update?" in password.text
assert "Kennwort (optional):" in password.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" in statuses
assert "admin_router_credentials_rotated" not in statuses
def test_admin_delete_normalizes_hostname_and_logs_audit_event(tmp_path: Path) -> None:
@@ -225,7 +251,7 @@ def test_admin_delete_normalizes_hostname_and_logs_audit_event(tmp_path: Path) -
client = TestClient(app)
response = client.post(
"/admin/accounts/delete",
"/admin/hostnames/delete",
auth=("admin", "admin"),
data={"csrf": admin_csrf_token(settings), "hostname": " HOME.EXAMPLE.NET. "},
follow_redirects=False,
tests/test_api.py
+225 -124
View File
@@ -1,81 +1,95 @@
from pathlib import Path
from urllib.parse import parse_qs, urlparse
from urllib.parse import urlparse
import pytest
from fastapi.testclient import TestClient
import router_dyndns.ddns_service as ddns_service
from router_dyndns.ddns import DdnsSettings, DdnsStore, DnsProviderError, make_app
import karldns.services.ddns as ddns_module
from karldns import DdnsSettings, DdnsStore, DnsProviderError, make_app
def test_generated_credentials_include_one_box_update_url(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
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=database_path,
database_path=tmp_path / "ddns.sqlite3",
admin_password="admin",
public_base_url="http://ddns.example.net",
hostname_suffix="ddns.example.net",
public_base_url="https://karldns.de",
)
)
client = TestClient(app)
response = client.post("/api/v1/hostnames/magic", json={"username": "router"})
assert response.status_code == 201
payload = response.json()
one_box_url = payload["credentialed_update_url"]
parsed = urlparse(one_box_url)
params = parse_qs(parsed.query)
assert parsed.path == "/nic/update"
assert params["hostname"] == [payload["hostname"]]
assert params["username"] == [payload["username"]]
assert params["password"] == [payload["password"]]
assert params["myip"] == ["<ipaddr>"]
assert params["myipv6"] == ["<ip6addr>"]
def test_api_docs_include_versioned_ddns_api(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
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 == 200
assert docs.headers["x-robots-tag"] == "noindex, nofollow"
assert redoc.headers["x-robots-tag"] == "noindex, nofollow"
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/&lt;secret&gt;" in docs.text
assert "GET or POST /api/v1/router/update/&lt;secret&gt;" in docs.text
assert "GET /api/v1/hostnames/&lt;secret&gt;" in docs.text
assert "POST /api/v1/hostnames/&lt;secret&gt;/rotate-update-url" in docs.text
assert "Rotate update URL" in docs.text
assert "GET /api/v1/admin/dashboard" in docs.text
assert "GET /api/v1/admin/events" 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/&lt;secret&gt;" 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 "unsafe-inline" in docs.headers["content-security-policy"]
assert "fonts.googleapis.com" in redoc.headers["content-security-policy"]
assert "fonts.gstatic.com" in redoc.headers["content-security-policy"]
assert "/api/v1/hostnames/magic" in schema["paths"]
assert "/api/v1/updates/{update_slug}" in schema["paths"]
assert "/u/{update_slug}" in schema["paths"]
assert "/nic/update" in schema["paths"]
assert "/ip" in schema["paths"]
assert "/healthz" in schema["paths"]
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/admin/dashboard",
"/api/v1/admin/events",
"/api/v1/admin/hostnames",
"/api/v1/admin/records",
"/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 "/" not in schema["paths"]
assert "/magic" not in schema["paths"]
assert "/accounts" not in schema["paths"]
assert "/m/{management_slug}" not in schema["paths"]
assert any(tag["name"] == "hostnames" for tag in schema["tags"])
assert all(tag["name"] != "admin" for tag in schema["tags"])
management_tag = next(tag for tag in schema["tags"] if tag["name"] == "management")
assert "Dashboard passwords protect only the web dashboard" in management_tag["description"]
management_get = schema["paths"]["/api/v1/management/{management_slug}"]["get"]
management_delete = schema["paths"]["/api/v1/management/{management_slug}"]["delete"]
assert "Dashboard passwords protect only the web dashboard" in management_get["description"]
assert "Dashboard passwords protect only the web dashboard" in management_delete["description"]
assert any(tag["name"] == "router" for tag in schema["tags"])
assert any(tag["name"] == "admin" for tag in schema["tags"])
def test_public_redoc_hides_custom_operator_routes(tmp_path: Path) -> None:
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",
@@ -85,78 +99,38 @@ def test_public_redoc_hides_custom_operator_routes(tmp_path: Path) -> None:
)
client = TestClient(app)
redoc = client.get("/redoc")
schema = client.get("/openapi.json").json()
forbidden_paths = {
"/ops",
"/ops/accounts/delete",
"/ops/accounts/rotate",
"/ops/hostnames/delete",
"/ops/hostnames/rotate",
"/ops/cleanup",
"/ops/events",
"/ops/records",
"/records",
"/events",
}
assert redoc.status_code == 200
assert client.get("/redoc").status_code == 404
assert forbidden_paths.isdisjoint(schema["paths"])
assert all("admin" not in tag["name"].lower() for tag in schema["tags"])
assert "/api/v1/admin/dashboard" in schema["paths"]
def test_public_redoc_only_exposes_api_and_router_endpoints(tmp_path: Path) -> None:
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"]
expected_public_paths = {
"/api/v1/hostnames/magic",
"/api/v1/management/{management_slug}",
"/api/v1/updates/{update_slug}",
"/healthz",
"/ip",
"/nic/update",
"/u/{update_slug}",
}
assert set(schema["paths"]) == expected_public_paths
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_openapi_marks_generated_update_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()
credentials = schema["components"]["schemas"]["HostnameCredentials"]["properties"]
managed = schema["components"]["schemas"]["ManagedRecord"]["properties"]
for name in ["password", "update_url", "credentialed_update_url", "management_url"]:
assert "Secret" in credentials[name]["description"]
assert "Secret" in managed["update_url"]["description"]
@pytest.mark.parametrize(
("path", "payload"),
[
("/api/v1/hostnames/magic", {"username": "router", "unexpected": True}),
],
)
def test_public_write_models_reject_extra_json_fields(tmp_path: Path, path: str, payload: dict[str, object]) -> 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(path, json=payload)
assert response.status_code == 422
assert response.json()["detail"][0]["type"] == "extra_forbidden"
def test_api_magic_hostname_lifecycle(tmp_path: Path) -> None:
def test_hostname_api_lifecycle_and_history(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
@@ -167,38 +141,165 @@ def test_api_magic_hostname_lifecycle(tmp_path: Path) -> None:
)
client = TestClient(app)
created = client.post("/api/v1/hostnames/magic", json={"username": "router"})
created = client.post("/api/v1/hostnames")
assert created.status_code == 201
assert created.headers["cache-control"] == "no-store"
assert created.headers["x-robots-tag"] == "noindex, nofollow"
body = created.json()
assert body["hostname"].endswith(".ddns.example.net")
assert body["update_url"].startswith("http://ddns.example.net/u/")
assert body["management_url"].startswith("http://ddns.example.net/m/")
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/")
management_slug = body["management_url"].rsplit("/", 1)[1]
managed = client.get(f"/api/v1/management/{management_slug}")
assert managed.status_code == 200
assert managed.json()["hostname"] == body["hostname"]
dashboard_secret = body["dashboard_url"].rsplit("/", 1)[1]
update_path = _path(body["update_url"])
register_path = _path(body["registration_url"])
deleted = client.delete(f"/api/v1/management/{management_slug}")
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/management/{management_slug}").status_code == 404
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_service, "publish_dns", lambda result, settings: published.append(result))
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/updates/{account['update_slug']}", params={"myip": "203.0.113.11"})
response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.11"})
assert response.status_code == 200
assert response.json() == {"status": "good", "hostname": "home.ddns.example.net", "ip": "203.0.113.11"}
assert response.text == "good 203.0.113.11"
assert len(published) == 1
assert store.list_records()[0]["ipv4"] == "203.0.113.11"
@@ -207,14 +308,14 @@ def test_api_dns_publish_failure_keeps_record_unchanged(tmp_path: Path, monkeypa
def fail_publish(result, settings):
raise DnsProviderError("dns down")
monkeypatch.setattr(ddns_service, "publish_dns", fail_publish)
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/updates/{account['update_slug']}", params={"myip": "203.0.113.11"})
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() == []
@@ -232,8 +333,8 @@ def test_api_creation_is_rate_limited(tmp_path: Path) -> None:
)
client = TestClient(app)
first = client.post("/api/v1/hostnames/magic", json={"username": "router"})
second = client.post("/api/v1/hostnames/magic", json={"username": "router"})
first = client.post("/api/v1/hostnames")
second = client.post("/api/v1/hostnames")
assert first.status_code == 201
assert second.status_code == 429
@@ -246,8 +347,8 @@ def test_api_invalid_ip_returns_clear_400(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/api/v1/updates/{account['update_slug']}", params={"myipv6": "203.0.113.9"})
response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myipv6": "203.0.113.9"})
assert response.status_code == 400
assert response.json()["detail"] == "myipv6 must be an IPv6 address"
assert response.text == "badip"
assert store.list_records() == []
tests/test_scheduler.py
+108 -8
View File
@@ -3,27 +3,29 @@ import logging
import sqlite3
import time
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
import router_dyndns.app as app_module
from router_dyndns.ddns import DdnsSettings, DdnsStore, make_app
import karldns.app as app_module
from karldns import DdnsSettings, DdnsStore, make_app
from karldns.security import hash_lookup_token
def test_rate_limit_reject_does_not_increment_bucket(tmp_path: Path) -> None:
store = DdnsStore(tmp_path / "ddns.sqlite3")
assert store.allow_rate_limit("198.51.100.1:/magic", 1) is True
assert store.allow_rate_limit("198.51.100.1:/magic", 1) is False
assert store.allow_rate_limit("198.51.100.1:/", 1) is True
assert store.allow_rate_limit("198.51.100.1:/", 1) is False
store.flush_rate_limits()
window = int(time.time()) // 60
with store.connect() as conn:
count = conn.execute(
"SELECT count FROM rate_limits WHERE key = ? AND window = ?",
("198.51.100.1:/magic", window),
("198.51.100.1:/", window),
).fetchone()[0]
assert count == 1
@@ -161,6 +163,52 @@ def test_cleanup_records_only_unused_hostname_count(tmp_path: Path) -> None:
assert set(store.list_cleanup_runs()[0]) == {"created_at", "unused_accounts_deleted"}
def test_cleanup_removes_pending_generated_setups_after_cutoff(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
with store.connect() as conn:
conn.execute(
"UPDATE accounts SET created_at = ? WHERE hostname = ?",
("2000-01-01T00:00:00+00:00", account["hostname"]),
)
result = store.cleanup(unused_account_hours=45 * 24)
assert result["unused_accounts"] == 1
assert store.get_account_by_slug(account["update_slug"]) is None
def test_cleanup_keeps_recent_pending_setup_until_pending_cutoff(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
recent = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
with store.connect() as conn:
conn.execute(
"UPDATE accounts SET created_at = ? WHERE hostname = ?",
(recent, account["hostname"]),
)
result = store.cleanup(unused_account_hours=1, pending_account_hours=24)
assert result["unused_accounts"] == 0
assert store.get_account_by_slug(account["update_slug"]) is not None
def test_pending_generated_setup_stays_pending_after_restart(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
store.close()
restarted = DdnsStore(database_path)
assert restarted.get_account_by_slug(account["update_slug"])["accepted_at"] is None
def test_sqlite_startup_tuning_is_idempotent(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
@@ -194,7 +242,7 @@ def test_sqlite_startup_tuning_is_idempotent(tmp_path: Path) -> None:
assert cache_size == -20_000
assert temp_store == 2
assert mmap_size == 256 * 1024 * 1024
assert user_version == 1
assert user_version == 3
assert {
"idx_accounts_created_at",
"idx_rate_limits_window",
@@ -203,6 +251,58 @@ def test_sqlite_startup_tuning_is_idempotent(tmp_path: Path) -> None:
}.issubset(index_names)
def test_accounts_schema_migration_removes_legacy_router_credentials(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
with sqlite3.connect(database_path) as conn:
conn.execute("PRAGMA user_version = 2")
conn.execute(
"""
CREATE TABLE accounts (
hostname TEXT PRIMARY KEY,
username TEXT NOT NULL,
token_hash TEXT NOT NULL,
update_slug TEXT NOT NULL UNIQUE,
management_hash TEXT UNIQUE,
dashboard_password_hash TEXT,
created_at TEXT NOT NULL,
accepted_at TEXT,
disabled INTEGER NOT NULL DEFAULT 0
)
"""
)
conn.execute(
"""
INSERT INTO accounts(
hostname, username, token_hash, update_slug, management_hash,
dashboard_password_hash, created_at, accepted_at, disabled
)
VALUES (?, ?, ?, ?, ?, NULL, ?, ?, 0)
""",
(
"home.ddns.example.net",
"router",
"old-token-hash",
"update-secret",
hash_lookup_token("dashboard-secret"),
"2026-05-21T00:00:00+00:00",
"2026-05-21T00:00:00+00:00",
),
)
store = DdnsStore(database_path)
account = store.get_account_by_slug("update-secret")
with store.connect() as conn:
columns = {row[1] for row in conn.execute("PRAGMA table_info(accounts)")}
user_version = conn.execute("PRAGMA user_version").fetchone()[0]
assert user_version == 3
assert "username" not in columns
assert "token_hash" not in columns
assert account
assert account["hostname"] == "home.ddns.example.net"
def test_rate_limit_flush_requeues_counts_after_database_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
store = DdnsStore(tmp_path / "ddns.sqlite3")
assert store.allow_rate_limit("198.51.100.1:/api", 60) is True
@@ -286,10 +386,10 @@ def test_scheduled_cleanup_failure_is_logged(tmp_path: Path, monkeypatch, caplog
store = DdnsStore(database_path)
store.ensure_scheduled_task(app_module.CLEANUP_TASK_NAME, settings.cleanup_interval_seconds)
def fail_cleanup(unused_account_hours, event_history_hours):
def fail_cleanup(unused_account_hours, event_history_hours, pending_account_hours):
raise RuntimeError("cleanup exploded")
caplog.set_level(logging.ERROR, logger="router_dyndns.app")
caplog.set_level(logging.ERROR, logger="karldns.app")
monkeypatch.setattr(store, "cleanup", fail_cleanup)
asyncio.run(app_module._run_cleanup_if_due(store, settings))
tests/test_scripts.py
+174 -4
View File
@@ -6,7 +6,7 @@ from pathlib import Path
from fastapi.testclient import TestClient
from router_dyndns.ddns import DdnsSettings, make_app
from karldns import DdnsSettings, make_app
ROOT = Path(__file__).parents[1]
@@ -40,7 +40,7 @@ def test_updater_strips_fritzbox_placeholders_and_caches_ip(tmp_path: Path) -> N
)
)
client = TestClient(app)
account = client.post("/api/v1/hostnames/magic", json={}).json()
account = client.post("/api/v1/hostnames").json()
update_url = account["update_url"]
script = tmp_path / "curl"
log_file = tmp_path / "curl.log"
@@ -51,7 +51,6 @@ def test_updater_strips_fritzbox_placeholders_and_caches_ip(tmp_path: Path) -> N
f"""#!/bin/sh
printf '%s\\n' "$*" >> {log_file}
case "$*" in
*'/ip'*) printf '203.0.113.25\\n' ;;
*'<ipaddr>'*|*'<ip6addr>'*) printf 'bad placeholder leaked\\n'; exit 1 ;;
*'myip=203.0.113.25'*) printf 'good 203.0.113.25\\n' ;;
*) printf 'bad request\\n'; exit 1 ;;
@@ -67,9 +66,9 @@ esac
env={
"KARLDNS_UPDATE_URL": update_url,
"KARLDNS_CURL": str(script),
"KARLDNS_IP_URL": "https://karldns.de/ip",
"KARLDNS_CACHE_FILE": str(cache_file),
"KARLDNS_LOCK_FILE": str(lock_file),
"KARLDNS_IPV4": "203.0.113.25",
"KARLDNS_IPV6": "off",
},
)
@@ -77,3 +76,174 @@ esac
assert "good 203.0.113.25" in result.stdout
assert cache_file.read_text(encoding="utf-8").strip() == "203.0.113.25|"
assert "<ipaddr>" not in log_file.read_text(encoding="utf-8")
def test_installer_registers_generated_hostname_before_use(tmp_path: Path) -> None:
bin_dir = tmp_path / "bin"
home_dir = tmp_path / "home"
log_file = tmp_path / "curl.log"
bin_dir.mkdir()
home_dir.mkdir()
curl = bin_dir / "curl"
curl.write_text(
f"""#!/bin/sh
printf '%s\\n' "$*" >> {log_file}
output_file=""
previous=""
for argument in "$@"; do
if [ "$previous" = "-o" ]; then
output_file="$argument"
fi
previous="$argument"
done
case "$*" in
*'/karldns-update.sh'*)
cat > "$output_file" <<'EOS'
#!/bin/sh
printf 'updater placeholder\\n'
EOS
;;
*'/api/v1/hostnames'*)
printf '%s\\n' '{{"hostname":"abc.karldns.test","status":"pending","update_url":"https://karldns.test/api/v1/router/update/update-secret?myip=<ipaddr>&myipv6=<ip6addr>","dashboard_url":"https://karldns.test/m/manage-secret","registration_url":"https://karldns.test/api/v1/router/register/manage-secret"}}'
;;
*'/api/v1/router/register/manage-secret'*)
printf '%s\\n' 'good registered'
;;
*)
printf 'unexpected curl call: %s\\n' "$*" >&2
exit 1
;;
esac
""",
encoding="utf-8",
)
curl.chmod(0o700)
result = run_script(
"sh",
str(ROOT / "scripts" / "install-linux.sh"),
"--mode",
"user",
"--no-cron",
"--no-run",
env={
"HOME": str(home_dir),
"PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
"KARLDNS_BASE_URL": "https://karldns.test",
"KARLDNS_RAW_BASE_URL": "https://raw.karldns.test",
},
)
log = log_file.read_text(encoding="utf-8")
assert "/api/v1/hostnames" in log
assert "/api/v1/router/register/manage-secret" in log
assert "KarlDNS: registering hostname" in result.stdout
config = home_dir / ".config" / "karldns" / "updater.env"
config_text = config.read_text(encoding="utf-8")
assert "KARLDNS_UPDATE_URL='https://karldns.test/api/v1/router/update/update-secret'" in config_text
assert "KARLDNS_HOSTNAME='abc.karldns.test'" in config_text
def test_user_installer_and_uninstaller_manage_cron_idempotently(tmp_path: Path) -> None:
bin_dir = tmp_path / "bin"
home_dir = tmp_path / "home"
log_file = tmp_path / "curl.log"
crontab_file = tmp_path / "crontab.txt"
bin_dir.mkdir()
home_dir.mkdir()
curl = bin_dir / "curl"
crontab = bin_dir / "crontab"
curl.write_text(
f"""#!/bin/sh
printf '%s\\n' "$*" >> {log_file}
output_file=""
previous=""
for argument in "$@"; do
if [ "$previous" = "-o" ]; then
output_file="$argument"
fi
previous="$argument"
done
case "$*" in
*'/karldns-update.sh'*)
cat > "$output_file" <<'EOS'
#!/bin/sh
printf 'updater placeholder\\n'
EOS
;;
*'/api/v1/hostnames'*)
printf '%s\\n' '{{"hostname":"abc.karldns.test","status":"pending","update_url":"https://karldns.test/api/v1/router/update/update-secret?myip=<ipaddr>&myipv6=<ip6addr>","dashboard_url":"https://karldns.test/m/manage-secret","registration_url":"https://karldns.test/api/v1/router/register/manage-secret"}}'
;;
*'/api/v1/router/register/manage-secret'*)
printf '%s\\n' 'good registered'
;;
*)
printf 'unexpected curl call: %s\\n' "$*" >&2
exit 1
;;
esac
""",
encoding="utf-8",
)
crontab.write_text(
f"""#!/bin/sh
if [ "$1" = "-l" ]; then
cat {crontab_file} 2>/dev/null || exit 1
exit 0
fi
cp "$1" {crontab_file}
""",
encoding="utf-8",
)
curl.chmod(0o700)
crontab.chmod(0o700)
env = {
"HOME": str(home_dir),
"PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
"KARLDNS_BASE_URL": "https://karldns.test",
"KARLDNS_RAW_BASE_URL": "https://raw.karldns.test",
}
for _ in range(2):
run_script(
"sh",
str(ROOT / "scripts" / "install-linux.sh"),
"--mode",
"user",
"--no-run",
env=env,
)
cron_text = crontab_file.read_text(encoding="utf-8")
log = log_file.read_text(encoding="utf-8")
assert cron_text.count("# karldns-update start") == 1
assert "KARLDNS_CONFIG=" in cron_text
assert "karldns-update" in cron_text
assert log.count("/api/v1/hostnames") == 1
assert log.count("/api/v1/router/register/manage-secret") == 1
assert (home_dir / ".local" / "bin" / "karldns-update").exists()
run_script("sh", str(ROOT / "scripts" / "uninstall-linux.sh"), "--mode", "user", env=env)
assert "# karldns-update start" not in crontab_file.read_text(encoding="utf-8")
assert not (home_dir / ".local" / "bin" / "karldns-update").exists()
assert (home_dir / ".config" / "karldns" / "updater.env").exists()
run_script("sh", str(ROOT / "scripts" / "uninstall-linux.sh"), "--mode", "user", "--delete-config", env=env)
assert not (home_dir / ".config" / "karldns").exists()
assert not (home_dir / ".cache" / "karldns").exists()
def test_hosted_script_routes_redirect_to_github(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
client = TestClient(app)
for path, target in {
"/install.sh": "install-linux.sh",
"/uninstall.sh": "uninstall-linux.sh",
"/karldns-update.sh": "karldns-update.sh",
}.items():
response = client.get(path, follow_redirects=False)
assert response.status_code == 302
assert response.headers["location"].endswith(f"/scripts/{target}")
tests/test_security.py
+31 -45
View File
@@ -6,10 +6,10 @@ from pathlib import Path
import pytest
from fastapi.testclient import TestClient
import router_dyndns.ddns_service as ddns_service
from router_dyndns.ddns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, UpdateResult, make_app
from router_dyndns.ddns_dns import validate_dns_provider_settings
from router_dyndns.ddns_security import DASHBOARD_UNLOCK_COOKIE
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
def test_generated_hostname_is_not_an_update_secret(tmp_path: Path) -> None:
@@ -20,14 +20,11 @@ def test_generated_hostname_is_not_an_update_secret(tmp_path: Path) -> None:
client = TestClient(app)
response = client.get(
"/nic/update",
params={
"hostname": "public-target.ddns.example.net",
"myip": "203.0.113.9",
},
f"/api/v1/router/update/{'a' * 43}",
params={"hostname": "public-target.ddns.example.net", "myip": "203.0.113.9"},
)
assert response.status_code == 401
assert response.status_code == 404
assert store.list_records() == []
@@ -39,16 +36,13 @@ def test_generated_account_cannot_update_other_hostname(tmp_path: Path) -> None:
client = TestClient(app)
response = client.get(
"/nic/update",
params={
"hostname": "other.example.net",
"myip": "203.0.113.9",
"username": account["username"],
"password": account["password"],
},
f"/api/v1/router/update/{account['update_slug']}",
params={"hostname": "other.example.net", "myip": "203.0.113.9"},
)
assert response.status_code == 401
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:
@@ -70,7 +64,7 @@ def test_bearer_link_routes_share_rate_limit_bucket(tmp_path: Path) -> None:
def test_redacted_access_log_masks_bearer_paths_and_query_strings(tmp_path: Path, caplog) -> None:
caplog.set_level(logging.INFO, logger="router_dyndns.access")
caplog.set_level(logging.INFO, logger="karldns.access")
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
@@ -81,30 +75,22 @@ def test_redacted_access_log_masks_bearer_paths_and_query_strings(tmp_path: Path
account = app.state.store.create_account("home.example.net", "router")
client = TestClient(app)
client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.1"})
client.get(
"/nic/update",
params={
"hostname": account["hostname"],
"username": account["username"],
"password": account["password"],
"myip": "203.0.113.2",
},
)
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 "/u/<secret>" in caplog.text
assert "/nic/update?<redacted>" in caplog.text
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["password"] not in caplog.text
assert account["management_slug"] not in caplog.text
@pytest.mark.parametrize(
"path",
[
"/u/short",
"/api/v1/updates/short",
"/api/v1/router/update/short",
"/api/v1/router/register/short",
"/api/v1/hostnames/short",
"/m/short",
"/api/v1/management/short",
],
)
def test_secret_path_tokens_are_constrained(tmp_path: Path, path: str) -> None:
@@ -174,7 +160,7 @@ def test_secret_html_responses_are_not_cached(tmp_path: Path) -> None:
)
client = TestClient(app)
created = client.post("/magic", data={"username": "router"})
created = client.post("/")
assert created.status_code == 200
assert created.headers["cache-control"] == "no-store"
@@ -192,7 +178,7 @@ def test_oversized_request_body_is_rejected(tmp_path: Path) -> None:
)
client = TestClient(app)
response = client.post("/magic", data={"username": "x" * 2000})
response = client.post("/", data={"username": "x" * 2000})
assert response.status_code == 413
@@ -208,7 +194,7 @@ def test_invalid_utf8_form_body_is_rejected(tmp_path: Path) -> None:
client = TestClient(app)
response = client.post(
"/magic",
"/",
content=b"username=\xff",
headers={"content-type": "application/x-www-form-urlencoded"},
)
@@ -229,7 +215,7 @@ def test_too_many_form_fields_are_rejected(tmp_path: Path) -> None:
body = "&".join(f"k{index}=v" for index in range(101))
response = client.post(
"/magic",
"/",
content=body.encode("ascii"),
headers={"content-type": "application/x-www-form-urlencoded"},
)
@@ -249,8 +235,8 @@ def test_untrusted_forwarded_for_does_not_bypass_rate_limit(tmp_path: Path) -> N
)
client = TestClient(app)
first = client.post("/magic", data={"username": "alice"}, headers={"x-forwarded-for": "198.51.100.10"})
second = client.post("/magic", data={"username": "bob"}, headers={"x-forwarded-for": "198.51.100.11"})
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
@@ -260,8 +246,8 @@ def test_dns_delete_failure_is_logged_without_secret_management_slug(tmp_path: P
def fail_delete(hostname, settings):
raise DnsProviderError("dns down")
caplog.set_level(logging.WARNING, logger="router_dyndns.ddns_service")
monkeypatch.setattr(ddns_service, "delete_dns_records", fail_delete)
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")
@@ -363,7 +349,7 @@ def test_required_rfc2136_provider_requires_server_and_zone(tmp_path: Path) -> N
def test_cloudflare_publish_reconciles_provider_state(monkeypatch) -> None:
import router_dyndns.dns.cloudflare as cloudflare
import karldns.dns.cloudflare as cloudflare
calls = {"find": 0}
@@ -394,7 +380,7 @@ def test_rfc2136_delete_checks_response_code(monkeypatch) -> None:
import dns.query
import dns.rcode
from router_dyndns.dns.rfc2136 import delete_rfc2136_records
from karldns.dns.rfc2136 import delete_rfc2136_records
class RefusedResponse:
def rcode(self) -> int:
tests/test_ui.py
+127 -83
View File
@@ -3,7 +3,7 @@ from pathlib import Path
from fastapi.testclient import TestClient
from router_dyndns.ddns import DdnsSettings, DdnsStore, make_app
from karldns import DdnsSettings, DdnsStore, make_app
def _csrf(html: str) -> str:
@@ -23,14 +23,12 @@ def test_self_service_managed_hostname_generates_random_account(tmp_path: Path)
)
client = TestClient(app)
response = client.post("/accounts", data={"mode": "managed", "username": "alice"})
response = client.post("/")
assert response.status_code == 200
assert "Update-URL:" in response.text
assert "Full update URL:" in response.text
assert "CNAME target:" in response.text
assert "Benutzername (optional):" in response.text
assert "Kennwort (optional):" in response.text
assert "Register URL:" in response.text
assert "Router fields" in response.text
assert "CNAME" in response.text
assert "Manual updates" in response.text
@@ -40,8 +38,8 @@ def test_self_service_managed_hostname_generates_random_account(tmp_path: Path)
assert "Public:" in response.text
assert "Secret:" in response.text
assert ".ddns.example.net" in response.text
assert "/u/" in response.text
assert "/nic/update?" in response.text
assert "/api/v1/router/update/" in response.text
assert "/api/v1/router/register/" in response.text
assert "/m/" in response.text
@@ -51,18 +49,18 @@ def test_legacy_custom_account_form_is_rejected(tmp_path: Path) -> None:
database_path=tmp_path / "ddns.sqlite3",
admin_password="admin",
hostname_suffix="ddns.example.net",
public_base_url="https://karldns.de",
)
)
client = TestClient(app)
response = client.post("/accounts", data={"mode": "custom", "hostname": "home.example.net", "claim_secret": "legacy"})
assert response.status_code == 400
assert response.json()["detail"] == "custom hostnames are not supported; create a KarlDNS hostname and use a CNAME"
assert response.status_code == 404
assert app.state.store.count_accounts() == 0
def test_magic_hostname_without_login_generates_update_and_management_links(tmp_path: Path) -> None:
def test_hostname_without_login_generates_update_and_dashboard_links(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
@@ -73,15 +71,16 @@ def test_magic_hostname_without_login_generates_update_and_management_links(tmp_
)
client = TestClient(app)
response = client.post("/magic")
response = client.post("/")
assert response.status_code == 200
assert "Update-URL:" in response.text
assert "Full update URL:" in response.text
assert "CNAME target:" in response.text
assert "Benutzername (optional):" in response.text
assert "Kennwort (optional):" in response.text
assert "Register URL:" in response.text
assert "Private dashboard:" in response.text
assert "Pending setup" in response.text
assert "Register hostname" in response.text
assert "Unregistered setups are removed after 24 hours." in response.text
assert "Router fields" in response.text
assert "CNAME" in response.text
assert "FRITZ!Box fills in" in response.text
@@ -91,17 +90,64 @@ def test_magic_hostname_without_login_generates_update_and_management_links(tmp_
assert ".ddns.example.net" in response.text
def test_magic_get_redirects_to_home_form(tmp_path: Path) -> None:
def test_public_examples_use_visitor_public_ip(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="https://karldns.de",
)
)
client = TestClient(app, client=("8.8.8.8", 12345))
home = client.get("/")
created = client.post("/")
assert home.status_code == 200
assert created.status_code == 200
assert "myip=8.8.8.8" in home.text
assert "good 8.8.8.8" in home.text
assert "myip=8.8.8.8" in created.text
def test_generated_hostname_must_be_activated_before_updates_work(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
app = make_app(
DdnsSettings(
database_path=database_path,
admin_password="admin",
hostname_suffix="ddns.example.net",
public_base_url="http://ddns.example.net",
)
)
client = TestClient(app)
created = client.post("/")
management_slug = re.search(r'action="/m/([^"]+)/activate"', created.text).group(1)
account = app.state.store.get_account_by_management_slug(management_slug)
assert account["accepted_at"] is None
assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404
activated = client.post(f"/m/{management_slug}/activate", data={"csrf": _csrf(created.text)})
assert activated.status_code == 200
assert "Hostname registered" in activated.text
assert app.state.store.get_account_by_management_slug(management_slug)["accepted_at"]
assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 200
def test_old_magic_route_is_gone(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
client = TestClient(app)
response = client.get("/magic", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/#create"
assert response.status_code == 404
def test_magic_management_link_can_delete_hostname(tmp_path: Path) -> None:
def test_dashboard_link_can_delete_hostname(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.ddns.example.net", "home")
@@ -113,8 +159,6 @@ def test_magic_management_link_can_delete_hostname(tmp_path: Path) -> None:
assert '<link href="/admin.css?v=2" rel="stylesheet">' in management.text
assert "home.ddns.example.net" in management.text
assert "CNAME target:" in management.text
assert "Benutzername (optional):" in management.text
assert "Kennwort (optional):" in management.text
assert "Router fields" in management.text
assert "CNAME" in management.text
assert "Manual updates" in management.text
@@ -126,7 +170,7 @@ def test_magic_management_link_can_delete_hostname(tmp_path: Path) -> None:
deleted = client.post(f"/m/{account['management_slug']}/delete", data={"csrf": _csrf(management.text)})
assert deleted.status_code == 200
update = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
update = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"})
assert update.status_code == 404
@@ -170,22 +214,10 @@ def test_management_dashboard_password_locks_dashboard_only(tmp_path: Path) -> N
assert "Update-URL:" not in locked.text
assert "correct horse" not in locked.text
update = locked_client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
update = locked_client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"})
assert update.status_code == 200
assert update.text == "good 203.0.113.9"
nic_update = locked_client.get(
"/nic/update",
params={
"hostname": account["hostname"],
"myip": "203.0.113.10",
"username": account["username"],
"password": account["password"],
},
)
assert nic_update.status_code == 200
assert nic_update.text == "good 203.0.113.10"
bad_unlock = locked_client.post(
f"/m/{account['management_slug']}/unlock",
data={"csrf": _csrf(locked.text), "dashboard_password": "wrong"},
@@ -333,7 +365,7 @@ def test_existing_dashboard_password_requires_current_password_to_change_or_remo
assert app.state.store.verify_dashboard_password("home.ddns.example.net", "correct horse")
def test_management_dashboard_rotates_update_url_and_router_credentials(tmp_path: Path) -> None:
def test_management_dashboard_rotates_update_url(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.ddns.example.net", "router")
@@ -349,7 +381,6 @@ def test_management_dashboard_rotates_update_url_and_router_credentials(tmp_path
dashboard = client.get(f"/m/{account['management_slug']}")
assert dashboard.status_code == 200
assert "New update URL" in dashboard.text
assert "New password" in dashboard.text
assert "data-secret-toggle" in dashboard.text
assert 'type="password"' in dashboard.text
@@ -357,26 +388,10 @@ def test_management_dashboard_rotates_update_url_and_router_credentials(tmp_path
assert rotated_update.status_code == 200
assert "Update URL changed. Update your router now." in rotated_update.text
assert "data-secret-input" in rotated_update.text
assert client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404
rotated_password = client.post(f"/m/{account['management_slug']}/rotate", data={"csrf": _csrf(rotated_update.text), "action": "password"})
assert rotated_password.status_code == 200
assert "Router password changed" in rotated_password.text
assert "Full update URL:" in rotated_password.text
assert "Kennwort (optional):" in rotated_password.text
assert "data-secret-toggle" in rotated_password.text
assert client.get(
"/nic/update",
params={
"hostname": "home.ddns.example.net",
"username": "router",
"password": account["password"],
"myip": "203.0.113.9",
},
).status_code == 401
assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404
statuses = {event["status"] for event in app.state.store.list_update_events_for_hostname("home.ddns.example.net", 10)}
assert "dashboard_update_url_rotated" in statuses
assert "dashboard_router_credentials_rotated" in statuses
assert "dashboard_router_credentials_rotated" not in statuses
def test_management_dashboard_actions_are_scoped_to_slug_owner(tmp_path: Path) -> None:
@@ -393,8 +408,8 @@ def test_management_dashboard_actions_are_scoped_to_slug_owner(tmp_path: Path) -
assert rotated.status_code == 200
assert store.get_account_by_hostname("first.ddns.example.net")["update_slug"] != first["update_slug"]
assert store.get_account_by_hostname("second.ddns.example.net")["update_slug"] == second["update_slug"]
assert client.get(f"/u/{first['update_slug']}", params={"myip": "203.0.113.1"}).status_code == 404
assert client.get(f"/u/{second['update_slug']}", params={"myip": "203.0.113.2"}).status_code == 200
assert client.get(f"/api/v1/router/update/{first['update_slug']}", params={"myip": "203.0.113.1"}).status_code == 404
assert client.get(f"/api/v1/router/update/{second['update_slug']}", params={"myip": "203.0.113.2"}).status_code == 200
def test_locked_dashboard_rejects_mutating_actions_without_unlock_cookie(tmp_path: Path) -> None:
@@ -437,30 +452,6 @@ def test_public_home_hides_admin_link_and_uses_automatic_router_login(tmp_path:
database_path=tmp_path / "ddns.sqlite3",
admin_password="admin",
hostname_suffix="ddns.example.net",
)
)
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert 'href="/admin"' not in response.text
assert "Router login name" not in response.text
assert 'name="username"' not in response.text
assert "You get a hostname, a private update URL, and router fields if your router needs them." in response.text
assert "Add a CNAME" in response.text
assert "Linux cron updater" in response.text
assert "curl -fsSL https://karldns.de/install.sh | sudo sh" in response.text
assert "curl -fsSL https://karldns.de/uninstall.sh | sudo sh" in response.text
assert "--delete-config" in response.text
def test_public_home_examples_use_configured_public_base_url(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
admin_password="admin",
hostname_suffix="karldns.de",
public_base_url="https://karldns.de",
)
)
@@ -469,11 +460,61 @@ def test_public_home_examples_use_configured_public_base_url(tmp_path: Path) ->
response = client.get("/")
assert response.status_code == 200
assert "https://karldns.de/u/&lt;secret&gt;" in response.text
assert 'href="/admin"' not in response.text
assert 'href="#create"' not in response.text
assert response.text.count(">Create hostname<") == 1
assert "Router login name" not in response.text
assert 'name="username"' not in response.text
assert "Create a hostname, save the private values, register it, done." in response.text
assert "Random hostname and private links." in response.text
assert "Activate the hostname when you are ready." in response.text
assert "home.example.com. CNAME" not in response.text
assert "Linux cron updater" in response.text
assert "curl -fsSL https://karldns.de/install.sh | sudo sh" in response.text
assert "curl -fsSL https://karldns.de/uninstall.sh | sudo sh" in response.text
assert "User install" not in response.text
assert "--delete-config" not in response.text
def test_public_home_examples_use_configured_public_base_url(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
admin_password="admin",
hostname_suffix="karldns.de",
public_base_url="https://ddns.example.net",
)
)
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert "https://ddns.example.net/api/v1/router/update/&lt;secret&gt;" in response.text
assert "https://karldns.de/api/v1/router/update/&lt;secret&gt;" not in response.text
assert "http://127.0.0.1" not in response.text
assert "http://localhost" not in response.text
def test_public_examples_use_karldns_domain_and_no_shell_prompt(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)
response = client.get("/")
assert response.status_code == 200
assert "https://karldns.de/api/v1/router/update/&lt;secret&gt;" in response.text
assert "http://127.0.0.1:8099/api/v1/router/update/&lt;secret&gt;" not in response.text
assert "$ curl" not in response.text
assert 'data-copy="curl -sS &#x27;https://karldns.de/api/v1/router/update/&lt;secret&gt;?myip=203.0.113.10&#x27;"' in response.text
def test_application_pages_use_stricter_script_csp(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
client = TestClient(app)
@@ -485,7 +526,7 @@ def test_application_pages_use_stricter_script_csp(tmp_path: Path) -> None:
assert '<script src="/clipboard.js?v=3" defer></script>' in response.text
assert '<script src="/admin-tables.js?v=2" defer></script>' not in response.text
assert '<link href="/bootstrap-subset.css?v=3" rel="stylesheet">' in response.text
assert '<link href="/app.css?v=5" rel="stylesheet">' in response.text
assert '<link href="/app.css?v=12" rel="stylesheet">' in response.text
assert '<link href="/admin.css" rel="stylesheet">' not in response.text
assert "cdn.jsdelivr.net/npm/bootstrap" not in response.text
assert "<style>" not in response.text
@@ -534,6 +575,9 @@ def test_footer_links_and_privacy_page_are_available(tmp_path: Path) -> None:
assert "Made with" in response.text
assert "❤️" in response.text
assert "by karl" in response.text
assert response.text.count('class="lucide-icon footer-icon"') == 5
assert 'class="lucide-icon theme-icon theme-icon-moon"' in response.text
assert 'class="lucide-icon copy-icon"' in response.text
assert 'href="https://github.com/StasonJatham/karldns.de"' in response.text
assert 'href="https://karl.fail"' in response.text
assert 'href="https://karlcom.de"' in response.text
@@ -546,7 +590,7 @@ def test_footer_links_and_privacy_page_are_available(tmp_path: Path) -> None:
assert "WebPage" in privacy.text
assert "Legal and privacy" in privacy.text
assert "Data processed" in privacy.text
assert "protected router passwords" in privacy.text
assert "private update links" in privacy.text
assert "Made with" in privacy.text
assert "❤️" in privacy.text
assert "by karl" in privacy.text
tests/test_updates.py
+50 -120
View File
@@ -1,15 +1,18 @@
import logging
from pathlib import Path
from urllib.parse import parse_qs, urlparse
import pytest
from fastapi.testclient import TestClient
import router_dyndns.ddns_service as ddns_service
from router_dyndns.ddns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, make_app
import karldns.services.ddns as ddns_module
from karldns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, make_app
def test_update_accepts_fritzbox_placeholders(tmp_path: Path) -> None:
def _update_path(account: dict[str, str]) -> str:
return f"/api/v1/router/update/{account['update_slug']}"
def test_router_update_get_accepts_fritzbox_placeholders(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.ddns.example.net", "router")
@@ -17,121 +20,70 @@ def test_update_accepts_fritzbox_placeholders(tmp_path: Path) -> None:
client = TestClient(app)
response = client.get(
"/nic/update",
_update_path(account),
params={
"hostname": "home.ddns.example.net",
"myip": "203.0.113.8",
"myipv6": "2001:db8::8",
"username": account["username"],
"password": account["password"],
},
)
assert response.status_code == 200
assert response.text == "good 203.0.113.8"
records = client.get("/records", auth=("admin", "admin")).json()
records = client.get("/api/v1/admin/records", auth=("admin", "admin")).json()
assert records["total"] == 1
assert records["items"][0]["hostname"] == "home.ddns.example.net"
assert records["items"][0]["ipv4"] == "203.0.113.8"
assert records["items"][0]["ipv6"] == "2001:db8::8"
def test_update_rejects_wrong_password(tmp_path: Path) -> None:
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", hostname_suffix="ddns.example.net"))
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": account["hostname"], "myip": "203.0.113.8", "username": account["username"], "password": "wrong"},
)
assert response.status_code == 401
def test_update_rejects_hostname_outside_provider_suffix(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.example.net", "router")
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin", hostname_suffix="ddns.example.net"))
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": account["hostname"], "myip": "203.0.113.8", "username": account["username"], "password": account["password"]},
)
assert response.status_code == 403
def test_generated_account_can_update_own_hostname(tmp_path: Path) -> None:
def test_router_update_post_accepts_form_and_json(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(
"/nic/update",
params={
"hostname": "home.example.net",
"myip": "203.0.113.9",
"username": account["username"],
"password": account["password"],
},
)
form_response = client.post(_update_path(account), data={"myip": "203.0.113.9"})
json_response = client.post(_update_path(account), json={"myip": "203.0.113.10"})
assert response.status_code == 200
assert response.text == "good 203.0.113.9"
slug_response = client.get(
f"/u/{account['update_slug']}",
params={"myip": "203.0.113.10"},
)
assert slug_response.status_code == 200
assert slug_response.text == "good 203.0.113.10"
assert form_response.status_code == 200
assert form_response.text == "good 203.0.113.9"
assert json_response.status_code == 200
assert json_response.text == "good 203.0.113.10"
assert store.list_records()[0]["ipv4"] == "203.0.113.10"
def test_one_box_update_url_can_authenticate_with_query_credentials(tmp_path: Path) -> None:
def test_router_update_requires_registered_hostname(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.example.net", "alice")
settings = DdnsSettings(
database_path=database_path,
admin_password="admin",
public_base_url="http://ddns.example.net",
)
app = make_app(settings)
account = store.create_account("home.example.net", "alice", accepted=False)
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
one_box_url = ddns_service.DdnsService(settings, store).credentialed_update_url(account)
parsed = urlparse(one_box_url)
params = parse_qs(parsed.query)
params["myip"] = ["203.0.113.11"]
params["myipv6"] = [""]
response = client.get(parsed.path, params={key: value[0] for key, value in params.items()})
before = client.get(_update_path(account), params={"myip": "203.0.113.9"})
registered = client.get(f"/api/v1/router/register/{account['management_slug']}", params={"myip": "203.0.113.9"})
after = client.get(_update_path(account), params={"myip": "203.0.113.10"})
assert response.status_code == 200
assert response.text == "good 203.0.113.11"
assert response.headers["cache-control"] == "no-store"
assert before.status_code == 404
assert registered.status_code == 200
assert registered.text == "good registered"
assert after.status_code == 200
assert after.text == "good 203.0.113.10"
def test_dns_publish_failure_does_not_update_record(tmp_path: Path, monkeypatch, caplog) -> None:
def fail_publish(result, settings):
raise DnsProviderError("dns down")
caplog.set_level(logging.WARNING, logger="router_dyndns.ddns_service")
monkeypatch.setattr(ddns_service, "publish_dns", fail_publish)
caplog.set_level(logging.WARNING, logger="karldns.services.ddns")
monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
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"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 500
assert store.list_records() == []
@@ -146,15 +98,15 @@ def test_dns_reconciliation_failure_is_logged_separately(tmp_path: Path, monkeyp
def fail_publish(result, settings):
raise DnsReconciliationError("provider state mismatch")
caplog.set_level(logging.WARNING, logger="router_dyndns.ddns_service")
monkeypatch.setattr(ddns_service, "publish_dns", fail_publish)
caplog.set_level(logging.WARNING, logger="karldns.services.ddns")
monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
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"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 500
assert store.list_records() == []
@@ -167,7 +119,7 @@ def test_programming_error_during_publish_is_not_mapped_to_dns_failure(tmp_path:
def fail_publish(result, settings):
raise AssertionError("publish bug")
monkeypatch.setattr(ddns_service, "publish_dns", fail_publish)
monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.example.net", "alice")
@@ -175,7 +127,7 @@ def test_programming_error_during_publish_is_not_mapped_to_dns_failure(tmp_path:
client = TestClient(app)
with pytest.raises(AssertionError, match="publish bug"):
client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert store.list_records() == []
assert store.list_update_events() == []
@@ -183,14 +135,14 @@ def test_programming_error_during_publish_is_not_mapped_to_dns_failure(tmp_path:
def test_dns_publish_success_updates_record(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "publish_dns", lambda result, settings: published.append(result))
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.example.net", "alice")
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 200
assert response.text == "good 203.0.113.9"
@@ -204,7 +156,7 @@ def test_dns_publish_success_updates_record(tmp_path: Path, monkeypatch) -> None
def test_unchanged_slug_update_is_read_only_for_history_and_record(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "publish_dns", lambda result, settings: published.append(result))
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.example.net", "alice")
@@ -213,8 +165,8 @@ def test_unchanged_slug_update_is_read_only_for_history_and_record(tmp_path: Pat
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
second_response = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
second_response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 200
assert response.text == "nochg 203.0.113.9"
@@ -233,7 +185,7 @@ def test_slug_update_rejects_hostname_outside_provider_suffix(tmp_path: Path) ->
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin", hostname_suffix="ddns.example.net"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 404
assert store.list_records() == []
@@ -241,7 +193,7 @@ def test_slug_update_rejects_hostname_outside_provider_suffix(tmp_path: Path) ->
def test_omitted_ip_family_is_preserved(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "publish_dns", lambda result, settings: published.append(result))
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.example.net", "alice")
@@ -249,7 +201,7 @@ def test_omitted_ip_family_is_preserved(tmp_path: Path, monkeypatch) -> None:
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
assert response.status_code == 200
record = store.list_records()[0]
@@ -267,7 +219,7 @@ def test_blank_ip_params_fall_back_to_source_ip(tmp_path: Path) -> None:
client = TestClient(app, client=("198.51.100.44", 50000))
response = client.get(
f"/u/{account['update_slug']}",
_update_path(account),
params={"myip": "", "myipv6": ""},
)
@@ -276,9 +228,9 @@ def test_blank_ip_params_fall_back_to_source_ip(tmp_path: Path) -> None:
assert store.list_records()[0]["ipv4"] == "198.51.100.44"
def test_invalid_legacy_ip_is_badip_and_does_not_change_record(tmp_path: Path, monkeypatch) -> None:
def test_invalid_ip_is_badip_and_does_not_change_record(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "publish_dns", lambda result, settings: published.append(result))
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.example.net", "alice")
@@ -286,7 +238,7 @@ def test_invalid_legacy_ip_is_badip_and_does_not_change_record(tmp_path: Path, m
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "not-an-ip"})
response = client.get(_update_path(account), params={"myip": "not-an-ip"})
assert response.status_code == 400
assert response.text == "badip"
@@ -296,7 +248,7 @@ def test_invalid_legacy_ip_is_badip_and_does_not_change_record(tmp_path: Path, m
assert record["ipv6"] == "2001:db8::8"
def test_wrong_family_legacy_ip_is_badip_and_preserves_records(tmp_path: Path) -> None:
def test_wrong_ip_family_is_badip_and_preserves_records(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.example.net", "alice")
@@ -304,32 +256,10 @@ def test_wrong_family_legacy_ip_is_badip_and_preserves_records(tmp_path: Path) -
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
client = TestClient(app)
response = client.get(f"/u/{account['update_slug']}", params={"myip": "2001:db8::9"})
response = client.get(_update_path(account), params={"myip": "2001:db8::9"})
assert response.status_code == 400
assert response.text == "badip"
record = store.list_records()[0]
assert record["ipv4"] == "203.0.113.8"
assert record["ipv6"] == "2001:db8::8"
def test_nic_update_invalid_ip_returns_badip(tmp_path: Path) -> None:
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", hostname_suffix="ddns.example.net"))
client = TestClient(app)
response = client.get(
"/nic/update",
params={
"hostname": account["hostname"],
"myipv6": "203.0.113.9",
"username": account["username"],
"password": account["password"],
},
)
assert response.status_code == 400
assert response.text == "badip"
assert store.list_records() == []