From d99100bf5998b7a895f05e97f9e5c0bffcf85d68 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Wed, 5 Aug 2026 18:48:17 +0200 Subject: [PATCH 1/7] docs(site_lock): design spec for optional site-wide password gate Staging/pre-launch gate behind one shared password, off by default. Middleware-only module sorting after Auth so it wraps outermost; DB-backed settings (admin UI only); logged-in admins bypass as the lockout escape hatch. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- .../specs/2026-08-05-site-lock-design.md | 215 ++++++++++++++++++ 1 file changed, 215 insertions(+) create mode 100644 docs/superpowers/specs/2026-08-05-site-lock-design.md diff --git a/docs/superpowers/specs/2026-08-05-site-lock-design.md b/docs/superpowers/specs/2026-08-05-site-lock-design.md new file mode 100644 index 00000000..39435770 --- /dev/null +++ b/docs/superpowers/specs/2026-08-05-site-lock-design.md @@ -0,0 +1,215 @@ +# Site Lock: optional site-wide password gate + +**Date:** 2026-08-05 +**Status:** approved-to-implement (proceeding under an active `/goal` directive) + +## Goal + +Let an operator hide the entire site behind a single shared password — a +staging / pre-launch gate. Anyone without the password sees nothing: not the +landing page, not the API, not even that a login form exists. + +**Off by default.** A deployment that never touches the setting behaves exactly +as it does today, and the disabled path must cost effectively nothing per +request. + +## Current state + +- Modules install middleware via `ModuleBase.register_middleware(app)`. The + host calls these in topological order inside `install_middleware` + (`framework/hosting/simple_module_hosting/_phase_helpers.py`). Because + Starlette's `add_middleware` is LIFO, **a module that sorts later wraps + outermost**. +- `AuthMiddleware` (`modules/auth/auth/middleware.py`) resolves the user via + the single registered `AuthProvider`, then redirects unauthenticated browser + requests to the provider's login URL and returns 401 JSON for `/api/*`. +- Public paths come from two places only: the `_FRAMEWORK_PUBLIC_*` constants + in the auth middleware, and `provider.get_public_paths()`. **There is no + registry a third module can add to** — this is what rules out mounting a + normal view route for the gate page without first extending the auth module. +- DB-backed per-module settings go through + `settings.registration.register_module_settings(app, package, cls, factory)`, + hydrate at startup, and hot-swap through + `settings.reload.apply_changes_and_reload` which publishes `SettingsReloaded`. +- `settings/_module_settings.py` masks any field whose name matches + `(password|secret|api[_-]?key|private[_-]?key|token[_-]?secret)` in the admin + UI. +- `UserContext` (`modules/auth/auth/contracts/schemas.py`) exposes + `roles: list[str]`; the admin role name is `"admin"` + (`users.constants.ADMIN_ROLE_NAME`). + +## Decisions + +Settled during brainstorming: + +1. **Staging / pre-launch gate** — one shared password in front of the whole + site, including the login page. Not a "public pages only" gate, not a + maintenance-mode banner. +2. **Configured from the admin UI only** (DB-backed). No `SM_SITE_LOCK_*` env + vars. +3. **Logged-in admins always bypass** the gate — this is the lockout escape + hatch, chosen over shipping a CLI command. +4. **Self-contained middleware-rendered gate page**, not an Inertia `.tsx` page. + This avoids extending the auth module with a public-paths registry, and + means a locked site serves exactly one page and leaks nothing else. + +## Design + +A new module `modules/site_lock/` that installs exactly one middleware. No +models, no migration, no `.tsx` pages, no routes registered with the app router. + +### 1. Placement in the pipeline + +```python +meta = ModuleMeta(name="SiteLock", depends_on=["Settings", "Auth"]) +``` + +- `Settings` — so `register_module_settings` can reach + `app.state.settings.module_registry` during `register_settings`. +- `Auth` — puts SiteLock after Auth in topological order, so its + `add_middleware` call happens later, so it **wraps outermost and executes + before `AuthMiddleware`**. + +Resulting order on a request: + +``` +CorrelationId → RequestLogging → SecurityHeaders → Session → [SiteLock] → Auth → Locale → InertiaLayoutData → app +``` + +Running before Auth is what makes an anonymous visitor see the gate rather than +a redirect to `/users/login`. `SessionMiddleware` sits *outside* SiteLock, so +`scope["session"]` is readable and writes are persisted on the way out — the +same mechanism `AuthMiddleware` already relies on for its `next` key. + +Raw ASGI class, not `BaseHTTPMiddleware`, per the convention stated in +`framework/hosting/simple_module_hosting/middleware.py`. + +### 2. Settings + +`SiteLockSettings(BaseSettings)`, registered as package `site_lock`, DB-backed +and hot-reloadable: + +| Field | Type | Default | Notes | +|---|---|---|---| +| `enabled` | `bool` | `False` | the off-by-default guarantee | +| `password` | `str` | `""` | name contains `password` → auto-masked in the settings UI | +| `message` | `str` | `""` | optional line shown on the gate page | + +A `@model_validator` rejects `enabled=True` with a blank/whitespace password, so +`apply_changes_and_reload` raises a `ValidationError` and the settings screen +refuses the change rather than gating the site behind an empty string. + +State object `SiteLockState(settings=...)` on `app.state.site_lock`, satisfying +`SM012`. + +### 3. Request handling + +In order: + +1. `scope["type"] != "http"` → pass through (websockets are not gated). +2. `not settings.enabled` → pass through. **The default-off fast path: one + attribute read and a boolean test — no `Request` construction, no + allocation.** +3. Path is `/health` (prefix) → pass through, always. Gating it would fail + Kubernetes liveness/readiness probes and get the pod killed. +4. Path is the unlock endpoint `/__unlock`: + - `GET` → serve the gate page (200). + - `POST` → read the form body, compare with `secrets.compare_digest`. On + success, write the session marker and `303` to the sanitised `next` + target. On failure, re-serve the page with an error and status `401`. + - Any other method → `405`. +5. `scope["session"].get("site_lock") == fingerprint` → pass through. The + fingerprint is a truncated `sha256` of the current password, so **rotating + the password invalidates every existing unlock session**. +6. Admin bypass: resolve via `app.state.auth.auth_provider.resolve_user(request)`. + If a user comes back and `"admin"` is in `user.roles`, stamp the session + marker and pass through. Stamping is what keeps this to **one resolve per + session rather than one per request**. Guarded for a `None` provider and for + a provider that raises. +7. Otherwise the request is gated: + - `/api/*` prefix, or an `Authorization` header present → `403` JSON + `{"detail": "Site is locked"}`. 403 rather than 401 so clients do not + start an auth flow that cannot succeed. + - Everything else → `302` to `/__unlock?next=`. + +All gate and gated responses carry `Cache-Control: no-store` so no proxy or CDN +caches either the gate page or a gated response. + +### 4. The gate page + +`site_lock/templates/unlock.html` — one self-contained file with inlined CSS, +read once at import time and cached in a module-level constant. Rendered by +plain string substitution (`string.Template`), not Jinja, to avoid wiring a +template loader for a single static asset. + +Every interpolated value (`message`, `next`, error text) is passed through +`html.escape`. Because the page needs no external CSS or JS, `/static/` needs +no exemption — a locked site serves exactly one page and nothing else. + +`next` is sanitised before being echoed into the form or used in a redirect: +only same-site absolute paths (starting with a single `/`, not `//`) are +accepted, anything else falls back to `/`. This prevents the gate from being +used as an open redirect. + +### 5. Brute-force protection + +One shared secret is the whole security boundary here, so the unlock endpoint +needs a limiter. `users` already has `LoginRateLimiter`, but importing it would +hard-couple `site_lock` to `users` and break under the `keycloak` provider. + +A small in-memory per-IP limiter lives in `site_lock/rate_limit.py`: **10 failed +attempts within 5 minutes** trigger a **15-minute cooldown** during which +`POST /__unlock` returns `429`. In-memory is consistent with the existing +`users` limiter and adequate for the single-process staging deployments this +feature targets. + +These three thresholds are **module-level constants, not settings fields** — +they are not part of the configurable surface. Keeping them out of +`SiteLockSettings` keeps the admin UI to the three fields that matter and +avoids offering an operator a way to weaken the only brute-force defence. + +Client IP comes from `scope["client"]`, which is what the app already trusts. + +## Known limitation: cold-start lockout + +The admin bypass rescues an admin who **already holds a live session**. That +covers the realistic footgun: you enable the gate, typo the password, and are +still holding the session that lets you go straight back to Settings and fix it. + +It does **not** cover a cold start — session expired, password forgotten, no +admin currently signed in. Recovery then requires deleting the settings +override row directly in the database. This was an explicit choice (admin +bypass was selected over shipping a `smpy site-lock disable` CLI command) and +is documented in the module README. Adding the CLI command later is a +self-contained follow-up. + +## Testing + +- Disabled by default: a request passes through untouched, and the middleware + does not touch the session or the auth provider. +- Enabled, no session: browser request → `302 /__unlock`; `/api/*` → `403` JSON. +- `/health` is never gated, enabled or not. +- Correct password → session marker set → the next request passes through. +- Wrong password → `401`, no session marker written. +- Rotating the password invalidates a previously-valid unlock session. +- An admin with a live session bypasses; an authenticated **non-admin** does not. +- The validator rejects `enabled=True` with a blank password. +- `next` sanitisation rejects `//evil.example` and absolute URLs. +- Rate limiter returns `429` after the configured number of failures. +- Middleware ordering: SiteLock wraps outside `AuthMiddleware` — follow the + existing `framework/hosting/tests/test_middleware_order.py` pattern. + +## Out of scope + +- Per-user or per-role gating (that is what the existing auth system is for). +- Persisting unlock state anywhere other than the signed session cookie. +- A distributed/Redis-backed rate limiter. +- Env-var configuration (explicitly declined — admin UI only). +- A `smpy site-lock disable` CLI command (see the lockout limitation above). + +## Non-effects + +No migration (no models). No `make gen-pages` run (no `.tsx`). Trips no +`make doctor` diagnostics: `SM012` is satisfied by `app.state.site_lock`, +`SM017` and `SM019` do not apply because the module ships no pages and +registers no view routes. From 4ebb4f9ebac1e2eb0cff77c2fb13a92e96a1f04f Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Wed, 5 Aug 2026 18:53:58 +0200 Subject: [PATCH 2/7] docs(site_lock): implementation plan Six TDD tasks: package scaffold + settings, attempt limiter, gate page, middleware, admin-bypass coverage, host wiring and docs. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- .../superpowers/plans/2026-08-05-site-lock.md | 1357 +++++++++++++++++ .../specs/2026-08-05-site-lock-design.md | 15 +- 2 files changed, 1366 insertions(+), 6 deletions(-) create mode 100644 docs/superpowers/plans/2026-08-05-site-lock.md diff --git a/docs/superpowers/plans/2026-08-05-site-lock.md b/docs/superpowers/plans/2026-08-05-site-lock.md new file mode 100644 index 00000000..984f596e --- /dev/null +++ b/docs/superpowers/plans/2026-08-05-site-lock.md @@ -0,0 +1,1357 @@ +# Site Lock Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship a `site_lock` module that hides the entire site behind one shared password — a staging / pre-launch gate — off by default. + +**Architecture:** A single raw-ASGI middleware installed by a new module that declares `depends_on=["Settings", "Auth"]`. Because the host installs module middleware in topological order and Starlette's `add_middleware` is LIFO, sorting after `Auth` makes SiteLock wrap outermost and execute *before* `AuthMiddleware` — so an anonymous visitor sees the gate rather than a redirect to the login page. Configuration is DB-backed through `register_module_settings`; unlock state lives in the signed session cookie. + +**Tech Stack:** Python 3.12, FastAPI/Starlette, pydantic-settings, pytest (`asyncio_mode=auto`), httpx `ASGITransport`, uv workspaces. + +**Spec:** `docs/superpowers/specs/2026-08-05-site-lock-design.md` + +## Global Constraints + +- **300-line cap** on every `.py` file, enforced by `scripts/check_file_size.py`. +- **SQLModel is the project-wide model standard** — not relevant here (this module has no DB models and adds no migration), but do not introduce `pydantic.BaseModel`. `SiteLockSettings` subclasses `pydantic_settings.BaseSettings`, matching every other module's settings class. +- **Raw ASGI middleware only** — never `BaseHTTPMiddleware`, per `framework/hosting/simple_module_hosting/middleware.py`. +- **Package version `0.0.26`**, and workspace deps pinned `==0.0.26`, matching every other module in this repo at this commit. +- **Module name is `SiteLock`; package/settings key is `site_lock`.** These exact strings appear in `app.state.site_lock`, the settings registry key, and `ModuleMeta.name`. +- **Never import from `users`** — `site_lock` must work under the `keycloak` auth provider too. The admin role name is duplicated as a local constant on purpose. +- `make lint` runs Ruff format-check, Ruff, `ty`, Biome, `tsc`, and the file-size cap. `make test-py` runs pytest. +- Settings fields whose name matches `(password|secret|api[_-]?key|private[_-]?key|token[_-]?secret)` are auto-masked in the admin UI. The field **must** be named `password` to get this for free. + +--- + +## File Structure + +**New package — `modules/site_lock/`:** + +| File | Responsibility | +|---|---| +| `pyproject.toml` | Package metadata + `simple_module` entry point | +| `README.md` | Usage, and the documented cold-start lockout limitation | +| `site_lock/__init__.py` | Empty marker | +| `site_lock/py.typed` | Typing marker | +| `site_lock/constants.py` | Every literal: paths, session key, admin role, rate-limit thresholds | +| `site_lock/settings.py` | `SiteLockSettings` + the enabled-requires-password validator | +| `site_lock/state.py` | `SiteLockState` mounted at `app.state.site_lock` | +| `site_lock/rate_limit.py` | `AttemptLimiter` — in-memory per-IP failure tracking | +| `site_lock/page.py` | `safe_next()` + `render_unlock_page()` | +| `site_lock/templates/unlock.html` | Self-contained gate page, inlined CSS | +| `site_lock/middleware.py` | `SiteLockMiddleware` — the gate itself | +| `site_lock/module.py` | `SiteLockModule` wiring settings + middleware | +| `tests/test_*.py` | One test file per unit above | + +**Modified:** + +| File | Change | +|---|---| +| `pyproject.toml` | Add `modules/site_lock` to workspace members and `modules/site_lock/tests` to `testpaths` | +| `host/pyproject.toml` | Add `simple_module_site_lock` to dependencies and `[tool.uv.sources]` | +| `framework/hosting/tests/test_middleware_order.py` | Insert `"SiteLockMiddleware"` into both expected tuples | +| `framework/cli/simple_module_cli/catalog.py` | Add the `site_lock` catalog entry | +| `docs/modules/site_lock.md`, `docs/.vitepress/config.ts`, `docs/index.md` | Docs page + nav | + +Splitting `page.py` out of `middleware.py` keeps both well under the 300-line cap and lets the escaping/`next`-sanitisation logic be unit-tested without an ASGI harness. + +--- + +### Task 1: Package scaffold + settings + +**Files:** +- Create: `modules/site_lock/pyproject.toml`, `modules/site_lock/README.md`, `modules/site_lock/site_lock/__init__.py`, `modules/site_lock/site_lock/py.typed`, `modules/site_lock/site_lock/constants.py`, `modules/site_lock/site_lock/settings.py`, `modules/site_lock/site_lock/state.py` +- Modify: `pyproject.toml` (workspace members + testpaths) +- Test: `modules/site_lock/tests/test_settings.py` + +**Interfaces:** +- Produces: `SiteLockSettings(enabled: bool, password: str, message: str)`; `SiteLockState(settings, limiter)`; constants module `site_lock.constants`. + +- [ ] **Step 1: Create the package skeleton** + +`modules/site_lock/pyproject.toml`: + +```toml +[project] +name = "simple_module_site_lock" +version = "0.0.26" +description = "Optional site-wide password gate (staging / pre-launch) for simple_module apps" +readme = "README.md" +license = "MIT" +requires-python = ">=3.12" +authors = [{ name = "Anto Subash", email = "antosubash@live.com" }] +keywords = ["simple-module", "password", "staging", "gate"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Framework :: FastAPI", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.12", + "Topic :: Internet :: WWW/HTTP", + "Typing :: Typed", +] +dependencies = [ + "simple_module_core==0.0.26", + "simple_module_hosting==0.0.26", + "simple_module_settings==0.0.26", + "simple_module_auth==0.0.26", +] + +[project.entry-points.simple_module] +site_lock = "site_lock.module:SiteLockModule" + +[project.urls] +Homepage = "https://github.com/antosubash/simple_module_python" +Repository = "https://github.com/antosubash/simple_module_python" +Issues = "https://github.com/antosubash/simple_module_python/issues" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["site_lock"] + +[tool.uv.sources] +simple_module_core = { workspace = true } +simple_module_hosting = { workspace = true } +simple_module_settings = { workspace = true } +simple_module_auth = { workspace = true } +``` + +Create empty `site_lock/__init__.py` and `site_lock/py.typed`. + +- [ ] **Step 2: Register the package in the workspace** + +In the root `pyproject.toml`, add `"modules/site_lock",` to `[tool.uv.workspace] members` (after `"modules/audit_log",`), and append `"modules/site_lock/tests"` to `[tool.pytest.ini_options] testpaths`. + +Then run: `uv sync --all-packages` +Expected: `simple_module_site_lock` installs as an editable workspace package. + +- [ ] **Step 3: Write the failing settings test** + +`modules/site_lock/tests/test_settings.py`: + +```python +"""SiteLockSettings defaults and the enabled-requires-password guard.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError +from site_lock.settings import SiteLockSettings + + +def test_disabled_by_default() -> None: + s = SiteLockSettings() + assert s.enabled is False + assert s.password == "" + assert s.message == "" + + +def test_enabled_with_password_is_valid() -> None: + s = SiteLockSettings(enabled=True, password="hunter2") + assert s.enabled is True + + +@pytest.mark.parametrize("password", ["", " "]) +def test_enabled_without_password_is_rejected(password: str) -> None: + with pytest.raises(ValidationError): + SiteLockSettings(enabled=True, password=password) + + +def test_password_field_is_masked_by_settings_ui() -> None: + # The admin UI masks fields by name; `password` must match that regex. + from settings._module_settings import is_secret_field + + assert is_secret_field("password") is True +``` + +- [ ] **Step 4: Run it and watch it fail** + +Run: `uv run pytest modules/site_lock/tests/test_settings.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'site_lock.settings'` + +- [ ] **Step 5: Write constants, settings, and state** + +`site_lock/constants.py`: + +```python +"""Every literal the site_lock module depends on.""" + +from __future__ import annotations + +MODULE_NAME = "SiteLock" +MODULE_PACKAGE = "site_lock" + +# Dependencies — names must match the other modules' ``ModuleMeta.name``. +MODULE_SETTINGS = "Settings" +MODULE_AUTH = "Auth" + +# The gate endpoint. Double-underscore prefix keeps it clear of app routes. +UNLOCK_PATH = "/__unlock" +SESSION_KEY = "site_lock" + +# Never gated: Kubernetes liveness/readiness probes must always succeed. +HEALTH_PREFIX = "/health" +API_PREFIX = "/api/" + +# Duplicated rather than imported from ``users`` — site_lock must also work +# under the ``keycloak`` provider, which has no dependency on ``users``. +ADMIN_ROLE = "admin" + +# Brute-force limits. Deliberately NOT settings fields: they are the only +# defence on a single shared secret and are not an operator-tunable surface. +MAX_FAILURES = 10 +WINDOW_SECONDS = 300 +COOLDOWN_SECONDS = 900 +``` + +`site_lock/settings.py`: + +```python +"""Site Lock module settings — DB-backed via ``register_module_settings``. + +The field is named ``password`` so the settings admin UI masks it +automatically (``settings._module_settings.is_secret_field``). +""" + +from __future__ import annotations + +from pydantic import model_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class SiteLockSettings(BaseSettings): + """Site-wide shared-password gate configuration.""" + + model_config = SettingsConfigDict(extra="ignore") + + enabled: bool = False + password: str = "" + message: str = "" + + @model_validator(mode="after") + def _password_required_when_enabled(self) -> SiteLockSettings: + """Refuse to gate the site behind an empty password. + + Without this, flipping ``enabled`` on before setting a password would + lock every visitor out from behind a secret that is the empty string. + Raising here makes ``apply_changes_and_reload`` reject the change so + the settings screen shows a validation error instead. + """ + if self.enabled and not self.password.strip(): + raise ValueError("password must be set before enabling the site lock") + return self +``` + +`site_lock/state.py`: + +```python +"""Module-owned state attached to ``app.state.site_lock``.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from site_lock import constants as c +from site_lock.rate_limit import AttemptLimiter +from site_lock.settings import SiteLockSettings + + +def _default_limiter() -> AttemptLimiter: + return AttemptLimiter( + max_failures=c.MAX_FAILURES, + window_seconds=c.WINDOW_SECONDS, + cooldown_seconds=c.COOLDOWN_SECONDS, + ) + + +@dataclass +class SiteLockState: + """Per-app site-lock state. + + ``settings`` is reassigned in place by ``settings.reload`` on a hot + reload, so this dataclass must stay mutable. ``limiter`` survives those + reloads, which is what keeps an in-flight brute-force cooldown from being + cleared by an unrelated settings save. + """ + + settings: SiteLockSettings + limiter: AttemptLimiter = field(default_factory=_default_limiter) + + +__all__ = ["SiteLockState"] +``` + +Note `state.py` imports `AttemptLimiter`, which Task 2 creates. Write `rate_limit.py` first if you are executing strictly top-to-bottom — or accept that `test_settings.py` is the only thing that must pass at the end of this task, and it does not import `state`. + +- [ ] **Step 6: Run the test to verify it passes** + +Run: `uv run pytest modules/site_lock/tests/test_settings.py -v` +Expected: PASS (4 tests, one parametrized twice → 5 cases) + +- [ ] **Step 7: Commit** + +```bash +git add modules/site_lock pyproject.toml +git commit -m "feat(site_lock): package scaffold and DB-backed settings" +``` + +--- + +### Task 2: Attempt limiter + +**Files:** +- Create: `modules/site_lock/site_lock/rate_limit.py` +- Test: `modules/site_lock/tests/test_rate_limit.py` + +**Interfaces:** +- Produces: `AttemptLimiter(max_failures: int, window_seconds: int, cooldown_seconds: int)` with `is_blocked(key: str, *, now: float | None = None) -> bool`, `record_failure(key: str, *, now: float | None = None) -> None`, `reset(key: str) -> None`. The injectable `now` is what lets tests cover expiry without sleeping. + +- [ ] **Step 1: Write the failing test** + +`modules/site_lock/tests/test_rate_limit.py`: + +```python +"""In-memory per-IP attempt limiter.""" + +from __future__ import annotations + +from site_lock.rate_limit import AttemptLimiter + + +def _limiter() -> AttemptLimiter: + return AttemptLimiter(max_failures=3, window_seconds=100, cooldown_seconds=500) + + +def test_fresh_key_is_not_blocked() -> None: + assert _limiter().is_blocked("1.2.3.4", now=0.0) is False + + +def test_blocks_after_max_failures() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is True + + +def test_under_the_limit_is_not_blocked() -> None: + lim = _limiter() + for i in range(2): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is False + + +def test_failures_outside_the_window_do_not_accumulate() -> None: + lim = _limiter() + lim.record_failure("1.2.3.4", now=0.0) + lim.record_failure("1.2.3.4", now=1.0) + # 500s later the first two have aged out of the 100s window. + lim.record_failure("1.2.3.4", now=500.0) + assert lim.is_blocked("1.2.3.4", now=500.0) is False + + +def test_cooldown_expires() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is True + assert lim.is_blocked("1.2.3.4", now=600.0) is False + + +def test_keys_are_independent() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.1.1.1", now=float(i)) + assert lim.is_blocked("1.1.1.1", now=3.0) is True + assert lim.is_blocked("2.2.2.2", now=3.0) is False + + +def test_reset_clears_a_block() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + lim.reset("1.2.3.4") + assert lim.is_blocked("1.2.3.4", now=3.0) is False +``` + +- [ ] **Step 2: Run it and watch it fail** + +Run: `uv run pytest modules/site_lock/tests/test_rate_limit.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'site_lock.rate_limit'` + +- [ ] **Step 3: Implement** + +`site_lock/rate_limit.py`: + +```python +"""In-memory per-IP attempt limiter for the unlock endpoint. + +Deliberately process-local: it mirrors the existing ``users`` +``LoginRateLimiter`` and is adequate for the single-process staging +deployments this module targets. ``users``' version is not reused because +importing it would couple ``site_lock`` to the ``users`` module, and the +site lock must also work under the ``keycloak`` provider. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass +class _Bucket: + failures: list[float] = field(default_factory=list) + blocked_until: float = 0.0 + + +class AttemptLimiter: + """Track failed unlock attempts per client key and impose a cooldown.""" + + def __init__( + self, + *, + max_failures: int, + window_seconds: int, + cooldown_seconds: int, + ) -> None: + self._max = max_failures + self._window = window_seconds + self._cooldown = cooldown_seconds + self._buckets: dict[str, _Bucket] = {} + + @staticmethod + def _now(now: float | None) -> float: + return time.monotonic() if now is None else now + + def is_blocked(self, key: str, *, now: float | None = None) -> bool: + bucket = self._buckets.get(key) + return bucket is not None and bucket.blocked_until > self._now(now) + + def record_failure(self, key: str, *, now: float | None = None) -> None: + moment = self._now(now) + bucket = self._buckets.setdefault(key, _Bucket()) + bucket.failures = [t for t in bucket.failures if moment - t < self._window] + bucket.failures.append(moment) + if len(bucket.failures) >= self._max: + bucket.blocked_until = moment + self._cooldown + bucket.failures.clear() + + def reset(self, key: str) -> None: + self._buckets.pop(key, None) + + +__all__ = ["AttemptLimiter"] +``` + +- [ ] **Step 4: Run the test to verify it passes** + +Run: `uv run pytest modules/site_lock/tests/test_rate_limit.py -v` +Expected: PASS (7 tests) + +- [ ] **Step 5: Commit** + +```bash +git add modules/site_lock/site_lock/rate_limit.py modules/site_lock/tests/test_rate_limit.py +git commit -m "feat(site_lock): in-memory per-IP attempt limiter" +``` + +--- + +### Task 3: Gate page — template, escaping, and `next` sanitisation + +**Files:** +- Create: `modules/site_lock/site_lock/templates/unlock.html`, `modules/site_lock/site_lock/page.py` +- Test: `modules/site_lock/tests/test_page.py` + +**Interfaces:** +- Produces: `safe_next(raw: str | None) -> str` and `render_unlock_page(*, message: str = "", error: str = "", next_url: str = "/") -> str`. + +- [ ] **Step 1: Write the failing test** + +`modules/site_lock/tests/test_page.py`: + +```python +"""Gate page rendering, escaping, and next-target sanitisation.""" + +from __future__ import annotations + +import pytest +from site_lock.page import render_unlock_page, safe_next + + +@pytest.mark.parametrize( + "raw", + [ + None, + "", + "//evil.example/path", + "https://evil.example", + "http://evil.example", + "/\\evil.example", + "evil", + "/ok\r\nSet-Cookie: x=1", + ], +) +def test_safe_next_rejects_offsite_and_injection(raw: str | None) -> None: + assert safe_next(raw) == "/" + + +@pytest.mark.parametrize("raw", ["/", "/dashboard/", "/users/me"]) +def test_safe_next_allows_same_site_paths(raw: str) -> None: + assert safe_next(raw) == raw + + +def test_message_is_html_escaped() -> None: + html = render_unlock_page(message="") + assert "" not in html + assert "<script>" in html + + +def test_error_is_rendered_and_escaped() -> None: + html = render_unlock_page(error="Bad password") + assert "<b>" in html + assert "password" not in html + + +def test_no_error_block_when_no_error() -> None: + assert 'role="alert"' not in render_unlock_page() + + +def test_next_is_embedded_as_a_hidden_field() -> None: + html = render_unlock_page(next_url="/dashboard/") + assert 'name="next"' in html + assert "/dashboard/" in html + + +def test_offsite_next_is_neutralised_in_the_form() -> None: + html = render_unlock_page(next_url="//evil.example") + assert "evil.example" not in html + + +def test_page_is_self_contained() -> None: + html = render_unlock_page() + # No external assets: a locked site must serve exactly one document. + assert " + + + + + +Protected + + + +
+

Protected

+

$message

+ $error +
+ + + + +
+
+ + +``` + +- [ ] **Step 4: Write the renderer** + +`site_lock/page.py`: + +```python +"""Render the standalone unlock page. + +Kept out of ``middleware.py`` so the escaping and ``next``-sanitisation rules +can be unit-tested without an ASGI harness, and so both files stay well under +the 300-line cap. +""" + +from __future__ import annotations + +import html +import importlib.resources +from string import Template + +_TEMPLATE = Template( + (importlib.resources.files(__package__) / "templates" / "unlock.html").read_text( + encoding="utf-8" + ) +) + +_DEFAULT_MESSAGE = "This site is not public yet. Enter the password to continue." +_ERROR_BLOCK = '' + + +def safe_next(raw: str | None) -> str: + """Return ``raw`` if it is a same-site absolute path, else ``/``. + + Rejects protocol-relative (``//host``) and backslash-prefixed (``/\\host``) + targets — browsers resolve both off-site — plus anything carrying CR/LF, + which could otherwise be smuggled into the redirect header. Without this + the gate would be an open redirect. + """ + if not raw or not raw.startswith("/"): + return "/" + if raw.startswith("//") or raw.startswith("/\\"): + return "/" + if "\r" in raw or "\n" in raw: + return "/" + return raw + + +def render_unlock_page( + *, + message: str = "", + error: str = "", + next_url: str = "/", +) -> str: + """Render the gate page. Every interpolated value is HTML-escaped.""" + return _TEMPLATE.safe_substitute( + message=html.escape(message or _DEFAULT_MESSAGE), + error=_ERROR_BLOCK.format(message=html.escape(error)) if error else "", + next_url=html.escape(safe_next(next_url), quote=True), + ) + + +__all__ = ["render_unlock_page", "safe_next"] +``` + +- [ ] **Step 5: Run the test to verify it passes** + +Run: `uv run pytest modules/site_lock/tests/test_page.py -v` +Expected: PASS (17 cases) + +If `test_page_is_self_contained` fails on `src=`, check that no CSS rule contains that substring. + +- [ ] **Step 6: Commit** + +```bash +git add modules/site_lock/site_lock/page.py modules/site_lock/site_lock/templates modules/site_lock/tests/test_page.py +git commit -m "feat(site_lock): self-contained gate page with escaping and next sanitisation" +``` + +--- + +### Task 4: The gate middleware + +**Files:** +- Create: `modules/site_lock/site_lock/middleware.py`, `modules/site_lock/site_lock/module.py` +- Test: `modules/site_lock/tests/test_middleware.py` + +**Interfaces:** +- Consumes: `SiteLockState` (Task 1), `AttemptLimiter` (Task 2), `render_unlock_page` / `safe_next` (Task 3). +- Produces: `SiteLockMiddleware(app)`; `password_fingerprint(password: str) -> str`; `SiteLockModule`. + +- [ ] **Step 1: Write the failing test** + +`modules/site_lock/tests/test_middleware.py`: + +```python +"""SiteLockMiddleware gating behaviour.""" + +from __future__ import annotations + +import httpx +import pytest +from fastapi import FastAPI +from site_lock import constants as c +from site_lock.middleware import SiteLockMiddleware, password_fingerprint +from site_lock.settings import SiteLockSettings +from site_lock.state import SiteLockState +from starlette.middleware.sessions import SessionMiddleware +from starlette.responses import JSONResponse + +SECRET = "test-site-lock-secret" +PASSWORD = "hunter2" + + +class _StubProvider: + """Minimal AuthProvider stand-in for the admin-bypass path.""" + + def __init__(self, user=None): + self._user = user + + async def resolve_user(self, request): + return self._user + + +def _build_app(settings: SiteLockSettings, *, provider=None) -> FastAPI: + app = FastAPI() + app.state.site_lock = SiteLockState(settings=settings) + app.state.auth = type("_AuthState", (), {"auth_provider": provider})() + + async def _handler(path: str = ""): + return JSONResponse({"ok": True}) + + app.add_api_route("/{path:path}", _handler, methods=["GET", "POST"]) + app.add_middleware(SiteLockMiddleware) + app.add_middleware(SessionMiddleware, secret_key=SECRET) + return app + + +def _client(app: FastAPI) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + follow_redirects=False, + ) + + +async def test_disabled_passes_everything_through() -> None: + app = _build_app(SiteLockSettings()) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 200 + assert response.json() == {"ok": True} + + +async def test_enabled_redirects_browser_to_the_gate() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + assert response.headers["location"].startswith(c.UNLOCK_PATH) + assert response.headers["cache-control"] == "no-store" + + +async def test_enabled_returns_403_json_for_api() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/api/users/me") + assert response.status_code == 403 + assert response.json() == {"detail": "Site is locked"} + + +async def test_bearer_request_gets_403_json_not_a_redirect() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/dashboard/", headers={"Authorization": "Bearer x"}) + assert response.status_code == 403 + + +async def test_health_is_never_gated() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/health") + assert response.status_code == 200 + + +async def test_gate_page_is_served() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD, message="Soon")) + async with _client(app) as client: + response = await client.get(c.UNLOCK_PATH) + assert response.status_code == 200 + assert 'name="password"' in response.text + assert "Soon" in response.text + + +async def test_correct_password_unlocks_and_persists() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post( + c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/dashboard/"} + ) + assert posted.status_code == 303 + assert posted.headers["location"] == "/dashboard/" + # The session cookie now carries the unlock marker. + followed = await client.get("/dashboard/") + assert followed.status_code == 200 + + +async def test_wrong_password_is_rejected() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post(c.UNLOCK_PATH, data={"password": "nope", "next": "/"}) + assert posted.status_code == 401 + followed = await client.get("/dashboard/") + assert followed.status_code == 302 + + +async def test_offsite_next_is_not_honoured_on_redirect() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post( + c.UNLOCK_PATH, data={"password": PASSWORD, "next": "//evil.example"} + ) + assert posted.headers["location"] == "/" + + +async def test_rotating_the_password_invalidates_existing_sessions() -> None: + settings = SiteLockSettings(enabled=True, password=PASSWORD) + app = _build_app(settings) + async with _client(app) as client: + await client.post(c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"}) + assert (await client.get("/dashboard/")).status_code == 200 + # Operator rotates the password via the settings UI. + app.state.site_lock.settings = SiteLockSettings(enabled=True, password="new-one") + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_unsupported_method_on_the_gate_is_405() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.request("PUT", c.UNLOCK_PATH) + assert response.status_code == 405 + + +async def test_rate_limited_after_repeated_failures() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + for _ in range(c.MAX_FAILURES): + await client.post(c.UNLOCK_PATH, data={"password": "nope", "next": "/"}) + response = await client.post( + c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"} + ) + assert response.status_code == 429 + + +def test_fingerprint_changes_with_the_password() -> None: + assert password_fingerprint("a") != password_fingerprint("b") + assert password_fingerprint("a") == password_fingerprint("a") +``` + +- [ ] **Step 2: Run it and watch it fail** + +Run: `uv run pytest modules/site_lock/tests/test_middleware.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'site_lock.middleware'` + +- [ ] **Step 3: Implement the middleware** + +`site_lock/middleware.py`: + +```python +"""Site-wide password gate. + +Runs outermost among module middleware — the module sorts after ``Auth``, and +Starlette's ``add_middleware`` is LIFO — so this executes *before* +``AuthMiddleware``. That ordering is the whole point: an anonymous visitor +sees the gate instead of being redirected to the login page, so a locked site +never reveals that a login form exists. + +``SessionMiddleware`` sits outside this one, so ``scope["session"]`` is +readable here and writes are persisted on the way out. +""" + +from __future__ import annotations + +import hashlib +import logging +import secrets +from urllib.parse import quote + +from starlette.requests import Request +from starlette.responses import ( + HTMLResponse, + JSONResponse, + RedirectResponse, + Response, +) +from starlette.types import ASGIApp, Receive, Scope, Send + +from site_lock import constants as c +from site_lock.page import render_unlock_page, safe_next + +logger = logging.getLogger(__name__) + +_NO_STORE = {"Cache-Control": "no-store"} +_ERR_WRONG = "Incorrect password." +_ERR_THROTTLED = "Too many attempts. Try again later." + + +def password_fingerprint(password: str) -> str: + """Short digest of the active password, stored in the session. + + Comparing this on each request is what makes a password rotation + invalidate every previously-unlocked session. + """ + return hashlib.sha256(password.encode("utf-8")).hexdigest()[:16] + + +class SiteLockMiddleware: + """Gate every request behind a single shared password.""" + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + state = scope["app"].state.site_lock + settings = state.settings + + # Default-off fast path: one attribute read, no Request construction. + if not settings.enabled: + await self.app(scope, receive, send) + return + + path: str = scope["path"] + + # Kubernetes probes must never be gated, or the pod gets killed. + if path.startswith(c.HEALTH_PREFIX): + await self.app(scope, receive, send) + return + + fingerprint = password_fingerprint(settings.password) + session = scope.get("session") + + if path == c.UNLOCK_PATH: + response = await self._unlock(scope, receive, state, fingerprint, session) + await response(scope, receive, send) + return + + if session is not None and session.get(c.SESSION_KEY) == fingerprint: + await self.app(scope, receive, send) + return + + if await self._is_admin(scope): + # Stamp the marker so the provider lookup costs once per session + # rather than once per request. + if session is not None: + session[c.SESSION_KEY] = fingerprint + await self.app(scope, receive, send) + return + + await self._gate(scope, receive, send, path) + + async def _gate(self, scope: Scope, receive: Receive, send: Send, path: str) -> None: + has_auth_header = any(k == b"authorization" for k, _ in scope.get("headers", ())) + if path.startswith(c.API_PREFIX) or has_auth_header: + # 403 rather than 401: a 401 would invite an auth flow that cannot + # succeed while the site is locked. + response: Response = JSONResponse( + {"detail": "Site is locked"}, status_code=403, headers=_NO_STORE + ) + else: + target = f"{c.UNLOCK_PATH}?next={quote(safe_next(path), safe='/')}" + response = RedirectResponse(target, status_code=302, headers=_NO_STORE) + await response(scope, receive, send) + + async def _unlock( + self, + scope: Scope, + receive: Receive, + state, + fingerprint: str, + session, + ) -> Response: + method: str = scope.get("method", "GET") + settings = state.settings + request = Request(scope, receive) + + if method == "GET": + return HTMLResponse( + render_unlock_page( + message=settings.message, + next_url=request.query_params.get("next", "/"), + ), + headers=_NO_STORE, + ) + if method != "POST": + return Response(status_code=405, headers=_NO_STORE) + + client = scope.get("client") + key = client[0] if client else "unknown" + if state.limiter.is_blocked(key): + return self._page(settings, _ERR_THROTTLED, "/", status=429) + + form = await request.form() + supplied = str(form.get("password", "")) + target = safe_next(str(form.get("next", "/"))) + + if secrets.compare_digest(supplied, settings.password): + state.limiter.reset(key) + if session is not None: + session[c.SESSION_KEY] = fingerprint + return RedirectResponse(target, status_code=303, headers=_NO_STORE) + + state.limiter.record_failure(key) + return self._page(settings, _ERR_WRONG, target, status=401) + + @staticmethod + def _page(settings, error: str, next_url: str, *, status: int) -> HTMLResponse: + return HTMLResponse( + render_unlock_page(message=settings.message, error=error, next_url=next_url), + status_code=status, + headers=_NO_STORE, + ) + + @staticmethod + async def _is_admin(scope: Scope) -> bool: + """True when the caller already holds a session for an admin user. + + This is the documented lockout escape hatch: an admin who enables the + gate and mistypes the password keeps access to the settings screen. + """ + auth_state = getattr(scope["app"].state, "auth", None) + provider = getattr(auth_state, "auth_provider", None) + if provider is None: + return False + try: + user = await provider.resolve_user(Request(scope)) + except Exception: + logger.exception("Site lock admin bypass failed; treating as anonymous") + return False + return user is not None and c.ADMIN_ROLE in getattr(user, "roles", ()) + + +__all__ = ["SiteLockMiddleware", "password_fingerprint"] +``` + +- [ ] **Step 4: Write the module definition** + +`site_lock/module.py`: + +```python +"""Site Lock module — optional site-wide password gate.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from simple_module_core.module import ModuleBase, ModuleMeta + +from site_lock import constants as c + +if TYPE_CHECKING: + from fastapi import FastAPI + + +class SiteLockModule(ModuleBase): + meta = ModuleMeta( + name=c.MODULE_NAME, + # ``Settings`` so register_module_settings can reach the module + # registry; ``Auth`` so this module sorts *after* auth and its + # middleware therefore wraps outermost, executing before + # AuthMiddleware. Both are load-bearing — see the module README. + depends_on=[c.MODULE_SETTINGS, c.MODULE_AUTH], + ) + + def register_settings(self, app: FastAPI) -> None: + import importlib + + from site_lock.settings import SiteLockSettings + from site_lock.state import SiteLockState + + # SM009 is AST-based: resolving via importlib matches the convention + # used by the other settings-backed modules. + register_module_settings = importlib.import_module( + "settings.registration" + ).register_module_settings + + register_module_settings( + app, + c.MODULE_PACKAGE, + SiteLockSettings, + lambda s: SiteLockState(settings=s), + ) + + def register_middleware(self, app: FastAPI) -> None: + from site_lock.middleware import SiteLockMiddleware + + app.add_middleware(SiteLockMiddleware) +``` + +- [ ] **Step 5: Run the tests to verify they pass** + +Run: `uv run pytest modules/site_lock/tests/ -v` +Expected: PASS (all tests across the three files) + +If `test_correct_password_unlocks_and_persists` fails with a form-parsing error, confirm `python-multipart` is installed — it is already a `host` dependency, and urlencoded bodies are parsed natively by Starlette regardless. + +- [ ] **Step 6: Commit** + +```bash +git add modules/site_lock +git commit -m "feat(site_lock): gate middleware and module wiring" +``` + +--- + +### Task 5: Admin bypass, verified end to end + +**Files:** +- Modify: `modules/site_lock/tests/test_middleware.py` (append) + +**Interfaces:** +- Consumes: `_build_app`, `_client`, `_StubProvider` from Task 4's test module. + +The bypass code already exists from Task 4; this task proves it, including the negative case that a non-admin does **not** get through. + +- [ ] **Step 1: Write the failing tests** + +Append to `modules/site_lock/tests/test_middleware.py`: + +```python +class _User: + def __init__(self, roles: list[str]) -> None: + self.id = "11111111-2222-3333-4444-555555555555" + self.roles = roles + + +async def test_admin_with_a_live_session_bypasses_the_gate() -> None: + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_StubProvider(_User(["admin"])), + ) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 200 + + +async def test_authenticated_non_admin_does_not_bypass() -> None: + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_StubProvider(_User(["user"])), + ) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_admin_bypass_stamps_the_session_once() -> None: + """After the first bypass the marker short-circuits further lookups.""" + calls = [] + + class _CountingProvider(_StubProvider): + async def resolve_user(self, request): + calls.append(1) + return _User(["admin"]) + + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_CountingProvider(), + ) + async with _client(app) as client: + await client.get("/dashboard/") + await client.get("/dashboard/") + await client.get("/dashboard/") + assert len(calls) == 1 + + +async def test_missing_provider_does_not_crash_the_gate() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD), provider=None) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_provider_that_raises_is_treated_as_anonymous() -> None: + class _BoomProvider: + async def resolve_user(self, request): + raise RuntimeError("provider exploded") + + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), provider=_BoomProvider() + ) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 +``` + +- [ ] **Step 2: Run the tests** + +Run: `uv run pytest modules/site_lock/tests/test_middleware.py -v -k "admin or provider"` +Expected: PASS. If `test_admin_bypass_stamps_the_session_once` fails with `len(calls) == 3`, the session marker is not being written in the bypass branch — check that `_build_app` adds `SessionMiddleware` *outside* `SiteLockMiddleware`. + +- [ ] **Step 3: Commit** + +```bash +git add modules/site_lock/tests/test_middleware.py +git commit -m "test(site_lock): cover admin bypass and provider failure modes" +``` + +--- + +### Task 6: Host wiring, middleware-order test, and docs + +**Files:** +- Modify: `host/pyproject.toml`, `framework/hosting/tests/test_middleware_order.py`, `framework/cli/simple_module_cli/catalog.py`, `docs/.vitepress/config.ts`, `docs/index.md` +- Create: `docs/modules/site_lock.md`, `modules/site_lock/README.md` + +**Interfaces:** +- Consumes: `SiteLockModule` entry point from Task 4. + +- [ ] **Step 1: Install the module into the host** + +In `host/pyproject.toml`, add `"simple_module_site_lock",` to `[project] dependencies` and `simple_module_site_lock = { workspace = true }` to `[tool.uv.sources]`. + +Run: `uv sync --all-packages` + +- [ ] **Step 2: Update the middleware-order test to expect the new entry** + +`framework/hosting/tests/test_middleware_order.py` pins the exact pipeline. Insert `"SiteLockMiddleware"` between `"SessionMiddleware"` and `"AuthMiddleware"` in **both** `_EXPECTED_MULTI_TENANT` and `_EXPECTED_SINGLE_TENANT`: + +```python +_EXPECTED_MULTI_TENANT = ( + "CorrelationIdMiddleware", + "RequestLoggingMiddleware", + "GZipMiddleware", + "SecurityHeadersMiddleware", + "SessionMiddleware", + "SiteLockMiddleware", + "AuthMiddleware", + "TenantMiddleware", + "LocaleMiddleware", + "InertiaLayoutDataMiddleware", +) + +_EXPECTED_SINGLE_TENANT = ( + "CorrelationIdMiddleware", + "RequestLoggingMiddleware", + "GZipMiddleware", + "SecurityHeadersMiddleware", + "SessionMiddleware", + "SiteLockMiddleware", + "AuthMiddleware", + "LocaleMiddleware", + "InertiaLayoutDataMiddleware", +) +``` + +Also extend that module's docstring to note that SiteLock must precede Auth so anonymous visitors get the gate rather than a login redirect. + +- [ ] **Step 3: Run the order test** + +Run: `uv run pytest framework/hosting/tests/test_middleware_order.py -v` +Expected: PASS. **A failure here where SiteLock lands *after* Auth means the `depends_on` is wrong** — that inversion silently breaks the feature (visitors would be bounced to the login page), so do not "fix" it by reordering the expected tuple. + +- [ ] **Step 4: Add the catalog entry** + +In `framework/cli/simple_module_cli/catalog.py`, add to `CATALOG` after the `audit_log` entry: + +```python + "site_lock": ModuleEntry( + "site_lock", + "simple_module_site_lock", + "Site Lock", + requires=("auth", "settings"), + ), +``` + +`PRESETS["full"]` is `tuple(CATALOG)`, so this is picked up automatically; `minimal` and `standard` are unaffected. + +- [ ] **Step 5: Write the module README** + +`modules/site_lock/README.md` — must cover: what it does, that it is off by default, how to enable it from Settings → Site Lock, the admin-bypass behaviour, and the cold-start lockout limitation with the SQL recovery: + +```sql +DELETE FROM settings_setting WHERE package = 'site_lock' AND name = 'enabled'; +``` + +Verify that table and column naming against `modules/settings/settings/models.py` before committing the README — do not guess. + +- [ ] **Step 6: Write the docs page and nav** + +Create `docs/modules/site_lock.md` following the shape of `docs/modules/file_storage.md`. Add `{ text: "site_lock", link: "/modules/site_lock" },` to `docs/.vitepress/config.ts` near the `file_storage` entry, and update the module count in `docs/index.md` line 29 ("Ten first-party modules" → "Eleven", adding `site_lock` to the list). + +- [ ] **Step 7: Full verification** + +Run each and confirm before claiming success: + +```bash +uv run pytest modules/site_lock/tests/ -v +make doctor +make lint +make test-py +``` + +Expected: all green. `make doctor` must report no new diagnostics — `SM012` is satisfied because `register_module_settings` mounts `app.state.site_lock`; `SM017`/`SM019` do not apply (no `.tsx`, no view routes). Redirect output to a file rather than piping into `grep`/`head`, or `$?` reports the pager's status instead of the real exit code. + +- [ ] **Step 8: Commit** + +```bash +git add -A +git commit -m "feat(site_lock): host wiring, catalog entry, and docs" +``` + +--- + +## Self-Review + +**Spec coverage:** + +| Spec section | Task | +|---|---| +| Placement in the pipeline (`depends_on`, order) | 4 (module.py), 6 (order test) | +| Settings: `enabled`/`password`/`message` + validator | 1 | +| Request handling steps 1–3 (websocket, fast path, `/health`) | 4 | +| Step 4 (unlock GET/POST/405) | 4 | +| Step 5 (session fingerprint, rotation) | 4 | +| Step 6 (admin bypass, one lookup per session) | 4 impl, 5 tests | +| Step 7 (403 JSON vs 302, `no-store`) | 4 | +| Gate page, escaping, `next` sanitisation | 3 | +| Brute-force limiter + fixed thresholds | 2, 4 | +| Cold-start lockout documented | 6 (README) | +| No migration / no `gen-pages` / no doctor diagnostics | 6 (verification) | + +**Type consistency:** `AttemptLimiter` methods (`is_blocked`/`record_failure`/`reset`) match between Tasks 2, 4, and the state factory in Task 1. `safe_next` and `render_unlock_page` signatures match between Tasks 3 and 4. `SiteLockState.settings`/`.limiter` match between Tasks 1 and 4. `c.MODULE_PACKAGE` (`"site_lock"`) is the single source for the `app.state` attribute and the settings registry key. + +**Known ordering wrinkle:** Task 1's `state.py` imports `AttemptLimiter` from Task 2. `test_settings.py` does not import `state`, so Task 1's tests pass regardless; if executing strictly in order, either write `rate_limit.py` during Task 1 or accept that `state.py` is only importable from Task 2 onward. diff --git a/docs/superpowers/specs/2026-08-05-site-lock-design.md b/docs/superpowers/specs/2026-08-05-site-lock-design.md index 39435770..83a307ae 100644 --- a/docs/superpowers/specs/2026-08-05-site-lock-design.md +++ b/docs/superpowers/specs/2026-08-05-site-lock-design.md @@ -23,10 +23,12 @@ request. - `AuthMiddleware` (`modules/auth/auth/middleware.py`) resolves the user via the single registered `AuthProvider`, then redirects unauthenticated browser requests to the provider's login URL and returns 401 JSON for `/api/*`. -- Public paths come from two places only: the `_FRAMEWORK_PUBLIC_*` constants - in the auth middleware, and `provider.get_public_paths()`. **There is no - registry a third module can add to** — this is what rules out mounting a - normal view route for the gate page without first extending the auth module. +- Public paths come from three places: the `_FRAMEWORK_PUBLIC_*` constants in + the auth middleware, the method-aware `app.state.public_routes` registry fed + by the `register_public_routes` module hook, and the legacy + `provider.get_public_paths()`. A module *could* therefore mount a normal + gated view route — but the design below does not need to, because the + middleware runs *before* `AuthMiddleware` and so never reaches that check. - DB-backed per-module settings go through `settings.registration.register_module_settings(app, package, cls, factory)`, hydrate at startup, and hot-swap through @@ -50,8 +52,9 @@ Settled during brainstorming: 3. **Logged-in admins always bypass** the gate — this is the lockout escape hatch, chosen over shipping a CLI command. 4. **Self-contained middleware-rendered gate page**, not an Inertia `.tsx` page. - This avoids extending the auth module with a public-paths registry, and - means a locked site serves exactly one page and leaks nothing else. + A locked site then serves exactly one page and leaks nothing else — no + menus, no branding, no JS bundle — and the gate keeps working even if the + frontend build is broken. ## Design From 64d51984316357b1f7241ed9742db940596179dc Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Wed, 5 Aug 2026 18:55:54 +0200 Subject: [PATCH 3/7] feat(site_lock): package scaffold, settings, and attempt limiter DB-backed SiteLockSettings (enabled=False by default) with a validator that refuses to enable the gate behind a blank password, plus an in-memory per-IP attempt limiter with injectable clock for testing. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- modules/site_lock/README.md | 83 ++++++++++++++++++++++ modules/site_lock/pyproject.toml | 47 ++++++++++++ modules/site_lock/site_lock/__init__.py | 0 modules/site_lock/site_lock/constants.py | 28 ++++++++ modules/site_lock/site_lock/py.typed | 0 modules/site_lock/site_lock/rate_limit.py | 58 +++++++++++++++ modules/site_lock/site_lock/settings.py | 33 +++++++++ modules/site_lock/site_lock/state.py | 34 +++++++++ modules/site_lock/tests/test_rate_limit.py | 60 ++++++++++++++++ modules/site_lock/tests/test_settings.py | 32 +++++++++ pyproject.toml | 3 +- 11 files changed, 377 insertions(+), 1 deletion(-) create mode 100644 modules/site_lock/README.md create mode 100644 modules/site_lock/pyproject.toml create mode 100644 modules/site_lock/site_lock/__init__.py create mode 100644 modules/site_lock/site_lock/constants.py create mode 100644 modules/site_lock/site_lock/py.typed create mode 100644 modules/site_lock/site_lock/rate_limit.py create mode 100644 modules/site_lock/site_lock/settings.py create mode 100644 modules/site_lock/site_lock/state.py create mode 100644 modules/site_lock/tests/test_rate_limit.py create mode 100644 modules/site_lock/tests/test_settings.py diff --git a/modules/site_lock/README.md b/modules/site_lock/README.md new file mode 100644 index 00000000..d06c7ed8 --- /dev/null +++ b/modules/site_lock/README.md @@ -0,0 +1,83 @@ +# simple_module_site_lock + +Optional site-wide password gate for `simple_module` apps — a staging / +pre-launch door. When enabled, every visitor must enter one shared password +before they can see anything: not the landing page, not the API, not even +that a login form exists. + +**Off by default.** Installing this module changes nothing until an operator +turns it on. + +## How it works + +The module installs a single middleware that runs *before* `AuthMiddleware`. +That ordering is the whole point — an anonymous visitor gets the gate rather +than a redirect to the login page, so a locked site never reveals that it has +one. + +It achieves that ordering by declaring `depends_on=["Settings", "Auth"]`: +modules are installed in topological order and Starlette's `add_middleware` +is LIFO, so sorting after `Auth` makes this middleware wrap outermost. + +Unlock state lives in the signed session cookie. The stored marker is a +fingerprint of the current password, so **rotating the password immediately +invalidates every unlocked session**. + +## Enabling it + +Settings → Site Lock: + +| Field | Default | Meaning | +|---|---|---| +| `enabled` | `false` | Master switch | +| `password` | `""` | The shared password. Masked in the admin UI | +| `message` | `""` | Optional line shown on the gate page | + +Changes apply immediately — no restart. Enabling with a blank password is +rejected by a validator, so you cannot accidentally gate the site behind the +empty string. + +## What stays reachable when locked + +- `/health` — always. Gating it would fail Kubernetes liveness/readiness + probes and get the pod killed. +- `/__unlock` — the gate page itself. + +Everything else is gated. Requests under `/api/`, and any request carrying an +`Authorization` header, get `403 {"detail": "Site is locked"}` rather than a +redirect — a `401` would invite an auth flow that cannot succeed while the +site is locked. Browser requests get a `302` to the gate. + +## Admin bypass, and the lockout you can still cause + +A user who already holds a session with the `admin` role skips the gate. +This is the intended escape hatch: if you enable the gate and mistype the +password, you are still holding the session that lets you go straight back to +Settings and fix it. + +**It only rescues a live session.** If no admin is currently signed in and the +password has been forgotten, there is no in-app recovery. Clear the override +directly in the database: + +```sql +DELETE FROM settings_setting + WHERE scope = 'system' AND key = 'site_lock.enabled'; +``` + +Then restart the app (or save any setting) so the module re-hydrates. + +## Brute-force protection + +The unlock endpoint tracks failures per client IP in memory: 10 failures +within 5 minutes trigger a 15-minute cooldown returning `429`. These +thresholds are module constants, not settings — they are the only defence on +a single shared secret, so they are not an operator-tunable surface. + +The limiter is process-local, which is adequate for the single-process +staging deployments this module targets. + +## What this is not + +This is not user authentication or authorisation — that is what the `auth`, +`users`, and `permissions` modules are for. The site lock is one shared +secret in front of everything, with no notion of identity. diff --git a/modules/site_lock/pyproject.toml b/modules/site_lock/pyproject.toml new file mode 100644 index 00000000..c8b4bf92 --- /dev/null +++ b/modules/site_lock/pyproject.toml @@ -0,0 +1,47 @@ +[project] +name = "simple_module_site_lock" +version = "0.0.26" +description = "Optional site-wide password gate (staging / pre-launch) for simple_module apps" +readme = "README.md" +license = "MIT" +requires-python = ">=3.12" +authors = [{ name = "Anto Subash", email = "antosubash@live.com" }] +keywords = ["simple-module", "password", "staging", "gate"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Framework :: FastAPI", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.12", + "Topic :: Internet :: WWW/HTTP", + "Typing :: Typed", +] +dependencies = [ + "simple_module_core==0.0.26", + "simple_module_hosting==0.0.26", + "simple_module_settings==0.0.26", + "simple_module_auth==0.0.26", +] + +[project.entry-points.simple_module] +site_lock = "site_lock.module:SiteLockModule" + +[project.urls] +Homepage = "https://github.com/antosubash/simple_module_python" +Repository = "https://github.com/antosubash/simple_module_python" +Issues = "https://github.com/antosubash/simple_module_python/issues" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["site_lock"] + +[tool.uv.sources] +simple_module_core = { workspace = true } +simple_module_hosting = { workspace = true } +simple_module_settings = { workspace = true } +simple_module_auth = { workspace = true } diff --git a/modules/site_lock/site_lock/__init__.py b/modules/site_lock/site_lock/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/modules/site_lock/site_lock/constants.py b/modules/site_lock/site_lock/constants.py new file mode 100644 index 00000000..eac52ca7 --- /dev/null +++ b/modules/site_lock/site_lock/constants.py @@ -0,0 +1,28 @@ +"""Every literal the site_lock module depends on.""" + +from __future__ import annotations + +MODULE_NAME = "SiteLock" +MODULE_PACKAGE = "site_lock" + +# Dependencies — names must match the other modules' ``ModuleMeta.name``. +MODULE_SETTINGS = "Settings" +MODULE_AUTH = "Auth" + +# The gate endpoint. Double-underscore prefix keeps it clear of app routes. +UNLOCK_PATH = "/__unlock" +SESSION_KEY = "site_lock" + +# Never gated: Kubernetes liveness/readiness probes must always succeed. +HEALTH_PREFIX = "/health" +API_PREFIX = "/api/" + +# Duplicated rather than imported from ``users`` — site_lock must also work +# under the ``keycloak`` provider, which has no dependency on ``users``. +ADMIN_ROLE = "admin" + +# Brute-force limits. Deliberately NOT settings fields: they are the only +# defence on a single shared secret and are not an operator-tunable surface. +MAX_FAILURES = 10 +WINDOW_SECONDS = 300 +COOLDOWN_SECONDS = 900 diff --git a/modules/site_lock/site_lock/py.typed b/modules/site_lock/site_lock/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/modules/site_lock/site_lock/rate_limit.py b/modules/site_lock/site_lock/rate_limit.py new file mode 100644 index 00000000..e84ef17e --- /dev/null +++ b/modules/site_lock/site_lock/rate_limit.py @@ -0,0 +1,58 @@ +"""In-memory per-IP attempt limiter for the unlock endpoint. + +Deliberately process-local: it mirrors the existing ``users`` +``LoginRateLimiter`` and is adequate for the single-process staging +deployments this module targets. ``users``' version is not reused because +importing it would couple ``site_lock`` to the ``users`` module, and the +site lock must also work under the ``keycloak`` provider. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + + +@dataclass +class _Bucket: + failures: list[float] = field(default_factory=list) + blocked_until: float = 0.0 + + +class AttemptLimiter: + """Track failed unlock attempts per client key and impose a cooldown.""" + + def __init__( + self, + *, + max_failures: int, + window_seconds: int, + cooldown_seconds: int, + ) -> None: + self._max = max_failures + self._window = window_seconds + self._cooldown = cooldown_seconds + self._buckets: dict[str, _Bucket] = {} + + @staticmethod + def _now(now: float | None) -> float: + return time.monotonic() if now is None else now + + def is_blocked(self, key: str, *, now: float | None = None) -> bool: + bucket = self._buckets.get(key) + return bucket is not None and bucket.blocked_until > self._now(now) + + def record_failure(self, key: str, *, now: float | None = None) -> None: + moment = self._now(now) + bucket = self._buckets.setdefault(key, _Bucket()) + bucket.failures = [t for t in bucket.failures if moment - t < self._window] + bucket.failures.append(moment) + if len(bucket.failures) >= self._max: + bucket.blocked_until = moment + self._cooldown + bucket.failures.clear() + + def reset(self, key: str) -> None: + self._buckets.pop(key, None) + + +__all__ = ["AttemptLimiter"] diff --git a/modules/site_lock/site_lock/settings.py b/modules/site_lock/site_lock/settings.py new file mode 100644 index 00000000..8699df58 --- /dev/null +++ b/modules/site_lock/site_lock/settings.py @@ -0,0 +1,33 @@ +"""Site Lock module settings — DB-backed via ``register_module_settings``. + +The field is named ``password`` so the settings admin UI masks it +automatically (``settings._module_settings.is_secret_field``). +""" + +from __future__ import annotations + +from pydantic import model_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class SiteLockSettings(BaseSettings): + """Site-wide shared-password gate configuration.""" + + model_config = SettingsConfigDict(extra="ignore") + + enabled: bool = False + password: str = "" + message: str = "" + + @model_validator(mode="after") + def _password_required_when_enabled(self) -> SiteLockSettings: + """Refuse to gate the site behind an empty password. + + Without this, flipping ``enabled`` on before setting a password would + lock every visitor out from behind a secret that is the empty string. + Raising here makes ``apply_changes_and_reload`` reject the change so + the settings screen shows a validation error instead. + """ + if self.enabled and not self.password.strip(): + raise ValueError("password must be set before enabling the site lock") + return self diff --git a/modules/site_lock/site_lock/state.py b/modules/site_lock/site_lock/state.py new file mode 100644 index 00000000..add309fd --- /dev/null +++ b/modules/site_lock/site_lock/state.py @@ -0,0 +1,34 @@ +"""Module-owned state attached to ``app.state.site_lock``.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from site_lock import constants as c +from site_lock.rate_limit import AttemptLimiter +from site_lock.settings import SiteLockSettings + + +def _default_limiter() -> AttemptLimiter: + return AttemptLimiter( + max_failures=c.MAX_FAILURES, + window_seconds=c.WINDOW_SECONDS, + cooldown_seconds=c.COOLDOWN_SECONDS, + ) + + +@dataclass +class SiteLockState: + """Per-app site-lock state. + + ``settings`` is reassigned in place by ``settings.reload`` on a hot + reload, so this dataclass must stay mutable. ``limiter`` survives those + reloads, which is what keeps an in-flight brute-force cooldown from being + cleared by an unrelated settings save. + """ + + settings: SiteLockSettings + limiter: AttemptLimiter = field(default_factory=_default_limiter) + + +__all__ = ["SiteLockState"] diff --git a/modules/site_lock/tests/test_rate_limit.py b/modules/site_lock/tests/test_rate_limit.py new file mode 100644 index 00000000..d270aefb --- /dev/null +++ b/modules/site_lock/tests/test_rate_limit.py @@ -0,0 +1,60 @@ +"""In-memory per-IP attempt limiter.""" + +from __future__ import annotations + +from site_lock.rate_limit import AttemptLimiter + + +def _limiter() -> AttemptLimiter: + return AttemptLimiter(max_failures=3, window_seconds=100, cooldown_seconds=500) + + +def test_fresh_key_is_not_blocked() -> None: + assert _limiter().is_blocked("1.2.3.4", now=0.0) is False + + +def test_blocks_after_max_failures() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is True + + +def test_under_the_limit_is_not_blocked() -> None: + lim = _limiter() + for i in range(2): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is False + + +def test_failures_outside_the_window_do_not_accumulate() -> None: + lim = _limiter() + lim.record_failure("1.2.3.4", now=0.0) + lim.record_failure("1.2.3.4", now=1.0) + # 500s later the first two have aged out of the 100s window. + lim.record_failure("1.2.3.4", now=500.0) + assert lim.is_blocked("1.2.3.4", now=500.0) is False + + +def test_cooldown_expires() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + assert lim.is_blocked("1.2.3.4", now=3.0) is True + assert lim.is_blocked("1.2.3.4", now=600.0) is False + + +def test_keys_are_independent() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.1.1.1", now=float(i)) + assert lim.is_blocked("1.1.1.1", now=3.0) is True + assert lim.is_blocked("2.2.2.2", now=3.0) is False + + +def test_reset_clears_a_block() -> None: + lim = _limiter() + for i in range(3): + lim.record_failure("1.2.3.4", now=float(i)) + lim.reset("1.2.3.4") + assert lim.is_blocked("1.2.3.4", now=3.0) is False diff --git a/modules/site_lock/tests/test_settings.py b/modules/site_lock/tests/test_settings.py new file mode 100644 index 00000000..3da339ec --- /dev/null +++ b/modules/site_lock/tests/test_settings.py @@ -0,0 +1,32 @@ +"""SiteLockSettings defaults and the enabled-requires-password guard.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError +from site_lock.settings import SiteLockSettings + + +def test_disabled_by_default() -> None: + s = SiteLockSettings() + assert s.enabled is False + assert s.password == "" + assert s.message == "" + + +def test_enabled_with_password_is_valid() -> None: + s = SiteLockSettings(enabled=True, password="hunter2") + assert s.enabled is True + + +@pytest.mark.parametrize("password", ["", " "]) +def test_enabled_without_password_is_rejected(password: str) -> None: + with pytest.raises(ValidationError): + SiteLockSettings(enabled=True, password=password) + + +def test_password_field_is_masked_by_settings_ui() -> None: + # The admin UI masks fields by name; `password` must match that regex. + from settings._module_settings import is_secret_field + + assert is_secret_field("password") is True diff --git a/pyproject.toml b/pyproject.toml index 55dc20a1..5d27fee1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,7 @@ extra-paths = [ "modules/feature_flags", "modules/keycloak", "modules/audit_log", + "modules/site_lock", "modules/branding", "host", "scripts", @@ -116,7 +117,7 @@ invalid-assignment = "ignore" [tool.pytest.ini_options] asyncio_mode = "auto" -testpaths = ["framework/cli/tests", "framework/core/tests", "framework/db/tests", "framework/hosting/tests", "framework/testing/tests", "host/tests", "modules/auth/tests", "modules/dashboard/tests", "modules/users/tests", "modules/permissions/tests", "modules/background_tasks/tests", "modules/file_storage/tests", "modules/settings/tests", "modules/feature_flags/tests", "modules/keycloak/tests", "modules/audit_log/tests", "modules/branding/tests", "scripts/tests", "tests/integration", "tests/e2e", "tests/benchmarks", "tests/perf"] +testpaths = ["framework/cli/tests", "framework/core/tests", "framework/db/tests", "framework/hosting/tests", "framework/testing/tests", "host/tests", "modules/auth/tests", "modules/dashboard/tests", "modules/users/tests", "modules/permissions/tests", "modules/background_tasks/tests", "modules/file_storage/tests", "modules/settings/tests", "modules/feature_flags/tests", "modules/keycloak/tests", "modules/audit_log/tests", "modules/branding/tests", "modules/site_lock/tests", "scripts/tests", "tests/integration", "tests/e2e", "tests/benchmarks", "tests/perf"] markers = [ "e2e: end-to-end tests requiring a live browser", "perf: performance benchmarks (opt-in; run via `make bench`)", From b55d78cf1c952301e86fb361e7c80598ebb3b444 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Wed, 5 Aug 2026 18:57:56 +0200 Subject: [PATCH 4/7] feat(site_lock): gate middleware, module wiring, and admin bypass Middleware runs before AuthMiddleware so anonymous visitors get the gate rather than a login redirect. Session marker is a password fingerprint, so rotating the password invalidates existing unlock sessions. Admins holding a live session bypass, stamped once per session. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- modules/site_lock/site_lock/middleware.py | 177 ++++++++++++++ modules/site_lock/site_lock/module.py | 47 ++++ modules/site_lock/site_lock/page.py | 55 +++++ .../site_lock/site_lock/templates/unlock.html | 64 +++++ modules/site_lock/tests/test_middleware.py | 223 ++++++++++++++++++ modules/site_lock/tests/test_page.py | 65 +++++ 6 files changed, 631 insertions(+) create mode 100644 modules/site_lock/site_lock/middleware.py create mode 100644 modules/site_lock/site_lock/module.py create mode 100644 modules/site_lock/site_lock/page.py create mode 100644 modules/site_lock/site_lock/templates/unlock.html create mode 100644 modules/site_lock/tests/test_middleware.py create mode 100644 modules/site_lock/tests/test_page.py diff --git a/modules/site_lock/site_lock/middleware.py b/modules/site_lock/site_lock/middleware.py new file mode 100644 index 00000000..6948ab25 --- /dev/null +++ b/modules/site_lock/site_lock/middleware.py @@ -0,0 +1,177 @@ +"""Site-wide password gate. + +Runs outermost among module middleware — the module sorts after ``Auth``, and +Starlette's ``add_middleware`` is LIFO — so this executes *before* +``AuthMiddleware``. That ordering is the whole point: an anonymous visitor +sees the gate instead of being redirected to the login page, so a locked site +never reveals that a login form exists. + +``SessionMiddleware`` sits outside this one, so ``scope["session"]`` is +readable here and writes are persisted on the way out. +""" + +from __future__ import annotations + +import hashlib +import logging +import secrets +from urllib.parse import quote + +from starlette.requests import Request +from starlette.responses import ( + HTMLResponse, + JSONResponse, + RedirectResponse, + Response, +) +from starlette.types import ASGIApp, Receive, Scope, Send + +from site_lock import constants as c +from site_lock.page import render_unlock_page, safe_next + +logger = logging.getLogger(__name__) + +_NO_STORE = {"Cache-Control": "no-store"} +_ERR_WRONG = "Incorrect password." +_ERR_THROTTLED = "Too many attempts. Try again later." + + +def password_fingerprint(password: str) -> str: + """Short digest of the active password, stored in the session. + + Comparing this on each request is what makes a password rotation + invalidate every previously-unlocked session. + """ + return hashlib.sha256(password.encode("utf-8")).hexdigest()[:16] + + +class SiteLockMiddleware: + """Gate every request behind a single shared password.""" + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + state = scope["app"].state.site_lock + settings = state.settings + + # Default-off fast path: one attribute read, no Request construction. + if not settings.enabled: + await self.app(scope, receive, send) + return + + path: str = scope["path"] + + # Kubernetes probes must never be gated, or the pod gets killed. + if path.startswith(c.HEALTH_PREFIX): + await self.app(scope, receive, send) + return + + fingerprint = password_fingerprint(settings.password) + session = scope.get("session") + + if path == c.UNLOCK_PATH: + response = await self._unlock(scope, receive, state, fingerprint, session) + await response(scope, receive, send) + return + + if session is not None and session.get(c.SESSION_KEY) == fingerprint: + await self.app(scope, receive, send) + return + + if await self._is_admin(scope): + # Stamp the marker so the provider lookup costs once per session + # rather than once per request. + if session is not None: + session[c.SESSION_KEY] = fingerprint + await self.app(scope, receive, send) + return + + await self._gate(scope, receive, send, path) + + async def _gate(self, scope: Scope, receive: Receive, send: Send, path: str) -> None: + has_auth_header = any(k == b"authorization" for k, _ in scope.get("headers", ())) + if path.startswith(c.API_PREFIX) or has_auth_header: + # 403 rather than 401: a 401 would invite an auth flow that cannot + # succeed while the site is locked. + response: Response = JSONResponse( + {"detail": "Site is locked"}, status_code=403, headers=_NO_STORE + ) + else: + target = f"{c.UNLOCK_PATH}?next={quote(safe_next(path), safe='/')}" + response = RedirectResponse(target, status_code=302, headers=_NO_STORE) + await response(scope, receive, send) + + async def _unlock( + self, + scope: Scope, + receive: Receive, + state, + fingerprint: str, + session, + ) -> Response: + method: str = scope.get("method", "GET") + settings = state.settings + request = Request(scope, receive) + + if method == "GET": + return HTMLResponse( + render_unlock_page( + message=settings.message, + next_url=request.query_params.get("next", "/"), + ), + headers=_NO_STORE, + ) + if method != "POST": + return Response(status_code=405, headers=_NO_STORE) + + client = scope.get("client") + key = client[0] if client else "unknown" + if state.limiter.is_blocked(key): + return self._page(settings, _ERR_THROTTLED, "/", status=429) + + form = await request.form() + supplied = str(form.get("password", "")) + target = safe_next(str(form.get("next", "/"))) + + if secrets.compare_digest(supplied, settings.password): + state.limiter.reset(key) + if session is not None: + session[c.SESSION_KEY] = fingerprint + return RedirectResponse(target, status_code=303, headers=_NO_STORE) + + state.limiter.record_failure(key) + return self._page(settings, _ERR_WRONG, target, status=401) + + @staticmethod + def _page(settings, error: str, next_url: str, *, status: int) -> HTMLResponse: + return HTMLResponse( + render_unlock_page(message=settings.message, error=error, next_url=next_url), + status_code=status, + headers=_NO_STORE, + ) + + @staticmethod + async def _is_admin(scope: Scope) -> bool: + """True when the caller already holds a session for an admin user. + + This is the documented lockout escape hatch: an admin who enables the + gate and mistypes the password keeps access to the settings screen. + """ + auth_state = getattr(scope["app"].state, "auth", None) + provider = getattr(auth_state, "auth_provider", None) + if provider is None: + return False + try: + user = await provider.resolve_user(Request(scope)) + except Exception: + logger.exception("Site lock admin bypass failed; treating as anonymous") + return False + return user is not None and c.ADMIN_ROLE in getattr(user, "roles", ()) + + +__all__ = ["SiteLockMiddleware", "password_fingerprint"] diff --git a/modules/site_lock/site_lock/module.py b/modules/site_lock/site_lock/module.py new file mode 100644 index 00000000..60b8d3a7 --- /dev/null +++ b/modules/site_lock/site_lock/module.py @@ -0,0 +1,47 @@ +"""Site Lock module — optional site-wide password gate.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from simple_module_core.module import ModuleBase, ModuleMeta + +from site_lock import constants as c + +if TYPE_CHECKING: + from fastapi import FastAPI + + +class SiteLockModule(ModuleBase): + meta = ModuleMeta( + name=c.MODULE_NAME, + # ``Settings`` so register_module_settings can reach the module + # registry; ``Auth`` so this module sorts *after* auth and its + # middleware therefore wraps outermost, executing before + # AuthMiddleware. Both are load-bearing — see the module README. + depends_on=[c.MODULE_SETTINGS, c.MODULE_AUTH], + ) + + def register_settings(self, app: FastAPI) -> None: + import importlib + + from site_lock.settings import SiteLockSettings + from site_lock.state import SiteLockState + + # SM009 is AST-based: resolving via importlib matches the convention + # used by the other settings-backed modules. + register_module_settings = importlib.import_module( + "settings.registration" + ).register_module_settings + + register_module_settings( + app, + c.MODULE_PACKAGE, + SiteLockSettings, + lambda s: SiteLockState(settings=s), + ) + + def register_middleware(self, app: FastAPI) -> None: + from site_lock.middleware import SiteLockMiddleware + + app.add_middleware(SiteLockMiddleware) diff --git a/modules/site_lock/site_lock/page.py b/modules/site_lock/site_lock/page.py new file mode 100644 index 00000000..4a76884f --- /dev/null +++ b/modules/site_lock/site_lock/page.py @@ -0,0 +1,55 @@ +"""Render the standalone unlock page. + +Kept out of ``middleware.py`` so the escaping and ``next``-sanitisation rules +can be unit-tested without an ASGI harness, and so both files stay well under +the 300-line cap. +""" + +from __future__ import annotations + +import html +import importlib.resources +from string import Template + +_TEMPLATE = Template( + (importlib.resources.files(__package__) / "templates" / "unlock.html").read_text( + encoding="utf-8" + ) +) + +_DEFAULT_MESSAGE = "This site is not public yet. Enter the password to continue." +_ERROR_BLOCK = '' + + +def safe_next(raw: str | None) -> str: + """Return ``raw`` if it is a same-site absolute path, else ``/``. + + Rejects protocol-relative (``//host``) and backslash-prefixed (``/\\host``) + targets — browsers resolve both off-site — plus anything carrying CR/LF, + which could otherwise be smuggled into the redirect header. Without this + the gate would be an open redirect. + """ + if not raw or not raw.startswith("/"): + return "/" + if raw.startswith("//") or raw.startswith("/\\"): + return "/" + if "\r" in raw or "\n" in raw: + return "/" + return raw + + +def render_unlock_page( + *, + message: str = "", + error: str = "", + next_url: str = "/", +) -> str: + """Render the gate page. Every interpolated value is HTML-escaped.""" + return _TEMPLATE.safe_substitute( + message=html.escape(message or _DEFAULT_MESSAGE), + error=_ERROR_BLOCK.format(message=html.escape(error)) if error else "", + next_url=html.escape(safe_next(next_url), quote=True), + ) + + +__all__ = ["render_unlock_page", "safe_next"] diff --git a/modules/site_lock/site_lock/templates/unlock.html b/modules/site_lock/site_lock/templates/unlock.html new file mode 100644 index 00000000..39f3a135 --- /dev/null +++ b/modules/site_lock/site_lock/templates/unlock.html @@ -0,0 +1,64 @@ + + + + + + +Protected + + + +
+

Protected

+

$message

+ $error +
+ + + + +
+
+ + diff --git a/modules/site_lock/tests/test_middleware.py b/modules/site_lock/tests/test_middleware.py new file mode 100644 index 00000000..54ee07b6 --- /dev/null +++ b/modules/site_lock/tests/test_middleware.py @@ -0,0 +1,223 @@ +"""SiteLockMiddleware gating behaviour.""" + +from __future__ import annotations + +import httpx +from fastapi import FastAPI +from site_lock import constants as c +from site_lock.middleware import SiteLockMiddleware, password_fingerprint +from site_lock.settings import SiteLockSettings +from site_lock.state import SiteLockState +from starlette.middleware.sessions import SessionMiddleware +from starlette.responses import JSONResponse + +SECRET = "test-site-lock-secret" +PASSWORD = "hunter2" + + +class _StubProvider: + """Minimal AuthProvider stand-in for the admin-bypass path.""" + + def __init__(self, user=None): + self._user = user + + async def resolve_user(self, request): + return self._user + + +def _build_app(settings: SiteLockSettings, *, provider=None) -> FastAPI: + app = FastAPI() + app.state.site_lock = SiteLockState(settings=settings) + app.state.auth = type("_AuthState", (), {"auth_provider": provider})() + + async def _handler(path: str = ""): + return JSONResponse({"ok": True}) + + app.add_api_route("/{path:path}", _handler, methods=["GET", "POST"]) + app.add_middleware(SiteLockMiddleware) + app.add_middleware(SessionMiddleware, secret_key=SECRET) + return app + + +def _client(app: FastAPI) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + follow_redirects=False, + ) + + +async def test_disabled_passes_everything_through() -> None: + app = _build_app(SiteLockSettings()) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 200 + assert response.json() == {"ok": True} + + +async def test_enabled_redirects_browser_to_the_gate() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + assert response.headers["location"].startswith(c.UNLOCK_PATH) + assert response.headers["cache-control"] == "no-store" + + +async def test_enabled_returns_403_json_for_api() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/api/users/me") + assert response.status_code == 403 + assert response.json() == {"detail": "Site is locked"} + + +async def test_bearer_request_gets_403_json_not_a_redirect() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/dashboard/", headers={"Authorization": "Bearer x"}) + assert response.status_code == 403 + + +async def test_health_is_never_gated() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.get("/health") + assert response.status_code == 200 + + +async def test_gate_page_is_served() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD, message="Soon")) + async with _client(app) as client: + response = await client.get(c.UNLOCK_PATH) + assert response.status_code == 200 + assert 'name="password"' in response.text + assert "Soon" in response.text + + +async def test_correct_password_unlocks_and_persists() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post( + c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/dashboard/"} + ) + assert posted.status_code == 303 + assert posted.headers["location"] == "/dashboard/" + # The session cookie now carries the unlock marker. + followed = await client.get("/dashboard/") + assert followed.status_code == 200 + + +async def test_wrong_password_is_rejected() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post(c.UNLOCK_PATH, data={"password": "nope", "next": "/"}) + assert posted.status_code == 401 + followed = await client.get("/dashboard/") + assert followed.status_code == 302 + + +async def test_offsite_next_is_not_honoured_on_redirect() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + posted = await client.post( + c.UNLOCK_PATH, data={"password": PASSWORD, "next": "//evil.example"} + ) + assert posted.headers["location"] == "/" + + +async def test_rotating_the_password_invalidates_existing_sessions() -> None: + settings = SiteLockSettings(enabled=True, password=PASSWORD) + app = _build_app(settings) + async with _client(app) as client: + await client.post(c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"}) + assert (await client.get("/dashboard/")).status_code == 200 + # Operator rotates the password via the settings UI. + app.state.site_lock.settings = SiteLockSettings(enabled=True, password="new-one") + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_unsupported_method_on_the_gate_is_405() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + response = await client.request("PUT", c.UNLOCK_PATH) + assert response.status_code == 405 + + +async def test_rate_limited_after_repeated_failures() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD)) + async with _client(app) as client: + for _ in range(c.MAX_FAILURES): + await client.post(c.UNLOCK_PATH, data={"password": "nope", "next": "/"}) + response = await client.post(c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"}) + assert response.status_code == 429 + + +def test_fingerprint_changes_with_the_password() -> None: + assert password_fingerprint("a") != password_fingerprint("b") + assert password_fingerprint("a") == password_fingerprint("a") + + +class _User: + def __init__(self, roles: list[str]) -> None: + self.id = "11111111-2222-3333-4444-555555555555" + self.roles = roles + + +async def test_admin_with_a_live_session_bypasses_the_gate() -> None: + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_StubProvider(_User(["admin"])), + ) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 200 + + +async def test_authenticated_non_admin_does_not_bypass() -> None: + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_StubProvider(_User(["user"])), + ) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_admin_bypass_stamps_the_session_once() -> None: + """After the first bypass the marker short-circuits further lookups.""" + calls = [] + + class _CountingProvider(_StubProvider): + async def resolve_user(self, request): + calls.append(1) + return _User(["admin"]) + + app = _build_app( + SiteLockSettings(enabled=True, password=PASSWORD), + provider=_CountingProvider(), + ) + async with _client(app) as client: + await client.get("/dashboard/") + await client.get("/dashboard/") + await client.get("/dashboard/") + assert len(calls) == 1 + + +async def test_missing_provider_does_not_crash_the_gate() -> None: + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD), provider=None) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 + + +async def test_provider_that_raises_is_treated_as_anonymous() -> None: + class _BoomProvider: + async def resolve_user(self, request): + raise RuntimeError("provider exploded") + + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD), provider=_BoomProvider()) + async with _client(app) as client: + response = await client.get("/dashboard/") + assert response.status_code == 302 diff --git a/modules/site_lock/tests/test_page.py b/modules/site_lock/tests/test_page.py new file mode 100644 index 00000000..e462a6a9 --- /dev/null +++ b/modules/site_lock/tests/test_page.py @@ -0,0 +1,65 @@ +"""Gate page rendering, escaping, and next-target sanitisation.""" + +from __future__ import annotations + +import pytest +from site_lock.page import render_unlock_page, safe_next + + +@pytest.mark.parametrize( + "raw", + [ + None, + "", + "//evil.example/path", + "https://evil.example", + "http://evil.example", + "/\\evil.example", + "evil", + "/ok\r\nSet-Cookie: x=1", + ], +) +def test_safe_next_rejects_offsite_and_injection(raw: str | None) -> None: + assert safe_next(raw) == "/" + + +@pytest.mark.parametrize("raw", ["/", "/dashboard/", "/users/me"]) +def test_safe_next_allows_same_site_paths(raw: str) -> None: + assert safe_next(raw) == raw + + +def test_message_is_html_escaped() -> None: + html = render_unlock_page(message="") + assert "" not in html + assert "<script>" in html + + +def test_error_is_rendered_and_escaped() -> None: + html = render_unlock_page(error="Bad password") + assert "<b>" in html + assert "password" not in html + + +def test_no_error_block_when_no_error() -> None: + assert 'role="alert"' not in render_unlock_page() + + +def test_next_is_embedded_as_a_hidden_field() -> None: + html = render_unlock_page(next_url="/dashboard/") + assert 'name="next"' in html + assert "/dashboard/" in html + + +def test_offsite_next_is_neutralised_in_the_form() -> None: + html = render_unlock_page(next_url="//evil.example") + assert "evil.example" not in html + + +def test_page_is_self_contained() -> None: + html = render_unlock_page() + # No external assets: a locked site must serve exactly one document. + assert " Date: Wed, 5 Aug 2026 19:06:48 +0200 Subject: [PATCH 5/7] feat(site_lock): host wiring, CLI catalog entry, and docs Installs the module in the host, pins SiteLockMiddleware ahead of AuthMiddleware in the middleware-order test, and adds the docs page + nav. Renames the module's test files with a site_lock_ prefix: pytest resolves test modules by basename with no __init__.py, so test_settings.py and test_rate_limit.py collided with the users module's files of the same name. Makes test_wizard_custom_picks_only_yes_answers derive its answer sequence from CATALOG instead of hardcoding a count -- the hardcoded sequence shifted onto the wrong module as soon as a new catalog entry landed. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- docs/.vitepress/config.ts | 1 + docs/index.md | 2 +- docs/modules/index.md | 1 + docs/modules/site_lock.md | 78 +++++++++++++++++++ .../superpowers/plans/2026-08-05-site-lock.md | 8 +- framework/cli/simple_module_cli/catalog.py | 6 ++ framework/cli/tests/test_cli_wizard.py | 8 +- .../hosting/tests/test_middleware_order.py | 10 ++- host/pyproject.toml | 2 + modules/site_lock/site_lock/page.py | 2 +- ...leware.py => test_site_lock_middleware.py} | 0 .../{test_page.py => test_site_lock_page.py} | 0 ..._limit.py => test_site_lock_rate_limit.py} | 0 ...settings.py => test_site_lock_settings.py} | 0 14 files changed, 108 insertions(+), 10 deletions(-) create mode 100644 docs/modules/site_lock.md rename modules/site_lock/tests/{test_middleware.py => test_site_lock_middleware.py} (100%) rename modules/site_lock/tests/{test_page.py => test_site_lock_page.py} (100%) rename modules/site_lock/tests/{test_rate_limit.py => test_site_lock_rate_limit.py} (100%) rename modules/site_lock/tests/{test_settings.py => test_site_lock_settings.py} (100%) diff --git a/docs/.vitepress/config.ts b/docs/.vitepress/config.ts index 6d493204..29de7b72 100644 --- a/docs/.vitepress/config.ts +++ b/docs/.vitepress/config.ts @@ -183,6 +183,7 @@ export default defineConfig({ { text: "branding", link: "/modules/branding" }, { text: "background_tasks", link: "/modules/background_tasks" }, { text: "audit_log", link: "/modules/audit_log" }, + { text: "site_lock", link: "/modules/site_lock" }, { text: "dashboard", link: "/modules/dashboard" }, ], }, diff --git a/docs/index.md b/docs/index.md index 164799f2..a7fda0e4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -26,7 +26,7 @@ features: link: /guide/first-module linkText: Build a module - title: Use a bundled module - details: Ten first-party modules ship with the framework — auth, users, keycloak, permissions, settings, file_storage, background_tasks, feature_flags, audit_log, dashboard. + details: Eleven first-party modules ship with the framework — auth, users, keycloak, permissions, settings, file_storage, background_tasks, feature_flags, audit_log, dashboard, site_lock. link: /modules/ linkText: Browse modules - title: Operate it in production diff --git a/docs/modules/index.md b/docs/modules/index.md index a7345be4..c8fce364 100644 --- a/docs/modules/index.md +++ b/docs/modules/index.md @@ -15,6 +15,7 @@ simple_module_python ships with eleven first-party modules. Each is a regular Py | [`background_tasks`](/modules/background_tasks) | `users` | Celery + Redis workers, persistent task history, retry, stuck-task sweep, live worker dashboard. | | [`audit_log`](/modules/audit_log) | `users` | Automatic field-level audit trail for SQLModel entities, with an admin UI to browse change history. | | [`dashboard`](/modules/dashboard) | `users` | Authenticated landing page with system overview (user counts, module list, health checks). | +| [`site_lock`](/modules/site_lock) | `settings`, `auth` | Optional site-wide shared-password gate for staging / pre-launch sites. Off by default. | ## How modules are wired in diff --git a/docs/modules/site_lock.md b/docs/modules/site_lock.md new file mode 100644 index 00000000..e297bc9d --- /dev/null +++ b/docs/modules/site_lock.md @@ -0,0 +1,78 @@ +# site_lock + +Optional site-wide password gate — a staging / pre-launch door. When enabled, every visitor must enter one shared password before they can see anything: not the landing page, not the API, not even that a login form exists. + +**Off by default.** Installing the module changes nothing until an operator turns it on. + +This is not user authentication — that is what [`auth`](/modules/auth), [`users`](/modules/users), and [`permissions`](/modules/permissions) are for. The site lock is one shared secret in front of everything, with no notion of identity. + +## ModuleMeta + +| Field | Value | +|---|---| +| `name` | `SiteLock` | +| `route_prefix` | *(none)* | +| `view_prefix` | *(none)* | +| `depends_on` | `["Settings", "Auth"]` | + +Both dependencies are load-bearing. `Settings` lets `register_module_settings` reach the module registry. `Auth` makes this module sort *after* the auth module, and since middleware is installed in topological order while Starlette's `add_middleware` is LIFO, sorting later means wrapping **outermost** — so `SiteLockMiddleware` executes *before* `AuthMiddleware`. + +That ordering is the point of the design: an anonymous visitor gets the gate rather than a redirect to the login page, so a locked site never reveals it has one. The order is pinned by `framework/hosting/tests/test_middleware_order.py`. + +## Routes + +The module registers **no routes**. The gate is served entirely from middleware at `/__unlock`, which keeps it reachable before auth runs and means a locked site serves exactly one document — no menus, no branding, no JS bundle. The gate keeps working even if the frontend build is broken. + +| Method + path | Response | +|---|---| +| `GET /__unlock` | The gate page (200) | +| `POST /__unlock` | 303 to `next` on success; 401 on a wrong password; 429 when rate-limited | +| Other methods on `/__unlock` | 405 | + +## Settings + +DB-backed and hot-reloadable — changes apply immediately, no restart. + +| Field | Type | Default | Meaning | +|---|---|---|---| +| `enabled` | `bool` | `false` | Master switch | +| `password` | `str` | `""` | The shared password. Masked in the admin UI | +| `message` | `str` | `""` | Optional line shown on the gate page | + +Enabling with a blank password is rejected by a model validator, so the settings screen shows an error rather than gating the site behind the empty string. + +## Behaviour when locked + +| Request | Result | +|---|---| +| `/health*` | **Always passes.** Gating it would fail Kubernetes probes and get the pod killed | +| `/__unlock` | The gate page | +| `/api/*`, or any request with an `Authorization` header | `403 {"detail": "Site is locked"}` | +| Anything else | `302` to `/__unlock?next=…` | + +A `403` rather than `401` is deliberate: a `401` would invite an auth flow that cannot succeed while the site is locked. Every gated response carries `Cache-Control: no-store`. + +Unlock state lives in the signed session cookie. The stored marker is a fingerprint of the current password, so **rotating the password immediately invalidates every unlocked session**. + +## Admin bypass and the lockout it does not cover + +A user already holding a session with the `admin` role skips the gate. This is the intended escape hatch: enable the gate, mistype the password, and you are still holding the session that lets you go back to Settings and fix it. The bypass stamps the session marker on first use, so the provider lookup costs once per session rather than once per request. + +**It only rescues a live session.** If no admin is signed in and the password has been forgotten, there is no in-app recovery — clear the override directly: + +```sql +DELETE FROM settings_setting + WHERE scope = 'system' AND key = 'site_lock.enabled'; +``` + +Then restart the app (or save any setting) so the module re-hydrates. + +## Brute-force protection + +The unlock endpoint tracks failures per client IP in memory: 10 failures within 5 minutes trigger a 15-minute cooldown returning `429`. These thresholds are module constants, not settings — they are the only defence on a single shared secret, so they are not an operator-tunable surface. + +The limiter is process-local, which suits the single-process staging deployments this module targets. It lives on the module state rather than inside the settings object, so an unrelated settings save cannot clear an in-flight cooldown. + +## Open-redirect protection + +The `next` target is sanitised before being echoed into the form or used in a redirect. Only same-site absolute paths are accepted; protocol-relative (`//host`), backslash-prefixed (`/\host`), and CR/LF-carrying values all fall back to `/`. diff --git a/docs/superpowers/plans/2026-08-05-site-lock.md b/docs/superpowers/plans/2026-08-05-site-lock.md index 984f596e..f0dbc62e 100644 --- a/docs/superpowers/plans/2026-08-05-site-lock.md +++ b/docs/superpowers/plans/2026-08-05-site-lock.md @@ -865,9 +865,7 @@ async def test_rate_limited_after_repeated_failures() -> None: async with _client(app) as client: for _ in range(c.MAX_FAILURES): await client.post(c.UNLOCK_PATH, data={"password": "nope", "next": "/"}) - response = await client.post( - c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"} - ) + response = await client.post(c.UNLOCK_PATH, data={"password": PASSWORD, "next": "/"}) assert response.status_code == 429 @@ -1208,9 +1206,7 @@ async def test_provider_that_raises_is_treated_as_anonymous() -> None: async def resolve_user(self, request): raise RuntimeError("provider exploded") - app = _build_app( - SiteLockSettings(enabled=True, password=PASSWORD), provider=_BoomProvider() - ) + app = _build_app(SiteLockSettings(enabled=True, password=PASSWORD), provider=_BoomProvider()) async with _client(app) as client: response = await client.get("/dashboard/") assert response.status_code == 302 diff --git a/framework/cli/simple_module_cli/catalog.py b/framework/cli/simple_module_cli/catalog.py index 43be0ba8..dfaf0660 100644 --- a/framework/cli/simple_module_cli/catalog.py +++ b/framework/cli/simple_module_cli/catalog.py @@ -57,6 +57,12 @@ class ModuleEntry: requires=("users",), recipe="background_tasks", ), + "site_lock": ModuleEntry( + "site_lock", + "simple_module_site_lock", + "Site Lock", + requires=("auth", "settings"), + ), } diff --git a/framework/cli/tests/test_cli_wizard.py b/framework/cli/tests/test_cli_wizard.py index 9bcc5735..e233a4b8 100644 --- a/framework/cli/tests/test_cli_wizard.py +++ b/framework/cli/tests/test_cli_wizard.py @@ -3,6 +3,7 @@ from __future__ import annotations import typer +from simple_module_cli.catalog import CATALOG from simple_module_cli.wizard import run_wizard from typer.testing import CliRunner @@ -52,7 +53,12 @@ def test_wizard_full_preset_includes_background_tasks() -> None: def test_wizard_custom_picks_only_yes_answers() -> None: - answers = ["", "", "4"] + ["n"] * 7 + ["y", ""] + # The custom path prompts once per catalog module, in CATALOG order. + # Derive the answers from CATALOG rather than hardcoding a count: a + # hardcoded sequence silently shifts onto the wrong module the next time + # one is added, which is how this test broke when `site_lock` landed. + picks = ["y" if name == "background_tasks" else "n" for name in CATALOG] + answers = ["", "", "4", *picks, ""] _, _, selected, out = _drive(answers) assert set(selected) == {"background_tasks", "users", "auth"} assert "Added users (required by background_tasks)" in out diff --git a/framework/hosting/tests/test_middleware_order.py b/framework/hosting/tests/test_middleware_order.py index 6ccd37de..ff184363 100644 --- a/framework/hosting/tests/test_middleware_order.py +++ b/framework/hosting/tests/test_middleware_order.py @@ -7,7 +7,13 @@ Tenant/Locale must see ``request.state.user`` set by AuthMiddleware so DB queries get filtered correctly; CorrelationId must wrap everything so -every log line carries its id. GZip sits inside the observability pair so +every log line carries its id. SiteLock must precede AuthMiddleware: it gates +anonymous visitors itself, and if Auth ran first they would be redirected to +the login page — revealing that a login form exists on a site that is meant +to be fully hidden. That inversion breaks the feature without failing any +site_lock unit test, which is why the order is pinned here. + +GZip sits inside the observability pair so those still see every request, but outside everything that produces a body — including the /static mount, which is where compression pays off most. Order matters and a swap is the kind of @@ -28,6 +34,7 @@ "GZipMiddleware", "SecurityHeadersMiddleware", "SessionMiddleware", + "SiteLockMiddleware", "AuthMiddleware", "TenantMiddleware", "LocaleMiddleware", @@ -40,6 +47,7 @@ "GZipMiddleware", "SecurityHeadersMiddleware", "SessionMiddleware", + "SiteLockMiddleware", "AuthMiddleware", "LocaleMiddleware", "InertiaLayoutDataMiddleware", diff --git a/host/pyproject.toml b/host/pyproject.toml index ef02ea95..a4976c9c 100644 --- a/host/pyproject.toml +++ b/host/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "simple_module_feature_flags", "simple_module_audit_log", "simple_module_branding", + "simple_module_site_lock", "python-multipart>=0.0.6", ] @@ -28,3 +29,4 @@ simple_module_settings = { workspace = true } simple_module_feature_flags = { workspace = true } simple_module_audit_log = { workspace = true } simple_module_branding = { workspace = true } +simple_module_site_lock = { workspace = true } diff --git a/modules/site_lock/site_lock/page.py b/modules/site_lock/site_lock/page.py index 4a76884f..6dface1d 100644 --- a/modules/site_lock/site_lock/page.py +++ b/modules/site_lock/site_lock/page.py @@ -31,7 +31,7 @@ def safe_next(raw: str | None) -> str: """ if not raw or not raw.startswith("/"): return "/" - if raw.startswith("//") or raw.startswith("/\\"): + if raw.startswith(("//", "/\\")): return "/" if "\r" in raw or "\n" in raw: return "/" diff --git a/modules/site_lock/tests/test_middleware.py b/modules/site_lock/tests/test_site_lock_middleware.py similarity index 100% rename from modules/site_lock/tests/test_middleware.py rename to modules/site_lock/tests/test_site_lock_middleware.py diff --git a/modules/site_lock/tests/test_page.py b/modules/site_lock/tests/test_site_lock_page.py similarity index 100% rename from modules/site_lock/tests/test_page.py rename to modules/site_lock/tests/test_site_lock_page.py diff --git a/modules/site_lock/tests/test_rate_limit.py b/modules/site_lock/tests/test_site_lock_rate_limit.py similarity index 100% rename from modules/site_lock/tests/test_rate_limit.py rename to modules/site_lock/tests/test_site_lock_rate_limit.py diff --git a/modules/site_lock/tests/test_settings.py b/modules/site_lock/tests/test_site_lock_settings.py similarity index 100% rename from modules/site_lock/tests/test_settings.py rename to modules/site_lock/tests/test_site_lock_settings.py From 58a41f4bf52acf8fecf334445584094c240865e3 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Wed, 5 Aug 2026 19:10:10 +0200 Subject: [PATCH 6/7] test(site_lock): boot-level integration coverage Exercises the real create_app pipeline: module state mounted, settings registered for the admin UI, middleware ordered ahead of Auth, and the gate's behaviour on a booted app both off and on. Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- .../tests/test_site_lock_integration.py | 75 +++++++++++++++++++ 1 file changed, 75 insertions(+) create mode 100644 modules/site_lock/tests/test_site_lock_integration.py diff --git a/modules/site_lock/tests/test_site_lock_integration.py b/modules/site_lock/tests/test_site_lock_integration.py new file mode 100644 index 00000000..665a6a56 --- /dev/null +++ b/modules/site_lock/tests/test_site_lock_integration.py @@ -0,0 +1,75 @@ +"""Boot-level wiring for the site_lock module. + +The unit tests in ``test_site_lock_middleware.py`` drive the middleware in a +hand-built ASGI stack. These exercise the real ``create_app`` pipeline so a +regression in discovery, settings registration, or middleware ordering is +caught even though every unit test would still pass. +""" + +from __future__ import annotations + +import pytest +from site_lock import constants as c +from site_lock.settings import SiteLockSettings + + +def test_module_state_is_mounted_and_disabled_by_default(app) -> None: + """The off-by-default guarantee, asserted against a really-booted app.""" + state = getattr(app.state, c.MODULE_PACKAGE) + assert state.settings.enabled is False + + +def test_site_lock_middleware_runs_before_auth(app) -> None: + """Ordering is load-bearing: if Auth ran first, anonymous visitors would + be redirected to the login page instead of seeing the gate.""" + names = [m.cls.__name__ for m in app.user_middleware] + assert names.index("SiteLockMiddleware") < names.index("AuthMiddleware") + + +def test_settings_are_registered_for_the_admin_ui(app) -> None: + registry = app.state.settings.module_registry + assert registry.get(c.MODULE_PACKAGE) is SiteLockSettings + + +async def test_disabled_gate_does_not_interfere(client) -> None: + """With the gate off the request reaches the app untouched. + + The test host mounts no landing route, so a 404 here is the *app's* own + answer — the point is that it is not a 302 to the gate. + """ + response = await client.get("/", follow_redirects=False) + assert response.status_code != 302 + assert c.UNLOCK_PATH not in response.headers.get("location", "") + + +@pytest.fixture +def locked(app): + """Turn the gate on the way the settings UI does, then restore it.""" + state = getattr(app.state, c.MODULE_PACKAGE) + original = state.settings + state.settings = SiteLockSettings(enabled=True, password="pw") + yield + state.settings = original + + +async def test_locked_site_gates_the_landing_page(client, locked) -> None: + response = await client.get("/", follow_redirects=False) + assert response.status_code == 302 + assert response.headers["location"].startswith(c.UNLOCK_PATH) + + +async def test_locked_site_still_answers_health_probes(client, locked) -> None: + response = await client.get("/health") + assert response.status_code == 200 + + +async def test_locked_site_serves_the_gate_page(client, locked) -> None: + response = await client.get(c.UNLOCK_PATH) + assert response.status_code == 200 + assert 'name="password"' in response.text + + +async def test_locked_site_returns_json_for_api_routes(client, locked) -> None: + response = await client.get("/api/users/me", follow_redirects=False) + assert response.status_code == 403 + assert response.json() == {"detail": "Site is locked"} From 0f66eaa29fda7394a528b60078ac992bf347d396 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 6 Aug 2026 10:38:32 +0200 Subject: [PATCH 7/7] fix(site_lock): surface the blank-password error in the settings UI Browser QA found that enabling the gate without a password looked like it worked. The backend correctly returned 422 and persisted nothing, but the admin saw no error: the toggle stayed on, "Reset to default" appeared, and the obvious reading was that the site was now locked. It was wide open. Cause: a bare `raise ValueError` in a `model_validator(mode="after")` produces `loc=[]`. `ModuleForm.onSave` keys field errors by `loc[-1]` and drops anything it cannot attach to a field, so the message was discarded before render. Raise a pydantic-core validation error pinned to `enabled` instead, so it lands under the toggle the admin just flipped. Worst possible failure mode for a security feature, so it gets a regression test asserting the error's `loc`, not just that it raises. Also add the Install/Usage sections `scripts/check_readmes.py` requires -- this was failing `make lint` since the module landed, missed because I had been gating on `make ci-python-lint` -- and document that the gate returns 403 to the login API too, so no programmatic client can get in while it is on. Verified: make lint (exit 0), make test-py (1562 passed). Claude-Session: https://claude.ai/code/session_01854CBXUPwhDSAkWbb81aRu --- modules/site_lock/README.md | 39 +++++++++++++++++++ modules/site_lock/site_lock/settings.py | 25 +++++++++++- .../tests/test_site_lock_settings.py | 17 ++++++++ 3 files changed, 80 insertions(+), 1 deletion(-) diff --git a/modules/site_lock/README.md b/modules/site_lock/README.md index d06c7ed8..fd6eabc8 100644 --- a/modules/site_lock/README.md +++ b/modules/site_lock/README.md @@ -8,6 +8,38 @@ that a login form exists. **Off by default.** Installing this module changes nothing until an operator turns it on. +## Install + +Add the package to your host and re-sync — discovery picks it up from the +`simple_module` entry point, no host code changes: + +```toml +# host/pyproject.toml +dependencies = ["simple_module_site_lock"] +``` + +```bash +uv sync --all-packages +``` + +Or scaffold a new app with it selected: `smpy new --modules site_lock`. + +## Usage + +1. Sign in as an admin and open **Settings → Modules → SiteLock**. +2. Set `password` (and optionally `message`), tick `enabled`, and **Save**. + Saving with a blank password is refused — the error appears on the + `enabled` field and nothing is persisted. +3. Anyone without the password now gets the gate page at `/__unlock` on every + URL. They enter the password once and are returned to where they were + heading. +4. To lift the gate, untick `enabled` and save. To rotate the password, set a + new one — every already-unlocked visitor is logged back out of the gate. + +Keep your own session alive while you do this: an admin with a live session +skips the gate, which is what lets you undo a mistake (see +[Admin bypass](#admin-bypass-and-the-lockout-you-can-still-cause) below). + ## How it works The module installs a single middleware that runs *before* `AuthMiddleware`. @@ -48,6 +80,13 @@ Everything else is gated. Requests under `/api/`, and any request carrying an redirect — a `401` would invite an auth flow that cannot succeed while the site is locked. Browser requests get a `302` to the gate. +That includes the login API itself, so while the gate is on there is no +programmatic way in: CI smoke tests, uptime monitors, mobile clients and +webhook senders all get `403` no matter what credentials they hold. Only a +browser that has passed the gate (or an admin holding a live session) can +reach anything. Pause those integrations before enabling the gate, or point +uptime checks at `/health`, which stays open. + ## Admin bypass, and the lockout you can still cause A user who already holds a session with the `admin` role skips the gate. diff --git a/modules/site_lock/site_lock/settings.py b/modules/site_lock/site_lock/settings.py index 8699df58..b39fa43d 100644 --- a/modules/site_lock/site_lock/settings.py +++ b/modules/site_lock/site_lock/settings.py @@ -7,8 +7,12 @@ from __future__ import annotations from pydantic import model_validator +from pydantic_core import InitErrorDetails, PydanticCustomError +from pydantic_core import ValidationError as CoreValidationError from pydantic_settings import BaseSettings, SettingsConfigDict +_BLANK_PASSWORD_MSG = "password must be set before enabling the site lock" + class SiteLockSettings(BaseSettings): """Site-wide shared-password gate configuration.""" @@ -27,7 +31,26 @@ def _password_required_when_enabled(self) -> SiteLockSettings: lock every visitor out from behind a secret that is the empty string. Raising here makes ``apply_changes_and_reload`` reject the change so the settings screen shows a validation error instead. + + The error is deliberately pinned to ``enabled`` rather than raised as a + bare ``ValueError``. A plain raise from a model validator carries + ``loc=[]``, and the shared settings form only renders errors it can + attach to a field (``ModuleForm.onSave`` keys them by ``loc[-1]``). + With an empty ``loc`` the 422 was silently swallowed: the admin saw the + toggle stay on, no error, and had every reason to believe the site was + locked when it was still wide open. """ if self.enabled and not self.password.strip(): - raise ValueError("password must be set before enabling the site lock") + raise CoreValidationError.from_exception_data( + title=type(self).__name__, + line_errors=[ + InitErrorDetails( + type=PydanticCustomError( + "site_lock_password_required", _BLANK_PASSWORD_MSG + ), + loc=("enabled",), + input=self.enabled, + ) + ], + ) return self diff --git a/modules/site_lock/tests/test_site_lock_settings.py b/modules/site_lock/tests/test_site_lock_settings.py index 3da339ec..39e02273 100644 --- a/modules/site_lock/tests/test_site_lock_settings.py +++ b/modules/site_lock/tests/test_site_lock_settings.py @@ -25,6 +25,23 @@ def test_enabled_without_password_is_rejected(password: str) -> None: SiteLockSettings(enabled=True, password=password) +@pytest.mark.parametrize("password", ["", " "]) +def test_blank_password_error_is_pinned_to_the_enabled_field(password: str) -> None: + """The 422 must name a field, or the admin never sees it. + + ``ModuleForm.onSave`` keys errors by ``loc[-1]`` and drops any error whose + ``loc`` is empty — which is what a bare ``raise ValueError`` in a model + validator produces. That combination silently swallowed this error: the + toggle stayed on with no message and the site stayed unlocked. + """ + with pytest.raises(ValidationError) as exc_info: + SiteLockSettings(enabled=True, password=password) + + errors = exc_info.value.errors() + assert [e["loc"] for e in errors] == [("enabled",)] + assert "password must be set" in errors[0]["msg"] + + def test_password_field_is_masked_by_settings_ui() -> None: # The admin UI masks fields by name; `password` must match that regex. from settings._module_settings import is_secret_field