From b9524bc9b2eef337da601887558b5bd9953ea943 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:07:05 +0200 Subject: [PATCH 1/7] users: revoke existing sessions when an admin disables, enables or changes a user's roles Closes #581 Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- .../test_users_admin_session_revocation.py | 106 ++++++++++++++++++ modules/users/users/admin/queries.py | 15 +++ modules/users/users/admin/service.py | 3 + modules/users/users/deps.py | 3 +- 4 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 modules/users/tests/test_users_admin_session_revocation.py diff --git a/modules/users/tests/test_users_admin_session_revocation.py b/modules/users/tests/test_users_admin_session_revocation.py new file mode 100644 index 000000000..fc00dde34 --- /dev/null +++ b/modules/users/tests/test_users_admin_session_revocation.py @@ -0,0 +1,106 @@ +"""Admin account changes strand cached browser sessions on every worker.""" + +import pytest +from _users_app_builders import user_row +from simple_module_core.invalidation import InvalidationBus +from users.session_revocation import subscribe +from users.session_version_cache import ( + clear_session_version_cache, + read_session_version, + store_session_version, +) + + +class RecordingTransport: + def __init__(self): + self.messages = [] + + async def publish(self, message): + self.messages.append(message) + + +@pytest.fixture(autouse=True) +def empty_cache(): + clear_session_version_cache() + yield + clear_session_version_cache() + + +@pytest.mark.parametrize("action", ["disable", "enable", "roles"]) +async def test_admin_change_revokes_cached_session( + users_app, anon_client, admin_client, alice, action +): + promotion = await admin_client.put( + f"/api/users/admin/{alice.id}/roles", json={"role_names": ["admin"]} + ) + assert promotion.status_code == 200 + login = await anon_client.post( + "/api/users/auth/login", + data={"username": alice.email, "password": "UserPass1!"}, + ) + assert login.status_code == 204 + user = await user_row(users_app, alice.email) + assert (await anon_client.get("/api/users/admin")).status_code == 200 + assert (await anon_client.get("/api/users/admin")).status_code == 200 + assert read_session_version(user.id) == (True, user.session_version) + + tokens = await anon_client.post( + "/api/users/auth/token", json={"email": alice.email, "password": "UserPass1!"} + ) + assert tokens.status_code == 200 + bearer = {"Authorization": f"Bearer {tokens.json()['access_token']}"} + stale_cookie = anon_client.cookies.get("session") + transport = RecordingTransport() + users_app.state.sm.invalidation.set_transport(transport) + path = f"/api/users/admin/{user.id}" + if action == "roles": + response = await admin_client.put(f"{path}/roles", json={"role_names": ["user"]}) + else: + response = await admin_client.patch(f"{path}/{action}") + assert response.status_code == 200, response.text + updated = await user_row(users_app, user.email) + assert updated.session_version == user.session_version + 1 + assert read_session_version(user.id) == (False, None) + assert (await anon_client.get("/api/users/admin")).status_code == 401 + + assert (await admin_client.get("/api/users/admin", headers=bearer)).status_code == 401 + + # Simulate a remote worker with its own stale counter and bus. + anon_client.cookies.clear() + anon_client.cookies.set("session", stale_cookie) + remote = InvalidationBus() + subscribe(remote) + store_session_version(user.id, user.session_version) + assert transport.messages + for message in transport.messages: + await remote.deliver(message) + assert read_session_version(user.id) == (False, None) + assert (await anon_client.get("/api/users/admin")).status_code == 401 + if action == "roles": + login = await anon_client.post( + "/api/users/auth/login", + data={"username": alice.email, "password": "UserPass1!"}, + ) + assert login.status_code == 204 + assert (await anon_client.get("/api/users/admin")).status_code == 403 + + +async def test_rollback_does_not_publish_revocation(users_app, alice): + from unittest.mock import Mock + + from users.admin.service import UserService + + transport = RecordingTransport() + users_app.state.sm.invalidation.set_transport(transport) + store_session_version(alice.id, alice.session_version) + async with users_app.state.sm.db.session_factory() as session: + service = UserService(session, Mock(), users_app) + changed = await service.disable(alice.id) + assert changed.session_version == alice.session_version + 1 + assert read_session_version(alice.id) == (True, alice.session_version) + assert transport.messages == [] + await session.rollback() + unchanged = await user_row(users_app, alice.email) + assert unchanged.is_active + assert unchanged.session_version == alice.session_version + assert transport.messages == [] diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py index 1dc494244..e285c3509 100644 --- a/modules/users/users/admin/queries.py +++ b/modules/users/users/admin/queries.py @@ -6,6 +6,7 @@ from datetime import UTC, datetime, timedelta from typing import Any, cast +from fastapi import FastAPI from simple_module_db import LIKE_ESCAPE_CHAR, like_contains_pattern from simple_module_db.pagination import PageParams from sqlalchemy import func, or_, select @@ -17,6 +18,8 @@ from users.exceptions import UserNotFoundError from users.manager import UserManager from users.models import Role, User, UserRole +from users.session_revocation import publish_revocation +from users.session_version_cache import forget_session_version def _live_lock(locked_until: datetime | None) -> datetime | None: @@ -35,9 +38,21 @@ def __init__( self, db: AsyncSession, user_manager: UserManager, + app: FastAPI | None = None, ) -> None: self._db = db self._manager = user_manager + self._app = app + + def _revoke_sessions(self, user: User) -> None: + """Persist the counter with the write; evict only after it commits.""" + user.session_version = int(user.session_version or 0) + 1 + user_id = user.id + app = self._app + if app is None: + self._db.on_commit(lambda: forget_session_version(user_id)) + else: + self._db.on_commit(lambda: publish_revocation(app, user_id)) # ── Helpers ───────────────────────────────────────────────── diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index 991caf6a3..2522d2f4a 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -210,6 +210,7 @@ async def disable(self, user_id: uuid.UUID) -> User: user = await self._require_user(user_id) user.disabled_at = datetime.now(UTC) user.is_active = False + self._revoke_sessions(user) await self._db.flush() return user @@ -217,6 +218,7 @@ async def enable(self, user_id: uuid.UUID) -> User: user = await self._require_user(user_id) user.disabled_at = None user.is_active = True + self._revoke_sessions(user) await self._db.flush() return user @@ -259,6 +261,7 @@ async def set_roles( ) ) + self._revoke_sessions(user) await self._db.flush() await self._db.refresh(user, attribute_names=["roles"]) return user diff --git a/modules/users/users/deps.py b/modules/users/users/deps.py index 564992652..cec16fa49 100644 --- a/modules/users/users/deps.py +++ b/modules/users/users/deps.py @@ -59,12 +59,13 @@ async def get_event_bus(request: Request) -> EventBus: async def get_user_service( + request: Request, db: AsyncSession = Depends(get_db), user_manager: UserManager = Depends(get_user_manager), ) -> UserService: from users.admin.service import UserService - return UserService(db, user_manager) + return UserService(db, user_manager, request.app) __all__ = [ From 6eb15a21c9602b1d17e8aeb054b1c225f958528c Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:07:05 +0200 Subject: [PATCH 2/7] file_storage: enforce the storage quota on plain uploads too Closes #586 Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- docs/modules/file_storage.md | 1 + .../file_storage/endpoints/api.py | 4 + modules/file_storage/file_storage/service.py | 4 +- modules/file_storage/file_storage/settings.py | 10 +- .../file_storage/upload_guards.py | 102 ++++++++++-------- .../tests/test_file_storage_plain_quota.py | 59 ++++++++++ .../tests/test_file_storage_uploads_pg.py | 21 ++++ 7 files changed, 150 insertions(+), 51 deletions(-) create mode 100644 modules/file_storage/tests/test_file_storage_plain_quota.py diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md index 7633826b1..1e3615b8a 100644 --- a/docs/modules/file_storage.md +++ b/docs/modules/file_storage.md @@ -286,6 +286,7 @@ DB-backed via `register_module_settings`; pydantic defaults seed at boot. Live e | `s3_signature_version` | `"s3v4"` | presign signature version; pinned so custom endpoints get SigV4 | | `s3_presign_ttl_seconds` | `300` | TTL for presigned `GET` URLs | | `max_file_size_bytes` | `100 * 1024 * 1024` | upload limit (100 MB) | +| `quota_bytes` | `None` | Enforced on plain and resumable uploads: stored bytes plus unexpired active session reservations; `None` means unlimited aggregate storage. Usage follows file-listing tenant scope; platform uploads count platform-owned files separately. PostgreSQL serializes quota admissions until the recording transaction commits. Rejected plain uploads delete the uploaded object. | | `allowed_content_types` | `None` | optional MIME-type whitelist; `None` ⇒ any | ## Backends diff --git a/modules/file_storage/file_storage/endpoints/api.py b/modules/file_storage/file_storage/endpoints/api.py index ac89e2118..34b41a273 100644 --- a/modules/file_storage/file_storage/endpoints/api.py +++ b/modules/file_storage/file_storage/endpoints/api.py @@ -38,6 +38,7 @@ from file_storage.contracts.service import StorageNotFoundError from file_storage.deps import OwnerScope, get_event_bus, get_file_storage_service from file_storage.downloads import content_disposition +from file_storage.endpoints.upload_errors import quota from file_storage.format import format_bytes from file_storage.ranges import file_body_response from file_storage.service import ( @@ -49,6 +50,7 @@ StreamDownload, ) from file_storage.serving import etag_for, not_found, sandbox_headers, thumbnail_response +from file_storage.upload_support import QuotaExceededError router = APIRouter() @@ -68,6 +70,8 @@ async def upload_file( ) -> StoredFileOut: try: out = await service.upload(file, public=public) + except QuotaExceededError as exc: + raise quota(t) from exc except FileTooLargeError as exc: # The limit belongs in the sentence: "too large" is not actionable to # someone holding a 40 MB file, and every client that shows this diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py index 3a1d26e41..086fab052 100644 --- a/modules/file_storage/file_storage/service.py +++ b/modules/file_storage/file_storage/service.py @@ -31,6 +31,7 @@ from file_storage.models import StoredFile from file_storage.reads import FileStorageReads from file_storage.scope import PLATFORM_TENANT_ID, owning_tenant, platform_scope +from file_storage.upload_guards import StorageQuota from file_storage.visibility import FileStoragePublic if TYPE_CHECKING: @@ -54,7 +55,7 @@ class StoredFileNotFoundError(Exception): """Raised when a StoredFile lookup misses; endpoints map this to HTTP 404.""" -class FileStorageService(FileStoragePublic, FileStorageReads, FileStorageDeletes): +class FileStorageService(FileStoragePublic, FileStorageReads, FileStorageDeletes, StorageQuota): """Orchestrates validation, hashing, backend IO, and DB lifecycle.""" def __init__( @@ -121,6 +122,7 @@ async def _hashing_stream() -> AsyncIterator[bytes]: # the cleanup error after logging; the orphaned key can be reaped # by a janitor sweep. try: + await self._check_quota(size, reserve=True, platform=platform) row = StoredFile( tenant_id=tenant_id, key=key, diff --git a/modules/file_storage/file_storage/settings.py b/modules/file_storage/file_storage/settings.py index 2406d8cb9..e25b5c191 100644 --- a/modules/file_storage/file_storage/settings.py +++ b/modules/file_storage/file_storage/settings.py @@ -82,17 +82,17 @@ class FileStorageSettings(DbBackedSettings): default=None, description="Whitelist of MIME types. None = any type allowed.", ) - # Reported, not enforced: the browse screen states "1.2 GB of 5 GB used" - # only for installs that have said what 5 GB means to them. Left unset, - # the screen reports usage without inventing a ceiling, and uploads are - # bounded by ``max_file_size_bytes`` alone as before. + # Enforced for plain uploads and resumable sessions: stored bytes plus + # unexpired active reservations. None leaves aggregate storage unlimited. # Positive, and no larger than a browser can hold exactly: the screen and # the settings editor read it as a JavaScript number (#548). quota_bytes: int | None = Field( default=None, gt=0, le=constants.MAX_JS_SAFE_INTEGER, - description="Total storage the browse screen measures usage against. None = unmeasured.", + description=( + "Storage ceiling for stored files and active upload reservations. None = unlimited." + ), ) public_files_cors: bool = Field( diff --git a/modules/file_storage/file_storage/upload_guards.py b/modules/file_storage/file_storage/upload_guards.py index 655f90a18..b8c3cc3af 100644 --- a/modules/file_storage/file_storage/upload_guards.py +++ b/modules/file_storage/file_storage/upload_guards.py @@ -1,4 +1,4 @@ -"""Cleanup and quota guards for upload sessions (#454).""" +"""Cleanup and shared quota guards for uploads.""" from __future__ import annotations @@ -12,7 +12,7 @@ from file_storage import constants from file_storage.contracts.service import StorageError from file_storage.models import StoredFile, UploadSession -from file_storage.scope import owning_tenant +from file_storage.scope import PLATFORM_TENANT_ID, owning_tenant, platform_scope from file_storage.upload_support import DirectUploadUnsupportedError, QuotaExceededError if TYPE_CHECKING: @@ -28,7 +28,61 @@ _QUOTA_LOCK_KEY = 0x4653544F << 32 -class UploadGuards: +class StorageQuota: + """Quota checks shared by plain uploads and resumable reservations.""" + + db: AsyncSession + settings: FileStorageSettings + + async def _serialize_reservations(self, *, platform: bool = False) -> None: + """Make concurrent quota admissions take turns on Postgres. + + Each upload sums stored + reserved bytes before recording a file or + reservation. Under READ COMMITTED, concurrent uploads can read the same + sum and all pass. The transaction-scoped lock lasts until the new row + becomes visible at commit. SQLite serialises writers already. + """ + bind = self.db.get_bind() + if bind.dialect.name != "postgresql": + return + tenant = owning_tenant(self.db, platform=True) if platform else owning_tenant(self.db) + await self.db.execute( + text("SELECT pg_advisory_xact_lock(:key)"), + {"key": _QUOTA_LOCK_KEY + zlib.crc32(tenant.encode())}, + ) + + async def _check_quota(self, size: int, *, reserve: bool, platform: bool = False) -> None: + """Refuse ``size`` bytes that would push stored (+ reserved) bytes past the quota. + + Read live, not from the browse screen's cached totals: a cache a minute + stale would let a burst of sessions each see an empty bucket. With + ``reserve``, sessions still open count at their declared size, so opening + many small-looking sessions cannot overbook the quota. + """ + quota = self.settings.quota_bytes + if quota is None: + return + if reserve: + if platform: + await self._serialize_reservations(platform=True) + else: + await self._serialize_reservations() + stored_query = select(func.coalesce(func.sum(StoredFile.size_bytes), 0)) + reserved_query = select(func.coalesce(func.sum(UploadSession.total_size), 0)).where( + UploadSession.status == constants.UploadStatus.ACTIVE, # ty: ignore[invalid-argument-type] + UploadSession.expires_at > datetime.now(UTC), # ty: ignore[invalid-argument-type] + ) + if platform: + stored_query = stored_query.where(StoredFile.tenant_id == PLATFORM_TENANT_ID) + reserved_query = reserved_query.where(UploadSession.tenant_id == PLATFORM_TENANT_ID) + async with platform_scope(self.db, platform): + stored = await self.db.scalar(stored_query) + reserved = await self.db.scalar(reserved_query) if reserve else 0 + if (stored or 0) + (reserved or 0) + size > quota: + raise QuotaExceededError(str(quota)) + + +class UploadGuards(StorageQuota): """Mixed into :class:`~file_storage.uploads.UploadSessionService`.""" files: FileStorageService @@ -73,45 +127,3 @@ async def purge_finished(self, *, owner: str | None) -> int: ) ) return result.rowcount or 0 - - async def _serialize_reservations(self) -> None: - """Make concurrent session openers take turns on Postgres. - - Each opener sums stored + reserved bytes and then inserts its own - reservation; under READ COMMITTED, N openers all read the same sum and - all pass. A transaction-scoped advisory lock (held until this request - commits, which is when the new reservation becomes visible) closes the - window. SQLite serialises writers already. - """ - bind = self.db.get_bind() - if bind.dialect.name != "postgresql": - return - await self.db.execute( - text("SELECT pg_advisory_xact_lock(:key)"), - {"key": _QUOTA_LOCK_KEY + zlib.crc32(owning_tenant(self.db).encode())}, - ) - - async def _check_quota(self, size: int, *, reserve: bool) -> None: - """Refuse ``size`` bytes that would push stored (+ reserved) bytes past the quota. - - Read live, not from the browse screen's cached totals: a cache a minute - stale would let a burst of sessions each see an empty bucket. With - ``reserve``, sessions still open count at their declared size, so opening - many small-looking sessions cannot overbook the quota. - """ - quota = self.settings.quota_bytes - if quota is None: - return - if reserve: - await self._serialize_reservations() - stored = await self.db.scalar(select(func.coalesce(func.sum(StoredFile.size_bytes), 0))) - reserved = 0 - if reserve: - reserved = await self.db.scalar( - select(func.coalesce(func.sum(UploadSession.total_size), 0)).where( - UploadSession.status == constants.UploadStatus.ACTIVE, # ty: ignore[invalid-argument-type] - UploadSession.expires_at > datetime.now(UTC), # ty: ignore[invalid-argument-type] - ) - ) - if (stored or 0) + (reserved or 0) + size > quota: - raise QuotaExceededError(str(quota)) diff --git a/modules/file_storage/tests/test_file_storage_plain_quota.py b/modules/file_storage/tests/test_file_storage_plain_quota.py new file mode 100644 index 000000000..1a09a59a5 --- /dev/null +++ b/modules/file_storage/tests/test_file_storage_plain_quota.py @@ -0,0 +1,59 @@ +"""Plain uploads share stored-byte limits and resumable reservations (#586).""" + +from file_storage import constants +from file_storage_uploads_helpers import API, open_session + + +async def upload(client, size): + return await client.post( + f"{API}/upload", files={"file": ("small.bin", b"x" * size, "application/octet-stream")} + ) + + +async def test_plain_uploads_enforce_aggregate_quota(authenticated_client, storage): + storage.settings.quota_bytes = 10 + assert (await upload(authenticated_client, 6)).status_code == 201 + assert (await upload(authenticated_client, 4)).status_code == 201 + rejected = await upload(authenticated_client, 1) + assert rejected.status_code == 413 + assert rejected.json()["detail"]["code"] == constants.ErrorCode.QUOTA_EXCEEDED + listing = (await authenticated_client.get(f"{API}/files")).json() + assert listing["total"] == 2 + assert len([p for p in storage.backend.root.rglob("*") if p.is_file()]) == 2 + + +async def test_plain_upload_honors_reservations_and_abort_releases_them( + authenticated_client, storage, checksum_calls +): + storage.settings.quota_bytes = 10 + response = await open_session(authenticated_client, total=8) + assert response.status_code == 201 + assert (await upload(authenticated_client, 3)).status_code == 413 + assert (await upload(authenticated_client, 2)).status_code == 201 + await authenticated_client.delete(f"{API}/uploads/{response.json()['id']}") + assert (await upload(authenticated_client, 8)).status_code == 201 + + +async def test_platform_quota_counts_only_platform_files(db_session, tmp_path): + from io import BytesIO + + import pytest + from fastapi import UploadFile + from file_storage.backends.filesystem import FilesystemBackend + from file_storage.service import FileStorageService + from file_storage.settings import FileStorageSettings + from file_storage.upload_support import QuotaExceededError + from simple_module_db import tenant_context + + backend = FilesystemBackend(root=tmp_path) + service = FileStorageService(db_session, backend, FileStorageSettings(quota_bytes=3)) + + def file(size): + return UploadFile(BytesIO(b"x" * size), filename="small.bin") + + with tenant_context("tenant-a"): + await service.upload(file(3)) + await service.upload(file(3), platform=True) + with pytest.raises(QuotaExceededError): + await service.upload(file(1), platform=True) + assert len([p for p in tmp_path.rglob("*") if p.is_file()]) == 2 diff --git a/modules/file_storage/tests/test_file_storage_uploads_pg.py b/modules/file_storage/tests/test_file_storage_uploads_pg.py index 1739b47a4..2ffd2a5a2 100644 --- a/modules/file_storage/tests/test_file_storage_uploads_pg.py +++ b/modules/file_storage/tests/test_file_storage_uploads_pg.py @@ -53,3 +53,24 @@ async def test_concurrent_completes_record_one_file(authenticated_client, storag assert len({r.json()["file"]["id"] for r in responses}) == 1 assert (await authenticated_client.get(f"{API}/files")).json()["total"] == 1 assert len(checksum_calls) == 1 + + +async def test_concurrent_plain_uploads_cannot_overbook_quota(authenticated_client, storage): + storage.settings.quota_bytes = 10 + responses = await asyncio.gather( + *( + authenticated_client.post( + f"{API}/upload", + files={"file": ("small.bin", b"x" * 5, "application/octet-stream")}, + ) + for _ in range(6) + ) + ) + assert sorted(r.status_code for r in responses) == [201, 201, 413, 413, 413, 413] + assert all( + r.json()["detail"]["code"] == constants.ErrorCode.QUOTA_EXCEEDED + for r in responses + if r.status_code == 413 + ) + assert (await authenticated_client.get(f"{API}/files")).json()["total"] == 2 + assert len([p for p in storage.backend.root.rglob("*") if p.is_file()]) == 2 From 60a294bd364e8d7ba0d8f4aea30ae02a14297311 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:11:09 +0200 Subject: [PATCH 3/7] changelog: batch 2 entries (#581 #586) Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e6971fc3..ed7ea250e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,16 @@ All notable changes to this project are documented in this file. The format is b ## [Unreleased] ### Fixed +- Users (security): disabling or enabling an account, or replacing its roles from the admin + screens, now bumps `session_version` in the same transaction and evicts the cached counter on + every worker after commit, so existing cookie sessions and bearer access tokens stop working + at once. Before, a disabled or demoted user kept the roles cached in their cookie until the + session expired. **Behaviour change:** a role change signs that user out of their existing + sessions (#581). +- File storage: **Behaviour change.** `quota_bytes` is now enforced for plain uploads as well as + resumable ones (stored bytes plus active reservations, serialised per tenant on Postgres); an + upload that would exceed it answers 413 and its object is removed. Previously only resumable + sessions were checked, and the setting was described as report-only (#586). - File storage: `DELETE /api/file_storage/files/{id}` and the bulk delete now drop the stored bytes only after the request transaction commits, so a failed commit no longer leaves a live row with no bytes. A new hourly `file_storage.reap_objects` beat task (needs From 925475f4544a4af2b4beb33f2d2cfb6b9f60fb32 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:13:02 +0200 Subject: [PATCH 4/7] refactor(optimize): drop duplicate attribute declarations in UploadGuards --- modules/file_storage/file_storage/upload_guards.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/modules/file_storage/file_storage/upload_guards.py b/modules/file_storage/file_storage/upload_guards.py index b8c3cc3af..bfd664ab0 100644 --- a/modules/file_storage/file_storage/upload_guards.py +++ b/modules/file_storage/file_storage/upload_guards.py @@ -86,8 +86,6 @@ class UploadGuards(StorageQuota): """Mixed into :class:`~file_storage.uploads.UploadSessionService`.""" files: FileStorageService - db: AsyncSession - settings: FileStorageSettings @property def backend(self) -> DirectUploadBackend: From 35aea2047da43ae088f4bafd617ed98dcfcd2345 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:18:12 +0200 Subject: [PATCH 5/7] fix: address code review findings (round 1, pass 1) Branding maps a full quota to 413; disabling an account also revokes its refresh tokens. Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- modules/branding/branding/endpoints/api.py | 8 ++++---- .../branding/branding/endpoints/tenant_api.py | 4 ++-- modules/branding/branding/images.py | 17 +++++++++++++++++ .../file_storage/file_storage/upload_guards.py | 7 ++----- .../tests/test_file_storage_upload_guards.py | 6 +++--- modules/users/users/admin/service.py | 10 +++++++++- 6 files changed, 37 insertions(+), 15 deletions(-) diff --git a/modules/branding/branding/endpoints/api.py b/modules/branding/branding/endpoints/api.py index e1bf9187b..12166e17e 100644 --- a/modules/branding/branding/endpoints/api.py +++ b/modules/branding/branding/endpoints/api.py @@ -10,7 +10,7 @@ from branding import constants from branding.contracts.schemas import BrandingOut, BrandingUpdate from branding.deps import BrandingServiceDep -from branding.images import validate_image +from branding.images import store_image, validate_image from branding.presets import find_preset router = APIRouter() @@ -67,7 +67,7 @@ async def upload_logo( storage: FileStorageService = Depends(get_file_storage_service), ) -> BrandingOut: await validate_image(file) - stored = await storage.upload(file, platform=True) + stored = await store_image(storage, file, platform=True) return await service.set_logo(str(stored.id)) @@ -91,7 +91,7 @@ async def upload_logo_dark( storage: FileStorageService = Depends(get_file_storage_service), ) -> BrandingOut: await validate_image(file) - stored = await storage.upload(file, platform=True) + stored = await store_image(storage, file, platform=True) return await service.set_logo_dark(str(stored.id)) @@ -107,7 +107,7 @@ async def upload_favicon( storage: FileStorageService = Depends(get_file_storage_service), ) -> BrandingOut: await validate_image(file) - stored = await storage.upload(file, platform=True) + stored = await store_image(storage, file, platform=True) return await service.set_favicon(str(stored.id)) diff --git a/modules/branding/branding/endpoints/tenant_api.py b/modules/branding/branding/endpoints/tenant_api.py index 790891fa4..76a073586 100644 --- a/modules/branding/branding/endpoints/tenant_api.py +++ b/modules/branding/branding/endpoints/tenant_api.py @@ -27,7 +27,7 @@ from branding.constants import PACKAGE, PATH_TENANT_ASSET, TENANT_ASSETS, TENANT_IMAGE_FIELDS from branding.contracts.schemas import BrandingOut -from branding.images import validate_image +from branding.images import store_image, validate_image from branding.reaper import schedule_reap from branding.service import to_out from branding.tenant_branding import merge, overrides_from @@ -83,7 +83,7 @@ async def upload_tenant_asset( field = _field(asset) tenant_id = active_tenant(request) await validate_image(file) - stored = await storage.upload(file) + stored = await store_image(storage, file) return await _swap(request, settings, storage, tenant_id, field, str(stored.id)) diff --git a/modules/branding/branding/images.py b/modules/branding/branding/images.py index 334758b6f..b9378cd52 100644 --- a/modules/branding/branding/images.py +++ b/modules/branding/branding/images.py @@ -96,3 +96,20 @@ async def validate_image(file: UploadFile) -> None: f"{content_type!r}. {ALLOWED_IMAGE_HINT}" ), ) + + +async def store_image(storage, file: UploadFile, *, platform: bool = False): + """Hand a validated image to ``file_storage``, answering 413 when the quota is full. + + ``file_storage`` enforces ``quota_bytes`` on every upload; its own endpoint + maps the refusal to 413, but branding calls the service directly, so an + unmapped ``QuotaExceededError`` would surface as a 500. + """ + from file_storage.upload_support import QuotaExceededError + + try: + return await storage.upload(file, platform=platform) + except QuotaExceededError: + raise HTTPException( + status_code=413, detail="The storage quota is full; the image was not saved." + ) from None diff --git a/modules/file_storage/file_storage/upload_guards.py b/modules/file_storage/file_storage/upload_guards.py index bfd664ab0..217b1df42 100644 --- a/modules/file_storage/file_storage/upload_guards.py +++ b/modules/file_storage/file_storage/upload_guards.py @@ -45,7 +45,7 @@ async def _serialize_reservations(self, *, platform: bool = False) -> None: bind = self.db.get_bind() if bind.dialect.name != "postgresql": return - tenant = owning_tenant(self.db, platform=True) if platform else owning_tenant(self.db) + tenant = owning_tenant(self.db, platform=platform) await self.db.execute( text("SELECT pg_advisory_xact_lock(:key)"), {"key": _QUOTA_LOCK_KEY + zlib.crc32(tenant.encode())}, @@ -63,10 +63,7 @@ async def _check_quota(self, size: int, *, reserve: bool, platform: bool = False if quota is None: return if reserve: - if platform: - await self._serialize_reservations(platform=True) - else: - await self._serialize_reservations() + await self._serialize_reservations(platform=platform) stored_query = select(func.coalesce(func.sum(StoredFile.size_bytes), 0)) reserved_query = select(func.coalesce(func.sum(UploadSession.total_size), 0)).where( UploadSession.status == constants.UploadStatus.ACTIVE, # ty: ignore[invalid-argument-type] diff --git a/modules/file_storage/tests/test_file_storage_upload_guards.py b/modules/file_storage/tests/test_file_storage_upload_guards.py index 5b844a686..5e0bd8bd3 100644 --- a/modules/file_storage/tests/test_file_storage_upload_guards.py +++ b/modules/file_storage/tests/test_file_storage_upload_guards.py @@ -233,7 +233,7 @@ async def test_reserve_takes_the_lock_before_reading_usage( order: list[str] = [] guards = _Guards(db_session, backend, quota_bytes=100) - async def fake_lock() -> None: + async def fake_lock(*, platform: bool = False) -> None: order.append("lock") monkeypatch.setattr(guards, "_serialize_reservations", fake_lock) @@ -261,9 +261,9 @@ async def test_is_a_no_op_off_postgres(self, backend, db_session): async def test_takes_a_per_tenant_advisory_lock_on_postgres(self, backend, monkeypatch): calls: list[tuple[str, Any]] = [] - monkeypatch.setattr(guards_module, "owning_tenant", lambda db: "acme") + monkeypatch.setattr(guards_module, "owning_tenant", lambda db, platform=False: "acme") await _Guards(_db_on("postgresql", calls), backend)._serialize_reservations() - monkeypatch.setattr(guards_module, "owning_tenant", lambda db: "globex") + monkeypatch.setattr(guards_module, "owning_tenant", lambda db, platform=False: "globex") await _Guards(_db_on("postgresql", calls), backend)._serialize_reservations() assert all("pg_advisory_xact_lock" in sql for sql, _ in calls) diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index 2522d2f4a..5dbf6d05b 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -6,7 +6,7 @@ import uuid from datetime import UTC, datetime -from sqlalchemy import delete, func, select +from sqlalchemy import delete, func, select, update from sqlalchemy.orm import noload from users.admin.queries import _UserServiceBase @@ -211,6 +211,14 @@ async def disable(self, user_id: uuid.UUID) -> User: user.disabled_at = datetime.now(UTC) user.is_active = False self._revoke_sessions(user) + # Refresh tokens are spent too, so re-enabling the account later does not + # revive credentials issued before it was disabled (as the self-service + # "sign out everywhere" does). + await self._db.execute( + update(RefreshToken) + .where(RefreshToken.user_id == user.id, RefreshToken.revoked_at.is_(None)) # ty: ignore[invalid-argument-type] + .values(revoked_at=datetime.now(UTC)) + ) await self._db.flush() return user From a4af0d7120797579b8551cd524989d8905d9a7f5 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 20:22:38 +0200 Subject: [PATCH 6/7] fix: address code review findings (round 1, pass 1 follow-up) session_version is incremented atomically in SQL on every revocation path. Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 10 +++--- .../tests/test_session_version_atomic.py | 35 +++++++++++++++++++ modules/users/users/admin/queries.py | 6 ++-- modules/users/users/admin/service.py | 6 ++-- .../users/users/auth_local/self_account.py | 14 ++------ modules/users/users/session_revocation.py | 23 +++++++++++- 6 files changed, 72 insertions(+), 22 deletions(-) create mode 100644 modules/users/tests/test_session_version_atomic.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ed7ea250e..282a3798d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,10 +18,12 @@ All notable changes to this project are documented in this file. The format is b ### Fixed - Users (security): disabling or enabling an account, or replacing its roles from the admin screens, now bumps `session_version` in the same transaction and evicts the cached counter on - every worker after commit, so existing cookie sessions and bearer access tokens stop working - at once. Before, a disabled or demoted user kept the roles cached in their cookie until the - session expired. **Behaviour change:** a role change signs that user out of their existing - sessions (#581). + every worker after commit (an atomic SQL increment, so overlapping changes cannot lose one), so + existing cookie sessions and bearer access tokens are refused on their next request; disabling + also revokes the account's refresh tokens. Before, a disabled or demoted user kept the roles + cached in their cookie until the session expired. **Behaviour change:** a role change signs that + user out of their existing sessions; a remembered sign-in re-authenticates with the new roles + (#581). - File storage: **Behaviour change.** `quota_bytes` is now enforced for plain uploads as well as resumable ones (stored bytes plus active reservations, serialised per tenant on Postgres); an upload that would exceed it answers 413 and its object is removed. Previously only resumable diff --git a/modules/users/tests/test_session_version_atomic.py b/modules/users/tests/test_session_version_atomic.py new file mode 100644 index 000000000..fb9b8c230 --- /dev/null +++ b/modules/users/tests/test_session_version_atomic.py @@ -0,0 +1,35 @@ +"""Revocations must increment the database counter rather than a stale copy.""" + +from sqlalchemy import event +from users.admin.queries import _UserServiceBase +from users.models import User + + +async def test_revocations_use_atomic_sql_and_refresh_counter(db_session): + user = User(email="atomic@example.com", hashed_password="unused", session_version=7) + db_session.add(user) + await db_session.flush() + statements = [] + + def record_sql(conn, cursor, statement, parameters, context, executemany): + if statement.lstrip().upper().startswith("UPDATE"): + statements.append(statement) + + engine = db_session.bind.sync_engine + event.listen(engine, "before_cursor_execute", record_sql) + try: + service = _UserServiceBase(db_session, None) + await service._revoke_sessions(user) + # A plain read must work without implicit async IO after the flush. + assert user.session_version == 8 + await service._revoke_sessions(user) + assert user.session_version == 9 + await db_session.refresh(user) + assert user.session_version == 9 + finally: + event.remove(engine, "before_cursor_execute", record_sql) + + # Each write derives its value from the database row, so overlapping + # requests with the same stale value cannot overwrite an increment. + assert len(statements) == 2 + assert all("session_version=(users_user.session_version +" in sql for sql in statements) diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py index e285c3509..f8e7b30fb 100644 --- a/modules/users/users/admin/queries.py +++ b/modules/users/users/admin/queries.py @@ -18,7 +18,7 @@ from users.exceptions import UserNotFoundError from users.manager import UserManager from users.models import Role, User, UserRole -from users.session_revocation import publish_revocation +from users.session_revocation import bump_session_version, publish_revocation from users.session_version_cache import forget_session_version @@ -44,9 +44,9 @@ def __init__( self._manager = user_manager self._app = app - def _revoke_sessions(self, user: User) -> None: + async def _revoke_sessions(self, user: User) -> None: """Persist the counter with the write; evict only after it commits.""" - user.session_version = int(user.session_version or 0) + 1 + await bump_session_version(self._db, user) user_id = user.id app = self._app if app is None: diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index 5dbf6d05b..d0ee7c8b0 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -210,7 +210,7 @@ async def disable(self, user_id: uuid.UUID) -> User: user = await self._require_user(user_id) user.disabled_at = datetime.now(UTC) user.is_active = False - self._revoke_sessions(user) + await self._revoke_sessions(user) # Refresh tokens are spent too, so re-enabling the account later does not # revive credentials issued before it was disabled (as the self-service # "sign out everywhere" does). @@ -226,7 +226,7 @@ async def enable(self, user_id: uuid.UUID) -> User: user = await self._require_user(user_id) user.disabled_at = None user.is_active = True - self._revoke_sessions(user) + await self._revoke_sessions(user) await self._db.flush() return user @@ -269,7 +269,7 @@ async def set_roles( ) ) - self._revoke_sessions(user) + await self._revoke_sessions(user) await self._db.flush() await self._db.refresh(user, attribute_names=["roles"]) return user diff --git a/modules/users/users/auth_local/self_account.py b/modules/users/users/auth_local/self_account.py index da4597d13..c236fd31b 100644 --- a/modules/users/users/auth_local/self_account.py +++ b/modules/users/users/auth_local/self_account.py @@ -29,7 +29,7 @@ from users.deps import fastapi_users, get_user_manager from users.manager import UserManager from users.models import RefreshToken, UserAccessToken -from users.session_revocation import publish_revocation +from users.session_revocation import bump_session_version, publish_revocation router = APIRouter() @@ -93,10 +93,7 @@ async def change_my_password( except fu_exceptions.InvalidPasswordException as exc: raise HTTPException(status_code=400, detail=exc.reason) from exc - # ``user`` is attached to this request's session (``get_user_db`` takes the - # same ``get_db``), so assigning is enough to mark it written — a Core - # UPDATE would be rolled back as a read-only request. - user.session_version = int(user.session_version or 0) + 1 + await bump_session_version(db, user) request.session[SESSION_VERSION_KEY] = user.session_version # Every worker caches the counter; without this they would keep honouring # the sessions this change was meant to strand. Hung on the commit rather @@ -168,12 +165,7 @@ async def revoke_all_my_sessions( session that ordered the revocation as the exception to it. """ now = datetime.now(UTC) - # Bumped through the ORM rather than as a bulk UPDATE: a Core statement - # does not mark the request session as written, so ``get_db`` would treat - # this handler as read-only and roll the whole revocation back. ``user`` - # is already attached to this session — ``get_user_db`` takes the same - # ``get_db`` dependency — so the assignment is enough. - user.session_version = int(user.session_version or 0) + 1 + await bump_session_version(db, user) # See the password change: the cached counter has to go with the bump, or # the workers keep letting the sessions this just revoked back in — and it # has to go *after* the commit, or a rollback restores the old counter diff --git a/modules/users/users/session_revocation.py b/modules/users/users/session_revocation.py index faf3a2b8c..a898f1f75 100644 --- a/modules/users/users/session_revocation.py +++ b/modules/users/users/session_revocation.py @@ -20,20 +20,41 @@ import uuid from typing import TYPE_CHECKING +from users.models import User from users.session_version_cache import clear_session_version_cache, forget_session_version if TYPE_CHECKING: from fastapi import FastAPI from simple_module_core.invalidation import Invalidation, InvalidationBus + from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) -__all__ = ["CHANNEL", "apply_invalidation", "publish_revocation", "subscribe"] +__all__ = [ + "CHANNEL", + "apply_invalidation", + "bump_session_version", + "publish_revocation", + "subscribe", +] CHANNEL = "users.session_version" """Carries one account's user id, or ``None`` for "forget every account".""" +async def bump_session_version(db: AsyncSession, user: User) -> None: + """Increment in SQL, then load the counter for session/token minting. + + The ORM assignment marks the request as written. A Python increment would + lose a revocation when two requests loaded the same old counter. + """ + user.session_version = User.session_version + 1 + await db.flush() + # SQL expressions expire the attribute on flush; async callers must not + # trigger implicit IO when they read the new counter later in the request. + await db.refresh(user, attribute_names=["session_version"]) + + def _cache_key(key: str): """Turn a wire key back into what the cache is keyed by. From 0824031a8a999a08d46719bca0450dfd73994c66 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 22 Oct 2026 21:28:13 +0200 Subject: [PATCH 7/7] fix: type branding store_image Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- modules/branding/branding/images.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/modules/branding/branding/images.py b/modules/branding/branding/images.py index b9378cd52..eb1323631 100644 --- a/modules/branding/branding/images.py +++ b/modules/branding/branding/images.py @@ -12,10 +12,14 @@ from __future__ import annotations -from typing import Final +from typing import TYPE_CHECKING, Final from fastapi import HTTPException, UploadFile +if TYPE_CHECKING: + from file_storage.contracts.schemas import StoredFileOut + from file_storage.service import FileStorageService + # Raster + icon only. SVG is excluded on purpose: it is an XML document that can # carry