security(07.3-02): add _rotate_tokens_on_algorithm_change lifespan hook + promote tests

- backend/main.py: add import logging, select; add _rotate_tokens_on_algorithm_change helper above lifespan; call from lifespan with try/except wrap
- backend/tests/test_auth_es256.py: promote ES256-04 and ES256-05 startup rotation tests from xfail to passing; suite reports 6 PASSED + 3 XFAILED

backend/main.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import time
 import uuid
 from contextlib import asynccontextmanager
@@ -12,7 +13,7 @@ from minio import Minio
 from slowapi import _rate_limit_exceeded_handler
 from slowapi.errors import RateLimitExceeded
 from slowapi.middleware import SlowAPIMiddleware
-from sqlalchemy import text
+from sqlalchemy import select, text
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.responses import Response as StarletteResponse
 from starlette.types import ASGIApp, Receive, Scope, Send
@@ -130,6 +131,49 @@ class CorrelationIDMiddleware:
         )
 
 
+# ── ES256 startup rotation ────────────────────────────────────────────────────
+
+async def _rotate_tokens_on_algorithm_change(session) -> None:
+    """Idempotent ES256 migration (Phase 7.3 D-04/D-05).
+
+    On every boot, compares the jwt_algorithm marker row in system_settings against
+    'ES256'. If absent or different, bulk-revokes all active refresh tokens via raw
+    SQL UPDATE and upserts the marker (with is_active=False so the AI provider loader
+    never returns this row). No-op on second boot.
+    """
+    from db.models import SystemSettings  # local import to avoid circular deps
+
+    stmt = select(SystemSettings).where(SystemSettings.provider_id == "jwt_algorithm")
+    result = await session.execute(stmt)
+    row = result.scalar_one_or_none()
+
+    if row is not None and row.model_name == "ES256":
+        return  # Already migrated — idempotent no-op
+
+    # Bulk-revoke all active refresh tokens (single raw SQL — no Python iteration)
+    await session.execute(
+        text("UPDATE refresh_tokens SET revoked = true WHERE revoked = false")
+    )
+    logging.getLogger(__name__).info(
+        "ES256 startup rotation: bulk-revoked all active refresh tokens"
+    )
+
+    if row is None:
+        session.add(SystemSettings(
+            id=uuid.uuid4(),
+            provider_id="jwt_algorithm",
+            model_name="ES256",
+            context_chars=0,
+            is_active=False,
+            api_key_enc=None,
+            base_url=None,
+        ))
+    else:
+        row.model_name = "ES256"
+
+    await session.commit()
+
+
 # ── Lifespan ──────────────────────────────────────────────────────────────────
 
 @asynccontextmanager
@@ -179,6 +223,18 @@ async def lifespan(app: FastAPI):
             "AI provider seed skipped (table may not exist yet): %s", _seed_exc
         )
 
+    # ES256 startup rotation (Phase 7.3 — D-04):
+    # If jwt_algorithm in system_settings differs from "ES256" (or is absent),
+    # bulk-revoke all refresh tokens and record the new algorithm.
+    # Wrapped in try/except so a missing table before migrations doesn't crash.
+    try:
+        async with AsyncSessionLocal() as session:
+            await _rotate_tokens_on_algorithm_change(session)
+    except Exception as _es256_exc:
+        logging.getLogger(__name__).warning(
+            "ES256 rotation check skipped (table may not exist yet): %s", _es256_exc
+        )
+
     yield
 
     # Shutdown: close pooled connections and Redis

backend/tests/test_auth_es256.py

@@ -79,18 +79,143 @@ def test_reset_token_uses_es256():
     assert decode_password_reset_token(token) == "u1"
 
 
-# ── Startup token rotation stubs ──────────────────────────────────────────────
+# ── Startup token rotation tests ──────────────────────────────────────────────
 
-@pytest.mark.xfail(strict=True, reason="ES256-04: not yet implemented")
 @pytest.mark.asyncio
 async def test_startup_rotation_revokes_tokens(db_session):
-    pytest.xfail("not yet implemented")
+    """ES256-04: on first run, all active refresh tokens are bulk-revoked and jwt_algorithm row is upserted."""
+    import secrets as _secrets
+    from main import _rotate_tokens_on_algorithm_change
+    from db.models import RefreshToken, SystemSettings, User
+    from sqlalchemy import select
+
+    # Create a minimal user row (required by RefreshToken FK)
+    user_id = uuid.uuid4()
+    user = User(
+        id=user_id,
+        handle=f"testrotate_{user_id.hex[:8]}",
+        email=f"testrotate_{user_id.hex[:8]}@example.com",
+        password_hash="fakehash",
+        role="user",
+        is_active=True,
+        password_must_change=False,
+    )
+    db_session.add(user)
+    await db_session.flush()
+
+    # Insert two active refresh tokens
+    now = datetime.now(timezone.utc)
+    tok1_id = uuid.uuid4()
+    tok2_id = uuid.uuid4()
+    tok1 = RefreshToken(
+        id=tok1_id,
+        user_id=user_id,
+        token_hash=hashlib.sha256(_secrets.token_urlsafe(32).encode()).hexdigest(),
+        expires_at=now + timedelta(days=1),
+        revoked=False,
+    )
+    tok2 = RefreshToken(
+        id=tok2_id,
+        user_id=user_id,
+        token_hash=hashlib.sha256(_secrets.token_urlsafe(32).encode()).hexdigest(),
+        expires_at=now + timedelta(days=1),
+        revoked=False,
+    )
+    db_session.add(tok1)
+    db_session.add(tok2)
+    await db_session.flush()
+
+    # Act: no jwt_algorithm row exists yet
+    await _rotate_tokens_on_algorithm_change(db_session)
+    # expire_all() required: raw SQL UPDATE bypasses ORM identity map; expire_on_commit=False
+    # (set in conftest.py) means SQLAlchemy won't auto-reload stale objects after commit.
+    db_session.expire_all()
+
+    # Assert: both tokens are now revoked
+    result = await db_session.execute(
+        select(RefreshToken).where(RefreshToken.id.in_([tok1_id, tok2_id]))
+    )
+    rows = result.scalars().all()
+    assert len(rows) == 2
+    assert all(r.revoked is True for r in rows)
+
+    # Assert: jwt_algorithm row was created correctly
+    ss_result = await db_session.execute(
+        select(SystemSettings).where(SystemSettings.provider_id == "jwt_algorithm")
+    )
+    ss_row = ss_result.scalar_one_or_none()
+    assert ss_row is not None
+    assert ss_row.model_name == "ES256"
+    assert ss_row.is_active is False
+    assert ss_row.context_chars == 0
 
 
-@pytest.mark.xfail(strict=True, reason="ES256-05: not yet implemented")
 @pytest.mark.asyncio
 async def test_startup_rotation_idempotent(db_session):
-    pytest.xfail("not yet implemented")
+    """ES256-05: when jwt_algorithm row already has model_name=ES256, no tokens are revoked."""
+    import secrets as _secrets
+    from main import _rotate_tokens_on_algorithm_change
+    from db.models import RefreshToken, SystemSettings, User
+    from sqlalchemy import select, func
+
+    # Create a user
+    user_id = uuid.uuid4()
+    user = User(
+        id=user_id,
+        handle=f"testidempotent_{user_id.hex[:8]}",
+        email=f"testidempotent_{user_id.hex[:8]}@example.com",
+        password_hash="fakehash",
+        role="user",
+        is_active=True,
+        password_must_change=False,
+    )
+    db_session.add(user)
+    await db_session.flush()
+
+    # Seed the jwt_algorithm row as already migrated
+    existing_marker = SystemSettings(
+        id=uuid.uuid4(),
+        provider_id="jwt_algorithm",
+        model_name="ES256",
+        context_chars=0,
+        is_active=False,
+        api_key_enc=None,
+        base_url=None,
+    )
+    db_session.add(existing_marker)
+
+    # Insert one fresh active token
+    now = datetime.now(timezone.utc)
+    tok_id = uuid.uuid4()
+    tok = RefreshToken(
+        id=tok_id,
+        user_id=user_id,
+        token_hash=hashlib.sha256(_secrets.token_urlsafe(32).encode()).hexdigest(),
+        expires_at=now + timedelta(days=1),
+        revoked=False,
+    )
+    db_session.add(tok)
+    await db_session.flush()
+
+    # Act
+    await _rotate_tokens_on_algorithm_change(db_session)
+    # expire_all() to clear identity map cache (same reason as test_startup_rotation_revokes_tokens)
+    db_session.expire_all()
+
+    # Assert: the fresh token was NOT revoked (bulk update did not fire)
+    result = await db_session.execute(
+        select(RefreshToken).where(RefreshToken.id == tok_id)
+    )
+    fresh_tok = result.scalar_one_or_none()
+    assert fresh_tok is not None
+    assert fresh_tok.revoked is False
+
+    # Assert: still exactly one jwt_algorithm row (no duplicate upsert)
+    count_result = await db_session.execute(
+        select(func.count()).where(SystemSettings.provider_id == "jwt_algorithm")
+    )
+    count = count_result.scalar_one()
+    assert count == 1
 
 
 # ── Remember-me TTL stubs ─────────────────────────────────────────────────────