import re from pathlib import Path from fastapi.testclient import TestClient from karldns import DdnsSettings, DdnsStore, make_app def _csrf(html: str) -> str: match = re.search(r'name="csrf" value="([^"]+)"', html) assert match return match.group(1) def test_self_service_managed_hostname_generates_random_account(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) response = client.post("/") assert response.status_code == 200 assert "/m/" in str(response.url) assert "Update-URL:" in response.text assert "CNAME target:" in response.text assert "Register URL:" in response.text assert "Router setup." in response.text assert "Use your own DNS name." in response.text assert "Manual updates" in response.text assert "curl -sS" in response.text assert "203.0.113.10" in response.text assert "2001:db8::10" in response.text assert "Public:" in response.text assert "Secret:" in response.text assert ".ddns.example.net" in response.text assert "/api/v1/router/update/" in response.text assert "/api/v1/router/register/" in response.text assert "/m/" in response.text assert 'type="password"' not in response.text assert "Copy setup values" not in response.text def test_legacy_custom_account_form_is_rejected(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) response = client.post("/accounts", data={"mode": "custom", "hostname": "home.example.net", "claim_secret": "legacy"}) assert response.status_code == 404 assert app.state.store.count_accounts() == 0 def test_hostname_without_login_generates_update_and_dashboard_links(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) response = client.post("/") assert response.status_code == 200 assert "/m/" in str(response.url) assert "Update-URL:" in response.text assert "CNAME target:" 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 "Pending setups are cleaned up after 24 hours." in response.text assert "Router fields" not in response.text assert "Copy setup values" not in response.text assert "Paste this exact URL into the FRITZ!Box" in response.text assert "Copy manual update commands" in response.text assert response.text.count("Secret") >= 3 assert response.text.count("Public") >= 3 assert ".ddns.example.net" in response.text assert 'type="password"' not in response.text 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 == 404 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") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) management = client.get(f"/m/{account['management_slug']}") assert management.status_code == 200 assert '' in management.text assert "home.ddns.example.net" in management.text assert "CNAME target:" in management.text assert "Router fields" in management.text assert "CNAME" in management.text assert "Manual updates" in management.text assert "curl -sS" in management.text assert "Public:" in management.text assert "Secret:" in management.text assert "Router updates" in management.text deleted = client.post(f"/m/{account['management_slug']}/delete", data={"csrf": _csrf(management.text)}) assert deleted.status_code == 200 update = client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}) assert update.status_code == 404 def test_management_dashboard_password_locks_dashboard_only(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") app = make_app( DdnsSettings( database_path=database_path, admin_password="admin", public_base_url="http://ddns.example.net", ) ) client = TestClient(app) dashboard = client.get(f"/m/{account['management_slug']}") set_password = client.post( f"/m/{account['management_slug']}/password", data={ "csrf": _csrf(dashboard.text), "dashboard_password": "correct horse", "dashboard_password_confirm": "correct horse", }, follow_redirects=False, ) assert set_password.status_code == 303 assert "httponly" in set_password.headers["set-cookie"].lower() assert "samesite=lax" in set_password.headers["set-cookie"].lower() assert "secure" not in set_password.headers["set-cookie"].lower() stored = app.state.store.get_account_by_management_slug(account["management_slug"]) assert stored["dashboard_password_hash"].startswith("pbkdf2_sha256$") assert stored["dashboard_password_hash"] != "correct horse" locked_client = TestClient(app) locked = locked_client.get(f"/m/{account['management_slug']}") assert locked.status_code == 200 assert "Dashboard locked" in locked.text assert "Update-URL:" not in locked.text assert "correct horse" not in locked.text 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" bad_unlock = locked_client.post( f"/m/{account['management_slug']}/unlock", data={"csrf": _csrf(locked.text), "dashboard_password": "wrong"}, ) assert bad_unlock.status_code == 401 assert "Incorrect password." in bad_unlock.text assert "Update-URL:" not in bad_unlock.text good_unlock = locked_client.post( f"/m/{account['management_slug']}/unlock", data={"csrf": _csrf(locked.text), "dashboard_password": "correct horse"}, follow_redirects=False, ) assert good_unlock.status_code == 303 assert "max-age=86400" in good_unlock.headers["set-cookie"].lower() unlocked = locked_client.get(f"/m/{account['management_slug']}") assert "Update-URL:" in unlocked.text assert "Password protection" in unlocked.text def test_management_dashboard_password_cookie_is_secure_on_https_public_url(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") app = make_app( DdnsSettings( database_path=database_path, admin_password="admin", public_base_url="https://ddns.example.net", ) ) client = TestClient(app) dashboard = client.get(f"/m/{account['management_slug']}") response = client.post( f"/m/{account['management_slug']}/password", data={ "csrf": _csrf(dashboard.text), "dashboard_password": "correct horse", "dashboard_password_confirm": "correct horse", }, follow_redirects=False, ) assert response.status_code == 303 assert "secure" in response.headers["set-cookie"].lower() def test_management_dashboard_password_can_be_removed(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") store.set_dashboard_password("home.ddns.example.net", "correct horse") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) unlock = client.post( f"/m/{account['management_slug']}/unlock", data={"csrf": _csrf(client.get(f"/m/{account['management_slug']}").text), "dashboard_password": "correct horse"}, ) assert unlock.status_code == 200 remove = client.post( f"/m/{account['management_slug']}/password/remove", data={"csrf": _csrf(unlock.text), "current_dashboard_password": "correct horse"}, ) assert remove.status_code == 200 assert "Not enabled" in remove.text fresh_client = TestClient(app) unlocked = fresh_client.get(f"/m/{account['management_slug']}") assert unlocked.status_code == 200 assert "Update-URL:" in unlocked.text def test_dashboard_stay_unlocked_cookie_uses_longer_lifetime(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") store.set_dashboard_password("home.ddns.example.net", "correct horse") app = make_app( DdnsSettings( database_path=database_path, admin_password="admin", dashboard_unlock_seconds=3600, dashboard_stay_unlocked_seconds=30 * 24 * 60 * 60, ) ) client = TestClient(app) locked = client.get(f"/m/{account['management_slug']}") response = client.post( f"/m/{account['management_slug']}/unlock", data={"csrf": _csrf(locked.text), "dashboard_password": "correct horse", "stay_unlocked": "1"}, follow_redirects=False, ) assert response.status_code == 303 assert "max-age=2592000" in response.headers["set-cookie"].lower() def test_unlocked_dashboard_posts_require_csrf(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.post(f"/m/{account['management_slug']}/rotate", data={"action": "update"}) assert response.status_code == 403 assert store.get_account_by_hostname("home.ddns.example.net")["update_slug"] == account["update_slug"] def test_existing_dashboard_password_requires_current_password_to_change_or_remove(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "home") store.set_dashboard_password("home.ddns.example.net", "correct horse") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) locked = client.get(f"/m/{account['management_slug']}") unlocked = client.post( f"/m/{account['management_slug']}/unlock", data={"csrf": _csrf(locked.text), "dashboard_password": "correct horse"}, ) change = client.post( f"/m/{account['management_slug']}/password", data={ "csrf": _csrf(unlocked.text), "current_dashboard_password": "wrong", "dashboard_password": "new password", "dashboard_password_confirm": "new password", }, ) remove = client.post( f"/m/{account['management_slug']}/password/remove", data={"csrf": _csrf(unlocked.text), "current_dashboard_password": "wrong"}, ) assert change.status_code == 400 assert remove.status_code == 400 assert "Current password is incorrect." in change.text assert app.state.store.verify_dashboard_password("home.ddns.example.net", "correct horse") 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") app = make_app( DdnsSettings( database_path=database_path, admin_password="admin", public_base_url="http://ddns.example.net", ) ) client = TestClient(app) dashboard = client.get(f"/m/{account['management_slug']}") assert dashboard.status_code == 200 assert "New update URL" in dashboard.text assert "data-secret-toggle" in dashboard.text assert 'type="password"' in dashboard.text rotated_update = client.post(f"/m/{account['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"}) 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"/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" not in statuses def test_management_dashboard_actions_are_scoped_to_slug_owner(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) first = store.create_account("first.ddns.example.net", "first") second = store.create_account("second.ddns.example.net", "second") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) dashboard = client.get(f"/m/{first['management_slug']}") rotated = client.post(f"/m/{first['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"}) 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"/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: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("locked.ddns.example.net", "router") store.set_dashboard_password("locked.ddns.example.net", "correct horse") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) rotate = client.post(f"/m/{account['management_slug']}/rotate", data={"action": "update"}) delete = client.post(f"/m/{account['management_slug']}/delete") assert rotate.status_code == 401 assert delete.status_code == 401 assert store.get_account_by_hostname("locked.ddns.example.net")["update_slug"] == account["update_slug"] assert store.get_account_by_slug(account["update_slug"]) is not None def test_management_page_shows_ip_history(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" store = DdnsStore(database_path) account = store.create_account("home.ddns.example.net", "router") store.log_update_event("home.ddns.example.net", "203.0.113.9", None, "updated", "ok", "198.51.100.1") store.log_update_event("home.ddns.example.net", "203.0.113.10", None, "updated", "ok", "198.51.100.1") app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) response = client.get(f"/m/{account['management_slug']}") assert response.status_code == 200 assert "Router updates" in response.text assert "203.0.113.9" in response.text assert "203.0.113.10" in response.text def test_public_home_hides_admin_link_and_uses_automatic_router_login(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) response = client.get("/") assert response.status_code == 200 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/<secret>" in response.text assert "https://karldns.de/api/v1/router/update/<secret>" 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/<secret>" in response.text assert "http://127.0.0.1:8099/api/v1/router/update/<secret>" not in response.text assert "$ curl" not in response.text assert 'data-copy="curl -sS 'https://karldns.de/api/v1/router/update/<secret>?myip=203.0.113.10'"' 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) response = client.get("/") assert response.status_code == 200 assert '' in response.text assert '' in response.text assert '' not in response.text assert '' in response.text assert '' in response.text assert '' not in response.text assert "cdn.jsdelivr.net/npm/bootstrap" not in response.text assert "