Archived
StasonJatham
5c46b47937
Preserve the GitHub history and the newer deployed source from ct_134_karldns. Exclude runtime data, credentials, AppleDouble files and local backup copies. Make Gitea the canonical source for documentation, package metadata and project links.
404 lines
16 KiB
Python
404 lines
16 KiB
Python
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import karldns.services.ddns as ddns_module
|
|
from karldns import DdnsSettings, DdnsStore, DnsProviderError, make_app
|
|
|
|
|
|
def _path(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
return parsed.path
|
|
|
|
|
|
def test_llms_txt_documents_public_api_without_creating_accounts(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="private-admin-password",
|
|
public_base_url="https://dns.example.net",
|
|
hostname_suffix="dns.example.net",
|
|
)
|
|
)
|
|
with TestClient(app) as client:
|
|
response = client.get("/llms.txt")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/plain")
|
|
assert response.text.startswith("# KarlDNS\n\n> ")
|
|
assert "https://dns.example.net/openapi.json" in response.text
|
|
assert "private-admin-password" not in response.text
|
|
assert "/api/v1/admin" not in response.text
|
|
assert "## Optional\n" in response.text
|
|
assert 'rel="describedby" href="/llms.txt"' in client.get("/").text
|
|
for path, operations in client.get("/openapi.json").json()["paths"].items():
|
|
assert path in response.text
|
|
for method in operations:
|
|
assert method.upper() in response.text
|
|
assert app.state.store.count_accounts() == 0
|
|
|
|
|
|
def test_custom_api_docs_include_only_current_api_surfaces(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
public_base_url="https://karldns.de",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
docs = client.get("/docs")
|
|
redoc = client.get("/redoc")
|
|
schema = client.get("/openapi.json").json()
|
|
|
|
assert docs.status_code == 200
|
|
assert redoc.status_code == 404
|
|
assert "Small API surface." in docs.text
|
|
assert "POST /api/v1/hostnames" in docs.text
|
|
assert "GET or POST /api/v1/router/register/<karl_secret>" in docs.text
|
|
assert "GET or POST /api/v1/router/update/<karl_secret>" in docs.text
|
|
assert "GET /api/v1/hostnames/<karl_secret>" in docs.text
|
|
assert "POST /api/v1/hostnames/<karl_secret>/rotate-update-url" in docs.text
|
|
assert "Rotate update URL" in docs.text
|
|
assert "Operator API" not in docs.text
|
|
assert "/api/v1/admin" not in docs.text
|
|
assert "admin:<admin-password>" not in docs.text
|
|
assert "https://karldns.de/api/v1/hostnames" in docs.text
|
|
assert "http://localhost:8080/api/v1/hostnames" not in docs.text
|
|
assert "registration_url" in docs.text
|
|
assert "/api/v1/management" not in docs.text
|
|
assert "/api/v1/hostnames/magic" not in docs.text
|
|
assert "/nic/update" not in docs.text
|
|
assert "/u/<secret>" not in docs.text
|
|
assert "swagger-ui" not in docs.text.lower()
|
|
assert "redoc" not in docs.text.lower()
|
|
assert "x-robots-tag" not in docs.headers
|
|
assert client.get("/openapi.json").headers["x-robots-tag"] == "noindex, nofollow"
|
|
assert "cdn.jsdelivr.net" not in docs.headers["content-security-policy"]
|
|
assert "fonts.googleapis.com" not in docs.headers["content-security-policy"]
|
|
|
|
expected_paths = {
|
|
"/api/v1/hostnames",
|
|
"/api/v1/hostnames/{dashboard_secret}",
|
|
"/api/v1/hostnames/{dashboard_secret}/rotate-update-url",
|
|
"/api/v1/router/register/{dashboard_secret}",
|
|
"/api/v1/router/update/{update_slug}",
|
|
}
|
|
assert set(schema["paths"]) == expected_paths
|
|
assert "/" not in schema["paths"]
|
|
assert "/admin" not in schema["paths"]
|
|
assert "/healthz" not in schema["paths"]
|
|
assert "/records" not in schema["paths"]
|
|
assert "/events" not in schema["paths"]
|
|
assert any(tag["name"] == "hostnames" for tag in schema["tags"])
|
|
assert any(tag["name"] == "router" for tag in schema["tags"])
|
|
assert all(tag["name"] != "admin" for tag in schema["tags"])
|
|
schema_names = set(schema["components"]["schemas"])
|
|
assert "AdminDashboardResponse" not in schema_names
|
|
assert "AdminHostnamesResponse" not in schema_names
|
|
assert "AdminEventsResponse" not in schema_names
|
|
assert "AdminRecordsResponse" not in schema_names
|
|
|
|
|
|
def test_custom_api_docs_use_visitor_public_ip(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
public_base_url="https://karldns.de",
|
|
)
|
|
)
|
|
client = TestClient(app, client=("8.8.4.4", 12345))
|
|
|
|
response = client.get("/docs")
|
|
|
|
assert response.status_code == 200
|
|
assert "myip=8.8.4.4" in response.text
|
|
|
|
|
|
def test_openapi_hides_custom_operator_routes(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
admin_path="/ops",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
schema = client.get("/openapi.json").json()
|
|
|
|
forbidden_paths = {
|
|
"/ops",
|
|
"/ops/hostnames/delete",
|
|
"/ops/hostnames/rotate",
|
|
"/ops/cleanup",
|
|
"/ops/events",
|
|
"/ops/records",
|
|
"/records",
|
|
"/events",
|
|
"/api/v1/admin/dashboard",
|
|
"/api/v1/admin/events",
|
|
"/api/v1/admin/hostnames",
|
|
"/api/v1/admin/records",
|
|
}
|
|
assert client.get("/redoc").status_code == 404
|
|
assert forbidden_paths.isdisjoint(schema["paths"])
|
|
assert not any(path.startswith("/api/v1/admin") for path in schema["paths"])
|
|
|
|
|
|
def test_openapi_marks_generated_links_as_secret(tmp_path: Path) -> None:
|
|
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
|
|
client = TestClient(app)
|
|
|
|
schema = client.get("/openapi.json").json()
|
|
created = schema["components"]["schemas"]["HostnameCreateResponse"]["properties"]
|
|
status = schema["components"]["schemas"]["HostnameStatusResponse"]["properties"]
|
|
|
|
for name in ["update_url", "dashboard_url", "registration_url"]:
|
|
assert "Secret" in created[name]["description"]
|
|
for name in ["update_url", "dashboard_url"]:
|
|
assert "Secret" in status[name]["description"]
|
|
|
|
|
|
def test_hostname_api_lifecycle_and_history(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
hostname_suffix="ddns.example.net",
|
|
public_base_url="http://ddns.example.net",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
created = client.post("/api/v1/hostnames")
|
|
|
|
assert created.status_code == 201
|
|
assert created.headers["cache-control"] == "no-store"
|
|
body = created.json()
|
|
assert body["hostname"].endswith(".ddns.example.net")
|
|
assert body["status"] == "pending"
|
|
assert body["update_url"].startswith("http://ddns.example.net/api/v1/router/update/")
|
|
assert body["dashboard_url"].startswith("http://ddns.example.net/m/")
|
|
assert body["registration_url"].startswith("http://ddns.example.net/api/v1/router/register/")
|
|
|
|
dashboard_secret = body["dashboard_url"].rsplit("/", 1)[1]
|
|
update_path = _path(body["update_url"])
|
|
register_path = _path(body["registration_url"])
|
|
assert dashboard_secret.startswith("karl_")
|
|
assert update_path.rsplit("/", 1)[1].startswith("karl_")
|
|
assert register_path.rsplit("/", 1)[1].startswith("karl_")
|
|
|
|
pending = client.get(f"/api/v1/hostnames/{dashboard_secret}")
|
|
assert pending.status_code == 200
|
|
assert pending.json()["status"] == "pending"
|
|
assert pending.json()["history"] == []
|
|
assert client.get(update_path, params={"myip": "203.0.113.11"}).status_code == 404
|
|
|
|
registered = client.post(register_path, params={"myip": "203.0.113.10"})
|
|
assert registered.status_code == 200
|
|
assert registered.text == "good registered"
|
|
|
|
update = client.get(update_path, params={"myip": "203.0.113.11"})
|
|
assert update.status_code == 200
|
|
assert update.text == "good 203.0.113.11"
|
|
|
|
active = client.get(f"/api/v1/hostnames/{dashboard_secret}", params={"history_limit": 10})
|
|
assert active.status_code == 200
|
|
payload = active.json()
|
|
assert payload["status"] == "active"
|
|
assert payload["ipv4"] == "203.0.113.11"
|
|
assert payload["history"][0]["hostname"] == body["hostname"]
|
|
assert "source_ip" not in payload["history"][0]
|
|
|
|
deleted = client.delete(f"/api/v1/hostnames/{dashboard_secret}")
|
|
assert deleted.status_code == 204
|
|
assert client.get(f"/api/v1/hostnames/{dashboard_secret}").status_code == 404
|
|
|
|
|
|
def test_hostname_api_can_rotate_update_url(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
hostname_suffix="ddns.example.net",
|
|
public_base_url="http://ddns.example.net",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
body = client.post("/api/v1/hostnames").json()
|
|
dashboard_secret = body["dashboard_url"].rsplit("/", 1)[1]
|
|
old_update_path = _path(body["update_url"])
|
|
|
|
pending_rotate = client.post(f"/api/v1/hostnames/{dashboard_secret}/rotate-update-url")
|
|
registered = client.post(_path(body["registration_url"]), params={"myip": "203.0.113.10"})
|
|
rotated = client.post(f"/api/v1/hostnames/{dashboard_secret}/rotate-update-url")
|
|
new_update_path = _path(rotated.json()["update_url"])
|
|
|
|
assert pending_rotate.status_code == 400
|
|
assert registered.status_code == 200
|
|
assert rotated.status_code == 200
|
|
assert rotated.json()["hostname"] == body["hostname"]
|
|
assert rotated.json()["update_url"] != body["update_url"]
|
|
assert new_update_path.rsplit("/", 1)[1].startswith("karl_")
|
|
assert client.get(old_update_path, params={"myip": "203.0.113.11"}).status_code == 404
|
|
assert client.get(new_update_path, params={"myip": "203.0.113.12"}).status_code == 200
|
|
|
|
|
|
def test_hostname_api_history_is_scoped_to_dashboard_secret(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
own = store.create_account("own.ddns.example.net", "own-router")
|
|
other = store.create_account("other.ddns.example.net", "other-router")
|
|
store.log_update_event("own.ddns.example.net", "203.0.113.21", None, "good", "updated", "198.51.100.10")
|
|
store.log_update_event("other.ddns.example.net", "203.0.113.22", None, "good", "updated", "198.51.100.11")
|
|
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
|
|
client = TestClient(app)
|
|
|
|
response = client.get(f"/api/v1/hostnames/{own['management_slug']}", params={"history_limit": 10})
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
payload = response.json()
|
|
assert len(payload["history"]) == 1
|
|
assert payload["history"][0]["hostname"] == "own.ddns.example.net"
|
|
assert "source_ip" not in payload["history"][0]
|
|
assert "other.ddns.example.net" not in response.text
|
|
assert client.get(f"/api/v1/hostnames/{other['management_slug']}", params={"history_limit": 101}).status_code == 422
|
|
assert client.get(f"/api/v1/hostnames/karl_{'a' * 32}").status_code == 404
|
|
|
|
|
|
def test_registration_publishes_initial_dns_when_provider_is_configured(tmp_path: Path, monkeypatch) -> None:
|
|
published = []
|
|
monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
hostname_suffix="ddns.example.net",
|
|
public_base_url="http://ddns.example.net",
|
|
dns_provider="cloudflare",
|
|
cloudflare_api_token="token",
|
|
cloudflare_zone_id="zone",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
body = client.post("/api/v1/hostnames").json()
|
|
registered = client.post(_path(body["registration_url"]), params={"myip": "203.0.113.77"})
|
|
status = client.get(f"/api/v1/hostnames/{body['dashboard_url'].rsplit('/', 1)[1]}")
|
|
|
|
assert registered.status_code == 200
|
|
assert status.json()["status"] == "active"
|
|
assert status.json()["ipv4"] == "203.0.113.77"
|
|
assert len(published) == 1
|
|
assert published[0].hostname == body["hostname"]
|
|
assert published[0].ipv4 == "203.0.113.77"
|
|
|
|
|
|
def test_registration_without_public_source_ip_publishes_placeholder_until_first_router_update(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
published = []
|
|
monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
hostname_suffix="ddns.example.net",
|
|
public_base_url="http://ddns.example.net",
|
|
dns_provider="cloudflare",
|
|
cloudflare_api_token="token",
|
|
cloudflare_zone_id="zone",
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
body = client.post("/api/v1/hostnames").json()
|
|
update_path = _path(body["update_url"])
|
|
registered = client.post(_path(body["registration_url"]))
|
|
status = client.get(f"/api/v1/hostnames/{body['dashboard_url'].rsplit('/', 1)[1]}")
|
|
update = client.get(update_path, params={"myip": "203.0.113.80"})
|
|
|
|
assert registered.status_code == 200
|
|
assert registered.text == "good registered"
|
|
assert status.json()["status"] == "active"
|
|
assert status.json()["ipv4"] == "192.0.2.1"
|
|
assert update.status_code == 200
|
|
assert update.text == "good 203.0.113.80"
|
|
assert len(published) == 2
|
|
assert published[0].ipv4 == "192.0.2.1"
|
|
assert published[0].update_ipv4 is True
|
|
assert published[0].update_ipv6 is False
|
|
assert published[1].ipv4 == "203.0.113.80"
|
|
|
|
|
|
def test_api_update_publishes_before_recording_ip(tmp_path: Path, monkeypatch) -> None:
|
|
published = []
|
|
monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("home.ddns.example.net", "router")
|
|
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
|
|
client = TestClient(app)
|
|
|
|
response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.11"})
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == "good 203.0.113.11"
|
|
assert len(published) == 1
|
|
assert store.list_records()[0]["ipv4"] == "203.0.113.11"
|
|
|
|
|
|
def test_api_dns_publish_failure_keeps_record_unchanged(tmp_path: Path, monkeypatch) -> None:
|
|
def fail_publish(result, settings):
|
|
raise DnsProviderError("dns down")
|
|
|
|
monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("home.ddns.example.net", "router")
|
|
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
|
|
client = TestClient(app)
|
|
|
|
response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.11"})
|
|
|
|
assert response.status_code == 500
|
|
assert store.list_records() == []
|
|
assert store.list_update_events()[0]["status"] == "dns_failed"
|
|
|
|
|
|
def test_api_creation_is_rate_limited(tmp_path: Path) -> None:
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
hostname_suffix="ddns.example.net",
|
|
rate_limit_per_minute=1,
|
|
)
|
|
)
|
|
client = TestClient(app)
|
|
|
|
first = client.post("/api/v1/hostnames")
|
|
second = client.post("/api/v1/hostnames")
|
|
|
|
assert first.status_code == 201
|
|
assert second.status_code == 429
|
|
|
|
|
|
def test_api_invalid_ip_returns_clear_400(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("home.example.net", "alice")
|
|
app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
|
|
client = TestClient(app)
|
|
|
|
response = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myipv6": "203.0.113.9"})
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "badip"
|
|
assert store.list_records() == []
|