feat(api): add POST /accounts/me/security/revoke-sessions

Sixth commit in the session-expiration-policy series. The kill-all- sessions endpoint folded into scope after the §4.11 design pass. - POST /accounts/me/security/revoke-sessions, owner-only. - Body: {"scope": "all" | "others"}. Default "all" includes the caller's own refresh token. "others" preserves the caller's sessions so an owner can sign everyone else out without logging themselves out. - Single SQL UPDATE through users.account_id -> refresh_tokens, with revoked_at IS NULL preserved as the gate so already-revoked rows don't get double-stamped (the idempotency property). - Caller's access token is not touched — it dies on its 5-minute timer. Frontend handles "scope=all" UX by clearing localStorage and redirecting after the response (commit 8). - Affected users' next /auth/refresh hits the existing atomic-revoke zero-rows path -> invalid_refresh_token (plain logout, no banner). - Writes one account.sessions_revoked_bulk audit event with {scope, revoked_count}. Tests added in test_session_policy.py (6 cases): - #17 scope=all kills caller's own session; their refresh -> 401 invalid_refresh_token. - #18 scope=others preserves caller's session; their refresh succeeds, member's refresh -> 401 invalid_refresh_token. - #19 account-scoped: test_admin in a different account is unaffected when test_user's owner runs revoke-all (revoked_count=1, not 2). - #20 engineer-role member -> 403. - #21 emits exactly one audit row with the expected payload. - #22 idempotent: second immediate POST returns revoked_count=0. 22/22 in test_session_policy. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-13 16:31:10 -04:00
parent 8cfaef6a9d
commit cabd745a2b
2 changed files with 327 additions and 1 deletions
--- a/backend/tests/test_session_policy.py
+++ b/backend/tests/test_session_policy.py
@@ -8,6 +8,7 @@ This file grows across commits:
 - Commit 3: claims embedded at login + response fields surfaced (#1, #14)
 - Commit 4: absolute-cap enforcement + grandfather path (#8, #9, #12)
 - Commit 5: GET/PATCH /accounts/me/security (#2, #3, #4, #5, #7, #16)
+- Commit 6: POST /accounts/me/security/revoke-sessions (#17-#22)
 """

 import uuid
@@ -378,6 +379,280 @@ class TestSessionPolicyEndpoint:
        assert entry.details["effective_old"]["absolute_minutes"] == 20160


+async def _seed_extra_account_user(
+    test_db, *, email: str, account_id, password_hash: str, role: str = "engineer"
+):
+    """Add a second user under an existing account for revoke-scope tests."""
+    from app.models.user import User
+
+    user = User(
+        email=email,
+        password_hash=password_hash,
+        name=email,
+        role="engineer",
+        is_super_admin=False,
+        is_active=True,
+        account_id=account_id,
+        account_role=role,
+        email_verified_at=datetime.now(timezone.utc),
+    )
+    test_db.add(user)
+    await test_db.commit()
+    return user
+
+
+class TestBulkRevoke:
+    """§6 tests #17-#22 — POST /accounts/me/security/revoke-sessions."""
+
+    @pytest.mark.asyncio
+    async def test_revoke_all_kills_callers_own_session(
+        self, client: AsyncClient, test_user: dict, test_db
+    ):
+        """§6 test #17 — scope=all includes the caller's own token. After
+        the response, the caller's refresh_token gets invalid_refresh_token
+        on next /auth/refresh.
+        """
+        from app.models.user import User
+        from sqlalchemy import select
+
+        owner = (
+            await test_db.execute(
+                select(User).where(User.email == test_user["email"])
+            )
+        ).scalar_one()
+        await _seed_extra_account_user(
+            test_db,
+            email="member-revoke-all@example.com",
+            account_id=owner.account_id,
+            password_hash=owner.password_hash,
+        )
+
+        # Owner logs in (also seeds owner's refresh-token row).
+        owner_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_user["email"], "password": test_user["password"]},
+        )
+        owner_refresh = owner_login.json()["refresh_token"]
+        owner_access = owner_login.json()["access_token"]
+
+        # Member also logs in so there's another active refresh-token row.
+        member_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={
+                "email": "member-revoke-all@example.com",
+                "password": test_user["password"],
+            },
+        )
+        assert member_login.status_code == 200
+
+        response = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "all"},
+        )
+        assert response.status_code == 200, response.json()
+        assert response.json()["revoked_count"] == 2
+
+        # Owner's own refresh now returns invalid_refresh_token.
+        retry = await client.post(
+            "/api/v1/auth/refresh",
+            headers={"Authorization": f"Bearer {owner_refresh}"},
+        )
+        assert retry.status_code == 401
+        assert retry.json()["detail"] == "invalid_refresh_token"
+
+    @pytest.mark.asyncio
+    async def test_revoke_others_preserves_callers_session(
+        self, client: AsyncClient, test_user: dict, test_db
+    ):
+        """§6 test #18 — scope=others excludes the caller's user_id from
+        the bulk update. Caller can still refresh; other users cannot.
+        """
+        from app.models.user import User
+        from sqlalchemy import select
+
+        owner = (
+            await test_db.execute(
+                select(User).where(User.email == test_user["email"])
+            )
+        ).scalar_one()
+        await _seed_extra_account_user(
+            test_db,
+            email="member-revoke-others@example.com",
+            account_id=owner.account_id,
+            password_hash=owner.password_hash,
+        )
+
+        owner_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_user["email"], "password": test_user["password"]},
+        )
+        owner_refresh = owner_login.json()["refresh_token"]
+        owner_access = owner_login.json()["access_token"]
+
+        member_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={
+                "email": "member-revoke-others@example.com",
+                "password": test_user["password"],
+            },
+        )
+        member_refresh = member_login.json()["refresh_token"]
+
+        response = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "others"},
+        )
+        assert response.status_code == 200
+        assert response.json()["revoked_count"] == 1
+
+        # Owner's refresh still works.
+        owner_retry = await client.post(
+            "/api/v1/auth/refresh",
+            headers={"Authorization": f"Bearer {owner_refresh}"},
+        )
+        assert owner_retry.status_code == 200
+
+        # Member's refresh is dead.
+        member_retry = await client.post(
+            "/api/v1/auth/refresh",
+            headers={"Authorization": f"Bearer {member_refresh}"},
+        )
+        assert member_retry.status_code == 401
+        assert member_retry.json()["detail"] == "invalid_refresh_token"
+
+    @pytest.mark.asyncio
+    async def test_revoke_is_account_scoped(
+        self, client: AsyncClient, test_user: dict, test_admin: dict
+    ):
+        """§6 test #19 — owner of account A cannot revoke tokens in account B.
+
+        test_admin lives in its own account. After test_user's owner runs
+        revoke-all, test_admin's session continues to work.
+        """
+        owner_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_user["email"], "password": test_user["password"]},
+        )
+        owner_access = owner_login.json()["access_token"]
+
+        admin_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_admin["email"], "password": test_admin["password"]},
+        )
+        admin_refresh = admin_login.json()["refresh_token"]
+
+        response = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "all"},
+        )
+        assert response.status_code == 200
+        # Only test_user's own session is revoked.
+        assert response.json()["revoked_count"] == 1
+
+        admin_retry = await client.post(
+            "/api/v1/auth/refresh",
+            headers={"Authorization": f"Bearer {admin_refresh}"},
+        )
+        assert admin_retry.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_revoke_engineer_forbidden(
+        self, client: AsyncClient, test_user: dict, test_db
+    ):
+        """§6 test #20 — engineer-role member gets 403."""
+        from app.models.user import User
+        from sqlalchemy import select
+
+        owner = (
+            await test_db.execute(
+                select(User).where(User.email == test_user["email"])
+            )
+        ).scalar_one()
+        await _seed_extra_account_user(
+            test_db,
+            email="engineer-revoke@example.com",
+            account_id=owner.account_id,
+            password_hash=owner.password_hash,
+        )
+
+        engineer_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={
+                "email": "engineer-revoke@example.com",
+                "password": test_user["password"],
+            },
+        )
+        engineer_access = engineer_login.json()["access_token"]
+
+        response = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {engineer_access}"},
+            json={"scope": "all"},
+        )
+        assert response.status_code == 403
+
+    @pytest.mark.asyncio
+    async def test_revoke_writes_audit_row(
+        self, client: AsyncClient, test_user: dict, test_db
+    ):
+        """§6 test #21 — emits one account.sessions_revoked_bulk event."""
+        from app.models.audit_log import AuditLog
+        from sqlalchemy import select
+
+        owner_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_user["email"], "password": test_user["password"]},
+        )
+        owner_access = owner_login.json()["access_token"]
+
+        response = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "all"},
+        )
+        assert response.status_code == 200
+
+        result = await test_db.execute(
+            select(AuditLog).where(AuditLog.action == "account.sessions_revoked_bulk")
+        )
+        rows = result.scalars().all()
+        assert len(rows) == 1
+        entry = rows[0]
+        assert entry.details["scope"] == "all"
+        assert entry.details["revoked_count"] == 1
+
+    @pytest.mark.asyncio
+    async def test_revoke_is_idempotent(
+        self, client: AsyncClient, test_user: dict
+    ):
+        """§6 test #22 — second immediate POST returns revoked_count=0
+        (no already-revoked rows get double-stamped or counted again).
+        """
+        owner_login = await client.post(
+            "/api/v1/auth/login/json",
+            json={"email": test_user["email"], "password": test_user["password"]},
+        )
+        owner_access = owner_login.json()["access_token"]
+
+        first = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "others"},  # owner's own session preserved
+        )
+        assert first.status_code == 200
+
+        second = await client.post(
+            "/api/v1/accounts/me/security/revoke-sessions",
+            headers={"Authorization": f"Bearer {owner_access}"},
+            json={"scope": "others"},
+        )
+        assert second.status_code == 200
+        assert second.json()["revoked_count"] == 0
+
+
 class TestAbsoluteCap:
    """§6 tests #8, #9, #12 — absolute-cap enforcement and grandfather path."""