From 3ad0a9503d7aadb7feaa1370773928ad72912271 Mon Sep 17 00:00:00 2001 From: StasonJatham Date: Wed, 20 May 2026 13:48:24 +0200 Subject: [PATCH] Initial router DynDNS service --- .env.example | 14 + .gitignore | 10 + Caddyfile.example | 4 + Dockerfile | 9 + LICENSE | 21 + README.md | 162 +++++++ docker-compose.yml | 18 + pyproject.toml | 41 ++ router_dyndns/__init__.py | 2 + router_dyndns/cli.py | 43 ++ router_dyndns/ddns.py | 784 +++++++++++++++++++++++++++++++++ router_dyndns/ddns_api.py | 260 +++++++++++ router_dyndns/ddns_dns.py | 155 +++++++ router_dyndns/ddns_models.py | 101 +++++ router_dyndns/ddns_security.py | 143 ++++++ router_dyndns/ddns_service.py | 251 +++++++++++ router_dyndns/ddns_store.py | 506 +++++++++++++++++++++ tests/test_ddns.py | 568 ++++++++++++++++++++++++ 18 files changed, 3092 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Caddyfile.example create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 README.md create mode 100644 docker-compose.yml create mode 100644 pyproject.toml create mode 100644 router_dyndns/__init__.py create mode 100644 router_dyndns/cli.py create mode 100644 router_dyndns/ddns.py create mode 100644 router_dyndns/ddns_api.py create mode 100644 router_dyndns/ddns_dns.py create mode 100644 router_dyndns/ddns_models.py create mode 100644 router_dyndns/ddns_security.py create mode 100644 router_dyndns/ddns_service.py create mode 100644 router_dyndns/ddns_store.py create mode 100644 tests/test_ddns.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..0efd832 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +DDNS_ADMIN_PASSWORD=change-me-long-random-admin-password +DDNS_SESSION_SECRET=change-me-long-random-session-secret +DDNS_PUBLIC_BASE_URL=https://ddns.example.net +DDNS_HOSTNAME_SUFFIX=ddns.example.net +DDNS_DATABASE=/data/ddns.sqlite3 +DDNS_DNS_PROVIDER=cloudflare +DDNS_DNS_ZONES=ddns.example.net +DDNS_REQUIRE_DNS_PROVIDER=1 +DDNS_CLOUDFLARE_API_TOKEN=change-me-cloudflare-token +DDNS_CLOUDFLARE_ZONE_ID=change-me-cloudflare-zone-id +DDNS_TTL=60 +DDNS_RATE_LIMIT_PER_MINUTE=60 +DDNS_CLEANUP_CHALLENGE_HOURS=72 +DDNS_TRUSTED_PROXY_IPS=127.0.0.1,::1 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6a3e92a --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.env +.venv/ +__pycache__/ +.pytest_cache/ +.ruff_cache/ +*.sqlite3 +*.db +dist/ +build/ +*.egg-info/ diff --git a/Caddyfile.example b/Caddyfile.example new file mode 100644 index 0000000..0b9961c --- /dev/null +++ b/Caddyfile.example @@ -0,0 +1,4 @@ +ddns.example.net { + encode zstd gzip + reverse_proxy ddns:8080 +} diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..cc1c5b9 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.12-slim + +WORKDIR /app +COPY pyproject.toml README.md /app/ +COPY router_dyndns /app/router_dyndns +RUN pip install --no-cache-dir . + +EXPOSE 8080 +CMD ["router-dyndns", "serve", "--host", "0.0.0.0", "--port", "8080"] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..14fac91 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..34d5287 --- /dev/null +++ b/README.md @@ -0,0 +1,162 @@ +# router-dyndns + +A small self-hosted DynDNS service for FRITZ!Box and compatible routers. + +It gives users FRITZ!Box-ready DynDNS settings without requiring a full account for provider-owned hostnames. For custom domains, users verify domain ownership with a DNS TXT challenge before creating hostnames. + +## Features + +- FastAPI app with `/docs`, `/redoc`, and `/openapi.json`. +- Anonymous magic URLs for provider-owned hostnames. +- Optional user sessions for custom-domain verification and hostname quotas. +- FRITZ!Box-compatible update URLs. +- SQLite persistence with WAL mode. +- SQLite-backed rate limiting for public creation and update endpoints. +- Cloudflare DNS publishing. +- RFC 2136 DNS publishing. +- DNS publish-before-store update behavior. +- Per-host update locking inside one server process. +- Custom-domain TXT verification. +- Trusted proxy configuration for `X-Forwarded-For`. +- Docker and Caddy examples. + +## Install + +```bash +python3 -m venv .venv +source .venv/bin/activate +pip install -e ".[dev]" +``` + +Run quality checks: + +```bash +ruff check router_dyndns tests +pytest +``` + +## Run Locally + +```bash +export DDNS_ADMIN_PASSWORD='long-random-admin-password' +export DDNS_SESSION_SECRET='another-long-random-secret' +export DDNS_PUBLIC_BASE_URL='https://ddns.example.net' +export DDNS_HOSTNAME_SUFFIX='ddns.example.net' +export DDNS_DATABASE='./ddns.sqlite3' +export DDNS_DNS_PROVIDER='cloudflare' +export DDNS_DNS_ZONES='ddns.example.net' +export DDNS_REQUIRE_DNS_PROVIDER=1 +export DDNS_CLOUDFLARE_API_TOKEN='cloudflare-api-token' +export DDNS_CLOUDFLARE_ZONE_ID='cloudflare-zone-id' +export DDNS_TRUSTED_PROXY_IPS='127.0.0.1,::1' + +router-dyndns serve --host 0.0.0.0 --port 8080 +``` + +Open: + +- Customer UI: `https://ddns.example.net/` +- Admin UI: `https://ddns.example.net/admin` +- Swagger UI: `https://ddns.example.net/docs` +- ReDoc: `https://ddns.example.net/redoc` + +Admin auth uses HTTP Basic auth. Use any username and `DDNS_ADMIN_PASSWORD` as the password. + +## FRITZ!Box Setup + +Users receive these fields: + +- `Update-URL` +- `Domainnamen` +- `Benutzername` +- `Kennwort` + +The generated update URL uses FRITZ!Box placeholders: + +```text +/u/?myip=&myipv6= +``` + +The compatibility endpoint is also available: + +```text +/nic/update?hostname=&myip=&myipv6=&username=&password= +``` + +The `/u/` endpoint is safer for public service use because the user cannot alter the hostname in the router request. + +## HTTP API + +The API is versioned under `/api/v1`. + +Core endpoints: + +- `POST /api/v1/hostnames/magic`: create a provider-owned random hostname. +- `GET /api/v1/management/{management_slug}`: inspect a hostname by private management link. +- `DELETE /api/v1/management/{management_slug}`: delete a hostname and DNS records. +- `POST /api/v1/domains/challenges`: create a TXT challenge. Requires login. +- `POST /api/v1/domains/verify`: verify a TXT challenge. Requires login. +- `POST /api/v1/hostnames/custom`: create a custom hostname below a verified and publishable domain. Requires login. +- `GET /api/v1/updates/{update_slug}`: JSON update endpoint. + +Example: + +```bash +curl -sS https://ddns.example.net/api/v1/hostnames/magic \ + -H 'content-type: application/json' \ + -d '{"username":"home-router"}' +``` + +## DNS Publishing + +For public service use, configure a real DNS backend and set: + +```bash +export DDNS_REQUIRE_DNS_PROVIDER=1 +``` + +Cloudflare: + +```bash +export DDNS_DNS_PROVIDER=cloudflare +export DDNS_DNS_ZONES=ddns.example.net +export DDNS_CLOUDFLARE_API_TOKEN='cloudflare-api-token' +export DDNS_CLOUDFLARE_ZONE_ID='cloudflare-zone-id' +export DDNS_TTL=60 +``` + +RFC 2136: + +```bash +export DDNS_DNS_PROVIDER=rfc2136 +export DDNS_DNS_ZONES=ddns.example.net +export DDNS_RFC2136_SERVER=127.0.0.1 +export DDNS_RFC2136_ZONE=ddns.example.net +export DDNS_RFC2136_KEY_NAME=ddns-key +export DDNS_RFC2136_KEY_SECRET='base64-tsig-secret' +export DDNS_TTL=60 +``` + +`DDNS_DNS_ZONES` defines which DNS zones this service may publish. A user proving TXT control of a domain is not enough; the hostname must also be inside a zone your DNS backend can update. + +## Docker + +```bash +docker compose up -d --build +``` + +Copy `.env.example` to `.env` and set real secrets first. + +## Operations + +- Run behind HTTPS. +- Use long random values for `DDNS_ADMIN_PASSWORD` and `DDNS_SESSION_SECRET`. +- Configure `DDNS_TRUSTED_PROXY_IPS` only for reverse proxies that strip user-supplied forwarding headers. +- Back up the SQLite database. +- Run one Uvicorn worker with SQLite. The per-host update lock is process-local. + +SQLite backup example: + +```bash +sqlite3 /data/ddns.sqlite3 ".backup '/backups/router-dyndns.sqlite3'" +``` diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..c9ba372 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,18 @@ +services: + ddns: + build: . + restart: unless-stopped + env_file: + - .env + ports: + - "8080:8080" + volumes: + - ddns-data:/data + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/healthz', timeout=2).read()"] + interval: 30s + timeout: 5s + retries: 3 + +volumes: + ddns-data: diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c7958b9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,41 @@ +[build-system] +requires = ["hatchling>=1.25"] +build-backend = "hatchling.build" + +[project] +name = "router-dyndns" +version = "0.1.0" +description = "A small self-hosted DynDNS service for FRITZ!Box and compatible routers." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "dnspython>=2.6", + "fastapi>=0.111", + "pydantic>=2.7", + "uvicorn[standard]>=0.30", +] + +[project.optional-dependencies] +dev = [ + "httpx>=0.28", + "pytest>=8.2", + "ruff>=0.8", +] + +[project.scripts] +router-dyndns = "router_dyndns.cli:main" + +[tool.hatch.build.targets.wheel] +packages = ["router_dyndns"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] +ignore = ["E501"] + diff --git a/router_dyndns/__init__.py b/router_dyndns/__init__.py new file mode 100644 index 0000000..9cbb057 --- /dev/null +++ b/router_dyndns/__init__.py @@ -0,0 +1,2 @@ +"""Self-hosted DynDNS service for FRITZ!Box and compatible routers.""" + diff --git a/router_dyndns/cli.py b/router_dyndns/cli.py new file mode 100644 index 0000000..55a7812 --- /dev/null +++ b/router_dyndns/cli.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import argparse + +from .ddns import DdnsSettings, make_app + + +def main() -> None: + parser = argparse.ArgumentParser(prog="router-dyndns") + subparsers = parser.add_subparsers(dest="command", required=True) + + render_url = subparsers.add_parser("render-fritz-url", help="Render a FRITZ!Box custom DynDNS update URL.") + render_url.add_argument("--base-url", required=True, help="Public base URL of this DDNS service.") + render_url.add_argument("--host", required=True, help="DNS hostname configured in FRITZ!Box.") + render_url.add_argument("--include-ipv6", action="store_true") + + serve = subparsers.add_parser("serve", help="Run the DDNS HTTP service.") + serve.add_argument("--host", default="0.0.0.0") + serve.add_argument("--port", type=int, default=8080) + + args = parser.parse_args() + if args.command == "render-fritz-url": + print(_render_fritz_url(args.base_url, args.host, args.include_ipv6)) + elif args.command == "serve": + import uvicorn + + settings = DdnsSettings.from_env() + if not settings.admin_password: + raise SystemExit("DDNS_ADMIN_PASSWORD must be set.") + uvicorn.run(make_app(settings), host=args.host, port=args.port) + + +def _render_fritz_url(base_url: str, host: str, include_ipv6: bool) -> str: + base = base_url.rstrip("/") + url = f"{base}/nic/update?hostname=&myip=&username=&password=" + if include_ipv6: + url += "&myipv6=" + return url.replace("", host) + + +if __name__ == "__main__": + main() + diff --git a/router_dyndns/ddns.py b/router_dyndns/ddns.py new file mode 100644 index 0000000..b04c434 --- /dev/null +++ b/router_dyndns/ddns.py @@ -0,0 +1,784 @@ +from __future__ import annotations + +import html +import secrets +import sqlite3 +import urllib.parse +from typing import Annotated + +from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, Response +from fastapi.responses import 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, + normalize_email, + require_admin_csrf, + set_session_cookie, + verify_session_cookie, +) +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.", + }, +] + + +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 not settings.session_secret: + settings = settings.model_copy(update={"session_secret": settings.admin_password or 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) + service = DdnsService(settings, store) + 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, + ) + 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'; 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 current_user(request: Request) -> dict[str, str | int] | None: + user_id = verify_session_cookie(request.cookies.get("ddns_session"), settings.session_secret) + return store.get_user(user_id) if user_id is not None else None + + 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_magic_page(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, None) + return _render_magic_page(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.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", + '

Hostname deleted

The update URL no longer works.

', + ) + ) + + @app.post("/request-domain", response_class=HTMLResponse) + async def request_domain(request: Request) -> str: + user = await run_in_threadpool(current_user, request) + if not user: + raise HTTPException(status_code=401, detail="authentication required") + form = _parse_form(await request.body()) + try: + challenge = await run_in_threadpool(service.create_domain_challenge, _first_form_value(form, "domain"), int(user["id"])) + except sqlite3.IntegrityError as exc: + raise HTTPException(status_code=409, detail="domain is already verified") from exc + return _render_magic_page(service, settings, None, None, challenge) + + @app.post("/verify-domain", response_class=HTMLResponse) + async def verify_domain(request: Request) -> str: + user = await run_in_threadpool(current_user, request) + if not user: + raise HTTPException(status_code=401, detail="authentication required") + 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, int(user["id"])) + if not found: + return _render_magic_page(service, settings, None, "TXT record not found yet. DNS propagation can take a few minutes.", challenge) + return _render_magic_page(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: + user = await run_in_threadpool(current_user, request) + if not user: + raise HTTPException(status_code=401, detail="authentication required") + 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, int(user["id"])) + 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, + int(user["id"]), + ) + return _render_magic_page(service, settings, account, None) + + @app.post("/register") + async def register(request: Request) -> Response: + form = _parse_form(await request.body()) + email = normalize_email(_first_form_value(form, "email")) + password = _first_form_value(form, "password") + invite = _first_form_value(form, "invite") + if not email or len(password) < 12: + return HTMLResponse(_render_public_page(service, settings, None, [], None, "Use a valid email and a password with at least 12 characters.")) + try: + user = await run_in_threadpool(store.create_user, email, password, invite, settings.require_invite) + except (sqlite3.IntegrityError, ValueError): + return HTMLResponse( + _render_public_page(service, settings, None, [], None, "Registration failed. Check the invite code and email address."), + status_code=400, + ) + response = RedirectResponse("/", status_code=303) + set_session_cookie(response, settings, int(user["id"])) + return response + + @app.post("/login") + async def login(request: Request) -> Response: + form = _parse_form(await request.body()) + user = await run_in_threadpool(store.authenticate_user, normalize_email(_first_form_value(form, "email")), _first_form_value(form, "password")) + if not user: + return HTMLResponse(_render_public_page(service, settings, None, [], None, "Login failed."), status_code=401) + response = RedirectResponse("/", status_code=303) + set_session_cookie(response, settings, int(user["id"])) + return response + + @app.post("/logout") + def logout() -> Response: + response = RedirectResponse("/", status_code=303) + response.delete_cookie("ddns_session") + return response + + @app.get("/admin", response_class=HTMLResponse) + def admin(_: None = Depends(authenticate_admin)) -> str: + return _render_admin_page(service, settings, store.list_accounts(), store.list_invites(), 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(), store.list_invites(), 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.post("/admin/invites") + async def create_invite(request: Request, _: None = Depends(authenticate_admin)) -> Response: + form = _parse_form(await request.body()) + require_admin_csrf(settings, _first_form_value(form, "csrf")) + store.create_invite() + 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() -> Response: + return Response(status_code=204) + + @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 + + +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_magic_page( + 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 = _created_account_panel(service, created_account) + challenge_html = _challenge_panel(service, challenge, message) + message_html = f'

{html.escape(message)}

' if message and not challenge else "" + managed_html = "" + if settings.hostname_suffix: + managed_html = """ +
+

Free DynDNS hostname

+ +

No account needed. You will receive a random hostname, a router update URL, and a separate magic management link. Keep the management link private.

+
+ """ + + return _page( + "DynDNS", + f""" +
+
+
+

FRITZ!Box compatible

+

Free DynDNS for {suffix}

+

Generate cryptographically random URLs, paste the update URL into your router, and keep the management link somewhere safe.

+
+ Admin +
+ {message_html} + {created_html} + {managed_html} + {challenge_html} +
+ """, + ) + + +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 = html.escape(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""" +
+
+
+

Magic management

+

{hostname}

+

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

+
+ Home +
+
+

FRITZ!Box fields

+
+ + + + +
+
+
+

Current IP

+ + + + + + +
IPv4{ipv4}
IPv6{ipv6}
Updated{updated}
+
+ +
+
+
+ """, + ) + + +def _render_public_page( + service: DdnsService, + settings: DdnsSettings, + user: dict[str, str | int] | None, + accounts: list[dict[str, str | int | None]], + challenge: dict[str, str | None] | None, + message: str | None, + created_account: dict[str, str] | None = None, +) -> str: + suffix = html.escape(settings.hostname_suffix or "your domain") + message_html = f'

{html.escape(message)}

' if message else "" + if not user: + invite_field = """ + + """ if settings.require_invite else "" + return _page( + "DynDNS", + f""" +
+
+
+

FRITZ!Box compatible

+

Managed DynDNS for {suffix}

+

Create a secure update URL for your FRITZ!Box or router. Sign in to generate provider hostnames or verify your own domain with DNS.

+
+ Admin +
+ {message_html} +
+
+

Login

+
+ + + +
+
+
+

Register

+
+ + + {invite_field} + +
+
+
+
+ """, + ) + return _render_magic_page(service, settings, created_account, message, challenge) + + +def _challenge_panel(service: DdnsService, challenge: dict[str, str | None] | None, message: str | None) -> str: + challenge_html = "" + claim_secret = "" + if challenge: + domain = html.escape(str(challenge["domain"])) + token = html.escape(str(challenge["token"])) + claim_secret = html.escape(str(challenge.get("claim_secret") or "")) + verification_name = html.escape(service.verification_name(str(challenge["domain"]))) + status = html.escape(message or "Add this TXT record, then verify.") + challenge_html = f""" +
+

{status}

+
+ + +
+
+ + + +
+
+ """ + + return f""" +
+

Custom domain

+ + {challenge_html} + +

The hostname must be inside a domain that has already passed TXT verification.

+
+ """ + + +def _created_account_panel(service: DdnsService, created_account: dict[str, str] | None) -> str: + if not created_account: + return "" + hostname = html.escape(created_account["hostname"]) + username = html.escape(created_account["username"]) + password = html.escape(created_account["password"]) + update_url = html.escape(service.fritz_update_url(created_account)) + management_url = html.escape(service.magic_management_url(created_account)) + return f""" +
+
+

Account generated

+

{hostname}

+
+
+ + + + + +
+

Save this now. The password and management link are only shown on this screen. The Update-URL and management link are separate cryptographically random bearer URLs.

+
+ """ + + +def _render_admin_page( + service: DdnsService, + settings: DdnsSettings, + accounts: list[dict[str, str | int | None]], + invites: list[dict[str, str | None]], + created_account: dict[str, str] | None, +) -> str: + csrf = html.escape(admin_csrf_token(settings)) + rows = "\n".join(_account_row(account, show_owner=True, csrf=csrf) for account in accounts) or """ + No accounts yet. + """ + invite_rows = "\n".join(_invite_row(invite) for invite in invites) or """ + No invites 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""" +
+
+
+

DynDNS aktiv

+

FRITZ!Box accounts

+
+
+ JSON records + Update events +
+
+ + {_created_account_panel(service, created_account)} + +
+

Invites

+
+ + +
+
+ + + {invite_rows} +
CodeCreatedUsed
+
+
+ +
+

Generate account

+

Geben Sie die Anmeldedaten für Ihren DynDNS-Anbieter an. Generated accounts produce the FRITZ!Box fields: Update-URL, Domainnamen, Benutzername, Kennwort.

+ +

{suffix_help}

+
+ +
+

Accounts

+
+ + + + + {rows} +
DomainUserOwnerIPv4IPv6Updated
+
+
+
+ """, + ) + + +def _account_row(account: dict[str, str | int | None], show_owner: bool, 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")) + owner = f"{html.escape(str(account.get('owner_user_id') or '-'))}" if show_owner else "" + return f""" + + {hostname}{username}{owner}{ipv4}{ipv6}{updated} + +
+ + + +
+ + + """ + + +def _invite_row(invite: dict[str, str | None]) -> str: + code = html.escape(str(invite["code"])) + created = html.escape(str(invite.get("created_at") or "-")) + used = html.escape(str(invite.get("used_at") or "-")) + return f"{created}{used}" + + +def _page(title: str, body: str) -> str: + return f""" + + + + + + {html.escape(title)} + + + {body} + + """ diff --git a/router_dyndns/ddns_api.py b/router_dyndns/ddns_api.py new file mode 100644 index 0000000..3073fd0 --- /dev/null +++ b/router_dyndns/ddns_api.py @@ -0,0 +1,260 @@ +import sqlite3 + +from fastapi import APIRouter, Depends, HTTPException, Request, Response, status +from pydantic import BaseModel, Field +from starlette.concurrency import run_in_threadpool + +from .ddns_models import DdnsSettings +from .ddns_security import verify_session_cookie +from .ddns_service import DdnsService, normalize_domain +from .ddns_store import DdnsStore + + +class ErrorResponse(BaseModel): + detail: str = Field(examples=["not found"]) + + +class MagicHostnameRequest(BaseModel): + username: str | None = Field(default=None, max_length=128, examples=["home-router"]) + + +class HostnameCredentials(BaseModel): + hostname: str + username: str + password: str + update_url: str + management_url: str + + +class ManagedRecord(BaseModel): + hostname: str + username: str + update_url: str + ipv4: str | None = None + ipv6: str | None = None + updated_at: str | None = None + + +class DomainChallengeRequest(BaseModel): + domain: str = Field(examples=["example.net"]) + + +class DomainChallengeResponse(BaseModel): + domain: str + txt_name: str + txt_value: str + claim_secret: str + + +class VerifyDomainRequest(BaseModel): + domain: str = Field(examples=["example.net"]) + claim_secret: str + + +class VerifyDomainResponse(BaseModel): + domain: str + verified: bool + message: str + claim_secret: str + + +class CustomHostnameRequest(BaseModel): + hostname: str = Field(examples=["home.example.net"]) + claim_secret: str + username: str | None = Field(default=None, max_length=128, examples=["home-router"]) + + +class RouterUpdateResponse(BaseModel): + status: str = Field(examples=["good"]) + hostname: str + ip: str + + +def create_api_router(settings: DdnsSettings, store: DdnsStore, service: DdnsService) -> APIRouter: + router = APIRouter(prefix="/api/v1") + + def optional_user(request: Request) -> dict[str, str | int] | None: + user_id = verify_session_cookie(request.cookies.get("ddns_session"), settings.session_secret) + return store.get_user(user_id) if user_id is not None else None + + def require_user(request: Request) -> dict[str, str | int]: + user = optional_user(request) + if not user: + raise HTTPException(status_code=401, detail="authentication required") + return user + + user_dependency = Depends(require_user) + + @router.post( + "/hostnames/magic", + response_model=HostnameCredentials, + status_code=status.HTTP_201_CREATED, + summary="Generate a provider-owned DynDNS hostname", + tags=["hostnames"], + responses={500: {"model": ErrorResponse}}, + ) + async def create_magic_hostname(payload: MagicHostnameRequest) -> HostnameCredentials: + account = await run_in_threadpool(service.create_managed_account, payload.username, None) + return _credentials_response(service, account) + + @router.get( + "/management/{management_slug}", + response_model=ManagedRecord, + summary="Read a hostname by magic management link", + tags=["management"], + responses={404: {"model": ErrorResponse}}, + ) + async def read_managed_hostname(management_slug: str) -> ManagedRecord: + account = await run_in_threadpool(store.get_account_by_management_slug, management_slug) + if not account or account["disabled"]: + raise HTTPException(status_code=404, detail="not found") + return _managed_record(service, account) + + @router.delete( + "/management/{management_slug}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete a hostname by magic management link", + tags=["management"], + responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + ) + async def delete_managed_hostname(management_slug: str) -> Response: + account = await run_in_threadpool(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: + await run_in_threadpool(service.delete_account, hostname, "api") + except Exception as exc: + raise HTTPException(status_code=500, detail="DNS delete failed; hostname was not deleted") from exc + return Response(status_code=status.HTTP_204_NO_CONTENT) + + @router.post( + "/domains/challenges", + response_model=DomainChallengeResponse, + status_code=status.HTTP_201_CREATED, + summary="Create a DNS TXT domain verification challenge", + tags=["domains"], + responses={400: {"model": ErrorResponse}}, + ) + async def create_domain_challenge( + payload: DomainChallengeRequest, + user: dict[str, str | int] = user_dependency, + ) -> DomainChallengeResponse: + domain = normalize_domain(payload.domain) + try: + challenge = await run_in_threadpool(service.create_domain_challenge, domain, int(user["id"])) + except sqlite3.IntegrityError as exc: + raise HTTPException(status_code=409, detail="domain is already verified") from exc + return DomainChallengeResponse( + domain=domain, + txt_name=service.verification_name(domain), + txt_value=challenge["token"], + claim_secret=challenge["claim_secret"], + ) + + @router.post( + "/domains/verify", + response_model=VerifyDomainResponse, + summary="Verify a domain TXT challenge", + tags=["domains"], + responses={404: {"model": ErrorResponse}}, + ) + async def verify_domain( + payload: VerifyDomainRequest, + user: dict[str, str | int] = user_dependency, + ) -> VerifyDomainResponse: + domain = normalize_domain(payload.domain) + found, challenge = await run_in_threadpool(service.verify_domain, domain, payload.claim_secret, int(user["id"])) + if not found: + return VerifyDomainResponse( + domain=domain, + verified=False, + message="TXT record not found yet", + claim_secret=payload.claim_secret, + ) + return VerifyDomainResponse( + domain=domain, + verified=True, + message="domain verified", + claim_secret=payload.claim_secret, + ) + + @router.post( + "/hostnames/custom", + response_model=HostnameCredentials, + status_code=status.HTTP_201_CREATED, + summary="Generate credentials for a verified custom hostname", + tags=["hostnames"], + responses={400: {"model": ErrorResponse}, 403: {"model": ErrorResponse}, 409: {"model": ErrorResponse}}, + ) + async def create_custom_hostname( + payload: CustomHostnameRequest, + user: dict[str, str | int] = user_dependency, + ) -> HostnameCredentials: + account = await run_in_threadpool( + service.create_custom_account, + payload.hostname, + payload.claim_secret, + payload.username, + int(user["id"]), + ) + return _credentials_response(service, account) + + @router.get( + "/updates/{update_slug}", + response_model=RouterUpdateResponse, + summary="Router-compatible update endpoint", + tags=["updates"], + responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + ) + async def update_hostname( + update_slug: str, + request: Request, + myip: str | None = None, + ipaddr: str | None = None, + myipv6: str | None = None, + ip6addr: str | None = None, + ) -> RouterUpdateResponse: + account = await run_in_threadpool(store.get_account_by_slug, update_slug) + if not account or account["disabled"]: + raise HTTPException(status_code=404, detail="not found") + + hostname = str(account["hostname"]) + try: + outcome = await run_in_threadpool( + service.apply_update, + request, + hostname, + myip, + ipaddr, + myipv6, + ip6addr, + ) + except Exception as exc: + raise HTTPException(status_code=500, detail="DNS publish failed") from exc + result = outcome.result + return RouterUpdateResponse(status="good" if result.changed else "nochg", hostname=hostname, ip=result.ipv4 or result.ipv6 or "") + + return router + + +def _credentials_response(service: DdnsService, account: dict[str, str]) -> HostnameCredentials: + return HostnameCredentials( + hostname=account["hostname"], + username=account["username"], + password=account["password"], + update_url=service.fritz_update_url(account), + management_url=service.magic_management_url(account), + ) + + +def _managed_record(service: DdnsService, account: dict[str, str | int | None]) -> ManagedRecord: + account_for_url = {"update_slug": str(account["update_slug"])} + return ManagedRecord( + hostname=str(account["hostname"]), + username=str(account["username"]), + update_url=service.fritz_update_url(account_for_url), + ipv4=str(account["ipv4"]) if account.get("ipv4") else None, + ipv6=str(account["ipv6"]) if account.get("ipv6") else None, + updated_at=str(account["updated_at"]) if account.get("updated_at") else None, + ) diff --git a/router_dyndns/ddns_dns.py b/router_dyndns/ddns_dns.py new file mode 100644 index 0000000..2674c8d --- /dev/null +++ b/router_dyndns/ddns_dns.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request + +from .ddns_models import DdnsSettings, UpdateResult + + +def publish_dns(result: UpdateResult, settings: DdnsSettings) -> None: + provider = settings.dns_provider.strip().lower() + if provider == "cloudflare": + publish_cloudflare(result, settings) + return + if settings.rfc2136_server or provider == "rfc2136": + publish_rfc2136(result, settings) + return + if settings.require_dns_provider: + raise RuntimeError("DNS provider is required but not configured") + + +def delete_dns_records(hostname: str, settings: DdnsSettings) -> None: + provider = settings.dns_provider.strip().lower() + if provider == "cloudflare": + delete_cloudflare_records(hostname, settings) + return + if settings.rfc2136_server or provider == "rfc2136": + delete_rfc2136_records(hostname, settings) + return + if settings.require_dns_provider: + raise RuntimeError("DNS provider is required but not configured") + + +def dns_backend_configured(settings: DdnsSettings) -> bool: + provider = settings.dns_provider.strip().lower() + return provider == "cloudflare" or provider == "rfc2136" or bool(settings.rfc2136_server) + + +def publish_rfc2136(result: UpdateResult, settings: DdnsSettings) -> None: + if not settings.rfc2136_server: + return + + import dns.query + import dns.tsigkeyring + import dns.update + + keyring = None + if settings.rfc2136_key_name and settings.rfc2136_key_secret: + keyring = dns.tsigkeyring.from_text({settings.rfc2136_key_name: settings.rfc2136_key_secret}) + + zone = settings.rfc2136_zone or ".".join(result.hostname.split(".")[1:]) + update = dns.update.Update(zone, keyring=keyring) + relative = result.hostname.removesuffix("." + zone).rstrip(".") or "@" + if result.update_ipv4 and result.ipv4: + update.replace(relative, settings.ttl, "A", result.ipv4) + elif result.update_ipv4: + update.delete(relative, "A") + if result.update_ipv6 and result.ipv6: + update.replace(relative, settings.ttl, "AAAA", result.ipv6) + elif result.update_ipv6: + update.delete(relative, "AAAA") + response = dns.query.tcp(update, settings.rfc2136_server, timeout=5) + rcode = response.rcode() + if rcode != 0: + import dns.rcode + + raise RuntimeError(f"RFC2136 update failed: {dns.rcode.to_text(rcode)}") + + +def delete_rfc2136_records(hostname: str, settings: DdnsSettings) -> None: + if not settings.rfc2136_server: + return + + import dns.query + import dns.tsigkeyring + import dns.update + + keyring = None + if settings.rfc2136_key_name and settings.rfc2136_key_secret: + keyring = dns.tsigkeyring.from_text({settings.rfc2136_key_name: settings.rfc2136_key_secret}) + + zone = settings.rfc2136_zone or ".".join(hostname.split(".")[1:]) + update = dns.update.Update(zone, keyring=keyring) + relative = hostname.removesuffix("." + zone).rstrip(".") or "@" + update.delete(relative, "A") + update.delete(relative, "AAAA") + dns.query.tcp(update, settings.rfc2136_server, timeout=5) + + +def publish_cloudflare(result: UpdateResult, settings: DdnsSettings) -> None: + if not settings.cloudflare_api_token or not settings.cloudflare_zone_id: + raise RuntimeError("DDNS_CLOUDFLARE_API_TOKEN and DDNS_CLOUDFLARE_ZONE_ID are required") + if result.update_ipv4: + _cloudflare_upsert_record(settings, result.hostname, "A", result.ipv4) + if result.update_ipv6: + _cloudflare_upsert_record(settings, result.hostname, "AAAA", result.ipv6) + + +def delete_cloudflare_records(hostname: str, settings: DdnsSettings) -> None: + if not settings.cloudflare_api_token or not settings.cloudflare_zone_id: + raise RuntimeError("DDNS_CLOUDFLARE_API_TOKEN and DDNS_CLOUDFLARE_ZONE_ID are required") + for record_type in ("A", "AAAA"): + record = _cloudflare_find_record(settings, hostname, record_type) + if record: + _cloudflare_request(settings, f"/dns_records/{record['id']}", method="DELETE") + + +def _cloudflare_upsert_record(settings: DdnsSettings, hostname: str, record_type: str, value: str | None) -> None: + record = _cloudflare_find_record(settings, hostname, record_type) + if value is None: + if record: + _cloudflare_request(settings, f"/dns_records/{record['id']}", method="DELETE") + return + + payload = { + "type": record_type, + "name": hostname, + "content": value, + "ttl": settings.ttl, + "proxied": False, + } + if record: + _cloudflare_request(settings, f"/dns_records/{record['id']}", method="PATCH", payload=payload) + else: + _cloudflare_request(settings, "/dns_records", method="POST", payload=payload) + + +def _cloudflare_find_record(settings: DdnsSettings, hostname: str, record_type: str) -> dict[str, str] | None: + query = urllib.parse.urlencode({"type": record_type, "name": hostname}) + response = _cloudflare_request(settings, f"/dns_records?{query}") + records = response.get("result", []) + return records[0] if records else None + + +def _cloudflare_request( + settings: DdnsSettings, + path: str, + method: str = "GET", + payload: dict[str, object] | None = None, +) -> dict[str, object]: + url = f"https://api.cloudflare.com/client/v4/zones/{settings.cloudflare_zone_id}{path}" + body = json.dumps(payload).encode("utf-8") if payload is not None else None + request = urllib.request.Request(url, data=body, method=method) + request.add_header("Authorization", f"Bearer {settings.cloudflare_api_token}") + request.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(request, timeout=10) as response: + parsed = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"Cloudflare API {exc.code}: {detail}") from exc + if not parsed.get("success", False): + raise RuntimeError(f"Cloudflare API error: {parsed}") + return parsed diff --git a/router_dyndns/ddns_models.py b/router_dyndns/ddns_models.py new file mode 100644 index 0000000..21a50f3 --- /dev/null +++ b/router_dyndns/ddns_models.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from pydantic import BaseModel, Field, field_validator + + +class DdnsSettings(BaseModel): + database_path: Path = Field(default=Path("ddns.sqlite3")) + shared_secret: str = Field(default="") + admin_password: str = Field(default="") + allowed_hosts: set[str] = Field(default_factory=set) + public_base_url: str = Field(default="http://localhost:8080") + hostname_suffix: str = Field(default="") + verification_prefix: str = Field(default="_router_dyndns-ddns") + session_secret: str = Field(default="") + require_invite: bool = Field(default=False) + max_hostnames_per_user: int = Field(default=5, ge=1, le=100) + rate_limit_per_minute: int = Field(default=60, ge=1, le=10_000) + trusted_proxy_ips: set[str] = Field(default_factory=set) + rfc2136_server: str | None = None + rfc2136_zone: str | None = None + rfc2136_key_name: str | None = None + rfc2136_key_secret: str | None = None + dns_provider: str = Field(default="") + dns_zones: set[str] = Field(default_factory=set) + require_dns_provider: bool = Field(default=False) + cloudflare_api_token: str = Field(default="") + cloudflare_zone_id: str = Field(default="") + cleanup_challenge_hours: int = Field(default=72, ge=1, le=24 * 365) + ttl: int = Field(default=60, ge=60, le=86400) + + @field_validator("public_base_url") + @classmethod + def validate_public_base_url(cls, value: str) -> str: + if not value.startswith(("http://", "https://")): + raise ValueError("public_base_url must start with http:// or https://") + return value.rstrip("/") + + @classmethod + def from_env(cls) -> DdnsSettings: + allowed = { + item.strip().lower() + for item in os.getenv("DDNS_ALLOWED_HOSTS", "").split(",") + if item.strip() + } + trusted_proxy_ips = _csv_set("DDNS_TRUSTED_PROXY_IPS") + dns_zones = _csv_set("DDNS_DNS_ZONES") + return cls( + database_path=Path(os.getenv("DDNS_DATABASE", "ddns.sqlite3")), + shared_secret=os.getenv("DDNS_SHARED_SECRET", ""), + admin_password=os.getenv("DDNS_ADMIN_PASSWORD", os.getenv("DDNS_SHARED_SECRET", "")), + allowed_hosts=allowed, + public_base_url=os.getenv("DDNS_PUBLIC_BASE_URL", "http://localhost:8080"), + hostname_suffix=os.getenv("DDNS_HOSTNAME_SUFFIX", ""), + verification_prefix=os.getenv("DDNS_VERIFICATION_PREFIX", "_router_dyndns-ddns"), + session_secret=os.getenv("DDNS_SESSION_SECRET", os.getenv("DDNS_ADMIN_PASSWORD", "")), + require_invite=os.getenv("DDNS_REQUIRE_INVITE", "0").strip().lower() not in {"0", "false", "no"}, + max_hostnames_per_user=_env_int("DDNS_MAX_HOSTNAMES_PER_USER", 5), + rate_limit_per_minute=_env_int("DDNS_RATE_LIMIT_PER_MINUTE", 60), + trusted_proxy_ips=trusted_proxy_ips, + rfc2136_server=os.getenv("DDNS_RFC2136_SERVER") or None, + rfc2136_zone=os.getenv("DDNS_RFC2136_ZONE") or None, + rfc2136_key_name=os.getenv("DDNS_RFC2136_KEY_NAME") or None, + rfc2136_key_secret=os.getenv("DDNS_RFC2136_KEY_SECRET") or None, + dns_provider=os.getenv("DDNS_DNS_PROVIDER", ""), + dns_zones=dns_zones, + require_dns_provider=os.getenv("DDNS_REQUIRE_DNS_PROVIDER", "0").strip().lower() not in {"0", "false", "no"}, + cloudflare_api_token=os.getenv("DDNS_CLOUDFLARE_API_TOKEN", ""), + cloudflare_zone_id=os.getenv("DDNS_CLOUDFLARE_ZONE_ID", ""), + cleanup_challenge_hours=_env_int("DDNS_CLEANUP_CHALLENGE_HOURS", 72), + ttl=_env_int("DDNS_TTL", 60), + ) + + +@dataclass(slots=True) +class UpdateResult: + hostname: str + ipv4: str | None + ipv6: str | None + changed: bool + update_ipv4: bool = True + update_ipv6: bool = True + + @property + def dyndns_response(self) -> str: + ip = self.ipv4 or self.ipv6 or "0.0.0.0" + return f"{'good' if self.changed else 'nochg'} {ip}" + + +def _csv_set(name: str) -> set[str]: + return {item.strip().lower() for item in os.getenv(name, "").split(",") if item.strip()} + + +def _env_int(name: str, default: int) -> int: + try: + return int(os.getenv(name, str(default))) + except ValueError: + return default diff --git a/router_dyndns/ddns_security.py b/router_dyndns/ddns_security.py new file mode 100644 index 0000000..0ef509e --- /dev/null +++ b/router_dyndns/ddns_security.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import re +import secrets +import time +from datetime import UTC, datetime +from ipaddress import ip_address, ip_network + +from fastapi import HTTPException, Request, Response + +from .ddns_models import DdnsSettings + + +def credentials_from_basic_auth(header: str) -> tuple[str | None, str | None]: + try: + raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8") + except Exception: + return None, None + if ":" not in raw: + return None, None + username, password = raw.split(":", 1) + return username, password + + +def hash_secret(token: str) -> str: + salt = secrets.token_hex(16) + digest = hashlib.pbkdf2_hmac("sha256", token.encode("utf-8"), salt.encode("ascii"), 200_000) + return f"pbkdf2_sha256${salt}${digest.hex()}" + + +def verify_secret(token: str, stored: str) -> bool: + try: + algorithm, salt, expected = stored.split("$", 2) + except ValueError: + return False + if algorithm != "pbkdf2_sha256": + return False + digest = hashlib.pbkdf2_hmac("sha256", token.encode("utf-8"), salt.encode("ascii"), 200_000) + return hmac.compare_digest(digest.hex(), expected) + + +def hash_lookup_token(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def normalize_email(value: str) -> str: + email = value.strip().lower() + if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", email): + return "" + return email + + +def make_session_cookie(user_id: int, secret: str) -> str: + timestamp = str(int(time.time())) + payload = f"{user_id}:{timestamp}" + signature = _session_signature(payload, secret) + return base64.urlsafe_b64encode(f"{payload}:{signature}".encode()).decode("ascii") + + +def verify_session_cookie(value: str | None, secret: str, max_age_seconds: int = 60 * 60 * 24 * 30) -> int | None: + if not value or not secret: + return None + try: + decoded = base64.urlsafe_b64decode(value.encode("ascii")).decode("utf-8") + user_id, timestamp, signature = decoded.split(":", 2) + except Exception: + return None + payload = f"{user_id}:{timestamp}" + if not hmac.compare_digest(signature, _session_signature(payload, secret)): + return None + if int(time.time()) - int(timestamp) > max_age_seconds: + return None + return int(user_id) + + +def set_session_cookie(response: Response, settings: DdnsSettings, user_id: int) -> None: + response.set_cookie( + "ddns_session", + make_session_cookie(user_id, settings.session_secret), + max_age=60 * 60 * 24 * 30, + httponly=True, + secure=settings.public_base_url.startswith("https://"), + samesite="strict", + ) + + +def admin_csrf_token(settings: DdnsSettings) -> str: + secret = settings.admin_password or settings.shared_secret or settings.session_secret + day = datetime.now(UTC).strftime("%Y-%m-%d") + return hmac.new(secret.encode("utf-8"), f"admin:{day}".encode(), hashlib.sha256).hexdigest() + + +def require_admin_csrf(settings: DdnsSettings, supplied: str) -> None: + if not supplied or not hmac.compare_digest(supplied, admin_csrf_token(settings)): + raise HTTPException(status_code=403, detail="invalid csrf token") + + +def client_ip(request: Request, settings: DdnsSettings | None = None) -> str: + peer = request.client.host if request.client else "unknown" + forwarded = request.headers.get("x-forwarded-for", "") + if forwarded and settings and _trusted_proxy(peer, settings.trusted_proxy_ips): + return forwarded.split(",", 1)[0].strip() + return peer + + +def is_rate_limited_path(path: str) -> bool: + limited_paths = { + "/magic", + "/login", + "/register", + "/accounts", + "/request-domain", + "/verify-domain", + "/nic/update", + } + return path in limited_paths or path.startswith(("/u/", "/api/v1/updates/")) or path.startswith( + ("/api/v1/hostnames/", "/api/v1/domains/") + ) + + +def _session_signature(payload: str, secret: str) -> str: + return hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() + + +def _trusted_proxy(peer: str, trusted: set[str]) -> bool: + if not trusted: + return False + try: + peer_ip = ip_address(peer) + except ValueError: + return False + for candidate in trusted: + try: + if "/" in candidate and peer_ip in ip_network(candidate, strict=False): + return True + if "/" not in candidate and peer_ip == ip_address(candidate): + return True + except ValueError: + continue + return False diff --git a/router_dyndns/ddns_service.py b/router_dyndns/ddns_service.py new file mode 100644 index 0000000..b88cb5d --- /dev/null +++ b/router_dyndns/ddns_service.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +import ipaddress +import re +import secrets +import sqlite3 +import threading +import urllib.parse +from dataclasses import dataclass + +from fastapi import HTTPException, Request + +from .ddns_dns import delete_dns_records, dns_backend_configured, publish_dns +from .ddns_models import DdnsSettings, UpdateResult +from .ddns_security import client_ip +from .ddns_store import DdnsStore + + +@dataclass(slots=True) +class UpdateOutcome: + result: UpdateResult + source_ip: str + + +class DdnsService: + def __init__(self, settings: DdnsSettings, store: DdnsStore) -> None: + self.settings = settings + self.store = store + self._locks_guard = threading.Lock() + self._locks: dict[str, threading.Lock] = {} + + def create_managed_account(self, username: str | None = None, owner_user_id: int | None = None) -> dict[str, str]: + if not self.settings.hostname_suffix: + raise HTTPException(status_code=500, detail="DDNS_HOSTNAME_SUFFIX is required") + if owner_user_id is not None: + self._enforce_account_quota(owner_user_id) + for _ in range(5): + try: + return self.store.create_account( + self.random_managed_hostname(), + username, + owner_user_id=owner_user_id, + ) + except sqlite3.IntegrityError: + continue + raise HTTPException(status_code=500, detail="could not allocate hostname") + + def create_custom_account( + self, + hostname: str, + claim_secret: str, + username: str | None = None, + owner_user_id: int | None = None, + ) -> dict[str, str]: + normalized = normalize_hostname(hostname, self.settings, expand_suffix=False) + if not normalized: + raise HTTPException(status_code=400, detail="valid custom hostname is required") + if not self.hostname_is_publishable(normalized): + raise HTTPException(status_code=403, detail="hostname is outside configured DNS publishing zones") + if not self.best_verified_parent(normalized, owner_user_id, claim_secret): + raise HTTPException(status_code=403, detail="verify the parent domain before creating this hostname") + if owner_user_id is not None: + self._enforce_account_quota(owner_user_id) + try: + return self.store.create_account(normalized, username, owner_user_id=owner_user_id) + except sqlite3.IntegrityError as exc: + raise HTTPException(status_code=409, detail="hostname already exists") from exc + + def create_domain_challenge(self, domain: str, owner_user_id: int | None = None) -> dict[str, str]: + normalized = normalize_domain(domain) + if not normalized: + raise HTTPException(status_code=400, detail="valid domain is required") + existing = self.store.get_domain_challenge(normalized, owner_user_id=owner_user_id) + if existing and existing.get("verified_at"): + raise HTTPException(status_code=409, detail="domain is already verified") + return self.store.create_domain_challenge(normalized, owner_user_id) + + def verify_domain(self, domain: str, claim_secret: str, owner_user_id: int | None = None) -> tuple[bool, dict[str, str | None]]: + normalized = normalize_domain(domain) + if not normalized: + raise HTTPException(status_code=400, detail="valid domain is required") + challenge = self.store.get_domain_challenge(normalized, owner_user_id=owner_user_id, claim_secret=claim_secret) + if not challenge: + raise HTTPException(status_code=404, detail="domain challenge not found") + found = dns_txt_contains(self.verification_name(normalized), str(challenge["token"])) + if not found: + challenge = dict(challenge) + challenge["claim_secret"] = claim_secret + return False, challenge + verified = self.store.verify_domain_challenge(normalized, owner_user_id=owner_user_id) + if verified is None: + raise HTTPException(status_code=404, detail="domain challenge not found") + verified = dict(verified) + verified["claim_secret"] = claim_secret + return True, verified + + def delete_account(self, hostname: str, source: str) -> None: + try: + delete_dns_records(hostname, self.settings) + except Exception as exc: + self.store.log_update_event(hostname, None, None, "dns_delete_failed", f"{type(exc).__name__}: {exc}", source) + raise + self.store.delete_account(hostname) + + def apply_update( + self, + request: Request, + hostname: str, + myip: str | None, + ipaddr: str | None, + myipv6: str | None, + ip6addr: str | None, + ) -> UpdateOutcome: + ipv4, ipv6, update_ipv4, update_ipv6 = parse_update_ips(request, myip, ipaddr, myipv6, ip6addr) + source_ip = client_ip(request, self.settings) + lock = self._lock_for(hostname) + with lock: + planned = self.store.preview_update(hostname, ipv4, ipv6, update_ipv4=update_ipv4, update_ipv6=update_ipv6) + if planned.changed: + try: + publish_dns(planned, self.settings) + except Exception as exc: + self.store.log_update_event(hostname, ipv4, ipv6, "dns_failed", f"{type(exc).__name__}: {exc}", source_ip) + raise + result = self.store.upsert(hostname, ipv4, ipv6, update_ipv4=update_ipv4, update_ipv6=update_ipv6) + self.store.log_update_event(hostname, result.ipv4, result.ipv6, "updated" if result.changed else "unchanged", "ok", source_ip) + return UpdateOutcome(result=result, source_ip=source_ip) + + def fritz_update_url(self, account: dict[str, str]) -> str: + base = self.settings.public_base_url.rstrip("/") + slug = urllib.parse.quote(account["update_slug"]) + return f"{base}/u/{slug}?myip=&myipv6=" + + def magic_management_url(self, account: dict[str, str]) -> str: + base = self.settings.public_base_url.rstrip("/") + slug = urllib.parse.quote(account["management_slug"]) + return f"{base}/m/{slug}" + + def random_managed_hostname(self) -> str: + suffix = self.settings.hostname_suffix.strip().lower().strip(".") + if not suffix: + raise HTTPException(status_code=500, detail="DDNS_HOSTNAME_SUFFIX is required for managed hostnames") + return f"{secrets.token_hex(6)}.{suffix}" + + def best_verified_parent( + self, + hostname: str, + owner_user_id: int | None = None, + claim_secret: str | None = None, + ) -> str | None: + labels = hostname.split(".") + for index in range(1, len(labels) - 1): + candidate = ".".join(labels[index:]) + if self.store.is_domain_verified(candidate, owner_user_id, claim_secret): + return candidate + if self.store.is_domain_verified(hostname, owner_user_id, claim_secret): + return hostname + return None + + def verification_name(self, domain: str) -> str: + prefix = self.settings.verification_prefix.strip().strip(".") or "_router_dyndns-ddns" + return f"{prefix}.{domain}" + + def hostname_is_publishable(self, hostname: str) -> bool: + if not dns_backend_configured(self.settings): + return not self.settings.require_dns_provider + zones = { + zone.strip().lower().strip(".") + for zone in self.settings.dns_zones + if zone.strip() + } + if not zones and self.settings.rfc2136_zone: + zones.add(self.settings.rfc2136_zone.strip().lower().strip(".")) + if not zones and self.settings.hostname_suffix: + zones.add(self.settings.hostname_suffix.strip().lower().strip(".")) + return bool(zones) and any(hostname == zone or hostname.endswith(f".{zone}") for zone in zones) + + def _enforce_account_quota(self, owner_user_id: int) -> None: + if self.store.count_user_accounts(owner_user_id) >= self.settings.max_hostnames_per_user: + raise HTTPException(status_code=403, detail="hostname quota exceeded") + + def _lock_for(self, hostname: str) -> threading.Lock: + with self._locks_guard: + lock = self._locks.get(hostname) + if lock is None: + lock = threading.Lock() + self._locks[hostname] = lock + return lock + + +def normalize_ip(value: str | None, version: int | None = None) -> str | None: + if not value: + return None + try: + ip = ipaddress.ip_address(value) + except ValueError: + return None + if version and ip.version != version: + return None + return str(ip) + + +def parse_update_ips( + request: Request, + myip: str | None, + ipaddr: str | None, + myipv6: str | None, + ip6addr: str | None, +) -> tuple[str | None, str | None, bool, bool]: + raw_ipv4 = myip if myip is not None else ipaddr + raw_ipv6 = myipv6 if myipv6 is not None else ip6addr + update_ipv4 = raw_ipv4 is not None + update_ipv6 = raw_ipv6 is not None + ipv4 = normalize_ip(raw_ipv4, version=4) + ipv6 = normalize_ip(raw_ipv6, version=6) + if not update_ipv4 and not update_ipv6: + peer = request.client.host if request.client else None + ipv4 = normalize_ip(peer, version=4) + ipv6 = normalize_ip(peer, version=6) + update_ipv4 = ipv4 is not None + update_ipv6 = ipv6 is not None + return ipv4, ipv6, update_ipv4, update_ipv6 + + +def normalize_hostname(value: str, settings: DdnsSettings, expand_suffix: bool = True) -> str: + hostname = value.strip().lower().rstrip(".") + suffix = settings.hostname_suffix.strip().lower().strip(".") + if expand_suffix and suffix and "." not in hostname: + hostname = f"{hostname}.{suffix}" + label = r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?" + if not re.fullmatch(rf"{label}(\.{label})+", hostname): + return "" + return hostname + + +def normalize_domain(value: str) -> str: + return normalize_hostname(value, DdnsSettings(), expand_suffix=False) + + +def dns_txt_contains(name: str, expected: str) -> bool: + try: + import dns.resolver + + answers = dns.resolver.resolve(name, "TXT", lifetime=5) + except Exception: + return False + for answer in answers: + for item in answer.strings: + if item.decode("utf-8", errors="replace") == expected: + return True + return False diff --git a/router_dyndns/ddns_store.py b/router_dyndns/ddns_store.py new file mode 100644 index 0000000..e74601d --- /dev/null +++ b/router_dyndns/ddns_store.py @@ -0,0 +1,506 @@ +from __future__ import annotations + +import secrets +import sqlite3 +import time +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from .ddns_models import UpdateResult +from .ddns_security import hash_lookup_token, hash_secret, verify_secret + + +class DdnsStore: + def __init__(self, database_path: Path) -> None: + self.database_path = database_path + self.database_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.database_path) + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("PRAGMA busy_timeout = 5000") + return conn + + def upsert( + self, + hostname: str, + ipv4: str | None, + ipv6: str | None, + update_ipv4: bool = True, + update_ipv6: bool = True, + ) -> UpdateResult: + now = datetime.now(UTC).isoformat() + with self.connect() as conn: + conn.row_factory = sqlite3.Row + existing = conn.execute("SELECT ipv4, ipv6 FROM records WHERE hostname = ?", (hostname,)).fetchone() + current_ipv4 = existing["ipv4"] if existing else None + current_ipv6 = existing["ipv6"] if existing else None + effective_ipv4 = ipv4 if update_ipv4 else current_ipv4 + effective_ipv6 = ipv6 if update_ipv6 else current_ipv6 + changed = existing is None or current_ipv4 != effective_ipv4 or current_ipv6 != effective_ipv6 + conn.execute( + """ + INSERT INTO records(hostname, ipv4, ipv6, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(hostname) DO UPDATE SET + ipv4=excluded.ipv4, + ipv6=excluded.ipv6, + updated_at=excluded.updated_at + """, + (hostname, effective_ipv4, effective_ipv6, now), + ) + return UpdateResult( + hostname=hostname, + ipv4=effective_ipv4, + ipv6=effective_ipv6, + changed=changed, + update_ipv4=update_ipv4, + update_ipv6=update_ipv6, + ) + + def preview_update( + self, + hostname: str, + ipv4: str | None, + ipv6: str | None, + update_ipv4: bool = True, + update_ipv6: bool = True, + ) -> UpdateResult: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + existing = conn.execute("SELECT ipv4, ipv6 FROM records WHERE hostname = ?", (hostname,)).fetchone() + current_ipv4 = existing["ipv4"] if existing else None + current_ipv6 = existing["ipv6"] if existing else None + effective_ipv4 = ipv4 if update_ipv4 else current_ipv4 + effective_ipv6 = ipv6 if update_ipv6 else current_ipv6 + changed = existing is None or current_ipv4 != effective_ipv4 or current_ipv6 != effective_ipv6 + return UpdateResult( + hostname=hostname, + ipv4=effective_ipv4, + ipv6=effective_ipv6, + changed=changed, + update_ipv4=update_ipv4, + update_ipv6=update_ipv6, + ) + + def create_account( + self, + hostname: str, + username: str | None = None, + owner_user_id: int | None = None, + ) -> dict[str, str]: + now = datetime.now(UTC).isoformat() + token = secrets.token_urlsafe(24) + update_slug = secrets.token_urlsafe(32) + management_slug = secrets.token_urlsafe(32) + username = username or _default_username(hostname) + token_hash = hash_secret(token) + management_hash = hash_lookup_token(management_slug) + with self.connect() as conn: + conn.execute( + """ + INSERT INTO accounts(hostname, username, token_hash, update_slug, management_hash, owner_user_id, created_at, disabled) + VALUES (?, ?, ?, ?, ?, ?, ?, 0) + """, + (hostname, username, token_hash, update_slug, management_hash, owner_user_id, now), + ) + return { + "hostname": hostname, + "username": username, + "password": token, + "update_slug": update_slug, + "management_slug": management_slug, + } + + def list_accounts(self, owner_user_id: int | None = None) -> list[dict[str, str | int | None]]: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + sql = """ + SELECT accounts.hostname, accounts.username, accounts.update_slug, accounts.owner_user_id, + accounts.created_at, accounts.disabled, records.ipv4, records.ipv6, records.updated_at + FROM accounts + LEFT JOIN records ON records.hostname = accounts.hostname + """ + params: tuple[int, ...] = () + if owner_user_id is not None: + sql += " WHERE accounts.owner_user_id = ?" + params = (owner_user_id,) + sql += " ORDER BY accounts.hostname" + rows = conn.execute(sql, params).fetchall() + return [dict(row) for row in rows] + + def get_account_by_slug(self, update_slug: str) -> dict[str, str | int | None] | None: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + account = conn.execute( + """ + SELECT hostname, username, token_hash, update_slug, disabled FROM accounts + WHERE update_slug = ? + """, + (update_slug,), + ).fetchone() + return dict(account) if account else None + + def get_account_by_management_slug(self, management_slug: str) -> dict[str, str | int | None] | None: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + account = conn.execute( + """ + SELECT accounts.hostname, accounts.username, accounts.update_slug, accounts.owner_user_id, + accounts.created_at, accounts.disabled, records.ipv4, records.ipv6, records.updated_at + FROM accounts + LEFT JOIN records ON records.hostname = accounts.hostname + WHERE accounts.management_hash = ? + """, + (hash_lookup_token(management_slug),), + ).fetchone() + return dict(account) if account else None + + def verify_account(self, hostname: str, username: str | None, token: str | None) -> bool: + if not username or not token: + return False + with self.connect() as conn: + conn.row_factory = sqlite3.Row + account = conn.execute( + """ + SELECT token_hash, disabled FROM accounts + WHERE hostname = ? AND username = ? + """, + (hostname, username), + ).fetchone() + if not account or account["disabled"]: + return False + return verify_secret(token, account["token_hash"]) + + def delete_account(self, hostname: str) -> None: + with self.connect() as conn: + conn.execute("DELETE FROM accounts WHERE hostname = ?", (hostname,)) + conn.execute("DELETE FROM records WHERE hostname = ?", (hostname,)) + + def create_domain_challenge(self, domain: str, owner_user_id: int | None = None) -> dict[str, str]: + token = f"ff-ddns-{secrets.token_urlsafe(24)}" + claim_secret = secrets.token_urlsafe(32) + claim_hash = hash_lookup_token(claim_secret) + now = datetime.now(UTC).isoformat() + with self.connect() as conn: + existing = conn.execute("SELECT owner_user_id, verified_at FROM domain_challenges WHERE domain = ?", (domain,)).fetchone() + if existing and existing[1]: + raise sqlite3.IntegrityError("domain is already verified") + if existing and owner_user_id is not None and existing[0] is not None and int(existing[0]) != owner_user_id: + raise sqlite3.IntegrityError("domain challenge belongs to another user") + conn.execute( + """ + INSERT INTO domain_challenges(domain, token, claim_hash, owner_user_id, created_at, verified_at) + VALUES (?, ?, ?, ?, ?, NULL) + ON CONFLICT(domain) DO UPDATE SET + token=excluded.token, + claim_hash=excluded.claim_hash, + owner_user_id=excluded.owner_user_id, + created_at=excluded.created_at, + verified_at=NULL + """, + (domain, token, claim_hash, owner_user_id, now), + ) + return {"domain": domain, "token": token, "claim_secret": claim_secret} + + def verify_domain_challenge(self, domain: str, owner_user_id: int | None = None) -> dict[str, str | None] | None: + now = datetime.now(UTC).isoformat() + with self.connect() as conn: + conn.row_factory = sqlite3.Row + sql = "SELECT domain, token, verified_at FROM domain_challenges WHERE domain = ?" + params: tuple[str, ...] | tuple[str, int] = (domain,) + if owner_user_id is not None: + sql += " AND owner_user_id = ?" + params = (domain, owner_user_id) + challenge = conn.execute(sql, params).fetchone() + if not challenge: + return None + if owner_user_id is not None: + conn.execute("UPDATE domain_challenges SET verified_at = ? WHERE domain = ? AND owner_user_id = ?", (now, domain, owner_user_id)) + else: + conn.execute("UPDATE domain_challenges SET verified_at = ? WHERE domain = ?", (now, domain)) + result = dict(challenge) + result["verified_at"] = now + return result + + def get_domain_challenge( + self, + domain: str, + owner_user_id: int | None = None, + claim_secret: str | None = None, + ) -> dict[str, str | None] | None: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + sql = "SELECT domain, token, verified_at FROM domain_challenges WHERE domain = ?" + params: tuple[str, ...] | tuple[str, int] = (domain,) + if owner_user_id is not None: + sql += " AND owner_user_id = ?" + params = (domain, owner_user_id) + elif claim_secret is not None: + sql += " AND claim_hash = ?" + params = (domain, hash_lookup_token(claim_secret)) + challenge = conn.execute(sql, params).fetchone() + return dict(challenge) if challenge else None + + def is_domain_verified( + self, + domain: str, + owner_user_id: int | None = None, + claim_secret: str | None = None, + ) -> bool: + with self.connect() as conn: + sql = "SELECT verified_at FROM domain_challenges WHERE domain = ?" + params: tuple[str, ...] | tuple[str, int] = (domain,) + if owner_user_id is not None: + sql += " AND owner_user_id = ?" + params = (domain, owner_user_id) + elif claim_secret is not None: + sql += " AND claim_hash = ?" + params = (domain, hash_lookup_token(claim_secret)) + verified_at = conn.execute(sql, params).fetchone() + return bool(verified_at and verified_at[0]) + + def count_user_accounts(self, owner_user_id: int) -> int: + with self.connect() as conn: + row = conn.execute("SELECT COUNT(*) FROM accounts WHERE owner_user_id = ? AND disabled = 0", (owner_user_id,)).fetchone() + return int(row[0]) + + def create_user(self, email: str, password: str, invite_code: str | None, require_invite: bool) -> dict[str, str | int]: + now = datetime.now(UTC).isoformat() + password_hash = hash_secret(password) + with self.connect() as conn: + if require_invite: + invite = conn.execute( + "SELECT code, used_at FROM invites WHERE code = ?", + (invite_code or "",), + ).fetchone() + if not invite or invite[1]: + raise ValueError("invalid invite code") + cursor = conn.execute( + """ + INSERT INTO users(email, password_hash, created_at, disabled) + VALUES (?, ?, ?, 0) + """, + (email, password_hash, now), + ) + user_id = int(cursor.lastrowid) + if require_invite and invite_code: + conn.execute("UPDATE invites SET used_at = ?, used_by_user_id = ? WHERE code = ?", (now, user_id, invite_code)) + return {"id": user_id, "email": email} + + def authenticate_user(self, email: str, password: str) -> dict[str, str | int] | None: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + user = conn.execute( + "SELECT id, email, password_hash, disabled FROM users WHERE email = ?", + (email,), + ).fetchone() + if not user or user["disabled"] or not verify_secret(password, user["password_hash"]): + return None + return {"id": int(user["id"]), "email": str(user["email"])} + + def get_user(self, user_id: int) -> dict[str, str | int] | None: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + user = conn.execute("SELECT id, email, disabled FROM users WHERE id = ?", (user_id,)).fetchone() + if not user or user["disabled"]: + return None + return {"id": int(user["id"]), "email": str(user["email"])} + + def create_invite(self) -> dict[str, str]: + code = secrets.token_urlsafe(18) + now = datetime.now(UTC).isoformat() + with self.connect() as conn: + conn.execute("INSERT INTO invites(code, created_at) VALUES (?, ?)", (code, now)) + return {"code": code} + + def list_invites(self) -> list[dict[str, str | None]]: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT code, created_at, used_at FROM invites ORDER BY created_at DESC LIMIT 50" + ).fetchall() + return [dict(row) for row in rows] + + def list_records(self) -> list[dict[str, str | None]]: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute("SELECT hostname, ipv4, ipv6, updated_at FROM records ORDER BY hostname").fetchall() + return [dict(row) for row in rows] + + def log_update_event( + self, + hostname: str, + ipv4: str | None, + ipv6: str | None, + status: str, + detail: str, + source_ip: str, + ) -> None: + with self.connect() as conn: + conn.execute( + """ + INSERT INTO update_events(hostname, ipv4, ipv6, status, detail, source_ip, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + (hostname, ipv4, ipv6, status, detail[:500], source_ip, datetime.now(UTC).isoformat()), + ) + + def list_update_events(self, limit: int = 100) -> list[dict[str, str | None]]: + with self.connect() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT hostname, ipv4, ipv6, status, detail, source_ip, created_at + FROM update_events + ORDER BY created_at DESC + LIMIT ? + """, + (limit,), + ).fetchall() + return [dict(row) for row in rows] + + def allow_rate_limit(self, key: str, limit: int) -> bool: + now = int(time.time()) + window = now // 60 + with self.connect() as conn: + conn.execute("DELETE FROM rate_limits WHERE window < ?", (window - 2,)) + row = conn.execute( + "SELECT count FROM rate_limits WHERE key = ? AND window = ?", + (key, window), + ).fetchone() + count = int(row[0]) if row else 0 + if count >= limit: + return False + conn.execute( + """ + INSERT INTO rate_limits(key, window, count) + VALUES (?, ?, 1) + ON CONFLICT(key, window) DO UPDATE SET count=count + 1 + """, + (key, window), + ) + return True + + def cleanup(self, challenge_hours: int) -> None: + cutoff = datetime.now(UTC) - timedelta(hours=challenge_hours) + with self.connect() as conn: + conn.execute( + "DELETE FROM domain_challenges WHERE verified_at IS NULL AND created_at < ?", + (cutoff.isoformat(),), + ) + + def _init_db(self) -> None: + with self.connect() as conn: + conn.execute("PRAGMA journal_mode = WAL") + conn.execute("PRAGMA synchronous = NORMAL") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS records ( + hostname TEXT PRIMARY KEY, + ipv4 TEXT, + ipv6 TEXT, + updated_at TEXT NOT NULL + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS update_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + hostname TEXT NOT NULL, + ipv4 TEXT, + ipv6 TEXT, + status TEXT NOT NULL, + detail TEXT NOT NULL, + source_ip TEXT NOT NULL, + created_at TEXT NOT NULL + ) + """ + ) + conn.execute("CREATE INDEX IF NOT EXISTS idx_update_events_created_at ON update_events(created_at)") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS rate_limits ( + key TEXT NOT NULL, + window INTEGER NOT NULL, + count INTEGER NOT NULL, + PRIMARY KEY(key, window) + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS accounts ( + hostname TEXT PRIMARY KEY, + username TEXT NOT NULL, + token_hash TEXT NOT NULL, + update_slug TEXT NOT NULL UNIQUE, + management_hash TEXT UNIQUE, + owner_user_id INTEGER, + created_at TEXT NOT NULL, + disabled INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY(owner_user_id) REFERENCES users(id) ON DELETE SET NULL + ) + """ + ) + _ensure_column(conn, "accounts", "update_slug", "TEXT") + _ensure_column(conn, "accounts", "management_hash", "TEXT") + _ensure_column(conn, "accounts", "owner_user_id", "INTEGER") + for row in conn.execute("SELECT hostname FROM accounts WHERE update_slug IS NULL OR update_slug = ''"): + conn.execute("UPDATE accounts SET update_slug = ? WHERE hostname = ?", (secrets.token_urlsafe(32), row[0])) + for row in conn.execute("SELECT hostname FROM accounts WHERE management_hash IS NULL OR management_hash = ''"): + conn.execute("UPDATE accounts SET management_hash = ? WHERE hostname = ?", (hash_lookup_token(secrets.token_urlsafe(32)), row[0])) + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_accounts_update_slug ON accounts(update_slug)") + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_accounts_management_hash ON accounts(management_hash)") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS domain_challenges ( + domain TEXT PRIMARY KEY, + token TEXT NOT NULL, + claim_hash TEXT, + owner_user_id INTEGER, + created_at TEXT NOT NULL, + verified_at TEXT, + FOREIGN KEY(owner_user_id) REFERENCES users(id) ON DELETE CASCADE + ) + """ + ) + _ensure_column(conn, "domain_challenges", "claim_hash", "TEXT") + _ensure_column(conn, "domain_challenges", "owner_user_id", "INTEGER") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + email TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL, + created_at TEXT NOT NULL, + disabled INTEGER NOT NULL DEFAULT 0 + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS invites ( + code TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + used_at TEXT, + used_by_user_id INTEGER, + FOREIGN KEY(used_by_user_id) REFERENCES users(id) ON DELETE SET NULL + ) + """ + ) + + +def _default_username(hostname: str) -> str: + label = hostname.split(".", 1)[0] + import re + + return re.sub(r"[^a-z0-9_.-]", "-", label.lower()) or "fritzbox" + + +def _ensure_column(conn: sqlite3.Connection, table: str, column: str, definition: str) -> None: + columns = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")} + if column not in columns: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}") diff --git a/tests/test_ddns.py b/tests/test_ddns.py new file mode 100644 index 0000000..38cd728 --- /dev/null +++ b/tests/test_ddns.py @@ -0,0 +1,568 @@ +from pathlib import Path + +from fastapi.testclient import TestClient + +import router_dyndns.ddns_service as ddns_service +from router_dyndns.ddns import DdnsSettings, DdnsStore, make_app + + +def login_test_user(client: TestClient) -> None: + response = client.post( + "/register", + data={"email": "alice@example.net", "password": "very-secure-password"}, + follow_redirects=False, + ) + assert response.status_code == 303 + + +def test_update_accepts_fritzbox_placeholders(tmp_path: Path) -> None: + app = make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + shared_secret="secret", + allowed_hosts={"home.example.net"}, + ) + ) + client = TestClient(app) + + response = client.get( + "/nic/update", + params={ + "hostname": "home.example.net", + "myip": "203.0.113.8", + "myipv6": "2001:db8::8", + "password": "secret", + }, + ) + + assert response.status_code == 200 + assert response.text == "good 203.0.113.8" + records = client.get("/records", auth=("admin", "secret")).json() + assert records[0]["hostname"] == "home.example.net" + assert records[0]["ipv4"] == "203.0.113.8" + assert records[0]["ipv6"] == "2001:db8::8" + + +def test_update_rejects_wrong_secret(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", shared_secret="secret")) + client = TestClient(app) + + response = client.get( + "/nic/update", + params={"hostname": "home.example.net", "myip": "203.0.113.8", "password": "wrong"}, + ) + + assert response.status_code == 401 + + +def test_update_rejects_unlisted_host(tmp_path: Path) -> None: + app = make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + shared_secret="secret", + allowed_hosts={"home.example.net"}, + ) + ) + client = TestClient(app) + + response = client.get( + "/nic/update", + params={"hostname": "other.example.net", "myip": "203.0.113.8", "password": "secret"}, + ) + + assert response.status_code == 403 + + +def test_generated_account_can_update_own_hostname(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"], + }, + ) + + 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" + + +def test_dns_publish_failure_does_not_update_record(tmp_path: Path, monkeypatch) -> None: + def fail_publish(result, settings): + raise RuntimeError("dns down") + + monkeypatch.setattr(ddns_service, "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"}) + + assert response.status_code == 500 + assert store.list_records() == [] + events = store.list_update_events() + assert events[0]["status"] == "dns_failed" + + +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)) + 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"}) + + assert response.status_code == 200 + assert response.text == "good 203.0.113.9" + assert len(published) == 1 + assert store.list_records()[0]["ipv4"] == "203.0.113.9" + + +def test_generated_account_cannot_update_other_hostname(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": "other.example.net", + "myip": "203.0.113.9", + "username": account["username"], + "password": account["password"], + }, + ) + + assert response.status_code == 401 + + +def test_admin_page_shows_account_form(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + + response = client.get("/admin", auth=("admin", "admin")) + + assert response.status_code == 200 + assert "Update-URL" in response.text + assert "Domainnamen" in response.text + assert "Benutzername" in response.text + assert "Kennwort" in response.text + + +def test_admin_posts_require_csrf(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + + response = client.post( + "/admin/accounts", + auth=("admin", "admin"), + data={"hostname": "home.example.net"}, + ) + + assert response.status_code == 403 + + +def test_admin_password_query_string_is_rejected(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + + response = client.get("/admin", params={"password": "admin"}) + + assert response.status_code == 401 + + +def test_registration_is_open_by_default_for_optional_accounts(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + + response = client.post( + "/register", + data={"email": "alice@example.net", "password": "very-secure-password"}, + follow_redirects=False, + ) + + assert response.status_code == 303 + + +def test_admin_invite_allows_registration(tmp_path: Path) -> None: + database_path = tmp_path / "ddns.sqlite3" + store = DdnsStore(database_path) + invite = store.create_invite() + app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) + client = TestClient(app) + + response = client.post( + "/register", + data={ + "email": "alice@example.net", + "password": "very-secure-password", + "invite": invite["code"], + }, + follow_redirects=False, + ) + + assert response.status_code == 303 + + +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", + require_invite=False, + ) + ) + client = TestClient(app) + login_test_user(client) + + response = client.post("/accounts", data={"mode": "managed", "username": "alice"}) + + assert response.status_code == 200 + assert "Update-URL:" in response.text + assert ".ddns.example.net" in response.text + assert "/u/" in response.text + assert "/m/" in response.text + + +def test_magic_hostname_without_login_generates_update_and_management_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("/magic", data={"username": "alice"}) + + assert response.status_code == 200 + assert "Update-URL:" in response.text + assert "Magic management link:" in response.text + assert ".ddns.example.net" in response.text + + +def test_magic_management_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 "home.ddns.example.net" in management.text + + deleted = client.post(f"/m/{account['management_slug']}/delete") + assert deleted.status_code == 200 + + update = client.get(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"}) + assert update.status_code == 404 + + +def test_custom_hostname_requires_verified_parent_domain(tmp_path: Path) -> None: + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin", require_invite=False)) + client = TestClient(app) + login_test_user(client) + + response = client.post("/accounts", data={"mode": "custom", "hostname": "home.example.net"}) + + assert response.status_code == 403 + + +def test_custom_hostname_after_dns_verification(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(ddns_service, "dns_txt_contains", lambda name, expected: True) + app = make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + public_base_url="http://ddns.example.net", + require_invite=False, + ) + ) + client = TestClient(app) + login_test_user(client) + + store = DdnsStore(tmp_path / "ddns.sqlite3") + challenge_data = store.create_domain_challenge("example.net", owner_user_id=1) + verify = client.post( + "/verify-domain", + data={"domain": "example.net", "claim_secret": challenge_data["claim_secret"]}, + ) + assert verify.status_code == 200 + + response = client.post( + "/accounts", + data={"mode": "custom", "hostname": "home.example.net", "claim_secret": challenge_data["claim_secret"]}, + ) + + assert response.status_code == 200 + assert "home.example.net" in response.text + assert "Update-URL:" in response.text + + +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) + + redoc = client.get("/redoc") + schema = client.get("/openapi.json").json() + + assert redoc.status_code == 200 + assert "/api/v1/hostnames/magic" in schema["paths"] + assert "/api/v1/updates/{update_slug}" in schema["paths"] + assert any(tag["name"] == "hostnames" for tag in schema["tags"]) + + +def test_api_magic_hostname_lifecycle(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/magic", json={"username": "router"}) + assert created.status_code == 201 + 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/") + + 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"] + + deleted = client.delete(f"/api/v1/management/{management_slug}") + assert deleted.status_code == 204 + assert client.get(f"/api/v1/management/{management_slug}").status_code == 404 + + +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)) + 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"}) + + assert response.status_code == 200 + assert response.json() == {"status": "good", "hostname": "home.ddns.example.net", "ip": "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 RuntimeError("dns down") + + monkeypatch.setattr(ddns_service, "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"}) + + assert response.status_code == 500 + assert store.list_records() == [] + assert store.list_update_events()[0]["status"] == "dns_failed" + + +def test_api_custom_hostname_requires_verified_domain(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(ddns_service, "dns_txt_contains", lambda name, expected: True) + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + login_test_user(client) + + challenge = client.post("/api/v1/domains/challenges", json={"domain": "example.net"}) + assert challenge.status_code == 201 + challenge_body = challenge.json() + + verified = client.post( + "/api/v1/domains/verify", + json={"domain": "example.net", "claim_secret": challenge_body["claim_secret"]}, + ) + assert verified.status_code == 200 + assert verified.json()["verified"] is True + + hostname = client.post( + "/api/v1/hostnames/custom", + json={ + "hostname": "home.example.net", + "claim_secret": challenge_body["claim_secret"], + "username": "router", + }, + ) + + assert hostname.status_code == 201 + assert hostname.json()["hostname"] == "home.example.net" + + +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/magic", json={"username": "router"}) + second = client.post("/api/v1/hostnames/magic", json={"username": "router"}) + + assert first.status_code == 201 + assert second.status_code == 429 + + +def test_untrusted_forwarded_for_does_not_bypass_rate_limit(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("/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"}) + + assert first.status_code == 200 + assert second.status_code == 429 + + +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)) + database_path = tmp_path / "ddns.sqlite3" + store = DdnsStore(database_path) + account = store.create_account("home.example.net", "alice") + store.upsert("home.example.net", "203.0.113.8", "2001:db8::8") + 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"}) + + assert response.status_code == 200 + record = store.list_records()[0] + assert record["ipv4"] == "203.0.113.9" + assert record["ipv6"] == "2001:db8::8" + assert published[0].update_ipv4 is True + assert published[0].update_ipv6 is False + + +def test_verified_domain_challenge_cannot_be_overwritten(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(ddns_service, "dns_txt_contains", lambda name, expected: True) + app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin")) + client = TestClient(app) + login_test_user(client) + + challenge = client.post("/api/v1/domains/challenges", json={"domain": "example.net"}).json() + verified = client.post( + "/api/v1/domains/verify", + json={"domain": "example.net", "claim_secret": challenge["claim_secret"]}, + ) + overwrite = client.post("/api/v1/domains/challenges", json={"domain": "example.net"}) + + assert verified.status_code == 200 + assert overwrite.status_code == 409 + + +def test_logged_in_hostname_quota_is_enforced(tmp_path: Path) -> None: + app = make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + hostname_suffix="ddns.example.net", + max_hostnames_per_user=1, + ) + ) + client = TestClient(app) + login_test_user(client) + + first = client.post("/accounts", data={"mode": "managed", "username": "one"}) + second = client.post("/accounts", data={"mode": "managed", "username": "two"}) + + assert first.status_code == 200 + assert second.status_code == 403 + + +def test_custom_hostname_must_be_inside_publishable_dns_zone(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(ddns_service, "dns_txt_contains", lambda name, expected: True) + app = make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + dns_provider="cloudflare", + cloudflare_api_token="token", + cloudflare_zone_id="zone", + dns_zones={"ddns.example.net"}, + ) + ) + client = TestClient(app) + login_test_user(client) + + challenge = client.post("/api/v1/domains/challenges", json={"domain": "example.net"}).json() + client.post( + "/api/v1/domains/verify", + json={"domain": "example.net", "claim_secret": challenge["claim_secret"]}, + ) + hostname = client.post( + "/api/v1/hostnames/custom", + json={"hostname": "home.example.net", "claim_secret": challenge["claim_secret"]}, + ) + + assert hostname.status_code == 403 + assert hostname.json()["detail"] == "hostname is outside configured DNS publishing zones" + + +def test_require_dns_provider_fails_startup_without_backend(tmp_path: Path) -> None: + try: + make_app( + DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + require_dns_provider=True, + ) + ) + except RuntimeError as exc: + assert "no DNS backend" in str(exc) + else: + raise AssertionError("expected startup failure")