from __future__ import annotations import asyncio import contextlib import html import secrets import sqlite3 import urllib.parse from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path from typing import Annotated from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, Response from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse, RedirectResponse from starlette.concurrency import run_in_threadpool from .ddns_api import create_api_router from .ddns_dns import delete_dns_records, dns_backend_configured, publish_dns from .ddns_models import DdnsSettings, UpdateResult from .ddns_security import ( admin_csrf_token, client_ip, credentials_from_basic_auth, is_rate_limited_path, require_admin_csrf, ) from .ddns_service import DdnsService, normalize_domain, normalize_hostname from .ddns_store import DdnsStore __all__ = [ "DdnsSettings", "DdnsStore", "UpdateResult", "delete_dns_records", "make_app", "publish_dns", ] OPENAPI_TAGS = [ { "name": "hostnames", "description": "Create provider-owned or verified custom DynDNS hostnames.", }, { "name": "domains", "description": "Issue and verify DNS TXT challenges for custom domains.", }, { "name": "updates", "description": "Router-facing IP update endpoints for FRITZ!Box and compatible clients.", }, { "name": "management", "description": "Bearer-link management for generated hostnames.", }, { "name": "admin", "description": "Operator-only HTML and JSON endpoints protected by the admin password.", }, ] ASSET_DIR = Path(__file__).with_name("assets") def make_app(settings: DdnsSettings | None = None) -> FastAPI: settings = settings or DdnsSettings.from_env() if not settings.admin_password and settings.shared_secret: settings = settings.model_copy(update={"admin_password": settings.shared_secret}) if settings.require_dns_provider and not dns_backend_configured(settings): raise RuntimeError("DDNS_REQUIRE_DNS_PROVIDER is set but no DNS backend is configured") store = DdnsStore(settings.database_path) store.cleanup(settings.cleanup_challenge_hours, settings.cleanup_unused_account_hours) service = DdnsService(settings, store) @asynccontextmanager async def lifespan(_: FastAPI) -> AsyncIterator[None]: cleanup_task = asyncio.create_task(_cleanup_loop(store, settings)) try: yield finally: cleanup_task.cancel() with contextlib.suppress(asyncio.CancelledError): await cleanup_task app = FastAPI( title="router_dyndns DDNS", summary="A small self-hosted DynDNS provider for FRITZ!Box routers.", description=( "Generate cryptographically random DynDNS endpoints, verify custom domains with DNS TXT " "records, and publish A/AAAA records through Cloudflare or RFC 2136." ), version="0.1.0", docs_url="/docs", redoc_url="/redoc", openapi_tags=OPENAPI_TAGS, lifespan=lifespan, ) app.include_router(create_api_router(settings, store, service)) @app.middleware("http") async def security_middleware(request: Request, call_next): if is_rate_limited_path(request.url.path): key = f"{client_ip(request, settings)}:{request.url.path}" allowed = await run_in_threadpool(store.allow_rate_limit, key, settings.rate_limit_per_minute) if not allowed: return PlainTextResponse("rate limit exceeded", status_code=429) response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["Referrer-Policy"] = "same-origin" response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()" response.headers["Content-Security-Policy"] = ( "default-src 'self'; script-src 'unsafe-inline' 'self'; style-src 'unsafe-inline' 'self'; form-action 'self'; frame-ancestors 'none'" ) if settings.public_base_url.startswith("https://"): response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" return response def authenticate_update( authorization: Annotated[str | None, Header()] = None, username: str | None = Query(default=None), password: str | None = Query(default=None), pass_: str | None = Query(default=None, alias="pass"), passwd: str | None = Query(default=None), ) -> tuple[str | None, str | None]: supplied_user = username supplied_token = password or pass_ or passwd if authorization and authorization.lower().startswith("basic "): supplied_user, supplied_token = credentials_from_basic_auth(authorization) return supplied_user, supplied_token def authenticate_admin( authorization: Annotated[str | None, Header()] = None, ) -> None: if not settings.admin_password: raise HTTPException(status_code=500, detail="DDNS_ADMIN_PASSWORD is required") supplied = None if authorization and authorization.lower().startswith("basic "): _, supplied = credentials_from_basic_auth(authorization) if not supplied or not secrets.compare_digest(supplied, settings.admin_password): raise HTTPException( status_code=401, detail="admin authentication required", headers={"WWW-Authenticate": 'Basic realm="router_dyndns-ddns"'}, ) def apply_update( request: Request, hostname: str, myip: str | None, ipaddr: str | None, myipv6: str | None, ip6addr: str | None, ) -> str | PlainTextResponse: try: return service.apply_update(request, hostname, myip, ipaddr, myipv6, ip6addr).result.dyndns_response except Exception: return PlainTextResponse("911 dns publish failed", status_code=500) @app.get("/", response_class=HTMLResponse) def index() -> str: return _render_public_home(service, settings, None, None) @app.post("/magic", response_class=HTMLResponse) async def magic_account(request: Request) -> str: form = _parse_form(await request.body()) username = _first_form_value(form, "username") or None account = await run_in_threadpool(service.create_managed_account, username) return _render_public_home(service, settings, account, None) @app.get("/m/{management_slug}", response_class=HTMLResponse) def manage_magic(management_slug: str) -> str: account = store.get_account_by_management_slug(management_slug) if not account or account["disabled"]: raise HTTPException(status_code=404, detail="not found") return _render_management_page(service, settings, management_slug, account) @app.get("/d/{claim_secret}", response_class=HTMLResponse) def manage_domain_claim(claim_secret: str) -> str: challenge = store.get_domain_challenge_by_secret(claim_secret) if not challenge: raise HTTPException(status_code=404, detail="not found") return _render_public_home(service, settings, None, None, challenge) @app.post("/m/{management_slug}/delete") def delete_magic(management_slug: str) -> Response: account = store.get_account_by_management_slug(management_slug) if not account or account["disabled"]: raise HTTPException(status_code=404, detail="not found") hostname = str(account["hostname"]) try: service.delete_account(hostname, "management-link") except Exception: return PlainTextResponse("DNS delete failed; hostname was not deleted", status_code=500) return HTMLResponse( _page( "DynDNS deleted", """

Deleted

Hostname deleted.

The update URL no longer works.

Home
""", ) ) @app.post("/request-domain", response_class=HTMLResponse) async def request_domain(request: Request) -> str: form = _parse_form(await request.body()) try: challenge = await run_in_threadpool( service.create_domain_challenge, _first_form_value(form, "domain"), ) except sqlite3.IntegrityError as exc: raise HTTPException(status_code=409, detail="domain is already verified") from exc return _render_public_home(service, settings, None, None, challenge) @app.post("/verify-domain", response_class=HTMLResponse) async def verify_domain(request: Request) -> str: form = _parse_form(await request.body()) domain = normalize_domain(_first_form_value(form, "domain")) claim_secret = _first_form_value(form, "claim_secret") found, challenge = await run_in_threadpool(service.verify_domain, domain, claim_secret) if not found: message = "TXT record not found yet. DNS propagation can take a few minutes." return _render_public_home(service, settings, None, message, challenge) return _render_public_home(service, settings, None, "Domain verified. You can now create a hostname under it.", challenge) @app.post("/accounts", response_class=HTMLResponse) async def self_service_account(request: Request) -> str: form = _parse_form(await request.body()) mode = _first_form_value(form, "mode") username = _first_form_value(form, "username") or None if mode == "managed": account = await run_in_threadpool(service.create_managed_account, username) else: claim_secret = _first_form_value(form, "claim_secret") account = await run_in_threadpool( service.create_custom_account, _first_form_value(form, "hostname"), claim_secret, username, ) return _render_public_home(service, settings, account, None) @app.get("/admin", response_class=HTMLResponse) def admin(_: None = Depends(authenticate_admin)) -> str: return _render_admin_page(service, settings, store.list_accounts(), None) @app.post("/admin/accounts") async def admin_create_account(request: Request, _: None = Depends(authenticate_admin)) -> Response: form = _parse_form(await request.body()) require_admin_csrf(settings, _first_form_value(form, "csrf")) hostname = normalize_hostname(_first_form_value(form, "hostname"), settings) username = _first_form_value(form, "username") or None if not hostname: raise HTTPException(status_code=400, detail="hostname is required") if settings.allowed_hosts and hostname not in settings.allowed_hosts: raise HTTPException(status_code=403, detail="hostname is not allowed") try: account = store.create_account(hostname, username) except sqlite3.IntegrityError as exc: raise HTTPException(status_code=409, detail="hostname already exists") from exc return HTMLResponse(_render_admin_page(service, settings, store.list_accounts(), account)) @app.post("/admin/accounts/delete") async def admin_delete_account(request: Request, _: None = Depends(authenticate_admin)) -> Response: form = _parse_form(await request.body()) require_admin_csrf(settings, _first_form_value(form, "csrf")) hostname = _first_form_value(form, "hostname") if hostname: try: service.delete_account(hostname, "admin") except Exception: return PlainTextResponse("DNS delete failed; hostname was not deleted", status_code=500) return RedirectResponse("/admin", status_code=303) @app.get("/records") def records(_: None = Depends(authenticate_admin)) -> list[dict[str, str | None]]: return store.list_records() @app.get("/events") def events(_: None = Depends(authenticate_admin)) -> list[dict[str, str | None]]: return store.list_update_events() @app.get("/healthz", response_class=PlainTextResponse) def healthz() -> str: return "ok" @app.get("/favicon.ico", include_in_schema=False) def favicon() -> FileResponse: return FileResponse(ASSET_DIR / "favicon.ico", media_type="image/x-icon") @app.get("/logo.svg", include_in_schema=False) def logo_svg() -> FileResponse: return FileResponse(ASSET_DIR / "logo.svg", media_type="image/svg+xml") @app.get("/logo.png", include_in_schema=False) def logo_png() -> FileResponse: return FileResponse(ASSET_DIR / "logo.png", media_type="image/png") @app.get("/u/{update_slug}", response_class=PlainTextResponse, response_model=None) def slug_update( update_slug: str, request: Request, myip: str | None = Query(default=None), ipaddr: str | None = Query(default=None), myipv6: str | None = Query(default=None), ip6addr: str | None = Query(default=None), ) -> str | PlainTextResponse: account = store.get_account_by_slug(update_slug) if not account or account["disabled"]: raise HTTPException(status_code=404, detail="not found") return apply_update(request, str(account["hostname"]), myip, ipaddr, myipv6, ip6addr) @app.get("/nic/update", response_class=PlainTextResponse, response_model=None) def update( request: Request, credentials: tuple[str | None, str | None] = Depends(authenticate_update), hostname: str | None = Query(default=None), domain: str | None = Query(default=None), myip: str | None = Query(default=None), ipaddr: str | None = Query(default=None), myipv6: str | None = Query(default=None), ip6addr: str | None = Query(default=None), ) -> str | PlainTextResponse: host = (hostname or domain or "").strip().lower().rstrip(".") if not host: return PlainTextResponse("nohost", status_code=400) if settings.allowed_hosts and host not in settings.allowed_hosts: return PlainTextResponse("nohost", status_code=403) supplied_user, supplied_token = credentials global_secret_ok = settings.shared_secret and supplied_token and secrets.compare_digest(supplied_token, settings.shared_secret) if not global_secret_ok and not store.verify_account(host, supplied_user, supplied_token): return PlainTextResponse("badauth", status_code=401) return apply_update(request, host, myip, ipaddr, myipv6, ip6addr) return app async def _cleanup_loop(store: DdnsStore, settings: DdnsSettings) -> None: while True: await asyncio.sleep(settings.cleanup_interval_seconds) await run_in_threadpool( store.cleanup, settings.cleanup_challenge_hours, settings.cleanup_unused_account_hours, ) def _parse_form(body: bytes) -> dict[str, list[str]]: return urllib.parse.parse_qs(body.decode("utf-8"), keep_blank_values=True) def _first_form_value(form: dict[str, list[str]], key: str) -> str: return form.get(key, [""])[0].strip() def _render_public_home( service: DdnsService, settings: DdnsSettings, created_account: dict[str, str] | None, message: str | None, challenge: dict[str, str | None] | None = None, ) -> str: suffix = html.escape(settings.hostname_suffix or "your DynDNS domain") created_html = _credentials_panel(service, created_account) message_html = _message_band(message) if message and not challenge else "" challenge_html = _custom_domain_flow(service, challenge, message) return _page( "router-dyndns", f"""
{_top_nav()}

FRITZ!Box compatible

DynDNS without the account ceremony.

Create a secure update endpoint under {suffix}. Provider hostnames are anonymous; custom domains require DNS proof.

{message_html} {created_html}

Free DynDNS hostname

Create a router update URL

Generate a random hostname, paste the update URL into your router, and keep the management link private.

{challenge_html}
""", ) def _top_nav(label: str | None = None, links: str = "") -> str: account_html = "" if label: account_html = f'{html.escape(label)}' if not links: account_html += 'Domain setup' return f""" """ def _custom_domain_flow(service: DdnsService, challenge: dict[str, str | None] | None, message: str | None) -> str: challenge_html = "" create_form = "" if challenge: domain = html.escape(str(challenge["domain"])) token = str(challenge["token"]) claim_secret = html.escape(str(challenge.get("claim_secret") or "")) verification_name = service.verification_name(str(challenge["domain"])) claim_url = service.domain_claim_url(challenge) status = html.escape(message or "Add this TXT record at your DNS provider, then press the check button.") challenge_html = f"""
2

Add the TXT record

{status}

{_copy_row("TXT name", verification_name)} {_copy_row("TXT value", token)} {_copy_row("Private claim link", claim_url)}
""" if challenge.get("verified_at"): create_form = f"""
3

Create router hostname

""" else: create_form = """
3

Create router hostname

This unlocks after the TXT record is visible in public DNS.

""" return f"""

Custom domain

Enter domain, add TXT, check DNS.

We only issue router credentials after the DNS record proves you control the domain.

1

Enter domain

{challenge_html} {create_form}
""" def _credentials_panel(service: DdnsService, created_account: dict[str, str] | None) -> str: if not created_account: return "" hostname = created_account["hostname"] username = created_account["username"] password = created_account["password"] update_url = service.fritz_update_url(created_account) management_url = service.magic_management_url(created_account) return f"""

Account generated

{html.escape(hostname)}

Save this now. The password and management link are only shown on this screen.

{_copy_row("Update-URL:", update_url)} {_copy_row("Domainnamen:", hostname)} {_copy_row("Benutzername:", username)} {_copy_row("Kennwort:", password)} {_copy_row("Magic management link:", management_url)}
""" def _created_account_panel(service: DdnsService, created_account: dict[str, str] | None) -> str: return _credentials_panel(service, created_account) def _copy_row(label: str, value: str) -> str: safe_label = html.escape(label) safe_value = html.escape(value) return f"""
{safe_label} {safe_value}
""" def _message_band(message: str | None) -> str: if not message: return "" return f"""

{html.escape(message)}

""" def _render_management_page(service: DdnsService, settings: DdnsSettings, management_slug: str, account: dict[str, str | int | None]) -> str: account_for_url = { "hostname": str(account["hostname"]), "username": str(account["username"]), "password": "", "update_slug": str(account["update_slug"]), "management_slug": management_slug, } update_url = service.fritz_update_url(account_for_url) hostname = html.escape(str(account["hostname"])) username = html.escape(str(account["username"])) ipv4 = html.escape(str(account.get("ipv4") or "-")) ipv6 = html.escape(str(account.get("ipv6") or "-")) updated = html.escape(str(account.get("updated_at") or "Never")) return _page( "DynDNS management", f"""
{_top_nav()}

Magic management

{hostname}

Anyone with this link can manage or delete this hostname. Keep it private.

Home

Router settings

FRITZ!Box fields

Use these values in the custom DynDNS provider fields.

{_copy_row("Update-URL:", update_url)} {_copy_row("Domainnamen:", hostname)} {_copy_row("Benutzername:", username)} {_copy_row("Kennwort:", "unchanged; only shown when generated")}

Current address

Status

IPv4{ipv4}
IPv6{ipv6}
Updated{updated}

Danger zone

Delete hostname

This removes the account and DNS records. The router update URL will stop working.

""", ) def _render_admin_page( service: DdnsService, settings: DdnsSettings, accounts: list[dict[str, str | int | None]], created_account: dict[str, str] | None, ) -> str: csrf = html.escape(admin_csrf_token(settings)) rows = "\n".join(_account_row(account, csrf=csrf) for account in accounts) or """ No accounts yet. """ suffix_help = ( f"Short names are expanded under {html.escape(settings.hostname_suffix)}." if settings.hostname_suffix else "Use a full hostname, for example home.example.net." ) return _page( "DynDNS Admin", f"""
{_top_nav("Operator console", 'RecordsEvents')}

DynDNS aktiv

FRITZ!Box accounts

Create, inspect, and retire router update credentials.

{_created_account_panel(service, created_account)}

Generate account

Router credentials

Generated accounts produce the FRITZ!Box fields: Update-URL, Domainnamen, Benutzername, Kennwort.

{suffix_help}

Accounts

Active credentials

{rows}
DomainUserIPv4IPv6Updated
""", ) def _account_row(account: dict[str, str | int | None], csrf: str = "") -> str: hostname = html.escape(str(account["hostname"])) username = html.escape(str(account["username"])) ipv4 = html.escape(str(account.get("ipv4") or "-")) ipv6 = html.escape(str(account.get("ipv6") or "-")) updated = html.escape(str(account.get("updated_at") or "Never")) return f""" {hostname}{username}{ipv4}{ipv6}{updated}
""" def _page(title: str, body: str) -> str: return f""" {html.escape(title)} {body} """