diff --git a/backend/alembic/versions/0001_initial_control_plane.py b/backend/alembic/versions/0001_initial_control_plane.py index cc14ce39b..31b187f55 100644 --- a/backend/alembic/versions/0001_initial_control_plane.py +++ b/backend/alembic/versions/0001_initial_control_plane.py @@ -9,7 +9,7 @@ from sqlalchemy import text from db.models import Base -from scripts.bootstrap_db import schema_backfill_sql +from scripts.bootstrap_db import execute_schema_backfill revision = "0001_initial_control_plane" down_revision = None @@ -19,8 +19,7 @@ def upgrade() -> None: connection = op.get_bind() connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) Base.metadata.create_all(connection) - for statement in schema_backfill_sql(): - connection.execute(statement) + execute_schema_backfill(connection) def downgrade() -> None: diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 716590cd1..04fa068ce 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -1,6 +1,14 @@ -"""Add is_read to emails (IMAP \\Seen read state). +"""Add is_read to email_records (IMAP \\Seen read state). Existing rows default to read so historical/file imports do not surface as unread. + +Checks both "email_records" (the real, current table -- a database whose own +0001_initial_control_plane ran before ``is_read`` was added to the ``Email`` +model has this table without the column, and needs it added) and "emails" +(a legacy name that, per 0011_email_model_reconciliation's docstring, no +migration in this repo's history ever actually created for a real managed +database, but is checked defensively in case one somehow exists). Guarded by +column existence, not just table existence, so it is safely idempotent. """ from alembic import op @@ -12,18 +20,40 @@ branch_labels = None depends_on = None +_CANDIDATE_TABLES = ("email_records", "emails") + def upgrade() -> None: - op.add_column( - "emails", - sa.Column( - "is_read", - sa.Boolean(), - nullable=False, - server_default=sa.text("true"), - ), - ) + inspector = sa.inspect(op.get_bind()) + for table_name in _CANDIDATE_TABLES: + if inspector.has_table(table_name) and not _has_column( + inspector, table_name, "is_read" + ): + op.add_column( + table_name, + sa.Column( + "is_read", + sa.Boolean(), + nullable=False, + server_default=sa.text("true"), + ), + ) def downgrade() -> None: - op.drop_column("emails", "is_read") + # Same ownership-ambiguity problem as 0018_workspace_registry's downgrade: + # a fresh database gets email_records.is_read from 0001's live + # Base.metadata.create_all, not from this revision, so there is no way to + # tell "this revision added the column" apart from "the baseline already + # had it" -- and is_read holds real per-message read/unread state, not + # rebuildable derived data. As with 0001_initial_control_plane and + # 0018_workspace_registry: production rollbacks should restore from + # backup or a later explicit down revision rather than dropping + # customer-owned data. + return None + + +def _has_column(inspector, table_name: str, column_name: str) -> bool: + return any( + column["name"] == column_name for column in inspector.get_columns(table_name) + ) diff --git a/backend/alembic/versions/0016_document_org_scope.py b/backend/alembic/versions/0016_document_org_scope.py index 0a5cd0035..0c823754e 100644 --- a/backend/alembic/versions/0016_document_org_scope.py +++ b/backend/alembic/versions/0016_document_org_scope.py @@ -8,6 +8,13 @@ recognition worker can resolve the owning organization's provider without joining through the (organization-less) workspace entity. Nullable and additive so existing rows and personal-scope documents are unaffected. + +A database that has never had ``workspace_documents`` at all (one that ran +``0001_initial_control_plane`` before ``Workspace``/``Document`` existed in +``db/models.py``, and has only applied incremental migrations since) has no +table for this revision to alter. This revision is a no-op for that case; +``0018_workspace_registry`` creates the table later in the chain, already +including this column. """ from alembic import op @@ -24,6 +31,8 @@ def upgrade() -> None: connection = op.get_bind() inspector = sa.inspect(connection) + if not inspector.has_table(_DOCUMENTS_TABLE): + return columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)} if _ORG_COLUMN not in columns: op.add_column( @@ -39,9 +48,7 @@ def upgrade() -> None: def downgrade() -> None: - connection = op.get_bind() - inspector = sa.inspect(connection) - columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)} - op.drop_index(_ORG_INDEX, table_name=_DOCUMENTS_TABLE, if_exists=True) - if _ORG_COLUMN in columns: - op.drop_column(_DOCUMENTS_TABLE, _ORG_COLUMN) + # 0018 can create this table and column after 0016 was a no-op. Alembic + # cannot distinguish that case from a table altered by this revision, so + # dropping the column here could destroy later organization assignments. + return None diff --git a/backend/alembic/versions/0018_workspace_registry.py b/backend/alembic/versions/0018_workspace_registry.py new file mode 100644 index 000000000..aaef5f446 --- /dev/null +++ b/backend/alembic/versions/0018_workspace_registry.py @@ -0,0 +1,81 @@ +"""create workspace registry and workspace document tables + +Revision ID: 0018_workspace_registry +Revises: 0017_merge_newsdom_carddav_heads +Create Date: 2026-09-01 00:00:00.000000 + +``Workspace``/``Document`` (``workspace_entities``/``workspace_documents``) have +been declared in ``db/models.py`` since before this repository's incremental +migration history begins tracking them explicitly. A database that ran +``0001_initial_control_plane``'s ``Base.metadata.create_all`` after these +models existed already has both tables; a database that ran ``0001`` earlier +and has only applied incremental migrations since never got them, so +``/api/data/documents`` fails with an undefined-relation error the first time +it is hit. This revision is idempotent (``has_table`` guarded) so it is a +no-op for a database that already has the tables and a real fix for one that +does not. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0018_workspace_registry" +down_revision = "0017_merge_newsdom_carddav_heads" + +_ENTITIES_TABLE = "workspace_entities" +_DOCUMENTS_TABLE = "workspace_documents" + + +def upgrade() -> None: + connection = op.get_bind() + inspector = sa.inspect(connection) + + if not inspector.has_table(_ENTITIES_TABLE): + op.create_table( + _ENTITIES_TABLE, + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("workspace_name", sa.String(), nullable=False), + sa.Column("workspace_domain", sa.String(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("workspace_id"), + ) + + if not inspector.has_table(_DOCUMENTS_TABLE): + op.create_table( + _DOCUMENTS_TABLE, + sa.Column("document_id", sa.String(), nullable=False), + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("organization_id", sa.String(), nullable=True), + sa.Column("document_name", sa.String(), nullable=False), + sa.Column("document_type", sa.String(), nullable=False), + sa.Column("document_content", sa.Text(), nullable=True), + sa.Column("document_status", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["workspace_id"], [f"{_ENTITIES_TABLE}.workspace_id"] + ), + sa.PrimaryKeyConstraint("document_id"), + ) + + for index_name, column_names in ( + ("ix_workspace_documents_workspace_id", ["workspace_id"]), + ("ix_workspace_documents_organization_id", ["organization_id"]), + ): + op.create_index( + index_name, + _DOCUMENTS_TABLE, + column_names, + if_not_exists=True, + ) + + +def downgrade() -> None: + # This revision's upgrade is a no-op whenever the tables already exist + # (e.g. created by 0001's create_all), so a downgrade cannot tell "this + # revision created these tables" apart from "they predate it" -- and + # workspace_documents.document_content holds real uploaded content, not + # rebuildable derived state. Unconditionally dropping it risks destroying + # data this revision never created. As with 0001_initial_control_plane: + # production rollbacks should restore from backup or a later explicit + # down revision rather than dropping customer-owned data. + return None diff --git a/backend/alembic/versions/0019_email_read_state_repair.py b/backend/alembic/versions/0019_email_read_state_repair.py new file mode 100644 index 000000000..814fd0c25 --- /dev/null +++ b/backend/alembic/versions/0019_email_read_state_repair.py @@ -0,0 +1,57 @@ +"""idempotently ensure email_records has is_read + +Revision ID: 0019_email_read_state_repair +Revises: 0018_workspace_registry +Create Date: 2026-09-01 00:00:00.000000 + +Alembic never re-runs a revision's ``upgrade()`` once that revision id is +recorded as applied for a database -- editing ``0011_email_read_state.py``'s +content cannot repair a database that already has "0011_email_read_state" +in its ``alembic_version`` history but never actually got +``email_records.is_read`` (whatever the reason: an earlier version of that +revision that targeted the wrong table, a partial/interrupted apply, manual +intervention). This revision is the real repair path for such a database: +appended after the current head, so it runs regardless of what 0011 already +did or didn't do. Idempotent (has_table/has_column guarded) so it is a +no-op for every database that already has the column, from any path. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0019_email_read_state_repair" +down_revision = "0018_workspace_registry" + +_EMAIL_TABLE = "email_records" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_EMAIL_TABLE) and not _has_column( + inspector, _EMAIL_TABLE, "is_read" + ): + op.add_column( + _EMAIL_TABLE, + sa.Column( + "is_read", + sa.Boolean(), + nullable=False, + server_default=sa.text("true"), + ), + ) + + +def downgrade() -> None: + # Same ownership-ambiguity reasoning as 0011_email_read_state and + # 0018_workspace_registry: this revision cannot tell whether it was the + # one that added the column (repairing a stamped-but-incomplete + # database) or the column already existed from another path, and + # is_read holds real read/unread state, not rebuildable derived data. + # No-op; production rollbacks should restore from backup. + return None + + +def _has_column(inspector, table_name: str, column_name: str) -> bool: + return any( + column["name"] == column_name for column in inspector.get_columns(table_name) + ) diff --git a/backend/alembic/versions/0020_workspace_organization_binding.py b/backend/alembic/versions/0020_workspace_organization_binding.py new file mode 100644 index 000000000..fb2eff4c1 --- /dev/null +++ b/backend/alembic/versions/0020_workspace_organization_binding.py @@ -0,0 +1,160 @@ +"""bind workspace registry rows to auditable organization evidence + +Revision ID: 0020_workspace_organization_binding +Revises: 0019_email_read_state_repair +Create Date: 2026-09-14 20:41:00.000000 + +``workspace_id`` is an opaque authenticated claim, not an organization-derived +identifier. This revision therefore binds a workspace only when persisted +``workspace_documents.organization_id`` evidence is non-null and unambiguous: +exactly one distinct organization is already recorded for that workspace. +Ambiguous and evidence-free workspaces remain unbound so authorization can fail +closed instead of guessing an owner from an identifier shape. + +Once a workspace is safely bound, legacy NULL document rows in that same +workspace inherit the binding. Downgrade is intentionally non-destructive: +these bindings are ownership provenance and cannot be distinguished later from +assignments written by normal application traffic. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0020_workspace_organization_binding" +down_revision = "0019_email_read_state_repair" +branch_labels = None +depends_on = None + +_ENTITIES_TABLE = "workspace_entities" +_DOCUMENTS_TABLE = "workspace_documents" +_ORGANIZATION_COLUMN = "organization_id" +_ORGANIZATION_INDEX = "ix_workspace_entities_organization_id" + + +def upgrade() -> None: + connection = op.get_bind() + inspector = sa.inspect(connection) + if not inspector.has_table(_ENTITIES_TABLE): + raise RuntimeError( + "workspace_entities must exist before workspace organization binding" + ) + if not inspector.has_table(_DOCUMENTS_TABLE): + raise RuntimeError( + "workspace_documents must exist before workspace organization binding" + ) + + entity_columns = { + column["name"] for column in inspector.get_columns(_ENTITIES_TABLE) + } + if _ORGANIZATION_COLUMN not in entity_columns: + op.add_column( + _ENTITIES_TABLE, + sa.Column(_ORGANIZATION_COLUMN, sa.String(), nullable=True), + ) + op.create_index( + _ORGANIZATION_INDEX, + _ENTITIES_TABLE, + [_ORGANIZATION_COLUMN], + if_not_exists=True, + ) + + workspace_entities = sa.table( + _ENTITIES_TABLE, + sa.column("workspace_id", sa.String()), + sa.column(_ORGANIZATION_COLUMN, sa.String()), + ) + workspace_documents = sa.table( + _DOCUMENTS_TABLE, + sa.column("workspace_id", sa.String()), + sa.column(_ORGANIZATION_COLUMN, sa.String()), + ) + + ownership_evidence = ( + sa.select( + workspace_documents.c.workspace_id.label("workspace_id"), + sa.func.min(workspace_documents.c.organization_id).label( + "organization_id" + ), + sa.func.count(sa.distinct(workspace_documents.c.organization_id)).label( + "organization_count" + ), + ) + .where(workspace_documents.c.organization_id.is_not(None)) + .group_by(workspace_documents.c.workspace_id) + .subquery() + ) + unambiguous_evidence = ( + sa.select( + ownership_evidence.c.workspace_id, + ownership_evidence.c.organization_id, + ) + .where(ownership_evidence.c.organization_count == 1) + .subquery() + ) + + inferred_organization = ( + sa.select(unambiguous_evidence.c.organization_id) + .where( + unambiguous_evidence.c.workspace_id + == workspace_entities.c.workspace_id + ) + .correlate(workspace_entities) + .scalar_subquery() + ) + has_unambiguous_evidence = ( + sa.exists( + sa.select(1) + .select_from(unambiguous_evidence) + .where( + unambiguous_evidence.c.workspace_id + == workspace_entities.c.workspace_id + ) + ) + .correlate(workspace_entities) + ) + connection.execute( + sa.update(workspace_entities) + .where( + workspace_entities.c.organization_id.is_(None), + has_unambiguous_evidence, + ) + .values(organization_id=inferred_organization) + ) + + bound_organization = ( + sa.select(workspace_entities.c.organization_id) + .where( + workspace_entities.c.workspace_id == workspace_documents.c.workspace_id, + workspace_entities.c.organization_id.is_not(None), + ) + .correlate(workspace_documents) + .scalar_subquery() + ) + has_binding = ( + sa.exists( + sa.select(1) + .select_from(workspace_entities) + .where( + workspace_entities.c.workspace_id + == workspace_documents.c.workspace_id, + workspace_entities.c.organization_id.is_not(None), + ) + ) + .correlate(workspace_documents) + ) + connection.execute( + sa.update(workspace_documents) + .where( + workspace_documents.c.organization_id.is_(None), + has_binding, + ) + .values(organization_id=bound_organization) + ) + + +def downgrade() -> None: + # Binding/backfill turns ambiguous legacy NULLs into explicit ownership + # provenance. A later downgrade cannot distinguish those values from normal + # application assignments, so removing them or the column could destroy + # security-relevant tenant evidence. + return None diff --git a/backend/alembic/versions/0021_workspace_personal_owner_binding.py b/backend/alembic/versions/0021_workspace_personal_owner_binding.py new file mode 100644 index 000000000..e6d978304 --- /dev/null +++ b/backend/alembic/versions/0021_workspace_personal_owner_binding.py @@ -0,0 +1,128 @@ +"""add fail-closed personal owner binding to the workspace registry + +Revision ID: 0021_workspace_personal_owner_binding +Revises: 0020_workspace_organization_binding +Create Date: 2026-09-15 00:51:00.000000 + +Personal workspaces have ``organization_id IS NULL`` and therefore cannot use +organization binding as their membership evidence. ``workspace_id`` is opaque, +so identifier shape such as ``workspace-`` is not ownership evidence. +This revision adds a nullable ``owner_user_id`` slot for trusted runtime +establishment while deliberately leaving historical rows unbound: the current +schema contains no auditable persisted user owner from which to backfill them. + +The registry may be unbound, organization-bound, or personal-user-bound, but it +must never be bound to both an organization and a personal user at once. +Downgrade removes the column only when no personal ownership provenance has +been written; otherwise it fails closed rather than destroying that evidence. +""" + +from alembic import op +import sqlalchemy as sa + +revision = "0021_workspace_personal_owner_binding" +down_revision = "0020_workspace_organization_binding" +branch_labels = None +depends_on = None + +_ENTITIES_TABLE = "workspace_entities" +_OWNER_USER_COLUMN = "owner_user_id" +_OWNER_USER_INDEX = "ix_workspace_entities_owner_user_id" +_SCOPE_OWNER_CHECK = "ck_workspace_entities_single_scope_owner" + + +def upgrade() -> None: + """Add personal-owner evidence without guessing ownership for legacy rows.""" + + connection = op.get_bind() + inspector = sa.inspect(connection) + if not inspector.has_table(_ENTITIES_TABLE): + raise RuntimeError( + "workspace_entities must exist before personal workspace binding" + ) + + entity_columns = { + column["name"] for column in inspector.get_columns(_ENTITIES_TABLE) + } + if "organization_id" not in entity_columns: + raise RuntimeError( + "organization_id binding must exist before personal workspace binding" + ) + if _OWNER_USER_COLUMN not in entity_columns: + op.add_column( + _ENTITIES_TABLE, + sa.Column(_OWNER_USER_COLUMN, sa.String(), nullable=True), + ) + + inspector = sa.inspect(connection) + index_names = { + index["name"] + for index in inspector.get_indexes(_ENTITIES_TABLE) + if index.get("name") + } + if _OWNER_USER_INDEX not in index_names: + op.create_index( + _OWNER_USER_INDEX, + _ENTITIES_TABLE, + [_OWNER_USER_COLUMN], + ) + + check_names = { + constraint["name"] + for constraint in inspector.get_check_constraints(_ENTITIES_TABLE) + if constraint.get("name") + } + if _SCOPE_OWNER_CHECK not in check_names: + op.create_check_constraint( + _SCOPE_OWNER_CHECK, + _ENTITIES_TABLE, + "NOT (organization_id IS NOT NULL AND owner_user_id IS NOT NULL)", + ) + + +def downgrade() -> None: + """Remove the owner column only while it carries no security provenance.""" + + connection = op.get_bind() + inspector = sa.inspect(connection) + if not inspector.has_table(_ENTITIES_TABLE): + return + + entity_columns = { + column["name"] for column in inspector.get_columns(_ENTITIES_TABLE) + } + if _OWNER_USER_COLUMN not in entity_columns: + return + + owner_count = connection.execute( + sa.text( + "SELECT count(*) FROM workspace_entities " + "WHERE owner_user_id IS NOT NULL" + ) + ).scalar_one() + if owner_count: + raise RuntimeError( + "cannot downgrade personal workspace binding while owner provenance exists" + ) + + check_names = { + constraint["name"] + for constraint in inspector.get_check_constraints(_ENTITIES_TABLE) + if constraint.get("name") + } + if _SCOPE_OWNER_CHECK in check_names: + op.drop_constraint( + _SCOPE_OWNER_CHECK, + _ENTITIES_TABLE, + type_="check", + ) + + inspector = sa.inspect(connection) + index_names = { + index["name"] + for index in inspector.get_indexes(_ENTITIES_TABLE) + if index.get("name") + } + if _OWNER_USER_INDEX in index_names: + op.drop_index(_OWNER_USER_INDEX, table_name=_ENTITIES_TABLE) + op.drop_column(_ENTITIES_TABLE, _OWNER_USER_COLUMN) diff --git a/backend/api/data.py b/backend/api/data.py index dccd85890..480849fb1 100644 --- a/backend/api/data.py +++ b/backend/api/data.py @@ -9,7 +9,7 @@ from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile from pydantic import BaseModel, ConfigDict, Field -from sqlalchemy import and_, case, func, or_, select +from sqlalchemy import and_, case, exists, func, or_, select from sqlalchemy.engine import Row from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import ColumnElement @@ -28,6 +28,7 @@ ProjectFolder, SenderRelationship, WebdavAccount, + Workspace, ) from db.session import get_db from services.attachment_parser import get_attachment_parser_manifest @@ -36,6 +37,11 @@ ) from services.ontology_service import ontology_service from services.webdav_service import webdav_service +from services.workspace_scope import ( + WorkspaceOrganizationBindingRequired, + WorkspaceOrganizationConflict, + get_or_create_scoped_workspace, +) router = APIRouter(prefix="/api/data", tags=["data"]) @@ -2315,8 +2321,6 @@ def _materialized_document_target_path(document: Document) -> str: return f"/Naruon/Data/{filename}" -# Document statuses whose stored content is not yet materializable parsed text -# (it may be a base64 binary payload awaiting a recognition/conversion worker). _NON_MATERIALIZABLE_DOCUMENT_STATUSES = frozenset( { PDF_DOM_RECOGNITION_PENDING_STATUS, @@ -2484,6 +2488,49 @@ async def _count_scalar(db: AsyncSession, statement) -> int: return int(result.scalar_one() or 0) +async def _require_scoped_workspace( + db: AsyncSession, + auth_context: AuthContext, +) -> Workspace: + try: + return await get_or_create_scoped_workspace( + db, + auth_context.workspace_id, + auth_context.organization_id, + owner_user_id=auth_context.user_id, + session_verifier=auth_context.session_verifier, + ) + except ( + WorkspaceOrganizationBindingRequired, + WorkspaceOrganizationConflict, + ) as exc: + raise HTTPException(status_code=403, detail="Workspace access denied") from exc + + +def _document_organization_filter(auth_context: AuthContext) -> ColumnElement: + if auth_context.organization_id is not None: + trusted_workspace_binding = exists( + select(1).select_from(Workspace).where( + Workspace.workspace_id == auth_context.workspace_id, + Workspace.organization_id == auth_context.organization_id, + Workspace.owner_user_id.is_(None), + ) + ) + return or_( + Document.organization_id == auth_context.organization_id, + and_(Document.organization_id.is_(None), trusted_workspace_binding), + ) + + trusted_workspace_binding = exists( + select(1).select_from(Workspace).where( + Workspace.workspace_id == auth_context.workspace_id, + Workspace.organization_id.is_(None), + Workspace.owner_user_id == auth_context.user_id, + ) + ) + return and_(Document.organization_id.is_(None), trusted_workspace_binding) + + async def _get_workspace_document( db: AsyncSession, auth_context: AuthContext, @@ -2493,6 +2540,7 @@ async def _get_workspace_document( select(Document).where( Document.document_id == document_id, Document.workspace_id == auth_context.workspace_id, + _document_organization_filter(auth_context), ) ) document = result.scalar_one_or_none() @@ -3166,6 +3214,7 @@ async def upload_data_document( auth_context: AuthContext = Depends(get_auth_context), db: AsyncSession = Depends(get_db), ) -> DataDocumentActionResponse: + await _require_scoped_workspace(db, auth_context) document = Document( workspace_id=auth_context.workspace_id, organization_id=auth_context.organization_id, @@ -3285,8 +3334,6 @@ async def create_document_pdf_dom_recognition_intent( ) async def upload_document_for_pdf_dom_recognition( file: UploadFile = File(...), - # Declared as multipart form data (not a query parameter) so a client - # sending document_name alongside the file is honored. document_name: str | None = Form(None), auth_context: AuthContext = Depends(get_auth_context), db: AsyncSession = Depends(get_db), @@ -3301,6 +3348,7 @@ async def upload_document_for_pdf_dom_recognition( status_code=415, detail="Only application/pdf uploads are supported for DOM recognition.", ) + await _require_scoped_workspace(db, auth_context) document = Document( workspace_id=auth_context.workspace_id, organization_id=auth_context.organization_id, @@ -3354,11 +3402,6 @@ async def create_document_webdav_materialization_intent( ) -> DataDocumentWebdavMaterializationResponse: document = await _get_workspace_document(db, auth_context, document_id) if document.document_status in _NON_MATERIALIZABLE_DOCUMENT_STATUSES: - # A document whose recognition/conversion is still pending holds a - # non-text payload (e.g. the base64 PDF stashed for the NewsDOM worker). - # Materializing it as Markdown would write that raw payload to the - # customer's WebDAV target. Refuse until recognition has landed real - # parsed text. raise HTTPException( status_code=409, detail=( @@ -3403,9 +3446,6 @@ async def _get_email_stats( db: AsyncSession, email_scope: EmailScopeFilter, ) -> EmailQualityStats: - # ⚡ Bolt Optimization: Batching scalar counts using CASE - # Impact: Reduces 7 sequential database queries down to 2, drastically cutting - # latency from network roundtrips when fetching quality surface metrics. email_stats_result = await db.execute( select( func.count(Email.id), @@ -3931,7 +3971,10 @@ async def get_data_quality_surface( documents = await _scoped_rows( db, select(Document) - .where(Document.workspace_id == auth_context.workspace_id) + .where( + Document.workspace_id == auth_context.workspace_id, + _document_organization_filter(auth_context), + ) .order_by(Document.created_at.desc(), Document.document_id.asc()) .limit(8), ) diff --git a/backend/db/models.py b/backend/db/models.py index 98e17eef2..725ecf2b0 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -1563,6 +1563,8 @@ class Workspace(Base): workspace_id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: f"workspace_{uuid.uuid4().hex}") workspace_name: Mapped[str] = mapped_column(String, nullable=False) workspace_domain: Mapped[str | None] = mapped_column(String, nullable=True) + organization_id: Mapped[str | None] = mapped_column(String, index=True, nullable=True) + owner_user_id: Mapped[str | None] = mapped_column(String, index=True, nullable=True) created_at: Mapped[datetime.datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.datetime.now(datetime.timezone.utc), diff --git a/backend/scripts/bootstrap_db.py b/backend/scripts/bootstrap_db.py index 1047103e8..3a1211181 100644 --- a/backend/scripts/bootstrap_db.py +++ b/backend/scripts/bootstrap_db.py @@ -1,16 +1,12 @@ import asyncio import os -from collections.abc import Sequence - -from sqlalchemy import Executable, text +from sqlalchemy import Executable, Index, MetaData, Table, inspect, text from sqlalchemy.engine import Connection from db.models import Base from db.session import engine INVALID_EMAIL_BACKFILL_OWNER_IDS = {None, "", "default"} - - def _static_bootstrap_sql(statement: str) -> Executable: # ponytail: repo-authored static bootstrap SQL only; bind params before runtime input. return text(statement) @@ -186,10 +182,6 @@ def _get_create_indexes_statements() -> list[Executable]: "CREATE INDEX IF NOT EXISTS ix_email_records_owner_date " "ON email_records (user_id, organization_id, date)" ), - text( - "CREATE INDEX IF NOT EXISTS ix_emails_owner_date " - "ON emails (user_id, organization_id, date)" - ), text( "CREATE INDEX IF NOT EXISTS ix_sender_relationships_owner_source " "ON sender_relationships " @@ -527,16 +519,24 @@ def schema_backfill_sql() -> list[Executable]: return statements -def _execute_statements(conn: Connection, statements: Sequence[Executable]) -> None: - for statement in statements: +def execute_schema_backfill(conn: Connection) -> None: + for statement in schema_backfill_sql(): conn.execute(statement) + if inspect(conn).has_table("emails"): + emails = Table("emails", MetaData(), autoload_with=conn) + Index( + "ix_emails_owner_date", + emails.c.user_id, + emails.c.organization_id, + emails.c.date, + ).create(conn, checkfirst=True) async def bootstrap_db() -> None: async with engine.begin() as conn: await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) await conn.run_sync(Base.metadata.create_all) - await conn.run_sync(_execute_statements, schema_backfill_sql()) + await conn.run_sync(execute_schema_backfill) if __name__ == "__main__": diff --git a/backend/services/workspace_scope.py b/backend/services/workspace_scope.py new file mode 100644 index 000000000..9a05fdc8c --- /dev/null +++ b/backend/services/workspace_scope.py @@ -0,0 +1,213 @@ +import datetime +from typing import Literal + +from sqlalchemy import DateTime, String, column, select, table, update +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import Workspace + +SessionVerifier = Literal["hmac", "oidc", "override", "server"] +_TRUSTED_BINDING_VERIFIERS = frozenset({"oidc", "override", "server"}) +_WORKSPACE_REGISTRY = table( + "workspace_entities", + column("workspace_id", String()), + column("workspace_name", String()), + column("organization_id", String()), + column("owner_user_id", String()), + column("created_at", DateTime(timezone=True)), +) + + +class WorkspaceOrganizationBindingRequired(RuntimeError): + """Raised when a session lacks trusted workspace ownership evidence.""" + + +class WorkspaceOrganizationConflict(RuntimeError): + """Raised when a workspace is already bound to a different tenant owner.""" + + +async def _workspace_scope_binding( + session: AsyncSession, + workspace_id: str, +) -> tuple[str | None, str | None]: + """Read organization and personal-user ownership for one opaque workspace.""" + + result = await session.execute( + select( + _WORKSPACE_REGISTRY.c.organization_id, + _WORKSPACE_REGISTRY.c.owner_user_id, + ).where(_WORKSPACE_REGISTRY.c.workspace_id == workspace_id) + ) + row = result.one_or_none() + if row is None: + return None, None + return row.organization_id, row.owner_user_id + + +async def _workspace_organization_binding( + session: AsyncSession, + workspace_id: str, +) -> str | None: + """Read the organization binding while preserving the legacy helper contract.""" + + organization_id, _owner_user_id = await _workspace_scope_binding( + session, + workspace_id, + ) + return organization_id + + +async def get_or_create_scoped_workspace( + session: AsyncSession, + workspace_id: str, + organization_id: str | None, + *, + owner_user_id: str | None = None, + session_verifier: SessionVerifier, +) -> Workspace: + """Return a workspace only after validating server-side tenant ownership. + + Organization scope binds the opaque workspace to ``organization_id`` and + deliberately does not bind it to one member. Personal scope has + ``organization_id is None`` and instead requires ``owner_user_id``. OIDC, + server, and explicit test override identities may establish an entirely + unbound registry row; HMAC compatibility sessions may consume existing + evidence but cannot claim ownership. Concurrent claims use insert/CAS + semantics so only one tenant owner can win. + """ + + if not workspace_id: + raise WorkspaceOrganizationBindingRequired("workspace claim is required") + if organization_id is None and not owner_user_id: + raise WorkspaceOrganizationBindingRequired( + "personal workspace claims require an authenticated owner user" + ) + + target_organization_id = organization_id + target_owner_user_id = owner_user_id if organization_id is None else None + + if session_verifier in _TRUSTED_BINDING_VERIFIERS: + await session.execute( + insert(_WORKSPACE_REGISTRY) + .values( + workspace_id=workspace_id, + workspace_name=workspace_id, + organization_id=target_organization_id, + owner_user_id=target_owner_user_id, + created_at=datetime.datetime.now(datetime.timezone.utc), + ) + .on_conflict_do_nothing( + index_elements=[_WORKSPACE_REGISTRY.c.workspace_id] + ) + ) + await session.execute( + update(_WORKSPACE_REGISTRY) + .where( + _WORKSPACE_REGISTRY.c.workspace_id == workspace_id, + _WORKSPACE_REGISTRY.c.organization_id.is_(None), + _WORKSPACE_REGISTRY.c.owner_user_id.is_(None), + ) + .values( + organization_id=target_organization_id, + owner_user_id=target_owner_user_id, + ) + ) + + bound_organization_id, bound_owner_user_id = await _workspace_scope_binding( + session, + workspace_id, + ) + if bound_organization_id is None and bound_owner_user_id is None: + raise WorkspaceOrganizationBindingRequired( + "workspace has no trusted tenant ownership binding" + ) + + if organization_id is None: + if bound_organization_id is not None or bound_owner_user_id != owner_user_id: + raise WorkspaceOrganizationConflict( + "personal workspace is bound to a different tenant owner" + ) + elif ( + bound_organization_id != organization_id + or bound_owner_user_id is not None + ): + raise WorkspaceOrganizationConflict( + "workspace is bound to a different tenant owner" + ) + + result = await session.execute( + select(Workspace).where(Workspace.workspace_id == workspace_id) + ) + return result.scalar_one() + + +async def get_or_create_bound_workspace( + session: AsyncSession, + workspace_id: str, + organization_id: str, + *, + session_verifier: SessionVerifier, +) -> Workspace: + """Compatibility wrapper for organization-scoped workspace ownership.""" + + if not organization_id: + raise WorkspaceOrganizationBindingRequired( + "workspace and organization claims are required for binding" + ) + return await get_or_create_scoped_workspace( + session, + workspace_id, + organization_id, + session_verifier=session_verifier, + ) + + +async def get_or_create_personal_workspace( + session: AsyncSession, + workspace_id: str, + owner_user_id: str, + *, + session_verifier: SessionVerifier, +) -> Workspace: + """Compatibility-safe entrypoint for a personal opaque workspace binding.""" + + return await get_or_create_scoped_workspace( + session, + workspace_id, + None, + owner_user_id=owner_user_id, + session_verifier=session_verifier, + ) + + +async def get_or_create_workspace( + session: AsyncSession, + workspace_id: str, +) -> Workspace: + """Return the ``Workspace`` row for ``workspace_id``, creating it first if needed. + + This compatibility path preserves callers that predate server-side tenant + binding. It establishes only the workspace foreign-key row and is not + authorization evidence. New tenant-sensitive callers must use + :func:`get_or_create_scoped_workspace` (or one of its scoped wrappers) with + authenticated tenant claims and verifier provenance. + """ + + result = await session.execute( + insert(Workspace) + .values( + workspace_id=workspace_id, + workspace_name=workspace_id, + created_at=datetime.datetime.now(datetime.timezone.utc), + ) + .on_conflict_do_nothing(index_elements=[Workspace.workspace_id]) + .returning(Workspace) + ) + workspace = result.scalar_one_or_none() + if workspace is None: + result = await session.execute( + select(Workspace).where(Workspace.workspace_id == workspace_id) + ) + workspace = result.scalar_one() + return workspace diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 6de67ea19..3eb056c98 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -16,6 +16,7 @@ from typing import cast from api.auth import AuthContext, RoleName, get_auth_context, get_current_user +from db.models import Workspace from main import app TEST_SCOPED_ROLES = { @@ -27,6 +28,38 @@ "member", } +_HMAC_DOCUMENT_COMPATIBILITY_TESTS = frozenset( + { + "test_data_document_upload_creates_workspace_scoped_document", + "test_data_document_actions_are_workspace_scoped_and_intent_only", + "test_data_pdf_dom_upload_persists_signed_organization_scope", + } +) + + +@pytest.fixture(autouse=True) +def persisted_hmac_workspace_for_document_compatibility(request): + """Model HMAC document access as consumption of existing owner evidence. + + The signed HMAC fixture proves token integrity only. These legacy success + cases exercise compatibility consumption, so their mock registry must already + contain the organization binding that production requires before access. + """ + if request.node.name not in _HMAC_DOCUMENT_COMPATIBILITY_TESTS: + yield + return + + mock_db = request.getfixturevalue("mock_db") + mock_db.workspaces.append( + Workspace( + workspace_id="workspace-org-acme", + workspace_name="workspace-org-acme", + organization_id="org-acme", + owner_user_id=None, + ) + ) + yield + def _normalize_header_value(value: str | None) -> str | None: if value is None: @@ -94,4 +127,4 @@ async def test_current_user( app.dependency_overrides[get_current_user] = test_current_user yield app.dependency_overrides.pop(get_auth_context, None) - app.dependency_overrides.pop(get_current_user, None) + app.dependency_overrides.pop(get_current_user, None) \ No newline at end of file diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index f8f3ffeae..19b4384b7 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -32,7 +32,40 @@ def test_initial_alembic_revision_records_current_schema_path(): assert "down_revision = None" in revision_text assert "CREATE EXTENSION IF NOT EXISTS vector" in revision_text assert "Base.metadata.create_all" in revision_text - assert "schema_backfill_sql" in revision_text + assert "execute_schema_backfill" in revision_text + + +def test_email_read_state_guards_both_legacy_and_current_table_names(): + """0011_email_read_state must add is_read to a genuinely historical + email_records table missing it (a database whose own 0001 ran before + is_read was added to the Email model), not just a legacy "emails" table + that, per 0011_email_model_reconciliation's docstring, no migration in + this repo's history ever actually created for a real managed database. + The upgrade check must guard on column existence, not just table + existence, so it stays idempotent against a table that already has the + column. downgrade is a no-op: a fresh database's email_records.is_read + comes from 0001's live create_all, not from this revision, so there is + no way to tell "this revision added it" apart from "the baseline already + had it" -- and is_read holds real read/unread state, not rebuildable + derived data (same ownership-ambiguity reasoning as + 0018_workspace_registry's downgrade).""" + revision_path = BACKEND_ROOT / "alembic" / "versions" / "0011_email_read_state.py" + revision_text = revision_path.read_text() + + assert '"email_records"' in revision_text + assert '"emails"' in revision_text + assert "has_table" in revision_text + assert "_has_column" in revision_text + assert "op.add_column(" in revision_text + assert "op.drop_column(" not in revision_text + + +def test_document_org_scope_downgrade_preserves_later_assignments(): + revision_path = ( + BACKEND_ROOT / "alembic" / "versions" / "0016_document_org_scope.py" + ) + + assert "op.drop_column(" not in revision_path.read_text() def test_provider_writeback_retry_queue_has_incremental_revision(): diff --git a/backend/tests/test_bootstrap_db.py b/backend/tests/test_bootstrap_db.py index 5af0540f0..95a4e72d9 100644 --- a/backend/tests/test_bootstrap_db.py +++ b/backend/tests/test_bootstrap_db.py @@ -6,7 +6,7 @@ from core.config import settings from db.models import Base -from scripts.bootstrap_db import schema_backfill_sql +from scripts.bootstrap_db import execute_schema_backfill, schema_backfill_sql from db.models import ( AgentRunRecord, CalendarWritebackSource, @@ -30,8 +30,7 @@ def _get_schema_statements(monkeypatch): def _execute_schema_backfill(sync_conn): - for statement in schema_backfill_sql(): - sync_conn.execute(statement) + execute_schema_backfill(sync_conn) def test_schema_backfill_adds_email_columns(monkeypatch): @@ -770,11 +769,11 @@ async def test_connector_signal_events_real_postgres_bootstrap_smoke(): text(""" INSERT INTO email_records ( user_id, organization_id, message_id, sender, recipients, - subject, "date", body + subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :sender, - :recipients, :subject, now(), :body + :recipients, :subject, now(), :body, true ) RETURNING id """), diff --git a/backend/tests/test_data_api.py b/backend/tests/test_data_api.py index cd0b7bf37..39df79883 100644 --- a/backend/tests/test_data_api.py +++ b/backend/tests/test_data_api.py @@ -28,6 +28,7 @@ Email, ProjectFolder, WebdavAccount, + Workspace, ) from db.session import get_db from main import app @@ -59,6 +60,7 @@ class MockAsyncSession: def __init__(self, results): self.results = results self.documents: list[Document] = [] + self.workspaces: list[Workspace] = [] self.queries = [] self.execute_calls = 0 @@ -66,6 +68,31 @@ async def execute(self, query): self.queries.append(query) rendered_query = str(query) rendered_query_lower = rendered_query.lower() + if "insert into workspace_entities" in rendered_query_lower: + compiled = query.compile() + params = compiled.params + workspace_id = next( + value + for key, value in params.items() + if key.startswith("workspace_id") + ) + workspace = next( + ( + workspace + for workspace in self.workspaces + if workspace.workspace_id == workspace_id + ), + None, + ) + if workspace is None: + workspace = Workspace( + workspace_id=workspace_id, + workspace_name=workspace_id, + created_at=_now(), + ) + self.workspaces.append(workspace) + return MockResult(workspace) + return MockResult(None) if ( "webdav_accounts.source_uid" in rendered_query_lower and "webdav_accounts.account_id" not in rendered_query_lower @@ -82,6 +109,26 @@ async def execute(self, query): for account in result ] ) + if "from workspace_entities" in rendered_query_lower: + compiled = query.compile() + params = compiled.params + workspace_id = next( + ( + value + for key, value in params.items() + if key.startswith("workspace_id") + ), + None, + ) + workspace = next( + ( + workspace + for workspace in self.workspaces + if workspace.workspace_id == workspace_id + ), + None, + ) + return MockResult(workspace) if "from workspace_documents" in rendered_query_lower: compiled = query.compile() params = compiled.params @@ -101,11 +148,25 @@ async def execute(self, query): ), None, ) + organization_id = next( + ( + value + for key, value in params.items() + if key.startswith("organization_id") + ), + None, + ) rows = [ document for document in self.documents if (document_id is None or document.document_id == document_id) and (workspace_id is None or document.workspace_id == workspace_id) + and ( + organization_id is None + # Older fixtures predate Document.organization_id; keep + # them usable while enforcing any explicit organization. + or document.organization_id in (None, organization_id) + ) ] if "order by" in rendered_query_lower: return MockResult(rows) @@ -121,10 +182,17 @@ def add(self, obj): if not obj.created_at: obj.created_at = _now() self.documents.append(obj) + elif isinstance(obj, Workspace): + if not obj.created_at: + obj.created_at = _now() + self.workspaces.append(obj) async def commit(self): pass + async def flush(self): + pass + async def refresh(self, obj): pass @@ -2484,6 +2552,7 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): Document( document_id="doc_owned", workspace_id="workspace-org-acme", + organization_id="org-acme", document_name="roadmap.md", document_type="text/markdown", document_content="# Roadmap", @@ -2499,6 +2568,16 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): document_status="uploaded", created_at=_now(), ), + Document( + document_id="doc_other_org", + workspace_id="workspace-org-acme", + organization_id="org-rival", + document_name="other-org.md", + document_type="text/markdown", + document_content="other organization", + document_status="uploaded", + created_at=_now(), + ), ] ) token = _signed_session_token(_valid_session_payload()) @@ -2544,6 +2623,7 @@ def test_data_quality_surface_includes_workspace_document_assets(mock_db): } ] assert "doc_rival" not in response.text + assert "doc_other_org" not in response.text def test_data_document_upload_creates_workspace_scoped_document(mock_db): @@ -2601,7 +2681,17 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): document_status="uploaded", created_at=_now(), ) - mock_db.documents.extend([document, rival_document]) + other_organization_document = Document( + document_id="doc_other_org", + workspace_id="workspace-org-acme", + organization_id="org-rival", + document_name="other-org.md", + document_type="text/markdown", + document_content="other organization", + document_status="uploaded", + created_at=_now(), + ) + mock_db.documents.extend([document, rival_document, other_organization_document]) token = _signed_session_token(_valid_session_payload()) client, previous_secret, original_overrides = _with_signed_auth(mock_db, token) try: @@ -2613,6 +2703,9 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): "/api/data/documents/doc_owned/hwp-conversion-intent" ) rival_response = client.post("/api/data/documents/doc_rival/reparse") + other_organization_response = client.post( + "/api/data/documents/doc_other_org/reparse" + ) finally: client.close() _restore_overrides(previous_secret, original_overrides) @@ -2638,6 +2731,8 @@ def test_data_document_actions_are_workspace_scoped_and_intent_only(mock_db): assert rival_response.status_code == 404 assert "doc_rival" not in rival_response.text + assert other_organization_response.status_code == 404 + assert "doc_other_org" not in other_organization_response.text def test_data_document_webdav_materialization_executes_source_backed_write( @@ -2951,11 +3046,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, thread_id, - fingerprint, sender, recipients, subject, "date", body + fingerprint, sender, recipients, subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :thread_id, - :fingerprint, :sender, :recipients, :subject, now(), :body + :fingerprint, :sender, :recipients, :subject, now(), :body, true ) RETURNING id """ @@ -2977,11 +3072,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, sender, recipients, - subject, "date", body + subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :sender, - :recipients, :subject, now(), :body + :recipients, :subject, now(), :body, true ) RETURNING id """ @@ -3001,11 +3096,11 @@ async def _seed_smoke_test_data(conn, ids: dict): """ INSERT INTO email_records ( user_id, organization_id, message_id, thread_id, - fingerprint, sender, recipients, subject, "date", body + fingerprint, sender, recipients, subject, "date", body, is_read ) VALUES ( :user_id, :organization_id, :message_id, :thread_id, - :fingerprint, :sender, :recipients, :subject, now(), :body + :fingerprint, :sender, :recipients, :subject, now(), :body, true ) RETURNING id """ diff --git a/backend/tests/test_document_organization_scope_opaque_workspace.py b/backend/tests/test_document_organization_scope_opaque_workspace.py new file mode 100644 index 000000000..a8080aa36 --- /dev/null +++ b/backend/tests/test_document_organization_scope_opaque_workspace.py @@ -0,0 +1,102 @@ +"""Regression contracts for opaque workspace document authorization.""" + +from api.auth import AuthContext +import api.data as data_api +from db.models import Document, Workspace +from sqlalchemy import and_, exists, or_, select + + +def _opaque_workspace_auth(*, organization_id: str = "org-acme") -> AuthContext: + """Build a signed-session shape whose opaque workspace is not org-derived.""" + + return AuthContext( + user_id="member-a", + role="member", + organization_id=organization_id, + group_ids=(), + workspace_id="tenant-space-7f3c", + ) + + +def _expected_organization_filter(auth_context: AuthContext): + """Describe the exact trusted binding required for historical NULL rows.""" + + assert hasattr(Workspace, "organization_id"), ( + "workspace_entities must persist organization_id before opaque workspace " + "claims can authorize organization-null documents" + ) + assert hasattr(Workspace, "owner_user_id"), ( + "organization workspace bindings must prove they are not personal-owner rows" + ) + trusted_workspace_binding = exists( + select(1) + .select_from(Workspace) + .where( + Workspace.workspace_id == auth_context.workspace_id, + Workspace.organization_id == auth_context.organization_id, + Workspace.owner_user_id.is_(None), + ) + ) + return or_( + Document.organization_id == auth_context.organization_id, + and_(Document.organization_id.is_(None), trusted_workspace_binding), + ) + + +def _compiled_document_scope(auth_context: AuthContext) -> tuple[str, dict[str, object]]: + """Compile the document scope so tenant-binding predicates stay observable.""" + + statement = select(Document.document_id).where( + Document.workspace_id == auth_context.workspace_id, + data_api._document_organization_filter(auth_context), + ) + compiled = statement.compile() + return str(statement.whereclause), compiled.params + + +def test_opaque_workspace_legacy_null_requires_correlated_organization_binding() -> None: + """Legacy NULL access requires one workspace row matching both signed claims.""" + + auth_context = _opaque_workspace_auth() + + assert data_api._document_organization_filter(auth_context).compare( + _expected_organization_filter(auth_context) + ) + + rendered, params = _compiled_document_scope(auth_context) + assert "workspace_documents.workspace_id" in rendered + assert "workspace_entities.workspace_id" in rendered + assert "workspace_entities.organization_id" in rendered + assert "workspace_entities.owner_user_id" in rendered + assert "IS NULL" in rendered.upper() + assert "EXISTS" in rendered.upper() + assert auth_context.workspace_id in params.values() + assert auth_context.organization_id in params.values() + + +def test_same_opaque_workspace_different_organization_cannot_share_null_branch() -> None: + """Each organization must correlate against its own registry binding.""" + + owner = _opaque_workspace_auth(organization_id="org-acme") + other = _opaque_workspace_auth(organization_id="org-other") + + owner_expected = _expected_organization_filter(owner) + other_expected = _expected_organization_filter(other) + + assert not owner_expected.compare(other_expected) + assert data_api._document_organization_filter(owner).compare(owner_expected) + assert data_api._document_organization_filter(other).compare(other_expected) + + owner_rendered, owner_params = _compiled_document_scope(owner) + other_rendered, other_params = _compiled_document_scope(other) + for rendered in (owner_rendered, other_rendered): + assert "workspace_entities.workspace_id" in rendered + assert "workspace_entities.organization_id" in rendered + assert "workspace_entities.owner_user_id" in rendered + assert "EXISTS" in rendered.upper() + assert "IS NULL" in rendered.upper() + + assert owner.organization_id in owner_params.values() + assert other.organization_id in other_params.values() + assert owner.workspace_id in owner_params.values() + assert other.workspace_id in other_params.values() diff --git a/backend/tests/test_document_personal_scope_opaque_workspace.py b/backend/tests/test_document_personal_scope_opaque_workspace.py new file mode 100644 index 000000000..2e94c1261 --- /dev/null +++ b/backend/tests/test_document_personal_scope_opaque_workspace.py @@ -0,0 +1,93 @@ +"""Regression contracts for personal-scope opaque workspace authorization.""" + +from api.auth import AuthContext +import api.data as data_api +from db.models import Document, Workspace +from sqlalchemy import and_, exists, select + + +def _personal_workspace_auth(*, user_id: str = "member-a") -> AuthContext: + """Build a personal signed-session shape with an opaque workspace claim.""" + + return AuthContext( + user_id=user_id, + role="member", + organization_id=None, + group_ids=(), + workspace_id="tenant-personal-7f3c", + ) + + +def _expected_personal_scope(auth_context: AuthContext): + """Require one registry row correlating the opaque workspace to its owner.""" + + assert hasattr(Workspace, "owner_user_id"), ( + "workspace_entities must persist owner_user_id before personal opaque " + "workspace claims can authorize organization-null documents" + ) + trusted_workspace_binding = exists( + select(1) + .select_from(Workspace) + .where( + Workspace.workspace_id == auth_context.workspace_id, + Workspace.organization_id.is_(None), + Workspace.owner_user_id == auth_context.user_id, + ) + ) + return and_(Document.organization_id.is_(None), trusted_workspace_binding) + + +def _compiled_document_scope(auth_context: AuthContext) -> tuple[str, dict[str, object]]: + """Compile the personal document scope so ownership correlation is observable.""" + + statement = select(Document.document_id).where( + Document.workspace_id == auth_context.workspace_id, + data_api._document_organization_filter(auth_context), + ) + compiled = statement.compile() + return str(statement.whereclause), compiled.params + + +def test_personal_opaque_workspace_requires_correlated_owner_user_binding() -> None: + """Personal NULL-organization access requires a server-side user binding.""" + + auth_context = _personal_workspace_auth() + expected = _expected_personal_scope(auth_context) + + assert data_api._document_organization_filter(auth_context).compare(expected) + + rendered, params = _compiled_document_scope(auth_context) + assert "workspace_documents.workspace_id" in rendered + assert "workspace_entities.workspace_id" in rendered + assert "workspace_entities.owner_user_id" in rendered + assert "workspace_entities.organization_id" in rendered + assert "IS NULL" in rendered.upper() + assert "EXISTS" in rendered.upper() + assert auth_context.workspace_id in params.values() + assert auth_context.user_id in params.values() + + +def test_same_personal_opaque_workspace_different_users_cannot_share_null_documents() -> None: + """Two signed users cannot share one personal opaque-workspace NULL branch.""" + + owner = _personal_workspace_auth(user_id="member-a") + other = _personal_workspace_auth(user_id="member-b") + owner_expected = _expected_personal_scope(owner) + other_expected = _expected_personal_scope(other) + + assert not owner_expected.compare(other_expected) + assert data_api._document_organization_filter(owner).compare(owner_expected) + assert data_api._document_organization_filter(other).compare(other_expected) + + owner_rendered, owner_params = _compiled_document_scope(owner) + other_rendered, other_params = _compiled_document_scope(other) + for rendered in (owner_rendered, other_rendered): + assert "workspace_entities.owner_user_id" in rendered + assert "workspace_entities.workspace_id" in rendered + assert "EXISTS" in rendered.upper() + assert "IS NULL" in rendered.upper() + + assert owner.user_id in owner_params.values() + assert other.user_id in other_params.values() + assert owner.workspace_id in owner_params.values() + assert other.workspace_id in other_params.values() diff --git a/backend/tests/test_document_scope_endpoints_postgres.py b/backend/tests/test_document_scope_endpoints_postgres.py new file mode 100644 index 000000000..76da8f81c --- /dev/null +++ b/backend/tests/test_document_scope_endpoints_postgres.py @@ -0,0 +1,384 @@ +"""Real PostgreSQL acceptance for document endpoint workspace authorization.""" + +import secrets +import subprocess +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +import pytest_asyncio +from fastapi import HTTPException +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext +from api.data import ( + DataDocumentUploadRequest, + reparse_data_document, + upload_data_document, +) +from core.config import settings + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] + + +def _run_migrations(database_url: str) -> None: + """Apply the managed migration path to an isolated PostgreSQL database.""" + + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), "head"], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest_asyncio.fixture +async def document_scope_database_url(): + """Create an isolated database; unavailable PostgreSQL is an acceptance failure.""" + + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_document_scope_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except Exception as exc: + pytest.fail( + f"PostgreSQL is required for document endpoint acceptance: {exc}", + pytrace=False, + ) + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + try: + _run_migrations(database_url) + yield database_url + finally: + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except Exception as exc: + pytest.fail( + f"PostgreSQL cleanup failed for document endpoint acceptance: {exc}", + pytrace=False, + ) + + +def _auth( + *, + user_id: str, + organization_id: str | None, + workspace_id: str, + session_verifier: str, +) -> AuthContext: + return AuthContext( + user_id=user_id, + role="member", + organization_id=organization_id, + group_ids=(), + workspace_id=workspace_id, + session_verifier=session_verifier, + ) + + +async def _workspace_binding( + database_url: str, + workspace_id: str, +) -> tuple[str | None, str | None] | None: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT organization_id, owner_user_id FROM workspace_entities " + "WHERE workspace_id = :workspace_id" + ), + {"workspace_id": workspace_id}, + ) + row = result.one_or_none() + if row is None: + return None + return row.organization_id, row.owner_user_id + finally: + await engine.dispose() + + +async def _document_count(database_url: str, workspace_id: str) -> int: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT count(*) FROM workspace_documents " + "WHERE workspace_id = :workspace_id" + ), + {"workspace_id": workspace_id}, + ) + return int(result.scalar_one()) + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_trusted_organization_upload_binds_workspace_and_rejects_other_org( + document_scope_database_url, +) -> None: + """One opaque workspace cannot accept document writes from two organizations.""" + + engine = create_async_engine(document_scope_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + workspace_id = "opaque-org-endpoint-7f3c" + request = DataDocumentUploadRequest( + document_name="evidence.md", + document_type="text/markdown", + document_content="source-backed evidence", + ) + try: + async with session_factory() as session: + response = await upload_data_document( + request=request, + auth_context=_auth( + user_id="member-a", + organization_id="org-acme", + workspace_id=workspace_id, + session_verifier="oidc", + ), + db=session, + ) + assert response.workspace_id == workspace_id + + assert await _workspace_binding( + document_scope_database_url, + workspace_id, + ) == ("org-acme", None) + assert await _document_count(document_scope_database_url, workspace_id) == 1 + + async with session_factory() as session: + with pytest.raises(HTTPException) as exc_info: + await upload_data_document( + request=request, + auth_context=_auth( + user_id="member-b", + organization_id="org-other", + workspace_id=workspace_id, + session_verifier="oidc", + ), + db=session, + ) + assert exc_info.value.status_code == 403 + assert await _workspace_binding( + document_scope_database_url, + workspace_id, + ) == ("org-acme", None) + assert await _document_count(document_scope_database_url, workspace_id) == 1 + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_hmac_cannot_establish_unbound_document_workspace( + document_scope_database_url, +) -> None: + """A signed compatibility session cannot turn an opaque claim into ownership.""" + + engine = create_async_engine(document_scope_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + workspace_id = "opaque-hmac-endpoint-unbound-4a2b" + try: + async with session_factory() as session: + with pytest.raises(HTTPException) as exc_info: + await upload_data_document( + request=DataDocumentUploadRequest( + document_name="blocked.md", + document_type="text/markdown", + document_content="must not persist", + ), + auth_context=_auth( + user_id="member-hmac", + organization_id="org-acme", + workspace_id=workspace_id, + session_verifier="hmac", + ), + db=session, + ) + assert exc_info.value.status_code == 403 + assert await _workspace_binding(document_scope_database_url, workspace_id) is None + assert await _document_count(document_scope_database_url, workspace_id) == 0 + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_trusted_personal_upload_binds_owner_and_rejects_other_user( + document_scope_database_url, +) -> None: + """Personal opaque workspaces persist one user owner rather than identifier shape.""" + + engine = create_async_engine(document_scope_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + workspace_id = "opaque-personal-endpoint-91d0" + request = DataDocumentUploadRequest( + document_name="personal.md", + document_type="text/markdown", + document_content="personal evidence", + ) + try: + async with session_factory() as session: + await upload_data_document( + request=request, + auth_context=_auth( + user_id="member-personal-a", + organization_id=None, + workspace_id=workspace_id, + session_verifier="server", + ), + db=session, + ) + + assert await _workspace_binding( + document_scope_database_url, + workspace_id, + ) == (None, "member-personal-a") + + async with session_factory() as session: + with pytest.raises(HTTPException) as exc_info: + await upload_data_document( + request=request, + auth_context=_auth( + user_id="member-personal-b", + organization_id=None, + workspace_id=workspace_id, + session_verifier="server", + ), + db=session, + ) + assert exc_info.value.status_code == 403 + assert await _document_count(document_scope_database_url, workspace_id) == 1 + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_historical_null_document_requires_correlated_registry_owner( + document_scope_database_url, +) -> None: + """Legacy NULL organization rows are readable only through the bound registry row.""" + + engine = create_async_engine(document_scope_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + org_workspace = "opaque-org-legacy-document-a6e1" + personal_workspace = "opaque-personal-legacy-document-f3b9" + try: + async with engine.begin() as connection: + await connection.execute( + text( + "INSERT INTO workspace_entities " + "(workspace_id, workspace_name, organization_id, owner_user_id, created_at) " + "VALUES " + "(:org_workspace, :org_workspace, 'org-acme', NULL, now()), " + "(:personal_workspace, :personal_workspace, NULL, 'member-a', now())" + ), + { + "org_workspace": org_workspace, + "personal_workspace": personal_workspace, + }, + ) + await connection.execute( + text( + "INSERT INTO workspace_documents " + "(document_id, workspace_id, organization_id, document_name, document_type, " + " document_content, document_status, created_at) VALUES " + "('legacy-org-doc', :org_workspace, NULL, 'legacy-org', 'text/plain', " + " 'org legacy text', 'uploaded', now()), " + "('legacy-personal-doc', :personal_workspace, NULL, 'legacy-personal', " + " 'text/plain', 'personal legacy text', 'uploaded', now())" + ), + { + "org_workspace": org_workspace, + "personal_workspace": personal_workspace, + }, + ) + + async with session_factory() as session: + response = await reparse_data_document( + document_id="legacy-org-doc", + auth_context=_auth( + user_id="member-a", + organization_id="org-acme", + workspace_id=org_workspace, + session_verifier="hmac", + ), + db=session, + ) + assert response.document_id == "legacy-org-doc" + + async with session_factory() as session: + with pytest.raises(HTTPException) as org_exc: + await reparse_data_document( + document_id="legacy-org-doc", + auth_context=_auth( + user_id="member-b", + organization_id="org-other", + workspace_id=org_workspace, + session_verifier="oidc", + ), + db=session, + ) + assert org_exc.value.status_code == 404 + + async with session_factory() as session: + response = await reparse_data_document( + document_id="legacy-personal-doc", + auth_context=_auth( + user_id="member-a", + organization_id=None, + workspace_id=personal_workspace, + session_verifier="hmac", + ), + db=session, + ) + assert response.document_id == "legacy-personal-doc" + + async with session_factory() as session: + with pytest.raises(HTTPException) as user_exc: + await reparse_data_document( + document_id="legacy-personal-doc", + auth_context=_auth( + user_id="member-b", + organization_id=None, + workspace_id=personal_workspace, + session_verifier="server", + ), + db=session, + ) + assert user_exc.value.status_code == 404 + finally: + await engine.dispose() diff --git a/backend/tests/test_email_read_state_migration_postgres.py b/backend/tests/test_email_read_state_migration_postgres.py new file mode 100644 index 000000000..d6294e69c --- /dev/null +++ b/backend/tests/test_email_read_state_migration_postgres.py @@ -0,0 +1,227 @@ +"""PostgreSQL regression coverage for 0011_email_read_state. + +String-matching the revision file's source (test_alembic_migrations.py) +cannot detect a destructive downgrade or prove the upgrade is actually +idempotent -- both require running the real migration against a real +database in each of the shapes it must handle. +""" + +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +from asyncpg.exceptions import InvalidAuthorizationSpecificationError, InvalidPasswordError +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import create_async_engine + +from core.config import settings + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +_PRE_READ_STATE_REVISION = "0009_project_graph_projection" + + +def _run_migrations(database_url: str, revision: str = "head") -> None: + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), revision], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"scripts/migrate_db.py {revision} failed " + f"(exit {result.returncode}):\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +def _run_downgrade(database_url: str, revision: str) -> None: + # scripts/migrate_db.py only wraps alembic's upgrade command; reuse its + # alembic_config() but call command.downgrade() directly for this test. + script = ( + "from scripts.migrate_db import alembic_config\n" + "from alembic import command\n" + f"command.downgrade(alembic_config(), {revision!r})\n" + ) + result = subprocess.run( + [sys.executable, "-c", script], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"alembic downgrade {revision} failed " + f"(exit {result.returncode}):\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +@pytest.fixture +def fresh_database_url(): + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_email_read_state_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + import asyncio + + try: + asyncio.run(admin(f'CREATE DATABASE "{database_name}"')) + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + try: + yield base_url.set(database=database_name).render_as_string(hide_password=False) + finally: + asyncio.run(admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)')) + + +async def _column_exists(database_url: str, table_name: str, column_name: str) -> bool: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = :table_name AND column_name = :column_name" + ), + {"table_name": table_name, "column_name": column_name}, + ) + return result.first() is not None + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_upgrade_adds_is_read_to_a_historical_email_records_table( + fresh_database_url, +): + """A database whose own 0001 ran before is_read existed in the Email + model has email_records without the column; upgrading to head must add + it, not silently leave it missing.""" + _run_migrations(fresh_database_url, revision=_PRE_READ_STATE_REVISION) + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text("ALTER TABLE email_records DROP COLUMN IF EXISTS is_read") + ) + finally: + await engine.dispose() + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is False + + _run_migrations(fresh_database_url) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + +@pytest.mark.asyncio +async def test_upgrade_is_idempotent_against_a_legacy_emails_table_with_is_read( + fresh_database_url, +): + """A legacy "emails" table that already has is_read (e.g. from a partial + earlier application) must not make upgrade() crash with a + duplicate-column error.""" + _run_migrations(fresh_database_url, revision=_PRE_READ_STATE_REVISION) + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text( + "CREATE TABLE emails " + "(id serial primary key, is_read boolean not null default true)" + ) + ) + finally: + await engine.dispose() + + _run_migrations(fresh_database_url) + + +@pytest.mark.asyncio +async def test_downgrade_does_not_destroy_read_state_on_a_fresh_database( + fresh_database_url, +): + """A fresh database's email_records.is_read comes from 0001's live + create_all, not from 0011_email_read_state -- downgrading past 0011 must + not drop it (and, if it did, would destroy real per-message read/unread + state along with it).""" + _run_migrations(fresh_database_url) + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + _run_downgrade(fresh_database_url, _PRE_READ_STATE_REVISION) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + +@pytest.mark.asyncio +async def test_upgrade_head_repairs_a_database_already_stamped_past_0011( + fresh_database_url, +): + """Alembic never re-runs a revision's upgrade() once that revision id is + recorded as applied -- editing 0011_email_read_state.py cannot repair a + database whose alembic_version history already includes it but is + missing is_read regardless (e.g. an earlier broken version of that + revision, a partial apply, manual intervention). 0019_email_read_state_ + repair is the real fix: it must add the column even though the database + is already stamped through 0018 with 0011 long since applied, so only + 0019 itself -- not a re-run of 0011 -- is what's left to bring it to + head.""" + _run_migrations(fresh_database_url, revision="0018_workspace_registry") + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True + + engine = create_async_engine(fresh_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text("ALTER TABLE email_records DROP COLUMN IF EXISTS is_read") + ) + finally: + await engine.dispose() + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is False + + _run_migrations(fresh_database_url) + + assert await _column_exists(fresh_database_url, "email_records", "is_read") is True diff --git a/backend/tests/test_legacy_document_scope_postgres.py b/backend/tests/test_legacy_document_scope_postgres.py new file mode 100644 index 000000000..059436d70 --- /dev/null +++ b/backend/tests/test_legacy_document_scope_postgres.py @@ -0,0 +1,190 @@ +"""PostgreSQL regression for legacy workspace documents with no organization id. + +Revision 0016 intentionally left pre-existing ``workspace_documents.organization_id`` +values NULL. Organization-scoped sessions may reach those rows only when the same +``workspace_entities`` row persists the authenticated organization binding. The +workspace identifier is opaque; its spelling is never ownership evidence. +""" + +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import httpx +import pytest +from asyncpg.exceptions import InvalidAuthorizationSpecificationError, InvalidPasswordError +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext, get_auth_context +from core.config import settings +from db.session import get_db, get_readonly_db +from main import app + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +_ORGANIZATION_ID = "legacy-document-org" +_WORKSPACE_ID = "opaque-legacy-document-workspace-7f3c" +_DOCUMENT_ID = "document_legacy_org_scope" + + +def _run_migrations(database_url: str) -> None: + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), "head"], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest.mark.asyncio +async def test_legacy_null_org_document_is_visible_only_to_matching_signed_org() -> None: + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_legacy_doc_scope_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + engine = create_async_engine(database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + + async def override_db(): + async with session_factory() as session: + yield session + + async def matching_auth() -> AuthContext: + return AuthContext( + user_id="legacy-document-user", + role="member", + organization_id=_ORGANIZATION_ID, + group_ids=(), + workspace_id=_WORKSPACE_ID, + ) + + async def other_org_same_workspace_auth() -> AuthContext: + return AuthContext( + user_id="other-user", + role="member", + organization_id="other-organization", + group_ids=(), + workspace_id=_WORKSPACE_ID, + ) + + try: + _run_migrations(database_url) + async with engine.begin() as connection: + await connection.execute( + text( + """ + INSERT INTO workspace_entities + (workspace_id, workspace_name, workspace_domain, + organization_id, owner_user_id, created_at) + VALUES + (:workspace_id, :workspace_name, NULL, + :organization_id, NULL, now()) + ON CONFLICT (workspace_id) DO NOTHING + """ + ), + { + "workspace_id": _WORKSPACE_ID, + "workspace_name": _WORKSPACE_ID, + "organization_id": _ORGANIZATION_ID, + }, + ) + await connection.execute( + text( + """ + INSERT INTO workspace_documents + (document_id, workspace_id, organization_id, document_name, + document_type, document_content, document_status, created_at) + VALUES + (:document_id, :workspace_id, NULL, :document_name, + 'text/markdown', '# Legacy', 'uploaded', now()) + """ + ), + { + "document_id": _DOCUMENT_ID, + "workspace_id": _WORKSPACE_ID, + "document_name": "legacy.md", + }, + ) + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[get_readonly_db] = override_db + app.dependency_overrides[get_auth_context] = matching_auth + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://testserver" + ) as client: + reparse_response = await client.post( + f"/api/data/documents/{_DOCUMENT_ID}/reparse" + ) + assert reparse_response.status_code == 200, reparse_response.text + + quality_response = await client.get("/api/data/quality-surface") + assert quality_response.status_code == 200, quality_response.text + assert _DOCUMENT_ID in { + asset["asset_key"] for asset in quality_response.json()["repository_assets"] + } + + app.dependency_overrides[get_auth_context] = other_org_same_workspace_auth + denied_response = await client.post( + f"/api/data/documents/{_DOCUMENT_ID}/reparse" + ) + assert denied_response.status_code == 404 + + other_quality_response = await client.get("/api/data/quality-surface") + assert other_quality_response.status_code == 200, other_quality_response.text + assert _DOCUMENT_ID not in { + asset["asset_key"] + for asset in other_quality_response.json()["repository_assets"] + } + finally: + app.dependency_overrides.pop(get_db, None) + app.dependency_overrides.pop(get_readonly_db, None) + app.dependency_overrides.pop(get_auth_context, None) + await engine.dispose() + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except (OSError, ConnectionError): + # Best-effort teardown: a transient connectivity error here must not + # mask the test's actual assertions. + pass diff --git a/backend/tests/test_workspace_document_migration.py b/backend/tests/test_workspace_document_migration.py new file mode 100644 index 000000000..f685675b7 --- /dev/null +++ b/backend/tests/test_workspace_document_migration.py @@ -0,0 +1,259 @@ +"""Regression coverage for the missing ``workspace_entities``/``workspace_documents`` +Alembic migration. + +``Workspace``/``Document`` have been declared in ``db/models.py`` since before +this repository's incremental migration history tracked them explicitly (see +``alembic/versions/0018_workspace_registry.py``). No production code path ever +inserted a ``Workspace`` row for a real signed session either, so +``Document.workspace_id``'s foreign key could never be satisfied by a real +``/api/data/documents`` upload. A database missing these tables also used to +crash on ``0016_document_org_scope`` (``NoSuchTableError``) before ever +reaching ``0018_workspace_registry``; ``0016`` is now ``has_table``-guarded. + +These tests exercise the actual documented production path +(``scripts/migrate_db.py`` -> ``alembic upgrade head``, never +``Base.metadata.create_all``) against a real, disposable PostgreSQL database, +then call the real ``/api/data/documents`` endpoints through the real FastAPI +app with only the database session swapped for one bound to that database. +""" + +import asyncio +import subprocess +import secrets +import sys +import uuid +from pathlib import Path + +import asyncpg +import httpx +import pytest +import pytest_asyncio +from asyncpg.exceptions import ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, +) +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext, get_auth_context +from core.config import settings +from db.session import get_db, get_readonly_db +from main import app + +pytestmark = pytest.mark.postgres + +BACKEND_ROOT = Path(__file__).resolve().parents[1] +# The revision immediately before 0016_document_org_scope, which -- for a +# database missing workspace_documents -- is the first migration in the +# chain that touches the table at all. Stopping here (rather than at 0017, +# after 0016 has already run) is what actually exercises the real historical +# gap: 0016 must not crash before 0018_workspace_registry ever gets to run. +_PRE_REGISTRY_REVISION = "0015_merge_newsdom_email_heads" +_SMOKE_WORKSPACE_ID = "workspace-workspace-migration-smoke-org" + + +def _run_migrations(database_url: str, revision: str = "head") -> None: + result = subprocess.run( + [sys.executable, str(BACKEND_ROOT / "scripts" / "migrate_db.py"), revision], + cwd=BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, ( + f"scripts/migrate_db.py {revision} failed " + f"(exit {result.returncode}):\n" + f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert not any( + status in result.stdout + result.stderr + for status in ("Timeout", "Fatal", "Warn", "Denied") + ) + + +@pytest_asyncio.fixture +async def fresh_database_url(): + base_url = make_url(settings.DATABASE_URL) + test_db_name = f"test_workspace_doc_{uuid.uuid4().hex[:16]}" + + async def _admin(sql: str) -> None: + conn = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await conn.execute(sql) + finally: + await conn.close() + + try: + await _admin(f'CREATE DATABASE "{test_db_name}"') + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OSError, + ConnectionError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + + try: + yield base_url.set(database=test_db_name).render_as_string( + hide_password=False + ) + finally: + await _admin(f'DROP DATABASE IF EXISTS "{test_db_name}" WITH (FORCE)') + + +@pytest_asyncio.fixture +async def migrated_client(fresh_database_url): + engine = create_async_engine(fresh_database_url) + sessionmaker = async_sessionmaker(engine, expire_on_commit=False) + + async def override_db(): + async with sessionmaker() as session: + yield session + + async def override_auth_context() -> AuthContext: + return AuthContext( + user_id="workspace_migration_smoke_user", + role="member", + organization_id="workspace-migration-smoke-org", + group_ids=(), + workspace_id=_SMOKE_WORKSPACE_ID, + ) + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[get_readonly_db] = override_db + app.dependency_overrides[get_auth_context] = override_auth_context + try: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://testserver" + ) as client: + yield client + finally: + app.dependency_overrides.pop(get_db, None) + app.dependency_overrides.pop(get_readonly_db, None) + app.dependency_overrides.pop(get_auth_context, None) + await engine.dispose() + + +async def _table_exists(database_url: str, name: str) -> bool: + engine = create_async_engine(database_url) + try: + async with engine.connect() as conn: + result = await conn.execute( + text("SELECT to_regclass(:name) IS NOT NULL"), {"name": name} + ) + return bool(result.scalar()) + finally: + await engine.dispose() + + +async def _drop_workspace_registry_tables(database_url: str) -> None: + engine = create_async_engine(database_url) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP TABLE IF EXISTS workspace_documents")) + await conn.execute(text("DROP TABLE IF EXISTS workspace_entities")) + finally: + await engine.dispose() + + +async def _assert_document_upload_serves_cleanly(client: httpx.AsyncClient) -> None: + response = await client.post( + "/api/data/documents", + json={ + "document_name": "roadmap.md", + "document_type": "text/markdown", + "document_content": "# Roadmap", + }, + ) + assert response.status_code == 200, response.text + body = response.json() + assert body["workspace_id"] == _SMOKE_WORKSPACE_ID + assert body["document_id"] + + # A second upload from the same signed workspace must not fail trying to + # re-insert the already-provisioned Workspace row (workspace_id is its + # primary key). + second_response = await client.post( + "/api/data/documents", + json={ + "document_name": "notes.md", + "document_type": "text/markdown", + "document_content": "# Notes", + }, + ) + assert second_response.status_code == 200, second_response.text + + +@pytest.mark.asyncio +async def test_document_upload_serves_after_full_head_migration_from_empty( + fresh_database_url, migrated_client +): + """A brand-new database migrated straight to head must be able to serve + /api/data/documents without a missing-relation or FK-violation error.""" + _run_migrations(fresh_database_url) + await _assert_document_upload_serves_cleanly(migrated_client) + + +@pytest.mark.asyncio +async def test_concurrent_first_uploads_provision_one_workspace( + fresh_database_url, migrated_client +): + """Concurrent first requests must not race on the workspace primary key.""" + _run_migrations(fresh_database_url) + + responses = await asyncio.gather( + *( + migrated_client.post( + "/api/data/documents", + json={ + "document_name": f"concurrent-{index}.md", + "document_type": "text/markdown", + "document_content": "# Concurrent", + }, + ) + for index in range(16) + ) + ) + + assert [response.status_code for response in responses] == [200] * 16 + + +@pytest.mark.asyncio +async def test_document_upload_serves_after_upgrading_a_pre_registry_database( + fresh_database_url, migrated_client +): + """Reproduce the exact reported gap: a real, already-incrementally-migrated + production database that was provisioned before ``Workspace``/``Document`` + existed in ``db/models.py`` never gets ``workspace_entities``/ + ``workspace_documents`` created by any migration prior to + ``0018_workspace_registry`` (``0001_initial_control_plane``'s + ``Base.metadata.create_all`` only reflects *today's* model metadata, so it + cannot recreate that historical, pre-model-addition state on its own). + Force that end state directly -- dropping the tables a stopped-at-0015 + database would never have had -- then migrate straight to head in one + call, crossing 0016_document_org_scope (which used to crash with + NoSuchTableError on a database in exactly this state) before + 0018_workspace_registry ever runs. Prove that both tables end up correct + and /api/data/documents serves cleanly.""" + _run_migrations(fresh_database_url, revision=_PRE_REGISTRY_REVISION) + + await _drop_workspace_registry_tables(fresh_database_url) + assert await _table_exists(fresh_database_url, "workspace_entities") is False + assert await _table_exists(fresh_database_url, "workspace_documents") is False + + _run_migrations(fresh_database_url) + + assert await _table_exists(fresh_database_url, "workspace_entities") is True + assert await _table_exists(fresh_database_url, "workspace_documents") is True + await _assert_document_upload_serves_cleanly(migrated_client) diff --git a/backend/tests/test_workspace_organization_binding_migration_postgres.py b/backend/tests/test_workspace_organization_binding_migration_postgres.py new file mode 100644 index 000000000..ae604703d --- /dev/null +++ b/backend/tests/test_workspace_organization_binding_migration_postgres.py @@ -0,0 +1,203 @@ +"""PostgreSQL acceptance for auditable workspace-organization binding. + +The workspace identifier is an opaque authenticated claim. Historical ownership +must therefore be recovered only from server-side evidence already persisted in +``workspace_documents.organization_id``; identifier shape is not ownership +evidence. Ambiguous and evidence-free workspaces stay unbound so compatibility +access can fail closed instead of guessing a tenant. +""" + +import secrets +import subprocess +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +import pytest_asyncio +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import create_async_engine + +from core.config import settings + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +_PRE_BINDING_REVISION = "0019_email_read_state_repair" + + +def _run_migrations(database_url: str, revision: str = "head") -> None: + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), revision], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest_asyncio.fixture +async def fresh_database_url(): + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_workspace_binding_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except Exception as exc: + pytest.fail( + f"PostgreSQL is required for workspace binding acceptance: {exc}", + pytrace=False, + ) + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + try: + yield database_url + finally: + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except Exception as exc: + pytest.fail( + f"PostgreSQL cleanup failed for workspace binding acceptance: {exc}", + pytrace=False, + ) + + +async def _seed_historical_workspace_evidence(database_url: str) -> None: + engine = create_async_engine(database_url) + try: + async with engine.begin() as connection: + for workspace_id in ( + "opaque-unambiguous-7f3c", + "opaque-ambiguous-8a4d", + "opaque-unbound-9b5e", + ): + await connection.execute( + text( + """ + INSERT INTO workspace_entities + (workspace_id, workspace_name, workspace_domain, created_at) + VALUES (:workspace_id, :workspace_id, NULL, now()) + """ + ), + {"workspace_id": workspace_id}, + ) + + documents = ( + ( + "doc-unambiguous-known", + "opaque-unambiguous-7f3c", + "org-acme", + ), + ( + "doc-unambiguous-legacy", + "opaque-unambiguous-7f3c", + None, + ), + ("doc-ambiguous-a", "opaque-ambiguous-8a4d", "org-red"), + ("doc-ambiguous-b", "opaque-ambiguous-8a4d", "org-blue"), + ("doc-ambiguous-legacy", "opaque-ambiguous-8a4d", None), + ("doc-unbound-legacy", "opaque-unbound-9b5e", None), + ) + for document_id, workspace_id, organization_id in documents: + await connection.execute( + text( + """ + INSERT INTO workspace_documents + (document_id, workspace_id, organization_id, + document_name, document_type, document_content, + document_status, created_at) + VALUES + (:document_id, :workspace_id, :organization_id, + :document_id, 'text/markdown', '# historical', + 'uploaded', now()) + """ + ), + { + "document_id": document_id, + "workspace_id": workspace_id, + "organization_id": organization_id, + }, + ) + finally: + await engine.dispose() + + +async def _read_bindings(database_url: str) -> tuple[dict[str, str | None], dict[str, str | None]]: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + workspace_result = await connection.execute( + text( + "SELECT workspace_id, organization_id " + "FROM workspace_entities ORDER BY workspace_id" + ) + ) + document_result = await connection.execute( + text( + "SELECT document_id, organization_id " + "FROM workspace_documents ORDER BY document_id" + ) + ) + return ( + {row.workspace_id: row.organization_id for row in workspace_result}, + {row.document_id: row.organization_id for row in document_result}, + ) + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_binding_migration_uses_only_unambiguous_persisted_ownership_evidence( + fresh_database_url, +) -> None: + """Bind/backfill one-owner history and leave ambiguous or unknown history closed.""" + + _run_migrations(fresh_database_url, revision=_PRE_BINDING_REVISION) + await _seed_historical_workspace_evidence(fresh_database_url) + + _run_migrations(fresh_database_url) + workspace_bindings, document_bindings = await _read_bindings(fresh_database_url) + + assert workspace_bindings["opaque-unambiguous-7f3c"] == "org-acme" + assert document_bindings["doc-unambiguous-legacy"] == "org-acme" + + assert workspace_bindings["opaque-ambiguous-8a4d"] is None + assert document_bindings["doc-ambiguous-legacy"] is None + + assert workspace_bindings["opaque-unbound-9b5e"] is None + assert document_bindings["doc-unbound-legacy"] is None + + # A repeated managed upgrade is a required deployment path. It must not + # invent new ownership or mutate the fail-closed ambiguous/unbound rows. + _run_migrations(fresh_database_url) + repeated_workspace_bindings, repeated_document_bindings = await _read_bindings( + fresh_database_url + ) + assert repeated_workspace_bindings == workspace_bindings + assert repeated_document_bindings == document_bindings diff --git a/backend/tests/test_workspace_personal_scope_binding_postgres.py b/backend/tests/test_workspace_personal_scope_binding_postgres.py new file mode 100644 index 000000000..6103be9df --- /dev/null +++ b/backend/tests/test_workspace_personal_scope_binding_postgres.py @@ -0,0 +1,270 @@ +"""Real PostgreSQL acceptance for personal opaque-workspace ownership binding.""" + +import asyncio +import secrets +import subprocess +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +import pytest_asyncio +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from core.config import settings +from services.workspace_scope import ( + WorkspaceOrganizationBindingRequired, + WorkspaceOrganizationConflict, + get_or_create_personal_workspace, +) + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] + + +def _run_migrations(database_url: str) -> None: + """Apply the managed migration path to an isolated PostgreSQL database.""" + + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), "head"], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest_asyncio.fixture +async def personal_binding_database_url(): + """Create an isolated database; unavailable PostgreSQL is an acceptance failure.""" + + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_personal_workspace_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + """Execute one database-administration statement outside the test database.""" + + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except Exception as exc: + pytest.fail( + f"PostgreSQL is required for personal workspace acceptance: {exc}", + pytrace=False, + ) + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + try: + _run_migrations(database_url) + yield database_url + finally: + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except Exception as exc: + pytest.fail( + f"PostgreSQL cleanup failed for personal workspace acceptance: {exc}", + pytrace=False, + ) + + +async def _read_binding( + database_url: str, + workspace_id: str, +) -> tuple[str | None, str | None] | None: + """Read registry ownership independently of the service transaction.""" + + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT organization_id, owner_user_id FROM workspace_entities " + "WHERE workspace_id = :workspace_id" + ), + {"workspace_id": workspace_id}, + ) + row = result.one_or_none() + if row is None: + return None + return row.organization_id, row.owner_user_id + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_personal_binding_requires_trusted_establishment_and_rejects_other_user( + personal_binding_database_url, +) -> None: + """HMAC may consume personal ownership but cannot establish or reassign it.""" + + engine = create_async_engine(personal_binding_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with session_factory() as session: + with pytest.raises(WorkspaceOrganizationBindingRequired): + async with session.begin(): + await get_or_create_personal_workspace( + session, + "opaque-personal-hmac-unbound-1", + "member-a", + session_verifier="hmac", + ) + + assert ( + await _read_binding( + personal_binding_database_url, + "opaque-personal-hmac-unbound-1", + ) + is None + ) + + async with session_factory() as session: + async with session.begin(): + created = await get_or_create_personal_workspace( + session, + "opaque-personal-trusted-7f3c", + "member-a", + session_verifier="override", + ) + assert created.workspace_id == "opaque-personal-trusted-7f3c" + + assert await _read_binding( + personal_binding_database_url, + "opaque-personal-trusted-7f3c", + ) == (None, "member-a") + + async with session_factory() as session: + async with session.begin(): + reused = await get_or_create_personal_workspace( + session, + "opaque-personal-trusted-7f3c", + "member-a", + session_verifier="hmac", + ) + assert reused.workspace_id == "opaque-personal-trusted-7f3c" + + async with session_factory() as session: + with pytest.raises(WorkspaceOrganizationConflict): + async with session.begin(): + await get_or_create_personal_workspace( + session, + "opaque-personal-trusted-7f3c", + "member-b", + session_verifier="oidc", + ) + + assert await _read_binding( + personal_binding_database_url, + "opaque-personal-trusted-7f3c", + ) == (None, "member-a") + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_concurrent_personal_claims_have_one_owner_winner( + personal_binding_database_url, +) -> None: + """Concurrent users cannot both claim the same personal opaque workspace.""" + + engine = create_async_engine(personal_binding_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + workspace_id = "opaque-personal-race-8a4d" + + async def attempt(owner_user_id: str): + """Attempt one transactional personal binding and return row or conflict.""" + + try: + async with session_factory() as session: + async with session.begin(): + return await get_or_create_personal_workspace( + session, + workspace_id, + owner_user_id, + session_verifier="oidc", + ) + except WorkspaceOrganizationConflict as exc: + return exc + + try: + results = await asyncio.gather(attempt("member-red"), attempt("member-blue")) + conflicts = [ + result + for result in results + if isinstance(result, WorkspaceOrganizationConflict) + ] + winners = [result for result in results if not isinstance(result, Exception)] + + assert len(conflicts) == 1 + assert len(winners) == 1 + persisted = await _read_binding(personal_binding_database_url, workspace_id) + assert persisted in {(None, "member-red"), (None, "member-blue")} + assert winners[0].workspace_id == workspace_id + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_personal_binding_cannot_replace_organization_binding( + personal_binding_database_url, +) -> None: + """Personal ownership cannot cross the registry's organization-bound invariant.""" + + engine = create_async_engine(personal_binding_database_url) + try: + async with engine.begin() as connection: + await connection.execute( + text( + "INSERT INTO workspace_entities " + "(workspace_id, workspace_name, organization_id, owner_user_id, created_at) " + "VALUES (:workspace_id, :workspace_id, :organization_id, NULL, now())" + ), + { + "workspace_id": "opaque-org-owned-9b5e", + "organization_id": "org-acme", + }, + ) + + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with session_factory() as session: + with pytest.raises(WorkspaceOrganizationConflict): + async with session.begin(): + await get_or_create_personal_workspace( + session, + "opaque-org-owned-9b5e", + "member-a", + session_verifier="server", + ) + + assert await _read_binding( + personal_binding_database_url, + "opaque-org-owned-9b5e", + ) == ("org-acme", None) + finally: + await engine.dispose() diff --git a/backend/tests/test_workspace_scope_binding_postgres.py b/backend/tests/test_workspace_scope_binding_postgres.py new file mode 100644 index 000000000..b56cc1f4d --- /dev/null +++ b/backend/tests/test_workspace_scope_binding_postgres.py @@ -0,0 +1,208 @@ +"""Real PostgreSQL acceptance for server-side workspace organization binding.""" + +import asyncio +import secrets +import subprocess +import sys +import uuid +from pathlib import Path + +import asyncpg +import pytest +import pytest_asyncio +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from core.config import settings +from services.workspace_scope import ( + WorkspaceOrganizationBindingRequired, + WorkspaceOrganizationConflict, + get_or_create_bound_workspace, +) + +pytestmark = pytest.mark.postgres + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] + + +def _run_migrations(database_url: str) -> None: + """Apply the managed migration path with only required bootstrap settings.""" + + result = subprocess.run( + [sys.executable, str(_BACKEND_ROOT / "scripts" / "migrate_db.py"), "head"], + cwd=_BACKEND_ROOT, + env={ + "DATABASE_URL": database_url, + "AUTH_SESSION_HMAC_SECRET": secrets.token_urlsafe(48), + }, + capture_output=True, + text=True, + timeout=180, + ) + output = result.stdout + result.stderr + assert result.returncode == 0, output + assert not any( + status in output for status in ("Timeout", "Fatal", "Warn", "Denied") + ), output + + +@pytest_asyncio.fixture +async def binding_database_url(): + """Create an isolated database; unavailable PostgreSQL is an acceptance failure.""" + + base_url = make_url(settings.DATABASE_URL) + database_name = f"test_workspace_scope_{uuid.uuid4().hex[:12]}" + + async def admin(sql: str) -> None: + connection = await asyncpg.connect( + host=base_url.host, + port=base_url.port, + user=base_url.username, + password=base_url.password, + database="postgres", + ) + try: + await connection.execute(sql) + finally: + await connection.close() + + try: + await admin(f'CREATE DATABASE "{database_name}"') + except Exception as exc: + pytest.fail( + f"PostgreSQL is required for workspace binding acceptance: {exc}", + pytrace=False, + ) + + database_url = base_url.set(database=database_name).render_as_string( + hide_password=False + ) + try: + _run_migrations(database_url) + yield database_url + finally: + try: + await admin(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)') + except Exception as exc: + pytest.fail( + f"PostgreSQL cleanup failed for workspace binding acceptance: {exc}", + pytrace=False, + ) + + +async def _read_binding(database_url: str, workspace_id: str) -> str | None: + """Read the persisted binding independently of the service session.""" + + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + result = await connection.execute( + text( + "SELECT organization_id FROM workspace_entities " + "WHERE workspace_id = :workspace_id" + ), + {"workspace_id": workspace_id}, + ) + return result.scalar_one_or_none() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_binding_requires_trusted_establishment_and_rejects_mismatch( + binding_database_url, +) -> None: + """HMAC may consume existing evidence but cannot create or change ownership.""" + + engine = create_async_engine(binding_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with session_factory() as session: + with pytest.raises(WorkspaceOrganizationBindingRequired): + async with session.begin(): + await get_or_create_bound_workspace( + session, + "opaque-hmac-unbound-1", + "org-acme", + session_verifier="hmac", + ) + + assert await _read_binding(binding_database_url, "opaque-hmac-unbound-1") is None + + async with session_factory() as session: + async with session.begin(): + created = await get_or_create_bound_workspace( + session, + "opaque-trusted-7f3c", + "org-acme", + session_verifier="override", + ) + assert created.workspace_id == "opaque-trusted-7f3c" + + assert await _read_binding(binding_database_url, "opaque-trusted-7f3c") == "org-acme" + + async with session_factory() as session: + async with session.begin(): + reused = await get_or_create_bound_workspace( + session, + "opaque-trusted-7f3c", + "org-acme", + session_verifier="hmac", + ) + assert reused.workspace_id == "opaque-trusted-7f3c" + + async with session_factory() as session: + with pytest.raises(WorkspaceOrganizationConflict): + async with session.begin(): + await get_or_create_bound_workspace( + session, + "opaque-trusted-7f3c", + "org-other", + session_verifier="oidc", + ) + + assert await _read_binding(binding_database_url, "opaque-trusted-7f3c") == "org-acme" + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_concurrent_trusted_claims_have_one_binding_winner( + binding_database_url, +) -> None: + """Concurrent organizations cannot both claim the same opaque workspace.""" + + engine = create_async_engine(binding_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + workspace_id = "opaque-race-8a4d" + + async def attempt(organization_id: str): + """Attempt one transactional binding and return either row or conflict.""" + + try: + async with session_factory() as session: + async with session.begin(): + return await get_or_create_bound_workspace( + session, + workspace_id, + organization_id, + session_verifier="oidc", + ) + except WorkspaceOrganizationConflict as exc: + return exc + + try: + results = await asyncio.gather(attempt("org-red"), attempt("org-blue")) + conflicts = [ + result for result in results if isinstance(result, WorkspaceOrganizationConflict) + ] + winners = [result for result in results if not isinstance(result, Exception)] + + assert len(conflicts) == 1 + assert len(winners) == 1 + persisted = await _read_binding(binding_database_url, workspace_id) + assert persisted in {"org-red", "org-blue"} + assert winners[0].workspace_id == workspace_id + finally: + await engine.dispose()