fix(auth): password policy + current-password check for the superuser (#1285)

CHANGELOG.md

@@ -9,6 +9,16 @@ and this project aims to follow [Semantic Versioning](https://semver.org/spec/v2
 
 ### Changed
 
+- **The admin account now has a password policy (#1285).** **Behavior change:**
+  a password must be at least 12 characters and must not be the email. This
+  applies at registration, on `PATCH /users/me` and `PATCH /users/{id}`, in
+  `codeframe auth set-password`, and on the web sign-up form. Changing the
+  password or email now requires `current_password`. Before this, the
+  bootstrap superuser could be registered with `a`, its password could be
+  emptied, and anyone holding its JWT could take the account permanently by
+  changing the password or email. Logout does not revoke a JWT. Existing
+  shorter passwords still log in; the rule applies only when a password is set.
+
 - **First-account registration refuses every proxied request, and rate limiting
   reads the real client IP behind a proxy (#1274).** **Behavior change:** with
   no `CODEFRAME_BOOTSTRAP_TOKEN` set, `POST /auth/register` now accepts only a

codeframe/auth/manager.py

@@ -4,7 +4,7 @@
 from typing import AsyncGenerator, Optional
 
 from fastapi import Depends, Request
-from fastapi_users import BaseUserManager, IntegerIDMixin, FastAPIUsers
+from fastapi_users import BaseUserManager, IntegerIDMixin, FastAPIUsers, exceptions
 from fastapi_users.authentication import (
     AuthenticationBackend,
     BearerTransport,
@@ -208,6 +208,19 @@ def get_async_session_maker():
     return _async_session_maker
 
 
+#: The bootstrap account is a superuser with terminal access (#1285).
+MIN_PASSWORD_LENGTH = 12
+
+
+def password_policy_error(password: str, email: str | None) -> str | None:
+    """Why ``password`` is unacceptable, or None. Shared with the offline CLI."""
+    if len(password) < MIN_PASSWORD_LENGTH:
+        return f"Password must be at least {MIN_PASSWORD_LENGTH} characters."
+    if email and password.strip().lower() == email.strip().lower():
+        return "Password must not be the account's email address."
+    return None
+
+
 class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
     """User manager for CodeFRAME."""
 
@@ -258,6 +271,50 @@ async def authenticate(self, credentials):
             )
         return user
 
+    async def validate_password(self, password: str, user) -> None:
+        """Runs on register and on every password PATCH, including ``""`` (#1285)."""
+        reason = password_policy_error(password, getattr(user, "email", None))
+        if reason:
+            raise exceptions.InvalidPasswordException(reason=reason)
+
+    async def update(self, user_update, user, safe=False, request=None):
+        """Changing the password or email requires the current password (#1285).
+
+        Otherwise a leaked JWT — valid for its whole lifetime, logout does not
+        revoke — becomes a permanent takeover of the one admin account. Checked
+        against the *target* user on both ``/users/me`` and ``/users/{id}``:
+        exempting the superuser route would let the sole superuser skip the
+        check by addressing its own id.
+        """
+        changes = user_update.model_dump(exclude_unset=True)
+        sensitive = changes.get("password") is not None or (
+            changes.get("email") is not None and changes["email"] != user.email
+        )
+        if sensitive and not await self._current_password_matches(
+            changes.get("current_password"), user
+        ):
+            raise exceptions.InvalidPasswordException(
+                reason="The current password is required to change the password or email."
+            )
+        # fastapi-users validates a new password against the *old* row, so a
+        # PATCH setting both fields to the same value would pass the email rule.
+        if changes.get("password") is not None and changes.get("email"):
+            reason = password_policy_error(changes["password"], changes["email"])
+            if reason:
+                raise exceptions.InvalidPasswordException(reason=reason)
+        return await super().update(user_update, user, safe=safe, request=request)
+
+    async def _current_password_matches(self, password: Optional[str], user: User) -> bool:
+        if not password:
+            return False
+        try:
+            verified, _ = self.password_helper.verify_and_update(
+                password, user.hashed_password
+            )
+        except UnknownHashError:  # the seeded '!DISABLED!' row (#938)
+            return False
+        return verified
+
     async def on_after_register(self, user: User, request: Optional[Request] = None):
         """Called after successful registration."""
         logger.info(

codeframe/auth/schemas.py

@@ -13,3 +13,6 @@ class UserCreate(schemas.BaseUserCreate):
 class UserUpdate(schemas.BaseUserUpdate):
     """Schema for updating users."""
     name: Optional[str] = None
+    #: Required to change password or email (#1285). Not a column, so the
+    #: update only sets it as a plain attribute; UserRead never returns it.
+    current_password: Optional[str] = None

codeframe/cli/auth_commands.py

@@ -469,7 +469,9 @@ def register(
             if "ALREADY_EXISTS" in str(error_detail).upper():
                 console.print("[red]Error:[/red] An account with this email already exists")
             else:
-                console.print(f"[red]Error:[/red] Registration failed: {error_detail}")
+                if isinstance(error_detail, dict):  # e.g. the password policy (#1285)
+                    error_detail = error_detail.get("reason") or error_detail
+                console.print(f"[red]Error:[/red] Registration failed: {escape(str(error_detail))}")
             raise typer.Exit(1)
 
         elif response.status_code == 422:
@@ -1186,6 +1188,14 @@ def set_password(
     if not password:
         password = typer.prompt("New password", hide_input=True, confirmation_prompt=True)
 
+    # Same policy as the API (#1285) — otherwise this is the way around it.
+    from codeframe.auth.manager import password_policy_error
+
+    reason = password_policy_error(password, email)
+    if reason:
+        print_error(reason)
+        raise typer.Exit(1)
+
     try:
         _set_user_password(get_db_for_cli(), email, password)
     except LookupError as e:

deploy/README.md

@@ -133,6 +133,11 @@ CODEFRAME_API_URL="http://127.0.0.1:${BACKEND_PORT:-8000}" \
 the first account"*: fill in email, password, and paste the token into
 **Bootstrap token**.
 
+The password must be at least 12 characters and must not be the email address
+(#1285). `codeframe auth set-password` applies the same rule. To change the
+password or email later through `PATCH /users/me`, send the current one in
+`current_password`. A token alone is not enough.
+
 After the account exists, remove `CODEFRAME_BOOTSTRAP_TOKEN` from the
 environment if you like — the route is closed either way, and a token left in
 place has no further use.

tests/auth/test_password_policy_1285.py

@@ -0,0 +1,196 @@
+"""Password policy on the sole superuser (#1285).
+
+The bootstrap account holds admin scope and terminal access, yet it could be
+registered with "a", emptied via ``PATCH /users/me {"password": ""}``, and its
+password or email changed with nothing but a (24h, unrevocable) JWT — so a
+leaked token became a permanent takeover.
+"""
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from codeframe.auth import router as auth_router
+from codeframe.auth.manager import reset_auth_engine
+from codeframe.platform_store.database import Database
+
+pytestmark = pytest.mark.v2
+
+EMAIL = "operator@example.com"
+PASSWORD = "correct-horse-battery"
+NEW_PASSWORD = "another-long-passphrase"
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db_path = tmp_path / "state.db"
+    monkeypatch.setenv("DATABASE_PATH", str(db_path))
+    monkeypatch.delenv("CODEFRAME_BOOTSTRAP_TOKEN", raising=False)
+    reset_auth_engine()
+    db = Database(db_path)
+    db.initialize()
+    db.close()
+
+    app = FastAPI()
+    app.include_router(auth_router.router)
+    yield TestClient(app, raise_server_exceptions=False, client=("127.0.0.1", 40000))
+    reset_auth_engine()
+
+
+def _register(client, password=PASSWORD, email=EMAIL):
+    return client.post("/auth/register", json={"email": email, "password": password})
+
+
+def _login(client, password=PASSWORD, email=EMAIL):
+    return client.post("/auth/jwt/login", data={"username": email, "password": password})
+
+
+@pytest.fixture
+def session(client):
+    """A registered bootstrap superuser: (client, auth headers, user id)."""
+    reg = _register(client)
+    assert reg.status_code == 201, reg.text
+    token = _login(client).json()["access_token"]
+    return client, {"Authorization": f"Bearer {token}"}, reg.json()["id"]
+
+
+class TestRegistration:
+    @pytest.mark.parametrize("password", ["a", "", "elevenchars"])
+    def test_a_short_password_is_refused(self, client, password):
+        resp = _register(client, password=password)
+        assert resp.status_code == 400, resp.text
+        assert resp.json()["detail"]["code"] == "REGISTER_INVALID_PASSWORD"
+
+    def test_the_email_as_password_is_refused(self, client):
+        resp = _register(client, password=EMAIL.upper())
+        assert resp.status_code == 400, resp.text
+
+    def test_twelve_characters_is_enough(self, client):
+        resp = _register(client, password="twelve-chars")
+        assert resp.status_code == 201, resp.text
+        assert resp.json()["is_superuser"] is True
+
+
+def _routes(user_id):
+    return ["/users/me", f"/users/{user_id}"]
+
+
+class TestPatchPasswordPolicy:
+    @pytest.mark.parametrize("route_index", [0, 1], ids=["me", "by-id"])
+    @pytest.mark.parametrize("password", ["", "short"])
+    def test_a_short_password_is_refused(self, session, route_index, password):
+        client, headers, uid = session
+        resp = client.patch(
+            _routes(uid)[route_index],
+            json={"password": password, "current_password": PASSWORD},
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert _login(client).status_code == 200, "the old password must still work"
+
+    @pytest.mark.parametrize("route_index", [0, 1], ids=["me", "by-id"])
+    def test_a_password_change_needs_the_current_password(self, session, route_index):
+        client, headers, uid = session
+        resp = client.patch(
+            _routes(uid)[route_index], json={"password": NEW_PASSWORD}, headers=headers
+        )
+        assert resp.status_code == 400, resp.text
+        assert _login(client, NEW_PASSWORD).status_code == 400
+
+    @pytest.mark.parametrize("route_index", [0, 1], ids=["me", "by-id"])
+    def test_a_wrong_current_password_is_refused(self, session, route_index):
+        client, headers, uid = session
+        resp = client.patch(
+            _routes(uid)[route_index],
+            json={"password": NEW_PASSWORD, "current_password": "not-the-password"},
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert _login(client, NEW_PASSWORD).status_code == 400
+
+    def test_the_right_current_password_changes_it(self, session):
+        client, headers, _ = session
+        resp = client.patch(
+            "/users/me",
+            json={"password": NEW_PASSWORD, "current_password": PASSWORD},
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        assert "current_password" not in resp.json()
+        assert _login(client, NEW_PASSWORD).status_code == 200
+        assert _login(client).status_code == 400
+
+
+    def test_an_explicit_null_password_changes_nothing(self, session):
+        """``null`` is "no change" in fastapi-users, not a way past the check —
+        and must not 500 on ``len(None)`` (claude-review)."""
+        client, headers, _ = session
+        resp = client.patch("/users/me", json={"password": None}, headers=headers)
+        assert resp.status_code == 200, resp.text
+        assert _login(client).status_code == 200
+
+
+class TestPatchEmail:
+    @pytest.mark.parametrize("route_index", [0, 1], ids=["me", "by-id"])
+    def test_an_email_change_needs_the_current_password(self, session, route_index):
+        client, headers, uid = session
+        resp = client.patch(
+            _routes(uid)[route_index], json={"email": "thief@example.com"}, headers=headers
+        )
+        assert resp.status_code == 400, resp.text
+        assert _login(client).status_code == 200
+
+    def test_the_right_current_password_changes_it(self, session):
+        client, headers, _ = session
+        resp = client.patch(
+            "/users/me",
+            json={"email": "moved@example.com", "current_password": PASSWORD},
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        assert _login(client, email="moved@example.com").status_code == 200
+
+    def test_the_new_password_cannot_be_the_new_email(self, session):
+        """fastapi-users validates against the *old* row, so setting both to the
+        same value in one PATCH slipped past the email rule (codex review)."""
+        client, headers, _ = session
+        resp = client.patch(
+            "/users/me",
+            json={
+                "email": "brand-new@example.com",
+                "password": "brand-new@example.com",
+                "current_password": PASSWORD,
+            },
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert _login(client).status_code == 200
+
+    def test_resending_the_same_email_is_not_a_change(self, session):
+        client, headers, _ = session
+        resp = client.patch("/users/me", json={"email": EMAIL}, headers=headers)
+        assert resp.status_code == 200, resp.text
+
+    def test_other_fields_need_no_current_password(self, session):
+        client, headers, _ = session
+        resp = client.patch("/users/me", json={"name": "Operator"}, headers=headers)
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["name"] == "Operator"
+
+
+class TestOfflineSetPassword:
+    """``cf auth set-password`` writes the hash directly — the same policy applies."""
+
+    def test_a_short_password_is_refused(self, monkeypatch):
+        from typer.testing import CliRunner
+
+        from codeframe.cli import auth_commands
+
+        monkeypatch.setattr(
+            auth_commands, "get_db_for_cli", lambda: pytest.fail("must refuse before the DB")
+        )
+        result = CliRunner().invoke(
+            auth_commands.auth_app, ["set-password", EMAIL, "--password", "short"]
+        )
+        assert result.exit_code == 1
+        assert "12" in result.output

tests/auth/test_registration_bootstrap.py

@@ -70,7 +70,7 @@ def _register(client, email="first@example.com", token=None, headers=None):
         request_headers["X-Bootstrap-Token"] = token
     return client.post(
         "/auth/register",
-        json={"email": email, "password": "secret123"},
+        json={"email": email, "password": "secret123-long-enough"},
         headers=request_headers,
     )
 
@@ -121,7 +121,7 @@ def test_concurrent_first_registrations_admit_exactly_one(self, auth_client):
         async def _register_async(client, email):
             resp = await client.post(
                 "/auth/register",
-                json={"email": email, "password": "secret123"},
+                json={"email": email, "password": "secret123-long-enough"},
             )
             statuses.append(resp.status_code)
 
@@ -221,7 +221,7 @@ def test_configured_token_not_bypassed_by_loopback(self, auth_client, monkeypatc
         request must not sidestep it."""
         monkeypatch.setenv("CODEFRAME_BOOTSTRAP_TOKEN", BOOTSTRAP_TOKEN)
         assert auth_client.post(
-            "/auth/register", json={"email": "a@example.com", "password": "secret123"}
+            "/auth/register", json={"email": "a@example.com", "password": "secret123-long-enough"}
         ).status_code == 403
 
     def test_second_registration_forbidden_even_with_correct_token(
@@ -308,7 +308,7 @@ def test_repeated_forwarded_for_headers_all_inspected(self, auth_client):
         address sitting in the second header."""
         resp = auth_client.post(
             "/auth/register",
-            json={"email": "a@example.com", "password": "secret123"},
+            json={"email": "a@example.com", "password": "secret123-long-enough"},
             headers=[
                 ("X-Forwarded-For", "127.0.0.1"),
                 ("X-Forwarded-For", "203.0.113.5"),

tests/ui/test_v2_auth_enforcement.py

@@ -231,6 +231,6 @@ def test_register_not_401(self, auth_app):
         client = TestClient(auth_app, raise_server_exceptions=False)
         resp = client.post(
             "/auth/register",
-            json={"email": "x@example.com", "password": "secret123"},
+            json={"email": "x@example.com", "password": "secret123-long-enough"},
         )
         assert resp.status_code != 401