fix(security): F03 (Astra P1) — Sitzungswiderruf in beiden Session-Stores durchsetzen

Vorher (Astra-Finding): Widerruf war inkonsistent ueber vier Pfade:
- Deaktivierung invalidierte nur den Berechtigungscache — Sessions
  liefen bis TTL (8h) weiter
- Loeschung invalidierte GAR NICHTS
- Passwortwechsel loeschte nur Redis-Sessions — PostgreSQL-Fallback-
  Sessions ueberlebten jeden Redis-Ausfall
- Fehlende UserTenant-Mitgliedschaft wurde durchgewinkt statt
  abgewiesen

Fix:
- Neuer zentraler Helfer revoke_user_sessions_all_stores (app/core/auth.py):
  Redis-Sessions loeschen UND PostgreSQL-Fallback-Sessions per
  expires_at=now() ablaufen lassen (Audit-Trail bleibt, Zugriff stirbt
  sofort — der DB-Fallback-Pfad prueft expires_at bereits)
- Alle 4 Widerrufsstellen verdrahtet: Deaktivierung + Loeschung
  (routes/users.py), Passwortwechsel (user_service.py), Passwort-Reset
  (auth_service.py)
- Membership-Check in get_current_user fail-closed: None (fehlende
  Mitgliedschaft) wird abgewiesen statt durchgelassen

Abnahme (Astra): Deaktivierung, Austritt und Passwortwechsel wirken
unmittelbar — auch bei Redis-Ausfall (Unit-Test beweist die
DB-Fallback-Abgelaufen-Rejection).

Tests: test_s1_security_guards.py 10/10 (3 neue F03-Tests) +
test_auth.py 11/11 + ruff clean.

app/core/auth.py

@@ -117,6 +117,55 @@ async def revoke_user_redis_sessions(user_id: str | uuid.UUID) -> int:
         return 0
 
 
+async def revoke_user_sessions_all_stores(user_id: str | uuid.UUID) -> None:
+    """F03 (Astra): revoke ALL sessions for a user in BOTH session stores.
+
+    Deactivation, deletion and password changes must take effect immediately —
+    including when Redis is down and requests fall back to the PostgreSQL
+    sessions table.
+
+    1. Redis runtime sessions are deleted (revoke_user_redis_sessions).
+    2. PostgreSQL session records are EXPIRED by setting ``expires_at = now()``
+       (not deleted — they stay as audit trail). The DB fallback path in
+       ``get_session_data`` rejects sessions whose ``expires_at`` is past.
+
+    Never raises — best-effort per store, but errors are logged loudly.
+    """
+    # 1. Redis runtime sessions
+    await revoke_user_redis_sessions(user_id)
+
+    # 2. PostgreSQL fallback sessions — expire instead of delete (audit trail)
+    try:
+        from datetime import UTC, datetime
+
+        from sqlalchemy import update
+
+        from app.core.db import get_session_factory
+        from app.models.session import Session as SessionModel
+
+        uid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id))
+        factory = get_session_factory()
+        async with factory() as db:
+            result = await db.execute(
+                update(SessionModel)
+                .where(
+                    SessionModel.user_id == uid,
+                    SessionModel.expires_at > datetime.now(UTC),
+                )
+                .values(expires_at=datetime.now(UTC))
+            )
+            await db.commit()
+            if result.rowcount:
+                logger.info(
+                    "F03: expired %d PostgreSQL fallback sessions for user %s",
+                    result.rowcount, uid,
+                )
+    except Exception:
+        logger.warning(
+            "F03: failed to expire PostgreSQL sessions for user %s", user_id, exc_info=True
+        )
+
+
 def hash_token(token: str) -> str:
     """SHA-256 hash a token for storage."""
     return hashlib.sha256(token.encode()).hexdigest()

app/deps.py

@@ -130,10 +130,20 @@ async def get_current_user(
     membership_row = membership_q.first()
     membership_status = membership_row[0] if membership_row else None
     role_id = membership_row[1] if membership_row else None
-    if membership_status is not None and membership_status != "active":
+    # F03 (Astra): a MISSING tenant membership must be rejected, not waved
+    # through. Previously `is not None` let membership-less sessions access
+    # the tenant's data via the RLS context set above.
+    if membership_status is None or membership_status != "active":
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
-            detail={"detail": f"Mitgliedschaft ist {membership_status}, Zugriff verweigert", "code": "membership_suspended"},
+            detail={
+                "detail": (
+                    f"Mitgliedschaft ist {membership_status}, Zugriff verweigert"
+                    if membership_status
+                    else "Keine aktive Mandanten-Mitgliedschaft, Zugriff verweigert"
+                ),
+                "code": "membership_suspended",
+            },
         )
 
     # Cache user principals for this request — avoids N+1 queries in visibility.py

app/routes/users.py

@@ -339,6 +339,15 @@ async def update_user(
         except Exception:
             pass  # Best-effort — don't fail the update if Redis is down
 
+    # F03 (Astra): deactivation must take effect IMMEDIATELY in BOTH session
+    # stores (Redis runtime + PostgreSQL fallback) — not just the permission
+    # cache. Otherwise a deactivated user keeps working until the session TTL
+    # (8h) expires, and DB-fallback sessions survive Redis outages entirely.
+    if body.is_active is False:
+        from app.core.auth import revoke_user_sessions_all_stores
+
+        await revoke_user_sessions_all_stores(uid)
+
     return {
         "id": str(user.id),
         "email": user.email,
@@ -379,6 +388,12 @@ async def delete_user(
     if not success:
         raise HTTPException(404, detail={"detail": "User not found", "code": "not_found"})
 
+    # F03 (Astra): deletion must revoke all sessions immediately — in BOTH
+    # stores (Redis runtime + PostgreSQL fallback).
+    from app.core.auth import revoke_user_sessions_all_stores
+
+    await revoke_user_sessions_all_stores(uid)
+
     await log_audit(
         db,
         tenant_id,

app/services/auth_service.py

@@ -328,11 +328,12 @@ class AuthService:
         reset_token.used_at = datetime.now(UTC)
         await db.flush()
 
-        # Invalidate all active Redis sessions for this user
-        # (shared helper — same mechanism as the profile/admin change path)
-        from app.core.auth import revoke_user_redis_sessions
+        # Invalidate all active sessions for this user — in BOTH stores
+        # (F03/Astra: PostgreSQL fallback sessions must not survive a Redis
+        # outage after a password change; shared helper for all password paths)
+        from app.core.auth import revoke_user_sessions_all_stores
 
-        await revoke_user_redis_sessions(user.id)
+        await revoke_user_sessions_all_stores(user.id)
 
         # Audit log entry for password reset — use separate API session (crm_api)
         # to avoid requiring audit_log INSERT grants on crm_auth

app/services/user_service.py

@@ -204,10 +204,12 @@ class UserService:
 
         # G2: after a profile/admin password change, kill all other sessions —
         # a stolen or lingering session must not survive the change.
+        # F03 (Astra): revoke in BOTH stores — PostgreSQL fallback sessions
+        # must not survive a Redis outage after a password change.
         if new_password is not None:
-            from app.core.auth import revoke_user_redis_sessions
+            from app.core.auth import revoke_user_sessions_all_stores
 
-            await revoke_user_redis_sessions(user.id)
+            await revoke_user_sessions_all_stores(user.id)
 
         return user, user_tenant
 

tests/test_s1_security_guards.py

@@ -111,3 +111,113 @@ class TestF10TokenScopesAreUpperBound:
         checker = require_permission("mail:write")
         ctx = _user_ctx(token_scopes=["mail:write"], is_system_admin=True)
         assert await checker(current_user=ctx) is ctx
+
+
+@pytest.mark.asyncio
+class TestF03SessionRevocation:
+    """F03 (Astra P1): revocation must work in BOTH session stores.
+
+    Deactivation, deletion and password changes kill Redis runtime sessions
+    AND expire PostgreSQL fallback sessions — a Redis outage must not let a
+    revoked session come back to life.
+    """
+
+    async def test_revoke_helper_expires_db_sessions(self, monkeypatch):
+        """revoke_user_sessions_all_stores expires PG sessions (expires_at=now)
+        instead of deleting them — audit trail stays, access dies."""
+        import json
+        from unittest.mock import AsyncMock, MagicMock
+
+        import app.core.auth as auth_mod
+
+        # Fake Redis with one live session for our user (async iteration —
+        # the real client's scan_iter is an async iterator)
+        user_id = str(uuid.uuid4())
+        redis = MagicMock()
+        redis.scan_iter = MagicMock(return_value=iter(["session:abc"]))
+        redis.get = AsyncMock(return_value=json.dumps({"user_id": user_id}))
+        redis.delete = AsyncMock()
+        monkeypatch.setattr(auth_mod, "get_redis", lambda: redis)
+        # Patch the inner iteration: real code does `async for key in scan_iter`
+
+        class _AsyncIter:
+            def __init__(self, items):
+                self._items = iter(items)
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                try:
+                    return next(self._items)
+                except StopIteration:
+                    raise StopAsyncIteration from None
+
+        redis.scan_iter = MagicMock(return_value=_AsyncIter(["session:abc"]))
+
+        # Fake DB factory recording the update
+        db = MagicMock()
+        db.execute = AsyncMock(return_value=MagicMock(rowcount=2))
+        db.commit = AsyncMock()
+        factory = MagicMock()
+        factory.return_value.__aenter__ = AsyncMock(return_value=db)
+        factory.return_value.__aexit__ = AsyncMock(return_value=False)
+        import app.core.db as db_mod
+
+        monkeypatch.setattr(db_mod, "get_session_factory", lambda: factory)
+
+        await auth_mod.revoke_user_sessions_all_stores(user_id)
+
+        # Redis session deleted
+        redis.delete.assert_awaited_once()
+        # DB update executed (expire) + committed
+        db.execute.assert_awaited_once()
+        db.commit.assert_awaited_once()
+
+    async def test_db_fallback_rejects_expired_session(self, monkeypatch):
+        """get_session_data (DB fallback) must reject sessions whose
+        expires_at is in the past — the revoke helper relies on this."""
+        from datetime import UTC, datetime, timedelta
+        from unittest.mock import AsyncMock, MagicMock
+
+        import app.core.auth as auth_mod
+        from app.models.session import Session as SessionModel
+
+        # Force DB fallback path: circuit says "cannot proceed"
+        circuit = MagicMock()
+        circuit.can_proceed = AsyncMock(return_value=False)
+        monkeypatch.setattr(
+            "app.core.resilience.get_circuit", lambda name: circuit
+        )
+
+        expired_session = MagicMock(spec=SessionModel)
+        expired_session.expires_at = datetime.now(UTC) - timedelta(hours=1)
+        db = MagicMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=lambda: expired_session))
+        factory = MagicMock()
+        factory.return_value.__aenter__ = AsyncMock(return_value=db)
+        factory.return_value.__aexit__ = AsyncMock(return_value=False)
+        import app.core.db as db_mod
+
+        monkeypatch.setattr(db_mod, "get_auth_session_factory", lambda: factory)
+
+        result = await auth_mod.get_session_data(MagicMock(), "some-session-id")
+        assert result is None
+
+    async def test_membership_check_fails_closed(self):
+        """F03: a session whose tenant membership is MISSING must be
+        rejected (403), not waved through.
+
+        Pins the decision rule used by get_current_user (deps.py):
+        reject when `membership_status is None or != 'active'`.
+        """
+        def _rejects(status: str | None) -> bool:
+            return status is None or status != "active"
+
+        # missing membership → reject (was waved through before the fix)
+        assert _rejects(None) is True
+        # suspended/invited/disabled membership → reject (pre-existing)
+        assert _rejects("suspended") is True
+        assert _rejects("invited") is True
+        # active membership → allow
+        assert _rejects("active") is False