Extract CORS/auth/secret helpers into digest.web.security

.docs/doing/web-security-helper-extraction/proposal.md

@@ -0,0 +1,36 @@
+# Web Security Helper Extraction Proposal
+
+## Why
+
+`src/digest/web/app.py` still holds the CORS, API-auth, and secret-redaction
+helpers as a cohesive top-of-file block. They are pure, environment-driven
+functions with their own configuration constants and no route or `create_app`
+state, so they can move into a focused module — continuing the
+helper-extraction pattern already applied for `feedback`, `schedule`,
+`run_progress`, and `sources`.
+
+## Scope
+
+- Move the security helpers and their constants into a new
+  `digest.web.security` module: `DEFAULT_WEB_CORS_ORIGINS`,
+  `DEFAULT_WEB_CORS_ORIGIN_REGEX`, `DEFAULT_WEB_API_AUTH_MODE`,
+  `DEFAULT_WEB_API_TOKEN_HEADER`, `ALLOWED_WEB_API_AUTH_MODES`,
+  `REDACTED_SECRET`, `SECRET_KEY_RE`, plus `_cors_allowed_origins`,
+  `_cors_allow_origin_regex`, `_web_api_auth_mode`, `_web_api_token`,
+  `_web_api_token_header`, `_api_auth_decision`, `_is_secret_key`,
+  `_redact_secrets`, and `_rehydrate_redacted_value`.
+- Re-import the eight route-facing helpers into `digest.web.app` so the CORS
+  middleware and auth middleware keep working. `_is_secret_key` and the
+  non-route constants stay internal to `digest.web.security`.
+- Update `test_web_security` and `test_web_cors` to import the security symbols
+  from `digest.web.security` (their new owner).
+- Verify security/CORS tests and the full backend/security checks.
+
+## Non-goals
+
+- No auth-decision, CORS, or redaction behavior changes.
+- No API shape changes.
+- No route or scheduler-loop changes.
+- No frontend changes.
+- The run-mode helpers (`_resolve_run_mode`, `_web_live_run_options`, …) and
+  `RUN_MODE_OPTIONS` stay in `digest.web.app`; they are a separate concern.

.docs/doing/web-security-helper-extraction/tasks.md

@@ -0,0 +1,22 @@
+# Tasks
+
+- [x] Create `digest.web.security` with the seven config constants and nine
+      CORS/auth/redaction helper functions.
+- [x] Remove the moved constants and functions from `digest.web.app` and import
+      the eight route-facing names back; drop now-unused `os`/`re`/`hmac`
+      imports.
+- [x] Update `test_web_security` and `test_web_cors` to import the security
+      symbols from `digest.web.security`.
+- [x] Confirm no circular import (`import digest.web.app` succeeds).
+- [x] `ruff check src tests` is clean.
+- [x] Full backend test suite passes (focus: `test_web_security`,
+      `test_web_cors`, `test_web_live_run_options`).
+
+## Result
+
+- `digest/web/app.py` shrank from 1875 to 1751 lines; new
+  `digest/web/security.py` holds the 143-line module.
+- Cumulative across the session's four extractions (schedule, run_progress,
+  sources, security), `app.py` dropped from 2589 to 1751 lines (-838, -32%).
+- `os`, `re`, and `hmac` are no longer imported in `app.py` — all of their uses
+  lived in the moved security cluster.

.secrets.baseline

@@ -136,13 +136,13 @@
         "line_number": 71
       }
     ],
-    "src/digest/web/app.py": [
+    "src/digest/web/security.py": [
       {
         "type": "Secret Keyword",
-        "filename": "src/digest/web/app.py",
+        "filename": "src/digest/web/security.py",
         "hashed_secret": "5bfb2389f53db8c241ab122ea096f85caaf7904b",
         "is_verified": false,
-        "line_number": 95
+        "line_number": 32
       }
     ],
     "tests/test_responses_api_summarizer.py": [
@@ -164,5 +164,5 @@
       }
     ]
   },
-  "generated_at": "2026-06-20T19:15:20Z"
+  "generated_at": "2026-06-20T19:27:42Z"
 }

src/digest/web/app.py

@@ -3,11 +3,8 @@
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
-import hmac
 import json
-import os
 from pathlib import Path
-import re
 import threading
 import uuid
 from typing import Any
@@ -67,6 +64,16 @@
     _schedule_due_slot_utc,
     _schedule_status_payload,
 )
+from digest.web.security import (
+    _api_auth_decision,
+    _cors_allow_origin_regex,
+    _cors_allowed_origins,
+    _redact_secrets,
+    _rehydrate_redacted_value,
+    _web_api_auth_mode,
+    _web_api_token,
+    _web_api_token_header,
+)
 from digest.web.sources import (
     _apply_blocked_source_preference,
     _build_source_health_items,
@@ -75,137 +82,6 @@
 )
 
 
-DEFAULT_WEB_CORS_ORIGINS = [
-    "http://127.0.0.1:5173",
-    "http://localhost:5173",
-    "http://127.0.0.1:4173",
-    "http://localhost:4173",
-]
-
-DEFAULT_WEB_CORS_ORIGIN_REGEX = (
-    r"^https?://((localhost|127\.0\.0\.1|0\.0\.0\.0)"
-    r"|(10\.\d+\.\d+\.\d+)"
-    r"|(192\.168\.\d+\.\d+)"
-    r"|(172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+))(:\d+)?$"
-)
-
-DEFAULT_WEB_API_AUTH_MODE = "required"
-DEFAULT_WEB_API_TOKEN_HEADER = "X-Digest-Api-Token"
-ALLOWED_WEB_API_AUTH_MODES = {"required", "optional", "off"}
-REDACTED_SECRET = "__REDACTED__"
-SECRET_KEY_RE = re.compile(r"(token|secret|password|api[_-]?key)", re.IGNORECASE)
-
-
-def _cors_allowed_origins() -> list[str]:
-    raw = str(os.getenv("DIGEST_WEB_CORS_ORIGINS", "") or "").strip()
-    if not raw:
-        return list(DEFAULT_WEB_CORS_ORIGINS)
-
-    values = [part.strip() for part in raw.split(",")]
-    return [value for value in values if value]
-
-
-def _cors_allow_origin_regex() -> str:
-    raw = str(os.getenv("DIGEST_WEB_CORS_ORIGIN_REGEX", "") or "").strip()
-    if raw:
-        return raw
-    return DEFAULT_WEB_CORS_ORIGIN_REGEX
-
-
-def _web_api_auth_mode() -> str:
-    raw = (
-        str(os.getenv("DIGEST_WEB_API_AUTH_MODE", DEFAULT_WEB_API_AUTH_MODE) or "")
-        .strip()
-        .lower()
-    )
-    if raw in ALLOWED_WEB_API_AUTH_MODES:
-        return raw
-    return DEFAULT_WEB_API_AUTH_MODE
-
-
-def _web_api_token() -> str:
-    return str(os.getenv("DIGEST_WEB_API_TOKEN", "") or "").strip()
-
-
-def _web_api_token_header() -> str:
-    raw = str(
-        os.getenv("DIGEST_WEB_API_TOKEN_HEADER", DEFAULT_WEB_API_TOKEN_HEADER) or ""
-    ).strip()
-    if not raw:
-        return DEFAULT_WEB_API_TOKEN_HEADER
-    return raw
-
-
-def _api_auth_decision(
-    *,
-    path: str,
-    method: str,
-    mode: str,
-    configured_token: str,
-    presented_token: str,
-) -> tuple[bool, int, str]:
-    if not path.startswith("/api/"):
-        return True, 200, ""
-    if method.upper() == "OPTIONS":
-        return True, 200, ""
-    if path == "/api/health":
-        return True, 200, ""
-    if mode == "off":
-        return True, 200, ""
-    if mode == "required" and not configured_token:
-        return (
-            False,
-            503,
-            "DIGEST_WEB_API_TOKEN is required when DIGEST_WEB_API_AUTH_MODE=required",
-        )
-    if configured_token:
-        if not presented_token:
-            return False, 401, "missing API token"
-        if not hmac.compare_digest(presented_token, configured_token):
-            return False, 401, "invalid API token"
-    return True, 200, ""
-
-
-def _is_secret_key(key: str) -> bool:
-    return bool(SECRET_KEY_RE.search((key or "").strip()))
-
-
-def _redact_secrets(value: Any) -> Any:
-    if isinstance(value, dict):
-        out: dict[str, Any] = {}
-        for key, item in value.items():
-            if _is_secret_key(str(key)):
-                if isinstance(item, str):
-                    out[key] = REDACTED_SECRET if item else ""
-                elif item is None:
-                    out[key] = None
-                else:
-                    out[key] = REDACTED_SECRET
-                continue
-            out[key] = _redact_secrets(item)
-        return out
-    if isinstance(value, list):
-        return [_redact_secrets(item) for item in value]
-    return value
-
-
-def _rehydrate_redacted_value(candidate: Any, current: Any) -> Any:
-    if isinstance(candidate, str) and candidate == REDACTED_SECRET:
-        return current
-    if isinstance(candidate, dict) and isinstance(current, dict):
-        out_dict: dict[str, Any] = {}
-        for key, item in candidate.items():
-            out_dict[key] = _rehydrate_redacted_value(item, current.get(key))
-        return out_dict
-    if isinstance(candidate, list) and isinstance(current, list):
-        out_list: list[Any] = []
-        for idx, item in enumerate(candidate):
-            ref = current[idx] if idx < len(current) else None
-            out_list.append(_rehydrate_redacted_value(item, ref))
-        return out_list
-    return candidate
-
-
 RUN_MODE_OPTIONS: dict[str, dict[str, bool]] = {
     "fresh_only": {
         "use_last_completed_window": True,

src/digest/web/security.py

@@ -0,0 +1,143 @@
+"""CORS, API-auth, and secret-redaction helpers for the web control plane.
+
+These are pure functions and configuration constants extracted from
+``digest.web.app``. They read environment configuration and operate on plain
+values; they hold no route or application state.
+"""
+
+from __future__ import annotations
+
+import hmac
+import os
+import re
+from typing import Any
+
+DEFAULT_WEB_CORS_ORIGINS = [
+    "http://127.0.0.1:5173",
+    "http://localhost:5173",
+    "http://127.0.0.1:4173",
+    "http://localhost:4173",
+]
+
+DEFAULT_WEB_CORS_ORIGIN_REGEX = (
+    r"^https?://((localhost|127\.0\.0\.1|0\.0\.0\.0)"
+    r"|(10\.\d+\.\d+\.\d+)"
+    r"|(192\.168\.\d+\.\d+)"
+    r"|(172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+))(:\d+)?$"
+)
+
+DEFAULT_WEB_API_AUTH_MODE = "required"
+DEFAULT_WEB_API_TOKEN_HEADER = "X-Digest-Api-Token"
+ALLOWED_WEB_API_AUTH_MODES = {"required", "optional", "off"}
+REDACTED_SECRET = "__REDACTED__"
+SECRET_KEY_RE = re.compile(r"(token|secret|password|api[_-]?key)", re.IGNORECASE)
+
+
+def _cors_allowed_origins() -> list[str]:
+    raw = str(os.getenv("DIGEST_WEB_CORS_ORIGINS", "") or "").strip()
+    if not raw:
+        return list(DEFAULT_WEB_CORS_ORIGINS)
+
+    values = [part.strip() for part in raw.split(",")]
+    return [value for value in values if value]
+
+
+def _cors_allow_origin_regex() -> str:
+    raw = str(os.getenv("DIGEST_WEB_CORS_ORIGIN_REGEX", "") or "").strip()
+    if raw:
+        return raw
+    return DEFAULT_WEB_CORS_ORIGIN_REGEX
+
+
+def _web_api_auth_mode() -> str:
+    raw = (
+        str(os.getenv("DIGEST_WEB_API_AUTH_MODE", DEFAULT_WEB_API_AUTH_MODE) or "")
+        .strip()
+        .lower()
+    )
+    if raw in ALLOWED_WEB_API_AUTH_MODES:
+        return raw
+    return DEFAULT_WEB_API_AUTH_MODE
+
+
+def _web_api_token() -> str:
+    return str(os.getenv("DIGEST_WEB_API_TOKEN", "") or "").strip()
+
+
+def _web_api_token_header() -> str:
+    raw = str(
+        os.getenv("DIGEST_WEB_API_TOKEN_HEADER", DEFAULT_WEB_API_TOKEN_HEADER) or ""
+    ).strip()
+    if not raw:
+        return DEFAULT_WEB_API_TOKEN_HEADER
+    return raw
+
+
+def _api_auth_decision(
+    *,
+    path: str,
+    method: str,
+    mode: str,
+    configured_token: str,
+    presented_token: str,
+) -> tuple[bool, int, str]:
+    if not path.startswith("/api/"):
+        return True, 200, ""
+    if method.upper() == "OPTIONS":
+        return True, 200, ""
+    if path == "/api/health":
+        return True, 200, ""
+    if mode == "off":
+        return True, 200, ""
+    if mode == "required" and not configured_token:
+        return (
+            False,
+            503,
+            "DIGEST_WEB_API_TOKEN is required when DIGEST_WEB_API_AUTH_MODE=required",
+        )
+    if configured_token:
+        if not presented_token:
+            return False, 401, "missing API token"
+        if not hmac.compare_digest(presented_token, configured_token):
+            return False, 401, "invalid API token"
+    return True, 200, ""
+
+
+def _is_secret_key(key: str) -> bool:
+    return bool(SECRET_KEY_RE.search((key or "").strip()))
+
+
+def _redact_secrets(value: Any) -> Any:
+    if isinstance(value, dict):
+        out: dict[str, Any] = {}
+        for key, item in value.items():
+            if _is_secret_key(str(key)):
+                if isinstance(item, str):
+                    out[key] = REDACTED_SECRET if item else ""
+                elif item is None:
+                    out[key] = None
+                else:
+                    out[key] = REDACTED_SECRET
+                continue
+            out[key] = _redact_secrets(item)
+        return out
+    if isinstance(value, list):
+        return [_redact_secrets(item) for item in value]
+    return value
+
+
+def _rehydrate_redacted_value(candidate: Any, current: Any) -> Any:
+    if isinstance(candidate, str) and candidate == REDACTED_SECRET:
+        return current
+    if isinstance(candidate, dict) and isinstance(current, dict):
+        out_dict: dict[str, Any] = {}
+        for key, item in candidate.items():
+            out_dict[key] = _rehydrate_redacted_value(item, current.get(key))
+        return out_dict
+    if isinstance(candidate, list) and isinstance(current, list):
+        out_list: list[Any] = []
+        for idx, item in enumerate(candidate):
+            ref = current[idx] if idx < len(current) else None
+            out_list.append(_rehydrate_redacted_value(item, ref))
+        return out_list
+    return candidate