feat(07.1): session revocation on privilege change — CR-01/CR-02/CR-03
- revoke_all_refresh_tokens: add skip_token_hash optional param (exclude
current session while revoking others)
- change_password, enable_totp, disable_totp: call revoke with skip hash
derived from refresh cookie; return sessions_revoked in response and
write to audit log metadata_
- 3 new tests: test_{change_password,enable_totp,disable_totp}_revokes_other_sessions
— all PASSED; 373 total passing, 0 regressions
- Frontend toasts: SettingsAccountTab + TotpEnrollment show
"Other sessions have been terminated." when sessions_revoked > 0
- Companion fixes: rate_limiting get_client_ip refactor, deps/auth.py
request.state.current_user, locustfile refresh-token task removal
- Version bump: 0.1.0 → 0.1.1
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
8d060a5da4
commit
c38c6b1c01
@@ -23,7 +23,7 @@ from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from db.models import BackupCode, Quota, User
|
||||
from db.models import BackupCode, Quota, RefreshToken, User
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
@@ -496,3 +496,107 @@ async def test_patch_preferences_requires_auth(async_client):
|
||||
json={"pdf_open_mode": "in_app"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
# ── Tests — sessions_revoked (CR-01, CR-02, CR-03) ───────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_change_password_revokes_other_sessions(authed_client, db_session: AsyncSession):
|
||||
"""change_password revokes other sessions and returns sessions_revoked >= 1."""
|
||||
from services import auth as auth_service
|
||||
|
||||
await _register(authed_client, handle="cpr1", email="cpr1@example.com")
|
||||
login_resp = await _login(authed_client, email="cpr1@example.com")
|
||||
token = login_resp.json()["access_token"]
|
||||
|
||||
result = await db_session.execute(select(User).where(User.email == "cpr1@example.com"))
|
||||
user = result.scalar_one()
|
||||
|
||||
# Insert a second session token (the "other device") directly in the DB
|
||||
await auth_service.create_refresh_token(db_session, user.id)
|
||||
|
||||
with patch("services.auth.check_hibp", return_value=False):
|
||||
resp = await authed_client.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": "ValidPass12!", "new_password": "NewStrong99!@"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["sessions_revoked"] >= 1
|
||||
|
||||
result2 = await db_session.execute(
|
||||
select(RefreshToken).where(RefreshToken.user_id == user.id)
|
||||
)
|
||||
rows = result2.scalars().all()
|
||||
assert any(r.revoked for r in rows), "Expected at least one revoked RefreshToken row"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_enable_totp_revokes_other_sessions(authed_client, db_session: AsyncSession):
|
||||
"""enable_totp revokes other sessions and returns sessions_revoked >= 1."""
|
||||
from services import auth as auth_service
|
||||
|
||||
await _register(authed_client, handle="etr1", email="etr1@example.com")
|
||||
login_resp = await _login(authed_client, email="etr1@example.com")
|
||||
token = login_resp.json()["access_token"]
|
||||
|
||||
result = await db_session.execute(select(User).where(User.email == "etr1@example.com"))
|
||||
user = result.scalar_one()
|
||||
user.totp_secret = "JBSWY3DPEHPK3PXP"
|
||||
await db_session.commit()
|
||||
|
||||
# Insert a second session token (the "other device")
|
||||
await auth_service.create_refresh_token(db_session, user.id)
|
||||
|
||||
with patch("services.auth.verify_totp", return_value=True):
|
||||
with patch("services.auth.store_backup_codes", return_value=None):
|
||||
resp = await authed_client.post(
|
||||
"/api/auth/totp/enable",
|
||||
json={"code": "123456"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["sessions_revoked"] >= 1
|
||||
|
||||
result2 = await db_session.execute(
|
||||
select(RefreshToken).where(RefreshToken.user_id == user.id)
|
||||
)
|
||||
rows = result2.scalars().all()
|
||||
assert any(r.revoked for r in rows), "Expected at least one revoked RefreshToken row"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disable_totp_revokes_other_sessions(authed_client, db_session: AsyncSession):
|
||||
"""disable_totp revokes other sessions and returns sessions_revoked >= 1."""
|
||||
from services import auth as auth_service
|
||||
|
||||
await _register(authed_client, handle="dtr1", email="dtr1@example.com")
|
||||
login_resp = await _login(authed_client, email="dtr1@example.com")
|
||||
token = login_resp.json()["access_token"]
|
||||
|
||||
result = await db_session.execute(select(User).where(User.email == "dtr1@example.com"))
|
||||
user = result.scalar_one()
|
||||
user.totp_enabled = True
|
||||
user.totp_secret = "JBSWY3DPEHPK3PXP"
|
||||
await db_session.commit()
|
||||
|
||||
# Insert a second session token (the "other device")
|
||||
await auth_service.create_refresh_token(db_session, user.id)
|
||||
|
||||
resp = await authed_client.delete(
|
||||
"/api/auth/totp",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["sessions_revoked"] >= 1
|
||||
|
||||
result2 = await db_session.execute(
|
||||
select(RefreshToken).where(RefreshToken.user_id == user.id)
|
||||
)
|
||||
rows = result2.scalars().all()
|
||||
assert any(r.revoked for r in rows), "Expected at least one revoked RefreshToken row"
|
||||
|
||||
Reference in New Issue
Block a user