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 starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.types import ASGIApp, Message, Receive, Scope, Send
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,
rate_limit_bucket,
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.",
},
]
ASSET_DIR = Path(__file__).with_name("assets")
APP_JS = """
(() => {
const applyTheme = () => {
const theme = window.matchMedia("(prefers-color-scheme: dark)").matches ? "dark" : "light";
document.documentElement.setAttribute("data-bs-theme", theme);
document.documentElement.style.colorScheme = theme;
};
applyTheme();
window.matchMedia("(prefers-color-scheme: dark)").addEventListener("change", applyTheme);
})();
document.addEventListener("click", async (event) => {
const button = event.target.closest("[data-copy]");
if (!button) return;
try {
await navigator.clipboard.writeText(button.dataset.copy || "");
const old = button.textContent;
button.textContent = "Copied";
setTimeout(() => {
button.textContent = old;
}, 1200);
} catch (_) {}
});
document.querySelectorAll("[data-admin-table]").forEach((tableShell) => {
const table = tableShell.querySelector("table");
const tbody = tableShell.querySelector("tbody");
const search = tableShell.querySelector("[data-table-search]");
const pageSize = tableShell.querySelector("[data-table-page-size]");
const count = tableShell.querySelector("[data-table-count]");
const pagination = tableShell.querySelector("[data-table-pagination]");
if (!table || !tbody || !search || !pageSize || !count || !pagination) return;
const headerCount = table.querySelectorAll("thead th").length || 1;
const originalRows = Array.from(tbody.querySelectorAll("tr"));
const dataRows = originalRows.filter((row) => !row.classList.contains("empty-row"));
let currentPage = 1;
const emptyRow = (message) => {
const row = document.createElement("tr");
row.className = "empty-row";
const cell = document.createElement("td");
cell.className = "empty";
cell.colSpan = headerCount;
cell.textContent = message;
row.append(cell);
return row;
};
const visibleRows = () => {
const query = search.value.trim().toLowerCase();
if (!query) return dataRows;
return dataRows.filter((row) => row.textContent.toLowerCase().includes(query));
};
const renderButton = (label, page, disabled = false, active = false) => {
const item = document.createElement("li");
item.className = `page-item${disabled ? " disabled" : ""}${active ? " active" : ""}`;
const button = document.createElement("button");
button.type = "button";
button.className = "page-link";
button.textContent = label;
button.disabled = disabled;
button.addEventListener("click", () => {
currentPage = page;
render();
});
item.append(button);
return item;
};
const renderPagination = (totalPages) => {
pagination.replaceChildren();
if (totalPages <= 1) return;
pagination.append(renderButton("Previous", Math.max(1, currentPage - 1), currentPage === 1));
for (let page = 1; page <= totalPages; page += 1) {
if (totalPages > 7 && page !== 1 && page !== totalPages && Math.abs(page - currentPage) > 1) {
if (page === 2 || page === totalPages - 1) pagination.append(renderButton("...", currentPage, true));
continue;
}
pagination.append(renderButton(String(page), page, false, page === currentPage));
}
pagination.append(renderButton("Next", Math.min(totalPages, currentPage + 1), currentPage === totalPages));
};
const render = () => {
const rows = visibleRows();
const size = Number.parseInt(pageSize.value, 10) || 25;
const totalPages = Math.max(1, Math.ceil(rows.length / size));
currentPage = Math.min(currentPage, totalPages);
const start = (currentPage - 1) * size;
const pageRows = rows.slice(start, start + size);
tbody.replaceChildren(...(pageRows.length ? pageRows : [emptyRow(dataRows.length ? "No matching rows." : "No rows yet.")]));
const shownStart = rows.length ? start + 1 : 0;
const shownEnd = rows.length ? Math.min(start + size, rows.length) : 0;
count.textContent = `${shownStart}-${shownEnd} of ${rows.length}`;
renderPagination(totalPages);
};
search.addEventListener("input", () => {
currentPage = 1;
render();
});
pageSize.addEventListener("change", () => {
currentPage = 1;
render();
});
render();
});
""".strip()
class RequestBodyTooLargeError(Exception):
pass
class RequestBodyLimitMiddleware:
def __init__(self, app: ASGIApp, max_body_bytes: int) -> None:
self.app = app
self.max_body_bytes = max_body_bytes
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
received = 0
async def limited_receive() -> Message:
nonlocal received
message = await receive()
if message["type"] == "http.request":
received += len(message.get("body", b""))
if received > self.max_body_bytes:
raise RequestBodyTooLargeError
return message
try:
await self.app(scope, limited_receive, send)
except RequestBodyTooLargeError:
response = PlainTextResponse("request body too large", status_code=413)
await response(scope, receive, send)
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")
settings.validate_launch_ready()
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.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.effective_trusted_hosts)
app.add_middleware(RequestBodyLimitMiddleware, max_body_bytes=settings.max_request_body_bytes)
app.include_router(create_api_router(settings, store, service))
@app.middleware("http")
async def security_middleware(request: Request, call_next):
try:
content_length = int(request.headers.get("content-length", "0") or "0")
except ValueError:
content_length = 0
if content_length > settings.max_request_body_bytes:
return PlainTextResponse("request body too large", status_code=413)
if is_rate_limited_path(request.url.path) or _is_admin_path(request.url.path):
limit = settings.admin_rate_limit_per_minute if _is_admin_path(request.url.path) else settings.rate_limit_per_minute
key = f"{client_ip(request, settings)}:{rate_limit_bucket(request.url.path)}"
allowed = await run_in_threadpool(store.allow_rate_limit, key, limit)
if not allowed:
return PlainTextResponse("rate limit exceeded", status_code=429)
response = await call_next(request)
if _is_secret_response(request):
response.headers["Cache-Control"] = "no-store"
response.headers["Pragma"] = "no-cache"
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"] = _content_security_policy(request.url.path)
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.get("/magic", include_in_schema=False)
def magic_form() -> RedirectResponse:
return RedirectResponse("/#create", status_code=303)
@app.post("/magic", response_class=HTMLResponse)
async def magic_account(request: Request) -> str:
form = await _request_form(request, settings)
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 = await _request_form(request, settings)
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 = await _request_form(request, settings)
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 = await _request_form(request, settings)
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, include_in_schema=False)
def admin(_: None = Depends(authenticate_admin)) -> str:
return _render_admin_page(
service,
settings,
store.list_accounts(),
store.list_domain_challenges(),
store.list_cleanup_runs(),
store.list_update_events(25),
None,
)
@app.post("/admin/accounts/delete", include_in_schema=False)
async def admin_delete_account(request: Request, _: None = Depends(authenticate_admin)) -> Response:
form = await _request_form(request, settings)
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.post("/admin/accounts/rotate", include_in_schema=False)
async def admin_rotate_account(request: Request, _: None = Depends(authenticate_admin)) -> Response:
form = await _request_form(request, settings)
require_admin_csrf(settings, _first_form_value(form, "csrf"))
hostname = normalize_hostname(_first_form_value(form, "hostname"), settings, expand_suffix=False)
action = _first_form_value(form, "action")
account = store.get_account_by_hostname(hostname)
if not account:
raise HTTPException(status_code=404, detail="hostname not found")
notice = None
if action == "update":
rotated = store.rotate_update_slug(hostname)
notice = _admin_secret_panel("Update URL rotated", [_copy_row("Update-URL:", service.fritz_update_url(rotated))]) if rotated else None
elif action == "management":
rotated = store.rotate_management_slug(hostname)
notice = _admin_secret_panel("Private status page rotated", [_copy_row("Private status page:", service.magic_management_url(rotated))]) if rotated else None
elif action == "password":
rotated = store.rotate_password(hostname)
if rotated:
notice = _admin_secret_panel(
"Router password rotated",
[
_copy_row("Domainnamen:", str(rotated["hostname"]), "Public"),
_copy_row("One-box update URL:", service.credentialed_update_url(rotated), "Secret"),
_copy_row("Benutzername:", str(rotated["username"]), "Public"),
_copy_row("Kennwort:", str(rotated["password"]), "Secret"),
],
)
else:
raise HTTPException(status_code=400, detail="invalid rotation action")
return HTMLResponse(
_render_admin_page(
service,
settings,
store.list_accounts(),
store.list_domain_challenges(),
store.list_cleanup_runs(),
store.list_update_events(25),
notice,
)
)
@app.post("/admin/cleanup", include_in_schema=False)
async def admin_run_cleanup(request: Request, _: None = Depends(authenticate_admin)) -> Response:
form = await _request_form(request, settings)
require_admin_csrf(settings, _first_form_value(form, "csrf"))
result = await run_in_threadpool(store.cleanup, settings.cleanup_challenge_hours, settings.cleanup_unused_account_hours)
notice = _admin_secret_panel(
"Cleanup completed",
[
_copy_row("Domain claims removed", str(result["domain_challenges"])),
_copy_row("Unused hostnames removed", str(result["unused_accounts"])),
],
)
return HTMLResponse(
_render_admin_page(
service,
settings,
store.list_accounts(),
store.list_domain_challenges(),
store.list_cleanup_runs(),
store.list_update_events(25),
notice,
)
)
@app.get("/records", include_in_schema=False)
def records(_: None = Depends(authenticate_admin)) -> list[dict[str, str | None]]:
return store.list_records()
@app.get("/events", include_in_schema=False)
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("/app.js", include_in_schema=False)
def app_js() -> Response:
return Response(APP_JS, media_type="application/javascript")
@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)
async def _request_form(request: Request, settings: DdnsSettings) -> dict[str, list[str]]:
body = await request.body()
if len(body) > settings.max_request_body_bytes:
raise HTTPException(status_code=413, detail="request body too large")
return _parse_form(body)
def _first_form_value(form: dict[str, list[str]], key: str) -> str:
return form.get(key, [""])[0].strip()
def _is_admin_path(path: str) -> bool:
return path == "/admin" or path.startswith("/admin/") or path in {"/records", "/events"}
def _is_secret_response(request: Request) -> bool:
path = request.url.path
if _is_admin_path(path) or path.startswith(("/m/", "/d/", "/u/")):
return True
if path.startswith(("/api/v1/hostnames/", "/api/v1/domains/")):
return request.method in {"POST", "DELETE"}
return (request.method == "POST" and path in {"/magic", "/request-domain", "/verify-domain", "/accounts"}) or path == "/nic/update"
def _content_security_policy(path: str) -> str:
if path in {"/docs", "/redoc", "/openapi.json"}:
return (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://fonts.googleapis.com; "
"img-src 'self' data: https://fastapi.tiangolo.com; "
"font-src 'self' https://cdn.jsdelivr.net https://fonts.gstatic.com; "
"connect-src 'self'; "
"form-action 'self'; "
"frame-ancestors 'none'"
)
return (
"default-src 'self'; "
"script-src 'self'; "
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
"img-src 'self' data:; "
"font-src 'self' https://cdn.jsdelivr.net; "
"connect-src 'self'; "
"form-action 'self'; "
"frame-ancestors 'none'"
)
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 signups.
Create a secure hostname under {suffix}. Use it directly, or point your own subdomain at it with a CNAME.
{message_html}
{created_html}
Free DynDNS hostname
Create your DynDNS target
Generate a random hostname, paste the update URL into your router, and keep the private status page. Your own DNS name can point to this hostname with a CNAME.
{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"""
Skip to content
"""
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 check when it has propagated.")
txt_bundle = f"TXT name: {verification_name}\nTXT value: {token}"
challenge_html = f"""
2
Add the TXT record
{status}
TXT name and value are public DNS records. The private claim link is secret.
{_copy_button("Copy TXT record", txt_bundle)}
{_copy_row("TXT name", verification_name, "Public")}
{_copy_row("TXT value", token, "Public")}
{_copy_row("Private claim link", claim_url, "Secret")}
"""
if challenge.get("verified_at"):
create_form = f"""
"""
else:
create_form = """
3
Create router hostname
This unlocks after the TXT record is visible in public DNS.
"""
return f"""
Direct custom-domain publishing
Verify DNS ownership
Use this only when RouterPulse should publish A/AAAA records directly inside your domain. If you just want your own subdomain, generate a hostname above and create a CNAME to it.
{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)
credentialed_update_url = service.credentialed_update_url(created_account)
management_url = service.magic_management_url(created_account)
cname_record = f"your-subdomain.example.com. CNAME {hostname}."
fritzbox_fields = _fritzbox_fields(update_url, hostname, username, password, management_url)
return f"""
Credentials generated
{html.escape(hostname)}
Save this now. The password, private status page, and one-box update URL are only shown on this screen. To use your own DNS name, create a CNAME from that name to this hostname.
{_secret_notice()}
{_copy_button("Copy FRITZ!Box fields", fritzbox_fields)}
{_copy_button("Copy one-box update URL", credentialed_update_url)}
{_copy_button("Copy CNAME record", cname_record)}
{_copy_row("Update-URL:", update_url, "Secret")}
{_copy_row("One-box update URL:", credentialed_update_url, "Secret")}
{_copy_row("Domainnamen:", hostname, "Public")}
{_copy_row("CNAME target:", hostname, "Public")}
{_copy_row("Benutzername:", username, "Public")}
{_copy_row("Kennwort:", password, "Secret")}
{_copy_row("Private status page:", management_url, "Secret")}
"""
def _admin_secret_panel(title: str, rows: list[str]) -> str:
return f"""
Admin action
{html.escape(title)}
Copy this now. Secret values are only shown in this response.
{"".join(rows)}
"""
def _copy_row(label: str, value: str, sensitivity: str | None = None) -> str:
safe_label = html.escape(label)
safe_value = html.escape(value)
badge = ""
if sensitivity:
safe_sensitivity = html.escape(sensitivity)
badge_class = "text-bg-warning" if sensitivity.lower() == "secret" else "text-bg-secondary"
badge = f'{safe_sensitivity} '
return f"""
{safe_label} {badge}
{safe_value}
Copy
"""
def _copy_button(label: str, value: str) -> str:
safe_label = html.escape(label)
safe_value = html.escape(value)
return f'{safe_label} '
def _fritzbox_fields(
update_url: str,
hostname: str,
username: str,
password: str,
management_url: str | None = None,
) -> str:
lines = [
f"Update-URL: {update_url}",
f"Domainnamen: {hostname}",
f"CNAME target: {hostname}",
f"Benutzername: {username}",
f"Kennwort: {password}",
]
if management_url:
lines.append(f"Private status page: {management_url}")
return "\n".join(lines)
def _secret_notice() -> str:
return """
Public: Domainnamen and CNAME target can be shared in DNS.
Secret: Update-URL, one-box update URL, Kennwort, private status page, and domain claim links control the hostname.
"""
def _message_band(message: str | None) -> str:
if not message:
return ""
return f"""
"""
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"))
fritzbox_fields = _fritzbox_fields(update_url, hostname, username, "unchanged; only shown when generated")
cname_record = f"your-subdomain.example.com. CNAME {hostname}."
history_rows = "\n".join(_user_event_row(event) for event in service.store.list_update_events_for_hostname(str(account["hostname"]), 20)) or """
No router updates yet.
"""
return _page(
"DynDNS management",
f"""
{_top_nav()}
Magic management
{hostname}
Anyone with this link can manage or delete this hostname. Keep it private.
Back home
Router settings
FRITZ!Box fields
Use the Update-URL directly if your router only has one URL field. The username and password are shown only when generated or rotated. If you use your own DNS name, point it at this hostname with a CNAME.
{_secret_notice()}
{_copy_button("Copy FRITZ!Box fields", fritzbox_fields)}
{_copy_button("Copy CNAME record", cname_record)}
{_copy_row("Update-URL:", update_url, "Secret")}
{_copy_row("Domainnamen:", hostname, "Public")}
{_copy_row("CNAME target:", hostname, "Public")}
{_copy_row("Benutzername:", username, "Public")}
{_copy_row("Kennwort:", "unchanged; only shown when generated", "Secret")}
IPv4 {ipv4}
IPv6 {ipv6}
Updated {updated}
History
Router updates
Recent IP changes and failed DNS publishes for this hostname.
Time Status IPv4 IPv6 Detail
{history_rows}
Danger zone
Delete hostname
This removes the hostname 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]],
domain_challenges: list[dict[str, str | None]],
cleanup_runs: list[dict[str, str | int]],
update_events: list[dict[str, str | None]],
notice_html: str | None,
) -> str:
csrf = html.escape(admin_csrf_token(settings))
rows = "\n".join(_account_row(account, csrf=csrf) for account in accounts) or """
No hostnames yet.
"""
domain_rows = "\n".join(_domain_claim_row(challenge) for challenge in domain_challenges) or """
No domain claims yet.
"""
cleanup_rows = "\n".join(_cleanup_row(run) for run in cleanup_runs) or """
No cleanup runs yet.
"""
event_rows = "\n".join(_event_row(event) for event in update_events) or """
No update events yet.
"""
return _page(
"DynDNS Admin",
f"""
{_top_nav("Operator console", 'Records Events ')}
DynDNS aktiv
Operator console
Inspect hostnames, verify domain claims, rotate links, run cleanup, and review router update events.
{notice_html or ""}
Service health
Scheduled cleanup
Cleanup runs every {html.escape(str(settings.cleanup_interval_seconds))} seconds. It removes unverified domain claims after {html.escape(str(settings.cleanup_challenge_hours))} hours and never-used generated hostnames after {html.escape(str(settings.cleanup_unused_account_hours))} hours.
Hostnames
Active credentials
{_admin_table_toolbar("hostnames", "Search hostnames")}
{_admin_table_pagination()}
Domain claims
Verification lifecycle
{_admin_table_toolbar("domain-claims", "Search domain claims")}
{_admin_table_pagination()}
{_admin_table_toolbar("cleanup-runs", "Search cleanup runs")}
{_admin_table_pagination()}
{_admin_table_toolbar("update-events", "Search update events")}
{_admin_table_pagination()}
""",
)
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 _admin_table_toolbar(table_id: str, search_label: str) -> str:
safe_id = html.escape(table_id)
safe_label = html.escape(search_label)
return f"""
"""
def _admin_table_pagination() -> str:
return """
"""
def _domain_claim_row(challenge: dict[str, str | None]) -> str:
domain = html.escape(str(challenge["domain"]))
created = html.escape(str(challenge.get("created_at") or "-"))
verified = html.escape(str(challenge.get("verified_at") or "-"))
status = "Verified" if challenge.get("verified_at") else "Pending"
return f"{domain} {status} {created} {verified} "
def _cleanup_row(run: dict[str, str | int]) -> str:
created = html.escape(str(run.get("created_at") or "-"))
domains = html.escape(str(run.get("domain_challenges_deleted") or 0))
accounts = html.escape(str(run.get("unused_accounts_deleted") or 0))
return f"{created} {domains} {accounts} "
def _event_row(event: dict[str, str | None]) -> str:
created = html.escape(str(event.get("created_at") or "-"))
hostname = html.escape(str(event.get("hostname") or "-"))
status = html.escape(str(event.get("status") or "-"))
ipv4 = html.escape(str(event.get("ipv4") or "-"))
ipv6 = html.escape(str(event.get("ipv6") or "-"))
detail = html.escape(str(event.get("detail") or "-"))
return f"{created} {hostname} {status} {ipv4} {ipv6} {detail} "
def _user_event_row(event: dict[str, str | None]) -> str:
created = html.escape(str(event.get("created_at") or "-"))
status = html.escape(str(event.get("status") or "-"))
ipv4 = html.escape(str(event.get("ipv4") or "-"))
ipv6 = html.escape(str(event.get("ipv6") or "-"))
detail = html.escape(str(event.get("detail") or "-"))
return f"{created} {status} {ipv4} {ipv6} {detail} "
def _page(title: str, body: str) -> str:
return f"""
{html.escape(title)}
{body}
"""