Refactor UI assets and harden quality checks

karldns/security.py

@@ -6,6 +6,7 @@ import hmac
 import secrets
 import time
 from collections.abc import Mapping
+from dataclasses import dataclass
 from datetime import UTC, datetime
 from ipaddress import ip_address, ip_network
 
@@ -32,6 +33,23 @@ PRIVATE_PROXY_NETWORKS = tuple(
 )
 
 
+@dataclass(frozen=True, slots=True)
+class _DashboardUnlockCookieParts:
+    timestamp: int
+    max_age_seconds: int
+    signature: str
+
+
+@dataclass(frozen=True, slots=True)
+class DashboardUnlockIdentity:
+    hostname: str
+    management_slug: str
+    password_hash: str
+
+    def payload(self, timestamp: int, max_age_seconds: int) -> bytes:
+        return f"{self.hostname}\0{self.management_slug}\0{self.password_hash}\0{timestamp}\0{max_age_seconds}".encode()
+
+
 def credentials_from_basic_auth(header: str) -> tuple[str | None, str | None]:
     try:
         raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
@@ -69,9 +87,7 @@ def new_bearer_secret() -> str:
 
 
 def sign_dashboard_unlock_cookie(
-    hostname: str,
-    management_slug: str,
-    password_hash: str,
+    identity: DashboardUnlockIdentity,
     signing_secret: str,
     max_age_seconds: int,
     now: int | None = None,
@@ -79,9 +95,7 @@ def sign_dashboard_unlock_cookie(
     timestamp = int(time.time() if now is None else now)
     bounded_max_age = _bounded_dashboard_unlock_seconds(max_age_seconds)
     signature = _dashboard_unlock_signature(
-        hostname,
-        management_slug,
-        password_hash,
+        identity,
         signing_secret,
         timestamp,
         bounded_max_age,
@@ -96,40 +110,63 @@ def verify_dashboard_unlock_cookie(
     signing_secret: str,
     cookie_value: str | None,
 ) -> bool:
-    if not password_hash or not signing_secret or not cookie_value:
+    identity = _dashboard_unlock_identity(hostname, management_slug, password_hash)
+    parts = _active_dashboard_unlock_cookie_parts(cookie_value)
+    if identity is None or not signing_secret or parts is None:
         return False
+    expected = _dashboard_unlock_signature(
+        identity,
+        signing_secret,
+        parts.timestamp,
+        parts.max_age_seconds,
+    )
+    return hmac.compare_digest(parts.signature, expected)
+
+
+def _dashboard_unlock_identity(
+    hostname: str,
+    management_slug: str,
+    password_hash: str | None,
+) -> DashboardUnlockIdentity | None:
+    if not password_hash:
+        return None
+    return DashboardUnlockIdentity(hostname, management_slug, password_hash)
+
+
+def _active_dashboard_unlock_cookie_parts(cookie_value: str | None) -> _DashboardUnlockCookieParts | None:
+    if cookie_value is None:
+        return None
+    parts = _dashboard_unlock_cookie_parts(cookie_value)
+    if parts is None:
+        return None
+    if _dashboard_unlock_cookie_expired(parts, int(time.time())):
+        return None
+    return parts
+
+
+def _dashboard_unlock_cookie_parts(cookie_value: str) -> _DashboardUnlockCookieParts | None:
     try:
-        version, raw_timestamp, raw_max_age, supplied = cookie_value.split(":", 3)
+        version, raw_timestamp, raw_max_age, signature = cookie_value.split(":", 3)
         timestamp = int(raw_timestamp)
         max_age_seconds = _bounded_dashboard_unlock_seconds(int(raw_max_age))
     except ValueError:
-        return False
+        return None
     if version != "v3":
-        return False
-    now = int(time.time())
-    if timestamp > now + 60 or now - timestamp > max_age_seconds:
-        return False
-    expected = _dashboard_unlock_signature(
-        hostname,
-        management_slug,
-        password_hash,
-        signing_secret,
-        timestamp,
-        max_age_seconds,
-    )
-    return hmac.compare_digest(supplied, expected)
+        return None
+    return _DashboardUnlockCookieParts(timestamp, max_age_seconds, signature)
+
+
+def _dashboard_unlock_cookie_expired(parts: _DashboardUnlockCookieParts, now: int) -> bool:
+    return parts.timestamp > now + 60 or now - parts.timestamp > parts.max_age_seconds
 
 
 def _dashboard_unlock_signature(
-    hostname: str,
-    management_slug: str,
-    password_hash: str,
+    identity: DashboardUnlockIdentity,
     signing_secret: str,
     timestamp: int,
     max_age_seconds: int,
 ) -> str:
-    payload = f"{hostname}\0{management_slug}\0{password_hash}\0{timestamp}\0{max_age_seconds}".encode()
-    return hmac.new(signing_secret.encode(), payload, hashlib.sha256).hexdigest()
+    return hmac.new(signing_secret.encode(), identity.payload(timestamp, max_age_seconds), hashlib.sha256).hexdigest()
 
 
 def dashboard_cookie_signing_secret(settings: DdnsSettings) -> str:
@@ -186,19 +223,20 @@ def rate_limit_bucket(path: str) -> str:
 def _trusted_proxy(peer: str, trusted: set[str]) -> bool:
     if not trusted:
         return False
+    return any(_trusted_proxy_candidate(peer, candidate) for candidate in trusted)
+
+
+def _trusted_proxy_candidate(peer: str, candidate: str) -> bool:
     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
+    try:
+        if "/" in candidate:
+            return peer_ip in ip_network(candidate, strict=False)
+        return peer_ip == ip_address(candidate)
+    except ValueError:
+        return False
 
 
 def _should_trust_proxy_headers(peer: str, settings: DdnsSettings | None) -> bool:
@@ -250,13 +288,28 @@ def _valid_header_ip(value: str) -> str | None:
 
 
 def _strip_header_ip(value: str) -> str:
-    cleaned = value.strip().strip('"').strip("'")
-    if not cleaned or cleaned.lower() == "unknown":
+    cleaned = _strip_header_ip_quotes(value)
+    if _empty_header_ip(cleaned):
         return ""
-    if cleaned.startswith("[") and "]" in cleaned:
-        return cleaned[1 : cleaned.index("]")]
-    if cleaned.count(":") == 1:
-        host, port = cleaned.rsplit(":", 1)
-        if port.isdigit():
-            return host
-    return cleaned
+    return _strip_bracketed_ip(cleaned) or _strip_host_port(cleaned) or cleaned
+
+
+def _strip_header_ip_quotes(value: str) -> str:
+    return value.strip().strip('"').strip("'")
+
+
+def _empty_header_ip(value: str) -> bool:
+    return not value or value.lower() == "unknown"
+
+
+def _strip_bracketed_ip(value: str) -> str:
+    if value.startswith("[") and "]" in value:
+        return value[1 : value.index("]")]
+    return ""
+
+
+def _strip_host_port(value: str) -> str:
+    if value.count(":") != 1:
+        return ""
+    host, port = value.rsplit(":", 1)
+    return host if port.isdigit() else ""