karl/karldns.de
Archived
1
0 Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove account-based DynDNS onboarding

Browse Source
This commit is contained in:
StasonJatham committed 2026-05-20 17:57:42 +02:00
1 parent 86060d6bdc
commit 148918b283
8 files changed
+67 -599
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+1 -1
View File
@@ -59,7 +59,7 @@ That means users always have a current hostname for VPN, remote access, self-hos
| Custom domains | TXT challenge verification before hostname creation |
| Persistence | SQLite with WAL mode |
| API | FastAPI, OpenAPI, Swagger UI, ReDoc |
| Auth model | Private magic links for hostnames and custom-domain claims; optional account workspace for dashboards |
| Auth model | No user accounts. Private bearer links for hostnames and custom-domain claims |
| Operations | Rate limiting, update logs, cleanup jobs, Docker, Caddy example |
## Quick Start
router_dyndns/ddns.py
+17 -271
View File
@@ -19,10 +19,7 @@ from .ddns_security import (
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
@@ -67,8 +64,6 @@ 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")
@@ -139,10 +134,6 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
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,
@@ -157,21 +148,14 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
return PlainTextResponse("911 dns publish failed", status_code=500)
@app.get("/", response_class=HTMLResponse)
def index(request: Request) -> str:
user = current_user(request)
if user:
return _render_user_home(service, settings, user, store.list_accounts(int(user["id"])), None, None, None)
def index() -> str:
return _render_public_home(service, settings, None, None)
@app.get("/login", response_class=HTMLResponse)
def login_page() -> str:
return _render_auth_page(settings, 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)
account = await run_in_threadpool(service.create_managed_account, username)
return _render_public_home(service, settings, account, None)
@app.get("/m/{management_slug}", response_class=HTMLResponse)
@@ -218,53 +202,34 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
@app.post("/request-domain", response_class=HTMLResponse)
async def request_domain(request: Request) -> str:
user = await run_in_threadpool(current_user, request)
form = _parse_form(await request.body())
try:
challenge = await run_in_threadpool(
service.create_domain_challenge,
_first_form_value(form, "domain"),
int(user["id"]) if user else None,
)
except sqlite3.IntegrityError as exc:
raise HTTPException(status_code=409, detail="domain is already verified") from exc
if user:
return _render_user_home(service, settings, user, store.list_accounts(int(user["id"])), None, None, challenge)
return _render_public_home(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)
form = _parse_form(await request.body())
domain = normalize_domain(_first_form_value(form, "domain"))
claim_secret = _first_form_value(form, "claim_secret")
owner_user_id = int(user["id"]) if user else None
found, challenge = await run_in_threadpool(service.verify_domain, domain, claim_secret, owner_user_id)
found, challenge = await run_in_threadpool(service.verify_domain, domain, claim_secret)
if not found:
message = "TXT record not found yet. DNS propagation can take a few minutes."
if not user:
return _render_public_home(service, settings, None, message, challenge)
return _render_user_home(
service,
settings,
user,
store.list_accounts(int(user["id"])),
None,
message,
challenge,
)
if not user:
return _render_public_home(service, settings, None, "Domain verified. You can now create a hostname under it.", challenge)
return _render_user_home(service, settings, user, store.list_accounts(int(user["id"])), None, "Domain verified. You can now create a hostname under it.", challenge)
return _render_public_home(service, settings, None, message, challenge)
return _render_public_home(service, settings, None, "Domain verified. You can now create a hostname under it.", challenge)
@app.post("/accounts", response_class=HTMLResponse)
async def self_service_account(request: Request) -> str:
user = await run_in_threadpool(current_user, request)
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"]) if user else None)
account = await run_in_threadpool(service.create_managed_account, username)
else:
claim_secret = _first_form_value(form, "claim_secret")
account = await run_in_threadpool(
@@ -272,50 +237,12 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
_first_form_value(form, "hostname"),
claim_secret,
username,
int(user["id"]) if user else None,
)
if not user:
return _render_public_home(service, settings, account, None)
return _render_user_home(service, settings, user, store.list_accounts(int(user["id"])), account, None, 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_auth_page(settings, "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_auth_page(settings, "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_auth_page(settings, "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
return _render_public_home(service, settings, account, None)
@app.get("/admin", response_class=HTMLResponse)
def admin(_: None = Depends(authenticate_admin)) -> str:
return _render_admin_page(service, settings, store.list_accounts(), store.list_invites(), None)
return _render_admin_page(service, settings, store.list_accounts(), None)
@app.post("/admin/accounts")
async def admin_create_account(request: Request, _: None = Depends(authenticate_admin)) -> Response:
@@ -331,7 +258,7 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
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))
return HTMLResponse(_render_admin_page(service, settings, store.list_accounts(), account))
@app.post("/admin/accounts/delete")
async def admin_delete_account(request: Request, _: None = Depends(authenticate_admin)) -> Response:
@@ -345,13 +272,6 @@ def make_app(settings: DdnsSettings | None = None) -> FastAPI:
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()
@@ -423,16 +343,6 @@ 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:
return _render_public_home(service, settings, created_account, message, challenge)
def _render_public_home(
service: DdnsService,
settings: DdnsSettings,
@@ -490,138 +400,11 @@ def _render_public_home(
)
def _render_user_home(
service: DdnsService,
settings: DdnsSettings,
user: dict[str, str | int],
accounts: list[dict[str, str | int | None]],
created_account: dict[str, str] | None,
message: str | None,
challenge: dict[str, str | None] | None = None,
) -> str:
created_html = _credentials_panel(service, created_account)
message_html = _message_band(message) if message and not challenge else ""
account_rows = "\n".join(_account_row(account, show_owner=False) for account in accounts) or """
<tr><td colspan="6" class="empty">No hostnames yet.</td></tr>
"""
return _page(
"router-dyndns dashboard",
f"""
<main>
{_top_nav(str(user["email"]), True)}
<section class="section">
<div class="container page-heading">
<p class="eyebrow">Dashboard</p>
<h1>Your router hostnames</h1>
<p class="lead">Create provider hostnames immediately, or add a domain and verify it with a DNS TXT record.</p>
</div>
</section>
{message_html}
{created_html}
<section class="section">
<div class="container tool-grid">
<div>
<p class="eyebrow">Provider hostname</p>
<h2>Create a random hostname</h2>
<p class="section-copy">Best for simple FRITZ!Box setups. The hostname and update link are generated for you.</p>
</div>
<form method="post" action="/accounts" class="tool-card">
<input type="hidden" name="mode" value="managed">
<label>Username
<input name="username" placeholder="optional">
</label>
<button type="submit">Create hostname</button>
</form>
</div>
</section>
{_custom_domain_flow(service, challenge, message)}
<section class="section">
<div class="container">
<div class="section-heading">
<div>
<p class="eyebrow">Inventory</p>
<h2>Hostnames</h2>
</div>
</div>
<div class="table-wrap">
<table>
<thead><tr><th>Domain</th><th>User</th><th>IPv4</th><th>IPv6</th><th>Updated</th><th></th></tr></thead>
<tbody>{account_rows}</tbody>
</table>
</div>
</div>
</section>
</main>
""",
)
def _render_auth_page(settings: DdnsSettings, message: str | None) -> str:
invite_field = """
<label>Invite code
<input name="invite" autocomplete="off" required>
</label>
""" if settings.require_invite else ""
return _page(
"Sign in",
f"""
<main>
{_top_nav()}
<section class="hero-band compact">
<div class="container hero-inner">
<p class="eyebrow">Custom domains</p>
<h1>Optional account workspace.</h1>
<p class="lead">Most users do not need this. Magic links handle normal hostname and domain setup; accounts are only for people who want a persistent dashboard.</p>
</div>
</section>
{_message_band(message) if message else ""}
<section class="section">
<div class="container auth-grid">
<div class="tool-card">
<h2>Login</h2>
<form method="post" action="/login" class="stack-form">
<label>Email<input name="email" type="email" autocomplete="email" required></label>
<label>Password<input name="password" type="password" autocomplete="current-password" required></label>
<button type="submit">Login</button>
</form>
</div>
<div class="tool-card secondary-card">
<h2>Register</h2>
<form method="post" action="/register" class="stack-form">
<label>Email<input name="email" type="email" autocomplete="email" required></label>
<label>Password<input name="password" type="password" autocomplete="new-password" minlength="12" required></label>
{invite_field}
<button type="submit">Create account</button>
</form>
</div>
</div>
</section>
</main>
""",
)
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:
if user:
return _render_user_home(service, settings, user, accounts, created_account, message, challenge)
return _render_public_home(service, settings, created_account, message, challenge)
def _top_nav(label: str | None = None, show_logout: bool = False, links: str = "") -> str:
def _top_nav(label: str | None = None, links: str = "") -> str:
account_html = ""
if label:
account_html = f'<span class="nav-label">{html.escape(label)}</span>'
if show_logout:
account_html += '<form method="post" action="/logout"><button class="nav-button" type="submit">Logout</button></form>'
elif not links:
if not links:
account_html += '<a class="nav-button" href="/#custom-domain">Domain setup</a>'
return f"""
<nav class="top-nav">
@@ -717,10 +500,6 @@ def _custom_domain_flow(service: DdnsService, challenge: dict[str, str | None] |
"""
def _challenge_panel(service: DdnsService, challenge: dict[str, str | None] | None, message: str | None) -> str:
return _custom_domain_flow(service, challenge, message)
def _credentials_panel(service: DdnsService, created_account: dict[str, str] | None) -> str:
if not created_account:
return ""
@@ -851,15 +630,11 @@ 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 """
<tr><td colspan="7" class="empty">No accounts yet.</td></tr>
"""
invite_rows = "\n".join(_invite_row(invite) for invite in invites) or """
<tr><td colspan="3" class="empty">No invites yet.</td></tr>
rows = "\n".join(_account_row(account, csrf=csrf) for account in accounts) or """
<tr><td colspan="6" class="empty">No accounts yet.</td></tr>
"""
suffix_help = (
f"Short names are expanded under {html.escape(settings.hostname_suffix)}."
@@ -870,7 +645,7 @@ def _render_admin_page(
"DynDNS Admin",
f"""
<main>
{_top_nav("Operator console", False, '<a href="/records">Records</a><a href="/events">Events</a>')}
{_top_nav("Operator console", '<a href="/records">Records</a><a href="/events">Events</a>')}
<section class="section admin-heading">
<div class="container">
<p class="eyebrow">DynDNS aktiv</p>
@@ -902,27 +677,6 @@ def _render_admin_page(
</div>
</section>
<section class="section">
<div class="container">
<div class="section-heading">
<div>
<p class="eyebrow">Invites</p>
<h2>Registration codes</h2>
</div>
<form method="post" action="/admin/invites">
<input type="hidden" name="csrf" value="{csrf}">
<button type="submit">Create invite</button>
</form>
</div>
<div class="table-wrap">
<table>
<thead><tr><th>Code</th><th>Created</th><th>Used</th></tr></thead>
<tbody>{invite_rows}</tbody>
</table>
</div>
</div>
</section>
<section class="section">
<div class="container">
<div class="section-heading">
@@ -934,7 +688,7 @@ def _render_admin_page(
<div class="table-wrap">
<table>
<thead>
<tr><th>Domain</th><th>User</th><th>Owner</th><th>IPv4</th><th>IPv6</th><th>Updated</th><th></th></tr>
<tr><th>Domain</th><th>User</th><th>IPv4</th><th>IPv6</th><th>Updated</th><th></th></tr>
</thead>
<tbody>{rows}</tbody>
</table>
@@ -946,16 +700,15 @@ def _render_admin_page(
)
def _account_row(account: dict[str, str | int | None], show_owner: bool, csrf: str = "") -> str:
def _account_row(account: dict[str, str | int | None], csrf: str = "") -> str:
hostname = html.escape(str(account["hostname"]))
username = html.escape(str(account["username"]))
ipv4 = html.escape(str(account.get("ipv4") or "-"))
ipv6 = html.escape(str(account.get("ipv6") or "-"))
updated = html.escape(str(account.get("updated_at") or "Never"))
owner = f"<td>{html.escape(str(account.get('owner_user_id') or '-'))}</td>" if show_owner else ""
return f"""
<tr>
<td>{hostname}</td><td>{username}</td>{owner}<td>{ipv4}</td><td>{ipv6}</td><td>{updated}</td>
<td>{hostname}</td><td>{username}</td><td>{ipv4}</td><td>{ipv6}</td><td>{updated}</td>
<td>
<form method="post" action="/admin/accounts/delete">
<input type="hidden" name="csrf" value="{html.escape(csrf)}">
@@ -967,13 +720,6 @@ def _account_row(account: dict[str, str | int | None], show_owner: bool, csrf: s
"""
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"<tr><td><input readonly value=\"{code}\"></td><td>{created}</td><td>{used}</td></tr>"
def _page(title: str, body: str) -> str:
return f"""
<!doctype html>
router_dyndns/ddns_api.py
+4 -15
View File
@@ -1,11 +1,10 @@
import sqlite3
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi import APIRouter, 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
@@ -73,12 +72,6 @@ class RouterUpdateResponse(BaseModel):
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
optional_user_dependency = Depends(optional_user)
@router.post(
"/hostnames/magic",
response_model=HostnameCredentials,
@@ -88,7 +81,7 @@ def create_api_router(settings: DdnsSettings, store: DdnsStore, service: DdnsSer
responses={500: {"model": ErrorResponse}},
)
async def create_magic_hostname(payload: MagicHostnameRequest) -> HostnameCredentials:
account = await run_in_threadpool(service.create_managed_account, payload.username, None)
account = await run_in_threadpool(service.create_managed_account, payload.username)
return _credentials_response(service, account)
@router.get(
@@ -132,11 +125,10 @@ def create_api_router(settings: DdnsSettings, store: DdnsStore, service: DdnsSer
)
async def create_domain_challenge(
payload: DomainChallengeRequest,
user: dict[str, str | int] | None = optional_user_dependency,
) -> DomainChallengeResponse:
domain = normalize_domain(payload.domain)
try:
challenge = await run_in_threadpool(service.create_domain_challenge, domain, int(user["id"]) if user else None)
challenge = await run_in_threadpool(service.create_domain_challenge, domain)
except sqlite3.IntegrityError as exc:
raise HTTPException(status_code=409, detail="domain is already verified") from exc
return DomainChallengeResponse(
@@ -155,10 +147,9 @@ def create_api_router(settings: DdnsSettings, store: DdnsStore, service: DdnsSer
)
async def verify_domain(
payload: VerifyDomainRequest,
user: dict[str, str | int] | None = optional_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 user else None)
found, challenge = await run_in_threadpool(service.verify_domain, domain, payload.claim_secret)
if not found:
return VerifyDomainResponse(
domain=domain,
@@ -183,14 +174,12 @@ def create_api_router(settings: DdnsSettings, store: DdnsStore, service: DdnsSer
)
async def create_custom_hostname(
payload: CustomHostnameRequest,
user: dict[str, str | int] | None = optional_user_dependency,
) -> HostnameCredentials:
account = await run_in_threadpool(
service.create_custom_account,
payload.hostname,
payload.claim_secret,
payload.username,
int(user["id"]) if user else None,
)
return _credentials_response(service, account)
router_dyndns/ddns_models.py
-6
View File
@@ -15,9 +15,6 @@ class DdnsSettings(BaseModel):
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
@@ -56,9 +53,6 @@ class DdnsSettings(BaseModel):
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,
router_dyndns/ddns_security.py
+2 -51
View File
@@ -3,13 +3,11 @@ 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 fastapi import HTTPException, Request
from .ddns_models import DdnsSettings
@@ -46,49 +44,8 @@ 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
secret = settings.admin_password or settings.shared_secret
day = datetime.now(UTC).strftime("%Y-%m-%d")
return hmac.new(secret.encode("utf-8"), f"admin:{day}".encode(), hashlib.sha256).hexdigest()
@@ -109,8 +66,6 @@ def client_ip(request: Request, settings: DdnsSettings | None = None) -> str:
def is_rate_limited_path(path: str) -> bool:
limited_paths = {
"/magic",
"/login",
"/register",
"/accounts",
"/request-domain",
"/verify-domain",
@@ -121,10 +76,6 @@ def is_rate_limited_path(path: str) -> bool:
)
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
router_dyndns/ddns_service.py
+13 -31
View File
@@ -29,18 +29,12 @@ class DdnsService:
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]:
def create_managed_account(self, username: str | 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,
)
return self.store.create_account(self.random_managed_hostname(), username)
except sqlite3.IntegrityError:
continue
raise HTTPException(status_code=500, detail="could not allocate hostname")
@@ -50,36 +44,33 @@ class DdnsService:
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):
if not self.best_verified_parent(normalized, 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)
return self.store.create_account(normalized, username)
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]:
def create_domain_challenge(self, domain: str) -> 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)
existing = self.store.get_domain_challenge(normalized)
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)
return self.store.create_domain_challenge(normalized)
def verify_domain(self, domain: str, claim_secret: str, owner_user_id: int | None = None) -> tuple[bool, dict[str, str | None]]:
def verify_domain(self, domain: str, claim_secret: str) -> 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)
challenge = self.store.get_domain_challenge(normalized, 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"]))
@@ -87,7 +78,7 @@ class DdnsService:
challenge = dict(challenge)
challenge["claim_secret"] = claim_secret
return False, challenge
verified = self.store.verify_domain_challenge(normalized, owner_user_id=owner_user_id)
verified = self.store.verify_domain_challenge(normalized)
if verified is None:
raise HTTPException(status_code=404, detail="domain challenge not found")
verified = dict(verified)
@@ -147,18 +138,13 @@ class DdnsService:
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:
def best_verified_parent(self, hostname: str, 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):
if self.store.is_domain_verified(candidate, claim_secret):
return candidate
if self.store.is_domain_verified(hostname, owner_user_id, claim_secret):
if self.store.is_domain_verified(hostname, claim_secret):
return hostname
return None
@@ -180,10 +166,6 @@ class DdnsService:
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)
router_dyndns/ddns_store.py
+28 -153
View File
@@ -84,12 +84,7 @@ class DdnsStore:
update_ipv6=update_ipv6,
)
def create_account(
self,
hostname: str,
username: str | None = None,
owner_user_id: int | None = None,
) -> dict[str, str]:
def create_account(self, hostname: str, username: str | None = None) -> dict[str, str]:
now = datetime.now(UTC).isoformat()
token = secrets.token_urlsafe(24)
update_slug = secrets.token_urlsafe(32)
@@ -100,10 +95,10 @@ class DdnsStore:
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)
INSERT INTO accounts(hostname, username, token_hash, update_slug, management_hash, created_at, disabled)
VALUES (?, ?, ?, ?, ?, ?, 0)
""",
(hostname, username, token_hash, update_slug, management_hash, owner_user_id, now),
(hostname, username, token_hash, update_slug, management_hash, now),
)
return {
"hostname": hostname,
@@ -113,21 +108,18 @@ class DdnsStore:
"management_slug": management_slug,
}
def list_accounts(self, owner_user_id: int | None = None) -> list[dict[str, str | int | None]]:
def list_accounts(self) -> 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,
rows = conn.execute(
"""
SELECT accounts.hostname, accounts.username, accounts.update_slug,
accounts.created_at, accounts.disabled, records.ipv4, records.ipv6, records.updated_at
FROM accounts
LEFT JOIN records ON records.hostname = accounts.hostname
ORDER BY 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()
).fetchall()
return [dict(row) for row in rows]
def get_account_by_slug(self, update_slug: str) -> dict[str, str | int | None] | None:
@@ -147,7 +139,7 @@ class DdnsStore:
conn.row_factory = sqlite3.Row
account = conn.execute(
"""
SELECT accounts.hostname, accounts.username, accounts.update_slug, accounts.owner_user_id,
SELECT accounts.hostname, accounts.username, accounts.update_slug,
accounts.created_at, accounts.disabled, records.ipv4, records.ipv6, records.updated_at
FROM accounts
LEFT JOIN records ON records.hostname = accounts.hostname
@@ -178,66 +170,47 @@ class DdnsStore:
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]:
def create_domain_challenge(self, domain: str) -> 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]:
existing = conn.execute("SELECT verified_at FROM domain_challenges WHERE domain = ?", (domain,)).fetchone()
if existing and existing[0]:
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)
INSERT INTO domain_challenges(domain, token, claim_hash, 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),
(domain, token, claim_hash, 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:
def verify_domain_challenge(self, domain: str) -> 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()
challenge = conn.execute("SELECT domain, token, verified_at FROM domain_challenges WHERE domain = ?", (domain,)).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))
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:
def get_domain_challenge(self, domain: str, 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:
params = (domain,)
if claim_secret is not None:
sql += " AND claim_hash = ?"
params = (domain, hash_lookup_token(claim_secret))
challenge = conn.execute(sql, params).fetchone()
@@ -256,86 +229,16 @@ class DdnsStore:
result["claim_secret"] = claim_secret
return result
def is_domain_verified(
self,
domain: str,
owner_user_id: int | None = None,
claim_secret: str | None = None,
) -> bool:
def is_domain_verified(self, domain: str, 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:
params = (domain,)
if 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
@@ -451,16 +354,13 @@ class DdnsStore:
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
disabled INTEGER NOT NULL DEFAULT 0
)
"""
)
_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 = ''"):
@@ -473,37 +373,12 @@ class DdnsStore:
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
verified_at TEXT
)
"""
)
_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:
tests/test_ddns.py
+2 -71
View File
@@ -7,15 +7,6 @@ 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": "[email protected]", "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(
@@ -194,39 +185,6 @@ def test_admin_password_query_string_is_rejected(tmp_path: Path) -> None:
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": "[email protected]", "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": "[email protected]",
"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(
@@ -234,11 +192,9 @@ def test_self_service_managed_hostname_generates_random_account(tmp_path: Path)
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"})
@@ -287,9 +243,8 @@ def test_magic_management_link_can_delete_hostname(tmp_path: Path) -> None:
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))
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", admin_password="admin"))
client = TestClient(app)
login_test_user(client)
response = client.post("/accounts", data={"mode": "custom", "hostname": "home.example.net"})
@@ -303,14 +258,12 @@ def test_custom_hostname_after_dns_verification(tmp_path: Path, monkeypatch) ->
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)
challenge_data = store.create_domain_challenge("example.net")
verify = client.post(
"/verify-domain",
data={"domain": "example.net", "claim_secret": challenge_data["claim_secret"]},
@@ -441,7 +394,6 @@ def test_api_custom_hostname_requires_verified_domain(tmp_path: Path, monkeypatc
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
@@ -527,7 +479,6 @@ def test_verified_domain_challenge_cannot_be_overwritten(tmp_path: Path, monkeyp
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(
@@ -540,25 +491,6 @@ def test_verified_domain_challenge_cannot_be_overwritten(tmp_path: Path, monkeyp
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(
@@ -572,7 +504,6 @@ def test_custom_hostname_must_be_inside_publishable_dns_zone(tmp_path: Path, mon
)
)
client = TestClient(app)
login_test_user(client)
challenge = client.post("/api/v1/domains/challenges", json={"domain": "example.net"}).json()
client.post(