diff --git a/.github/workflows/app-ci.yml b/.github/workflows/app-ci.yml index e8f445748..c3ce0fb39 100644 --- a/.github/workflows/app-ci.yml +++ b/.github/workflows/app-ci.yml @@ -129,9 +129,14 @@ jobs: cd frontend && pnpm run build - name: Install Playwright Chromium - run: cd frontend && pnpm exec playwright install --with-deps chromium + timeout-minutes: 10 + # The hosted Ubuntu runner already provides the system libraries needed by + # Chromium. Installing them again with --with-deps can block on apt for + # hours and turn a real frontend result into a cancelled check. + run: cd frontend && pnpm exec playwright install chromium - name: Run full product smoke + timeout-minutes: 15 env: NEXT_TELEMETRY_DISABLED: "1" NARUON_FULL_PRODUCT_BASE_URL: "http://127.0.0.1:3001" diff --git a/backend/alembic/versions/0018_disksage_file_lineage.py b/backend/alembic/versions/0018_disksage_file_lineage.py new file mode 100644 index 000000000..ae44860cf --- /dev/null +++ b/backend/alembic/versions/0018_disksage_file_lineage.py @@ -0,0 +1,86 @@ +"""persist scoped, encrypted DiskSage file lineage envelopes + +Revision ID: 0018_disksage_file_lineage +Revises: 0017_merge_newsdom_carddav_heads +Create Date: 2026-08-13 00:00:00.000000 + +The Rust DiskSage verifier remains authoritative for copy/provider proof. Naruon +stores the validated envelope for workspace-scoped provenance and exposes only +the redacted graph projection in list responses. +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0018_disksage_file_lineage" +down_revision = "0017_merge_newsdom_carddav_heads" +branch_labels = None +depends_on = None + +_TABLE = "disksage_file_lineage_records" + + +def upgrade() -> None: + connection = op.get_bind() + inspector = sa.inspect(connection) + if not inspector.has_table(_TABLE): + op.create_table( + _TABLE, + sa.Column("lineage_record_uid", sa.String(length=96), nullable=False), + sa.Column("user_id", sa.String(), nullable=False), + sa.Column("organization_id", sa.String(), nullable=True), + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("lineage_fingerprint", sa.String(length=64), nullable=False), + sa.Column("envelope_sha256", sa.String(length=64), nullable=False), + sa.Column("schema_version", sa.Integer(), nullable=False), + sa.Column("schema_kind", sa.String(length=96), nullable=False), + sa.Column("source_kind", sa.String(length=64), nullable=False), + sa.Column("archive_kind", sa.String(length=64), nullable=False), + sa.Column("raw_content_sha256", sa.String(length=64), nullable=False), + sa.Column("raw_content_blake3", sa.String(length=64), nullable=False), + sa.Column("content_bytes", sa.BigInteger(), nullable=False), + sa.Column("ontology_class", sa.String(length=256), nullable=False), + sa.Column("ontology_relation_count", sa.Integer(), nullable=False), + sa.Column("ontology_predicates", sa.JSON(), nullable=False), + sa.Column("provider_name", sa.String(length=32), nullable=False), + sa.Column("provider_sync_confirmed", sa.Boolean(), nullable=False), + sa.Column( + "provider_sync_state", + sa.String(length=32), + nullable=False, + server_default="unknown", + ), + sa.Column("envelope_json_encrypted", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("lineage_record_uid"), + sa.UniqueConstraint( + "user_id", + "workspace_id", + "lineage_fingerprint", + name="uq_disksage_lineage_workspace_fingerprint", + ), + ) + + op.create_index( + "ix_disksage_lineage_scope_time", + _TABLE, + ["user_id", "organization_id", "workspace_id", "created_at"], + if_not_exists=True, + ) + op.create_index( + "ix_disksage_lineage_ontology_class", + _TABLE, + ["workspace_id", "ontology_class"], + if_not_exists=True, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_disksage_lineage_ontology_class", table_name=_TABLE, if_exists=True + ) + op.drop_index("ix_disksage_lineage_scope_time", table_name=_TABLE, if_exists=True) + connection = op.get_bind() + if sa.inspect(connection).has_table(_TABLE): + op.drop_table(_TABLE) diff --git a/backend/alembic/versions/0019_disksage_provider_sync_state.py b/backend/alembic/versions/0019_disksage_provider_sync_state.py new file mode 100644 index 000000000..314a8b2ba --- /dev/null +++ b/backend/alembic/versions/0019_disksage_provider_sync_state.py @@ -0,0 +1,43 @@ +"""persist provider-native sync state for DiskSage lineage envelopes + +Revision ID: 0019_disksage_sync_state +Revises: 0018_disksage_file_lineage +Create Date: 2026-08-13 00:00:00.000000 +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0019_disksage_sync_state" +down_revision = "0018_disksage_file_lineage" +branch_labels = None +depends_on = None + +_TABLE = "disksage_file_lineage_records" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_TABLE) and not any( + column["name"] == "provider_sync_state" + for column in inspector.get_columns(_TABLE) + ): + op.add_column( + _TABLE, + sa.Column( + "provider_sync_state", + sa.String(length=32), + nullable=False, + server_default="unknown", + ), + ) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_TABLE) and any( + column["name"] == "provider_sync_state" + for column in inspector.get_columns(_TABLE) + ): + op.drop_column(_TABLE, "provider_sync_state") diff --git a/backend/alembic/versions/0020_disksage_lineage_column_names.py b/backend/alembic/versions/0020_disksage_lineage_column_names.py new file mode 100644 index 000000000..326a69452 --- /dev/null +++ b/backend/alembic/versions/0020_disksage_lineage_column_names.py @@ -0,0 +1,39 @@ +"""rename ambiguous DiskSage lineage column names + +Revision ID: 0020_disksage_columns +Revises: 0019_disksage_sync_state +Create Date: 2026-08-13 00:00:00.000000 +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0020_disksage_columns" +down_revision = "0019_disksage_sync_state" +branch_labels = None +depends_on = None + +_TABLE = "disksage_file_lineage_records" + + +def _rename_if_present(inspector: sa.Inspector, old: str, new: str) -> None: + columns = {column["name"] for column in inspector.get_columns(_TABLE)} + if old in columns and new not in columns: + op.alter_column(_TABLE, old, new_column_name=new) + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_TABLE): + _rename_if_present(inspector, "bytes", "content_bytes") + inspector = sa.inspect(op.get_bind()) + _rename_if_present(inspector, "provider", "provider_name") + + +def downgrade() -> None: + # This revision repairs legacy installations that still have the old + # single-word names. Revision 0019 and the canonical 0018 schema both + # use content_bytes/provider_name, so a downgrade must not reintroduce + # the legacy shape or leave the model and migration chain inconsistent. + pass diff --git a/backend/alembic/versions/0021_disksage_lineage_scope_index.py b/backend/alembic/versions/0021_disksage_lineage_scope_index.py new file mode 100644 index 000000000..bfb6c33eb --- /dev/null +++ b/backend/alembic/versions/0021_disksage_lineage_scope_index.py @@ -0,0 +1,39 @@ +"""align the DiskSage lineage scope index with the list query""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0021_disksage_scope_idx" +down_revision = "0020_disksage_columns" +branch_labels = None +depends_on = None + +_TABLE = "disksage_file_lineage_records" +_INDEX = "ix_disksage_lineage_scope_time" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table(_TABLE): + return + op.drop_index(_INDEX, table_name=_TABLE, if_exists=True) + op.create_index( + _INDEX, + _TABLE, + ["user_id", "workspace_id", "created_at"], + if_not_exists=True, + ) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table(_TABLE): + return + op.drop_index(_INDEX, table_name=_TABLE, if_exists=True) + op.create_index( + _INDEX, + _TABLE, + ["user_id", "organization_id", "workspace_id", "created_at"], + if_not_exists=True, + ) diff --git a/backend/alembic/versions/0022_disksage_organization_lineage.py b/backend/alembic/versions/0022_disksage_organization_lineage.py new file mode 100644 index 000000000..6df7adeee --- /dev/null +++ b/backend/alembic/versions/0022_disksage_organization_lineage.py @@ -0,0 +1,51 @@ +"""persist path-free DiskSage organization lineage batches""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0022_disksage_org_lineage" +down_revision = "0021_disksage_scope_idx" +branch_labels = None +depends_on = None + +_TABLE = "disksage_organization_lineage_records" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table(_TABLE): + return + op.create_table( + _TABLE, + sa.Column("organization_lineage_record_uid", sa.String(length=96), nullable=False), + sa.Column("user_id", sa.String(), nullable=False), + sa.Column("organization_id", sa.String(), nullable=True), + sa.Column("workspace_id", sa.String(), nullable=False), + sa.Column("batch_fingerprint_sha256", sa.String(length=64), nullable=False), + sa.Column("envelope_sha256", sa.String(length=64), nullable=False), + sa.Column("schema_version", sa.Integer(), nullable=False), + sa.Column("item_count", sa.Integer(), nullable=False), + sa.Column("ontology_classes", sa.JSON(), nullable=False), + sa.Column("envelope_json_encrypted", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("organization_lineage_record_uid"), + sa.UniqueConstraint( + "user_id", + "workspace_id", + "batch_fingerprint_sha256", + name="uq_disksage_org_lineage_workspace_fingerprint", + ), + ) + op.create_index( + "ix_disksage_org_lineage_scope_time", + _TABLE, + ["user_id", "workspace_id", "created_at"], + if_not_exists=True, + ) + + +def downgrade() -> None: + op.drop_index("ix_disksage_org_lineage_scope_time", table_name=_TABLE, if_exists=True) + if sa.inspect(op.get_bind()).has_table(_TABLE): + op.drop_table(_TABLE) diff --git a/backend/api/dav.py b/backend/api/dav.py index d618c25f3..b20ec7c0e 100644 --- a/backend/api/dav.py +++ b/backend/api/dav.py @@ -159,6 +159,7 @@ async def _handle_project_propfind( @router.api_route( "/{path:path}", methods=["PROPFIND", "REPORT", "MKCOL", "GET", "PUT", "DELETE", "OPTIONS"], + include_in_schema=False, ) async def dav_handler( request: Request, diff --git a/backend/api/disksage.py b/backend/api/disksage.py new file mode 100644 index 000000000..f07451b20 --- /dev/null +++ b/backend/api/disksage.py @@ -0,0 +1,325 @@ +import datetime +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.exc import IntegrityError, StatementError + +from api.auth import AuthContext, get_auth_context +from db.models import ( + DiskSageFileLineageRecord, + DiskSageOrganizationLineageRecord, +) +from db.session import get_db +from services.disksage_file_lineage import ( + FileLineageEnvelope, + FileLineageSummary, + canonical_envelope_json, + canonical_envelope_sha256, + ontology_predicates, +) +from services.disksage_organization_lineage import ( + OrganizationLineageBatch, + OrganizationLineageSummary, + canonical_batch_json, + canonical_batch_sha256, +) +from core.runtime_secrets import EncryptionKeyMissingError + +router = APIRouter(prefix="/api/disksage", tags=["disksage"]) + + +def _summary(record: DiskSageFileLineageRecord) -> FileLineageSummary: + return FileLineageSummary( + lineage_record_uid=record.lineage_record_uid, + lineage_fingerprint=record.lineage_fingerprint, + schema_version=record.schema_version, + source_kind=record.source_kind, + archive_kind=record.archive_kind, + raw_content_sha256=record.raw_content_sha256, + raw_content_blake3=record.raw_content_blake3, + content_bytes=record.content_bytes, + ontology_class=record.ontology_class, + ontology_relation_count=record.ontology_relation_count, + ontology_predicates=list(record.ontology_predicates or []), + provider_name=record.provider_name, + provider_sync_confirmed=record.provider_sync_confirmed, + provider_sync_state=record.provider_sync_state, + created_at=record.created_at.isoformat(), + ) + + +def _encrypted_envelope_json(envelope: FileLineageEnvelope) -> str: + return canonical_envelope_json(envelope) + + +def _lineage_scope(auth_context: AuthContext) -> tuple[object, ...]: + organization_filter = ( + DiskSageFileLineageRecord.organization_id == auth_context.organization_id + if auth_context.organization_id is not None + else DiskSageFileLineageRecord.organization_id.is_(None) + ) + return ( + DiskSageFileLineageRecord.user_id == auth_context.user_id, + organization_filter, + DiskSageFileLineageRecord.workspace_id == auth_context.workspace_id, + ) + + +def _lineage_identity_scope(auth_context: AuthContext) -> tuple[object, ...]: + """Return the columns covered by the lineage uniqueness constraint.""" + + return ( + DiskSageFileLineageRecord.user_id == auth_context.user_id, + DiskSageFileLineageRecord.workspace_id == auth_context.workspace_id, + ) + + +def _organization_lineage_scope(auth_context: AuthContext) -> tuple[object, ...]: + organization_filter = ( + DiskSageOrganizationLineageRecord.organization_id == auth_context.organization_id + if auth_context.organization_id is not None + else DiskSageOrganizationLineageRecord.organization_id.is_(None) + ) + return ( + DiskSageOrganizationLineageRecord.user_id == auth_context.user_id, + organization_filter, + DiskSageOrganizationLineageRecord.workspace_id == auth_context.workspace_id, + ) + + +def _organization_lineage_identity_scope(auth_context: AuthContext) -> tuple[object, ...]: + """Return the columns covered by the organization-lineage uniqueness constraint.""" + + return ( + DiskSageOrganizationLineageRecord.user_id == auth_context.user_id, + DiskSageOrganizationLineageRecord.workspace_id == auth_context.workspace_id, + ) + + +def _organization_summary( + record: DiskSageOrganizationLineageRecord, +) -> OrganizationLineageSummary: + return OrganizationLineageSummary( + organization_lineage_record_uid=record.organization_lineage_record_uid, + batch_fingerprint_sha256=record.batch_fingerprint_sha256, + schema_version=record.schema_version, + item_count=record.item_count, + ontology_classes=list(record.ontology_classes or []), + created_at=record.created_at.isoformat(), + ) + + +def _is_missing_encryption_key(error: BaseException) -> bool: + """Recognize direct and SQLAlchemy-wrapped missing-key failures.""" + + return isinstance(error, EncryptionKeyMissingError) or ( + isinstance(error, StatementError) + and isinstance(error.orig, EncryptionKeyMissingError) + ) + + +@router.post("/file-lineage", response_model=FileLineageSummary, status_code=201) +async def ingest_file_lineage( + envelope: FileLineageEnvelope, + auth_context: AuthContext = Depends(get_auth_context), + db: AsyncSession = Depends(get_db), +) -> FileLineageSummary: + """Persist one validated DiskSage envelope without authorizing source eviction.""" + + envelope_sha256 = canonical_envelope_sha256(envelope) + lineage_fingerprint = envelope.cloud_copy.lineage_fingerprint + existing_result = await db.execute( + select(DiskSageFileLineageRecord).where( + *_lineage_scope(auth_context), + DiskSageFileLineageRecord.lineage_fingerprint == lineage_fingerprint, + ) + ) + existing = existing_result.scalar_one_or_none() + if existing is not None: + if existing.envelope_sha256 != envelope_sha256: + raise HTTPException( + status_code=409, + detail="lineage fingerprint is already bound to different evidence", + ) + return _summary(existing) + + try: + record = DiskSageFileLineageRecord( + lineage_record_uid=f"disksage_lineage_{uuid.uuid4().hex}", + user_id=auth_context.user_id, + organization_id=auth_context.organization_id, + workspace_id=auth_context.workspace_id, + lineage_fingerprint=lineage_fingerprint, + envelope_sha256=envelope_sha256, + schema_version=envelope.schema_version, + schema_kind=envelope.schema_kind, + source_kind=envelope.source_kind, + archive_kind=envelope.archive_kind, + raw_content_sha256=envelope.raw_content_sha256, + raw_content_blake3=envelope.raw_content_blake3, + content_bytes=envelope.bytes, + ontology_class=envelope.ontology_class, + ontology_relation_count=len(envelope.ontology_relations), + ontology_predicates=ontology_predicates(envelope), + provider_name=envelope.cloud_copy.provider, + provider_sync_confirmed=envelope.cloud_copy.provider_sync_confirmed, + provider_sync_state=envelope.cloud_copy.provider_sync_state or "unknown", + envelope_json_encrypted=_encrypted_envelope_json(envelope), + created_at=datetime.datetime.now(datetime.timezone.utc), + ) + db.add(record) + await db.commit() + await db.refresh(record) + except IntegrityError: + await db.rollback() + replayed_result = await db.execute( + select(DiskSageFileLineageRecord).where( + *_lineage_identity_scope(auth_context), + DiskSageFileLineageRecord.lineage_fingerprint == lineage_fingerprint, + ) + ) + replayed = replayed_result.scalar_one_or_none() + if replayed is None: + raise + if replayed.organization_id != auth_context.organization_id: + raise HTTPException( + status_code=409, + detail="lineage fingerprint is already bound to a different organization", + ) from None + if replayed.envelope_sha256 != envelope_sha256: + raise HTTPException( + status_code=409, + detail="lineage fingerprint is already bound to different evidence", + ) from None + return _summary(replayed) + except (EncryptionKeyMissingError, StatementError) as error: + if not _is_missing_encryption_key(error): + raise + await db.rollback() + raise HTTPException( + status_code=503, + detail="Server encryption key is not configured. Contact your workspace administrator.", + ) from error + return _summary(record) + + +@router.get("/file-lineage", response_model=list[FileLineageSummary]) +async def list_file_lineage( + limit: int = Query(default=50, ge=1, le=100), + auth_context: AuthContext = Depends(get_auth_context), + db: AsyncSession = Depends(get_db), +) -> list[FileLineageSummary]: + """List redacted lineage graph projections; encrypted envelope values stay server-side.""" + + result = await db.execute( + select(DiskSageFileLineageRecord) + .where( + *_lineage_scope(auth_context), + ) + .order_by(DiskSageFileLineageRecord.created_at.desc()) + .limit(limit) + ) + return [_summary(record) for record in result.scalars().all()] + + +@router.post( + "/organization-lineage", + response_model=OrganizationLineageSummary, + status_code=201, +) +async def ingest_organization_lineage( + envelope: OrganizationLineageBatch, + auth_context: AuthContext = Depends(get_auth_context), + db: AsyncSession = Depends(get_db), +) -> OrganizationLineageSummary: + """Persist a path-free ontology organization plan without authorizing moves.""" + + envelope_sha256 = canonical_batch_sha256(envelope) + batch_fingerprint = envelope.batch_fingerprint_sha256 + existing_result = await db.execute( + select(DiskSageOrganizationLineageRecord).where( + *_organization_lineage_scope(auth_context), + DiskSageOrganizationLineageRecord.batch_fingerprint_sha256 + == batch_fingerprint, + ) + ) + existing = existing_result.scalar_one_or_none() + if existing is not None: + if existing.envelope_sha256 != envelope_sha256: + raise HTTPException( + status_code=409, + detail="organization lineage fingerprint is already bound to different evidence", + ) + return _organization_summary(existing) + + try: + record = DiskSageOrganizationLineageRecord( + organization_lineage_record_uid=f"disksage_org_lineage_{uuid.uuid4().hex}", + user_id=auth_context.user_id, + organization_id=auth_context.organization_id, + workspace_id=auth_context.workspace_id, + batch_fingerprint_sha256=batch_fingerprint, + envelope_sha256=envelope_sha256, + schema_version=envelope.version, + item_count=len(envelope.items), + ontology_classes=sorted({item.ontology_class for item in envelope.items}), + envelope_json_encrypted=canonical_batch_json(envelope), + created_at=datetime.datetime.now(datetime.timezone.utc), + ) + db.add(record) + await db.commit() + await db.refresh(record) + except IntegrityError: + await db.rollback() + replayed_result = await db.execute( + select(DiskSageOrganizationLineageRecord).where( + *_organization_lineage_identity_scope(auth_context), + DiskSageOrganizationLineageRecord.batch_fingerprint_sha256 + == batch_fingerprint, + ) + ) + replayed = replayed_result.scalar_one_or_none() + if replayed is None: + raise + if replayed.organization_id != auth_context.organization_id: + raise HTTPException( + status_code=409, + detail="organization lineage fingerprint is already bound to a different organization", + ) from None + if replayed.envelope_sha256 != envelope_sha256: + raise HTTPException( + status_code=409, + detail="organization lineage fingerprint is already bound to different evidence", + ) from None + return _organization_summary(replayed) + except (EncryptionKeyMissingError, StatementError) as error: + if not _is_missing_encryption_key(error): + raise + await db.rollback() + raise HTTPException( + status_code=503, + detail="Server encryption key is not configured. Contact your workspace administrator.", + ) from error + return _organization_summary(record) + + +@router.get( + "/organization-lineage", + response_model=list[OrganizationLineageSummary], +) +async def list_organization_lineage( + limit: int = Query(default=50, ge=1, le=100), + auth_context: AuthContext = Depends(get_auth_context), + db: AsyncSession = Depends(get_db), +) -> list[OrganizationLineageSummary]: + """List redacted organization lineage summaries; paths remain encrypted.""" + + result = await db.execute( + select(DiskSageOrganizationLineageRecord) + .where(*_organization_lineage_scope(auth_context)) + .order_by(DiskSageOrganizationLineageRecord.created_at.desc()) + .limit(limit) + ) + return [_organization_summary(record) for record in result.scalars().all()] diff --git a/backend/core/runtime_secrets.py b/backend/core/runtime_secrets.py index 1bf1496d1..d6c820117 100644 --- a/backend/core/runtime_secrets.py +++ b/backend/core/runtime_secrets.py @@ -18,6 +18,12 @@ ENCRYPTION_KEY_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$") +class EncryptionKeyMissingError(RuntimeError): + """Raised when encryption is requested without the configured active key.""" + + error_code = "encryption-key-missing" + + @dataclass(frozen=True) class RuntimeEncryptionKey: key_id: str @@ -147,7 +153,7 @@ def build_encryption_keyring( previous_keys_value: str | None = None, ) -> EncryptionKeyRing: if active_key_value is None or not active_key_value.strip(): - raise RuntimeError( + raise EncryptionKeyMissingError( "ENCRYPTION_KEY is required. Refusing to encrypt without a configured key." ) diff --git a/backend/db/models.py b/backend/db/models.py index 98e17eef2..2db34ddb7 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -7,6 +7,7 @@ from cryptography.fernet import Fernet, InvalidToken from pgvector.sqlalchemy import Vector from sqlalchemy import ( + BigInteger, Boolean, DateTime, ForeignKey, @@ -23,7 +24,11 @@ from sqlalchemy.types import TypeDecorator from core.config import settings -from core.runtime_secrets import EncryptionKeyRing, build_encryption_keyring +from core.runtime_secrets import ( + EncryptionKeyMissingError, + EncryptionKeyRing, + build_encryption_keyring, +) logger = logging.getLogger(__name__) @@ -33,7 +38,7 @@ def _validated_fernet_key() -> bytes: if settings.ENCRYPTION_KEY is None: - raise RuntimeError( + raise EncryptionKeyMissingError( "ENCRYPTION_KEY is required. Refusing to encrypt without a configured key." ) @@ -1102,6 +1107,108 @@ class KnowledgeGraphEdgeRecord(Base): ) +class DiskSageFileLineageRecord(Base): + """Encrypted DiskSage provenance scoped to the authenticated workspace.""" + + __tablename__ = "disksage_file_lineage_records" + __table_args__ = ( + UniqueConstraint( + "user_id", + "workspace_id", + "lineage_fingerprint", + name="uq_disksage_lineage_workspace_fingerprint", + ), + Index( + "ix_disksage_lineage_scope_time", + "user_id", + "workspace_id", + "created_at", + ), + Index("ix_disksage_lineage_ontology_class", "workspace_id", "ontology_class"), + ) + + lineage_record_uid: Mapped[str] = mapped_column(String(96), primary_key=True) + user_id: Mapped[str] = mapped_column(String, index=True, nullable=False) + organization_id: Mapped[str | None] = mapped_column( + String, index=True, nullable=True + ) + workspace_id: Mapped[str] = mapped_column(String, index=True, nullable=False) + lineage_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + envelope_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + schema_version: Mapped[int] = mapped_column(Integer, nullable=False) + schema_kind: Mapped[str] = mapped_column(String(96), nullable=False) + source_kind: Mapped[str] = mapped_column(String(64), nullable=False) + archive_kind: Mapped[str] = mapped_column(String(64), nullable=False) + raw_content_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + raw_content_blake3: Mapped[str] = mapped_column(String(64), nullable=False) + content_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False) + ontology_class: Mapped[str] = mapped_column(String(256), nullable=False) + ontology_relation_count: Mapped[int] = mapped_column(Integer, nullable=False) + ontology_predicates: Mapped[list[str]] = mapped_column( + JSON, default=list, nullable=False + ) + provider_name: Mapped[str] = mapped_column(String(32), nullable=False) + provider_sync_confirmed: Mapped[bool] = mapped_column(Boolean, nullable=False) + provider_sync_state: Mapped[str] = mapped_column( + String(32), nullable=False, default="unknown", server_default="unknown" + ) + # Source paths and metadata evidence are intentionally not queryable plaintext. + envelope_json_encrypted: Mapped[str] = mapped_column( + EncryptedString, nullable=False + ) + created_at: Mapped[datetime.datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.datetime.now(datetime.timezone.utc), + nullable=False, + ) + + +class DiskSageOrganizationLineageRecord(Base): + """Encrypted, path-free local organization lineage scoped to a workspace.""" + + __tablename__ = "disksage_organization_lineage_records" + __table_args__ = ( + UniqueConstraint( + "user_id", + "workspace_id", + "batch_fingerprint_sha256", + name="uq_disksage_org_lineage_workspace_fingerprint", + ), + Index( + "ix_disksage_org_lineage_scope_time", + "user_id", + "workspace_id", + "created_at", + ), + ) + + organization_lineage_record_uid: Mapped[str] = mapped_column( + String(96), primary_key=True + ) + user_id: Mapped[str] = mapped_column(String, index=True, nullable=False) + organization_id: Mapped[str | None] = mapped_column( + String, index=True, nullable=True + ) + workspace_id: Mapped[str] = mapped_column(String, index=True, nullable=False) + batch_fingerprint_sha256: Mapped[str] = mapped_column( + String(64), nullable=False + ) + envelope_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + schema_version: Mapped[int] = mapped_column(Integer, nullable=False) + item_count: Mapped[int] = mapped_column(Integer, nullable=False) + ontology_classes: Mapped[list[str]] = mapped_column( + JSON, default=list, nullable=False + ) + envelope_json_encrypted: Mapped[str] = mapped_column( + EncryptedString, nullable=False + ) + created_at: Mapped[datetime.datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.datetime.now(datetime.timezone.utc), + nullable=False, + ) + + class ProjectGraphObjectRecord(Base): __tablename__ = "project_graph_objects" __table_args__ = ( diff --git a/backend/main.py b/backend/main.py index 51b054dbf..0569673a6 100644 --- a/backend/main.py +++ b/backend/main.py @@ -32,6 +32,7 @@ from api.ai_hub import router as ai_hub_router from api.projects import router as projects_router from api.session import router as auth_session_router +from api.disksage import router as disksage_router from core.config import canonical_origin, settings from core.telemetry import setup_telemetry from core.version import get_release_version @@ -239,6 +240,7 @@ async def add_security_headers(request: Request, call_next): app.include_router(ai_hub_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(projects_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(auth_session_router, dependencies=PRIVATE_API_DEPENDENCIES) +app.include_router(disksage_router, dependencies=PRIVATE_API_DEPENDENCIES) @app.get("/") diff --git a/backend/scripts/disksage_copy_readiness_handoff.py b/backend/scripts/disksage_copy_readiness_handoff.py index 1036d1bb1..2604864fc 100644 --- a/backend/scripts/disksage_copy_readiness_handoff.py +++ b/backend/scripts/disksage_copy_readiness_handoff.py @@ -32,7 +32,7 @@ EXIT_VERIFIER_UNAVAILABLE = 66 EXIT_EXECUTION_FAILED = 70 -SUCCESS_FIELDS = frozenset( +BASE_SUCCESS_FIELDS = frozenset( { "ok", "schema_kind", @@ -49,6 +49,14 @@ "source_eviction_authorized", } ) +NATIVE_STATUS_FIELDS = frozenset( + { + "icloud_native_status_observed", + "icloud_native_sync_state", + "icloud_native_status_timed_out", + } +) +SUCCESS_FIELDS = BASE_SUCCESS_FIELDS | NATIVE_STATUS_FIELDS FAILURE_FIELDS = frozenset({"ok", "error_code"}) FALSE_CLAIM_FIELDS = ( "local_paths_included", @@ -61,11 +69,12 @@ READINESS_STATES = frozenset( {"no-candidates", "blocked", "partially-ready", "ready-without-new-review"} ) -# DiskSage schema v5 adds path-free provider-global-sync evidence while retaining the same -# success contract consumed by this handoff. Keep v3/v4 readable for already-issued evidence +# DiskSage schema v6 is the current path-free provider-global-sync envelope and retains the same +# success contract consumed by this handoff. Keep v3-v5 readable for already-issued evidence # records; newer envelopes must be added here deliberately and tested. -SUPPORTED_READINESS_SCHEMA_VERSIONS = frozenset({3, 4, 5}) +SUPPORTED_READINESS_SCHEMA_VERSIONS = frozenset({3, 4, 5, 6}) ERROR_CODE_PATTERN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*") +NATIVE_STATUS_TOKEN_PATTERN = re.compile(r"[A-Za-z0-9|_.-]{1,128}") class HandoffError(Exception): @@ -353,9 +362,15 @@ def _decode_protocol(result: VerifierResult) -> dict[str, object]: raise HandoffError("disksage-verifier-protocol-invalid") if result.returncode == 0: + schema_version = payload.get("schema_version") + fields_valid = ( + frozenset(payload) == SUCCESS_FIELDS + if schema_version == 6 + else frozenset(payload) in (BASE_SUCCESS_FIELDS, SUCCESS_FIELDS) + ) valid = ( not result.stderr - and frozenset(payload) == SUCCESS_FIELDS + and fields_valid and payload.get("ok") is True and payload.get("schema_kind") == "disksage.naruon.cloud-copy-readiness" and type(payload.get("schema_version")) is int @@ -368,6 +383,33 @@ def _decode_protocol(result: VerifierResult) -> dict[str, object]: and payload["candidate_bytes"] >= 0 and _is_lower_hex_64(payload.get("readiness_fingerprint_sha256")) and all(payload.get(field) is False for field in FALSE_CLAIM_FIELDS) + and ( + frozenset(payload) == BASE_SUCCESS_FIELDS + or ( + type(payload.get("icloud_native_status_observed")) is bool + or payload.get("icloud_native_status_observed") is None + ) + ) + and ( + frozenset(payload) == BASE_SUCCESS_FIELDS + or ( + type(payload.get("icloud_native_status_timed_out")) is bool + or payload.get("icloud_native_status_timed_out") is None + ) + ) + and ( + frozenset(payload) == BASE_SUCCESS_FIELDS + or ( + payload.get("icloud_native_sync_state") is None + or ( + type(payload.get("icloud_native_sync_state")) is str + and NATIVE_STATUS_TOKEN_PATTERN.fullmatch( + payload["icloud_native_sync_state"] + ) + is not None + ) + ) + ) ) elif result.returncode in (64, 65): error_code = payload.get("error_code") diff --git a/backend/services/disksage_file_lineage.py b/backend/services/disksage_file_lineage.py new file mode 100644 index 000000000..3223e1067 --- /dev/null +++ b/backend/services/disksage_file_lineage.py @@ -0,0 +1,382 @@ +"""Strict, scope-neutral validation for DiskSage file-lineage envelopes. + +DiskSage performs the filesystem and provider proof work locally. Naruon only +accepts the resulting immutable envelope; it never treats a local File +Provider copy as a provider API write or as permission to evict the source. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import UTC, datetime +from pathlib import PureWindowsPath +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +HEX64_PATTERN = r"^[0-9a-fA-F]{64}$" +ONTOLOGY_CLASS_PATTERN = ( + r"^https://disksage\.app/ontology#[A-Za-z][A-Za-z0-9_-]{0,127}$" +) +PROVIDER_VALUES = Literal["icloud", "onedrive", "google-drive"] +ARCHIVE_KIND_VALUES = Literal[ + "document", + "media", + "archive", + "dataset", + "backup", + "creative", + "incomplete-download", +] +REVIEW_DISPOSITION_VALUES = Literal["approved", "held"] +SYNC_KIND_VALUES = Literal["provider-api", "provider-native-status"] +COPY_APPROVAL_ACTION_VALUES = Literal["copy-only", "adopt-existing-copy"] +PROVIDER_SYNC_STATE_VALUES = Literal[ + "complete", + "pending-upload", + "not-ubiquitous", + "not-local-current", + "uploading", + "excluded-from-sync", + "sync-paused", + "remote-unavailable", + "content-mismatch", + "unknown", +] +PROVIDER_SYNC_TIMELINESS_VALUES = Literal["complete", "pending", "overdue"] +UNKNOWN_ONTOLOGY_CLASS = "https://disksage.app/ontology#Unknown" +EVIDENCE_PRECEDENCE = ( + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", +) + + +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class FileLineageRelation(_StrictModel): + subject: str = Field(min_length=1, max_length=2048) + predicate: str = Field(min_length=1, max_length=256) + object: str = Field(min_length=1, max_length=2048) + source: str = Field(min_length=1, max_length=256) + + @model_validator(mode="after") + def reject_control_values(self) -> FileLineageRelation: + if any( + any(ord(character) < 32 for character in value) + for value in (self.subject, self.predicate, self.object, self.source) + ): + raise ValueError("lineage relation contains a control character") + return self + + +class FileMetadataEvidence(_StrictModel): + field: str = Field(min_length=1, max_length=128) + value: str = Field(max_length=2048) + source: str = Field(min_length=1, max_length=256) + confidence: Literal["high", "medium", "low", "unknown"] + + +class ProductionTimeLineage(_StrictModel): + selected_value_ms: int = Field(ge=0) + selected_source: str = Field(min_length=1, max_length=256) + confidence: Literal["high", "medium", "low", "unknown"] + evidence_precedence: list[str] = Field(min_length=1, max_length=8) + + +class FilesystemTimeLineage(_StrictModel): + created_at_ms: int = Field(ge=0) + modified_at_ms: int = Field(ge=0) + + +class ReviewLineage(_StrictModel): + candidate_fingerprint: str = Field(pattern=HEX64_PATTERN) + review_fingerprint: str = Field(pattern=HEX64_PATTERN) + requires_review: bool + reason_codes: list[str] = Field(max_length=64) + decision_id: str | None = Field(default=None, max_length=256) + disposition: REVIEW_DISPOSITION_VALUES | None = None + reviewed_at_ms: int | None = Field(default=None, ge=0) + reviewed_by: str | None = Field(default=None, max_length=256) + rationale: str | None = Field(default=None, max_length=2000) + + @model_validator(mode="after") + def bind_decision_fields(self) -> ReviewLineage: + if self.disposition is not None and ( + self.decision_id is None + or self.reviewed_at_ms is None + or not self.reviewed_by + or not self.rationale + ): + raise ValueError("review disposition is missing its decision evidence") + return self + + +class RemoteContentProof(_StrictModel): + object_id: str = Field(min_length=1, max_length=512) + revision: str = Field(min_length=1, max_length=512) + algorithm: Literal["sha256", "quick-xor"] + checksum: str = Field(min_length=1, max_length=256) + location_bound: bool + location_proof: str | None = Field(default=None, max_length=2048) + + +class CloudCopyLineage(_StrictModel): + receipt_id: str = Field(pattern=HEX64_PATTERN) + lineage_fingerprint: str = Field(pattern=HEX64_PATTERN) + provider: PROVIDER_VALUES + destination_account_scope: Literal["personal", "organization", "shared", "unknown"] + destination: str = Field(min_length=1, max_length=4096) + copied_at_ms: int = Field(ge=0) + copy_verification_method: Literal[ + "copied-by-disk-sage", "copied-by-provider-api", "adopted-existing" + ] + # DiskSage v2 binds the copy to one attributed human action. These remain + # optional so v1 and pre-approval receipts can still be ingested. + copy_approval_id: str | None = Field(default=None, pattern=HEX64_PATTERN) + copy_approval_action: COPY_APPROVAL_ACTION_VALUES | None = None + copy_approved_at_ms: int | None = Field(default=None, ge=0) + copy_approved_by: str | None = Field(default=None, max_length=256) + copy_approval_rationale: str | None = Field(default=None, max_length=2000) + local_copy_verified: bool + provider_write_executed: bool + provider_sync_confirmed: bool + # Optional keeps version-1 envelopes backwards compatible; new DiskSage + # exports preserve provider-native states such as pending-upload. + provider_sync_state: PROVIDER_SYNC_STATE_VALUES | None = None + sync_evidence_record_id: str | None = Field(default=None, max_length=256) + sync_evidence_kind: SYNC_KIND_VALUES | None = None + sync_evidence_id: str | None = Field(default=None, max_length=512) + sync_confirmed_at_ms: int | None = Field(default=None, ge=0) + remote_object_id: str | None = Field(default=None, max_length=512) + remote_revision: str | None = Field(default=None, max_length=512) + remote_location_bound: bool | None = None + # DiskSage's diagnostic timeliness projection is retained inside the encrypted envelope; + # it never grants source-eviction authority. + sync_timeliness: PROVIDER_SYNC_TIMELINESS_VALUES | None = None + sync_pending_age_ms: int | None = Field(default=None, ge=0) + sync_overdue_after_ms: int | None = Field(default=None, ge=0) + sync_reason_codes: list[str] = Field(default_factory=list, max_length=32) + + @model_validator(mode="after") + def bind_provider_evidence(self) -> CloudCopyLineage: + if not self.local_copy_verified: + raise ValueError("lineage copy is not locally verified") + if self.provider_write_executed: + raise ValueError("Naruon cannot accept a provider-write claim") + evidence_fields = ( + self.sync_evidence_record_id, + self.sync_evidence_kind, + self.sync_evidence_id, + self.sync_confirmed_at_ms, + ) + if self.provider_sync_confirmed and any( + value is None + or (isinstance(value, str) and not value.strip()) + for value in evidence_fields + ): + raise ValueError("confirmed sync is missing provider evidence") + if self.provider_sync_state not in (None, "unknown") and ( + self.provider_sync_confirmed != (self.provider_sync_state == "complete") + ): + raise ValueError("provider sync state does not match confirmation") + if self.remote_location_bound is True and ( + self.remote_object_id is None or not self.remote_object_id.strip() + ): + raise ValueError("remote location binding is missing its object id") + if self.sync_timeliness == "complete" and ( + not self.provider_sync_confirmed or self.provider_sync_state != "complete" + ): + raise ValueError("complete sync timeliness is not provider-confirmed") + if self.sync_timeliness in {"pending", "overdue"} and self.provider_sync_confirmed: + raise ValueError("incomplete sync timeliness is provider-confirmed") + approval_fields = ( + self.copy_approval_id, + self.copy_approval_action, + self.copy_approved_at_ms, + self.copy_approved_by, + self.copy_approval_rationale, + ) + approval_present = [value is not None for value in approval_fields] + if any(approval_present) and not all(approval_present): + raise ValueError("copy approval evidence must be complete or absent") + if any(approval_present) and ( + not self.copy_approved_by.strip() or not self.copy_approval_rationale.strip() + ): + raise ValueError("copy approval attribution is missing") + return self + + +class FileLineageEnvelope(_StrictModel): + # DiskSage v3 makes ontology projection and provider-native state explicit; + # v1/v2 remain readable with deterministic legacy defaults. + schema_version: Literal[1, 2, 3] + schema_kind: Literal["disksage.file-lineage"] + source_kind: Literal["file"] + archive_kind: ARCHIVE_KIND_VALUES + source_filename: str = Field(min_length=1, max_length=512) + source_relative_path: str = Field(min_length=1, max_length=4096) + source_context: str = Field(min_length=1, max_length=1024) + ontology_class: str = Field( + default=UNKNOWN_ONTOLOGY_CLASS, pattern=ONTOLOGY_CLASS_PATTERN + ) + ontology_relations: list[FileLineageRelation] = Field(default_factory=list, max_length=256) + raw_content_sha256: str = Field(pattern=HEX64_PATTERN) + raw_content_blake3: str = Field(pattern=HEX64_PATTERN) + bytes: int = Field(ge=0) + production_time: ProductionTimeLineage + filesystem_time: FilesystemTimeLineage + metadata_evidence: list[FileMetadataEvidence] = Field(max_length=128) + content_title: str | None = Field(default=None, max_length=1024) + content_authors: list[str] = Field(max_length=32) + content_context: list[str] = Field(max_length=64) + duration_ms: int | None = Field(default=None, ge=0) + review: ReviewLineage + cloud_copy: CloudCopyLineage + + @model_validator(mode="before") + @classmethod + def require_v3_ontology_projection(cls, values: object) -> object: + if isinstance(values, dict) and values.get("schema_version") == 3: + if not values.get("ontology_class") or "ontology_relations" not in values: + raise ValueError("schema version 3 requires ontology projection") + return values + + @model_validator(mode="after") + def validate_path_and_relations(self) -> FileLineageEnvelope: + relative = self.source_relative_path + if "\\" in relative or PureWindowsPath(relative).drive: + raise ValueError("source relative path is not normalized") + parts = relative.split("/") + if relative.startswith("/") or any(part in {"", ".", ".."} for part in parts): + raise ValueError("source relative path is not a normalized relative path") + if self.source_filename != parts[-1]: + raise ValueError("source filename does not match source relative path") + if any(ord(character) < 32 for character in relative): + raise ValueError("source relative path contains a control character") + _validate_production_evidence(self.production_time, self.metadata_evidence) + approval_fields = ( + self.cloud_copy.copy_approval_id, + self.cloud_copy.copy_approval_action, + self.cloud_copy.copy_approved_at_ms, + self.cloud_copy.copy_approved_by, + self.cloud_copy.copy_approval_rationale, + ) + if self.schema_version == 1 and any(value is not None for value in approval_fields): + raise ValueError("schema version 1 cannot carry copy approval evidence") + return self + + +class FileLineageSummary(_StrictModel): + lineage_record_uid: str + lineage_fingerprint: str + schema_version: int + source_kind: str + archive_kind: str + raw_content_sha256: str + raw_content_blake3: str + content_bytes: int + ontology_class: str + ontology_relation_count: int + ontology_predicates: list[str] + provider_name: PROVIDER_VALUES + provider_sync_confirmed: bool + provider_sync_state: PROVIDER_SYNC_STATE_VALUES + created_at: str + + +def _date_value(epoch_ms: int) -> str: + try: + return datetime.fromtimestamp(epoch_ms / 1000, tz=UTC).date().isoformat() + except (OverflowError, OSError, ValueError) as error: + raise ValueError("production time is out of bounds") from error + + +def _production_source_class(source: str) -> str: + if source.startswith("embedded:"): + return "embedded_metadata" + return { + "filename:path-token": "explicit_filename_date", + "filesystem:created": "filesystem_created_at", + "filesystem:modified-fallback": "filesystem_modified_at", + }.get(source, "") + + +def _evidence_source_class(evidence: FileMetadataEvidence) -> str | None: + if evidence.field == "production-date" and evidence.source.startswith("embedded:"): + return "embedded_metadata" + return { + ("filename-date-hint", "filename:path-token"): "explicit_filename_date", + ("filesystem-created-date", "filesystem:created"): "filesystem_created_at", + ( + "filesystem-modified-date", + "filesystem:modified", + ): "filesystem_modified_at", + }.get((evidence.field, evidence.source)) + + +def _validate_production_evidence( + production: ProductionTimeLineage, + metadata: list[FileMetadataEvidence], +) -> None: + if production.evidence_precedence != list(EVIDENCE_PRECEDENCE): + raise ValueError("production evidence precedence is not canonical") + source_class = _production_source_class(production.selected_source) + if not source_class: + raise ValueError("production time source is unsupported") + if source_class != "embedded_metadata" and production.confidence != "low": + raise ValueError("non-embedded production evidence must be low confidence") + expected = { + "embedded_metadata": ("production-date", production.selected_source), + "explicit_filename_date": ("filename-date-hint", "filename:path-token"), + "filesystem_created_at": ("filesystem-created-date", "filesystem:created"), + "filesystem_modified_at": ( + "filesystem-modified-date", + "filesystem:modified", + ), + }[source_class] + selected_date = _date_value(production.selected_value_ms) + if not any( + evidence.field == expected[0] + and evidence.source == expected[1] + and evidence.value == selected_date + for evidence in metadata + ): + raise ValueError("selected production evidence is missing or mismatched") + selected_rank = EVIDENCE_PRECEDENCE.index(source_class) + if any( + EVIDENCE_PRECEDENCE.index(evidence_class) < selected_rank + for evidence_class in ( + _evidence_source_class(evidence) for evidence in metadata + ) + if evidence_class is not None + ): + raise ValueError("production evidence violates precedence") + + +def canonical_envelope_json(envelope: FileLineageEnvelope) -> str: + """Serialize the validated envelope deterministically for hashing and storage.""" + + return json.dumps( + envelope.model_dump(mode="json"), + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def canonical_envelope_sha256(envelope: FileLineageEnvelope) -> str: + """Hash the exact validated JSON payload for idempotent, tamper-evident ingest.""" + + return hashlib.sha256(canonical_envelope_json(envelope).encode("utf-8")).hexdigest() + + +def ontology_predicates(envelope: FileLineageEnvelope) -> list[str]: + """Return a deterministic public projection without path/object values.""" + + return sorted({relation.predicate for relation in envelope.ontology_relations}) diff --git a/backend/services/disksage_organization_lineage.py b/backend/services/disksage_organization_lineage.py new file mode 100644 index 000000000..feca8f302 --- /dev/null +++ b/backend/services/disksage_organization_lineage.py @@ -0,0 +1,89 @@ +"""Strict validation for DiskSage's path-free local organization handoff.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +HEX64_PATTERN = r"^[0-9a-f]{64}$" +ONTOLOGY_CLASS_PATTERN = ( + r"^https://disksage\.app/ontology#[A-Za-z][A-Za-z0-9_-]{0,127}$" +) +PRODUCTION_SOURCES = ( + "embedded:", + "filename:path-token", + "filesystem:created", + "filesystem:modified-fallback", +) + + +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class OrganizationLineageItem(_StrictModel): + lineage_fingerprint: str = Field(pattern=HEX64_PATTERN) + source_size: int = Field(ge=0) + source_mtime_ms: int = Field(ge=0) + production_time_ms: int = Field(gt=0) + production_time_source: str = Field(min_length=1, max_length=256) + production_time_confidence: Literal["high", "medium", "low", "unknown"] + ontology_class: str = Field(pattern=ONTOLOGY_CLASS_PATTERN) + destination_relation: Literal["targetFolder"] + action: Literal["move"] + + @model_validator(mode="after") + def validate_production_source(self) -> OrganizationLineageItem: + if not any( + self.production_time_source == source + or self.production_time_source.startswith(source) + for source in PRODUCTION_SOURCES + ): + raise ValueError("organization lineage production source is unsupported") + if not self.production_time_source.isprintable(): + raise ValueError("organization lineage production source contains control characters") + return self + + +class OrganizationLineageBatch(_StrictModel): + schema_kind: Literal["disksage.organization-lineage-batch"] = Field(alias="schema") + version: Literal[1] + generated_at_ms: int = Field(gt=0) + complete: Literal[True] + batch_fingerprint_sha256: str = Field(pattern=HEX64_PATTERN) + items: list[OrganizationLineageItem] = Field(min_length=1, max_length=200) + + @model_validator(mode="after") + def validate_batch(self) -> OrganizationLineageBatch: + fingerprints = [item.lineage_fingerprint for item in self.items] + if len(set(fingerprints)) != len(fingerprints): + raise ValueError("organization lineage fingerprints must be unique") + if self.generated_at_ms > 253_402_300_799_999: + raise ValueError("organization lineage generated time is out of bounds") + return self + + +class OrganizationLineageSummary(_StrictModel): + organization_lineage_record_uid: str + batch_fingerprint_sha256: str + schema_version: int + item_count: int + ontology_classes: list[str] + created_at: str + + +def canonical_batch_json(batch: OrganizationLineageBatch) -> str: + return json.dumps( + batch.model_dump(mode="json", by_alias=True), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + +def canonical_batch_sha256(batch: OrganizationLineageBatch) -> str: + return hashlib.sha256(canonical_batch_json(batch).encode("utf-8")).hexdigest() diff --git a/backend/tests/test_disksage_copy_readiness_handoff.py b/backend/tests/test_disksage_copy_readiness_handoff.py index 471cc0090..ed7fc2ce8 100644 --- a/backend/tests/test_disksage_copy_readiness_handoff.py +++ b/backend/tests/test_disksage_copy_readiness_handoff.py @@ -36,6 +36,9 @@ def _success_payload() -> dict[str, object]: "raw_metadata_values_included": False, "cloud_write_executed": False, "source_eviction_authorized": False, + "icloud_native_status_observed": None, + "icloud_native_sync_state": None, + "icloud_native_status_timed_out": None, } @@ -111,7 +114,7 @@ def test_main_delegates_to_absolute_verifier_without_shell_env_or_input_read( assert not (tmp_path / "must-not-exist").exists() -@pytest.mark.parametrize("schema_version", [4, 5]) +@pytest.mark.parametrize("schema_version", [4, 5, 6]) def test_main_accepts_current_disksage_schema_versions(tmp_path, capsys, schema_version): payload = _success_payload() payload["schema_version"] = schema_version @@ -121,6 +124,17 @@ def test_main_accepts_current_disksage_schema_versions(tmp_path, capsys, schema_ assert json.loads(capsys.readouterr().out) == payload +def test_main_accepts_legacy_success_protocol_for_prior_schema(tmp_path, capsys): + payload = _success_payload() + payload["schema_version"] = 5 + for field in handoff.NATIVE_STATUS_FIELDS: + payload.pop(field) + verifier = _json_verifier(tmp_path / "verifier", payload, 0) + + assert handoff.main(_handoff_args(verifier, tmp_path / "readiness.json")) == 0 + assert json.loads(capsys.readouterr().out) == payload + + @pytest.mark.parametrize("exit_code", [64, 65]) def test_main_preserves_valid_disksage_failure_protocol(tmp_path, capsys, exit_code): payload = { @@ -765,6 +779,25 @@ def test_protocol_decoder_rejects_invalid_or_ambiguous_transport(result): assert error.value.error_code == "disksage-verifier-protocol-invalid" +@pytest.mark.parametrize( + "field_value", + [ + {"icloud_native_status_observed": "yes"}, + {"icloud_native_status_timed_out": 1}, + {"icloud_native_sync_state": "/private/source"}, + ], +) +def test_protocol_decoder_rejects_invalid_native_status_summary(field_value): + payload = _success_payload() + payload.update(field_value) + with pytest.raises(handoff.HandoffError) as error: + handoff._decode_protocol( + handoff.VerifierResult(0, json.dumps(payload).encode(), b"") + ) + + assert error.value.error_code == "disksage-verifier-protocol-invalid" + + @pytest.mark.parametrize("stream_name", ["stdout", "stderr"]) def test_main_kills_oversized_output_without_echoing_it(tmp_path, capsys, stream_name): verifier = _python_verifier( diff --git a/backend/tests/test_disksage_file_lineage.py b/backend/tests/test_disksage_file_lineage.py new file mode 100644 index 000000000..cf78d25b8 --- /dev/null +++ b/backend/tests/test_disksage_file_lineage.py @@ -0,0 +1,427 @@ +import pytest +from pydantic import ValidationError + +from services.disksage_file_lineage import ( + FileLineageEnvelope, + canonical_envelope_json, + canonical_envelope_sha256, + ontology_predicates, +) + + +def _envelope(**overrides: object) -> dict[str, object]: + payload: dict[str, object] = { + "schema_version": 1, + "schema_kind": "disksage.file-lineage", + "source_kind": "file", + "archive_kind": "media", + "source_filename": "Video 1.mov", + "source_relative_path": "DaVinci Resolve/Video 1.mov", + "source_context": "DaVinci Resolve", + "ontology_class": "https://disksage.app/ontology#Media", + "ontology_relations": [ + { + "subject": "/Users/example/Movies/Video 1.mov", + "predicate": "https://disksage.app/ontology#archivedTo", + "object": "/Users/example/iCloud/DiskSage Archive/Video 1.mov", + "source": "archive-destination-planner", + }, + { + "subject": "source", + "predicate": "https://disksage.app/ontology#archivedTo", + "object": "destination", + "source": "test", + }, + ], + "raw_content_sha256": "a" * 64, + "raw_content_blake3": "b" * 64, + "bytes": 160085038, + "production_time": { + "selected_value_ms": 1, + "selected_source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + "filesystem_time": {"created_at_ms": 2, "modified_at_ms": 3}, + "metadata_evidence": [ + { + "field": "production-date", + "value": "1970-01-01", + "source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + } + ], + "content_title": "Video 1", + "content_authors": [], + "content_context": ["DaVinci Resolve"], + "duration_ms": 60000, + "review": { + "candidate_fingerprint": "c" * 64, + "review_fingerprint": "d" * 64, + "requires_review": True, + "reason_codes": ["destination-account-scope-unknown"], + "decision_id": "decision-1", + "disposition": "approved", + "reviewed_at_ms": 4, + "reviewed_by": "human:local:test", + "rationale": "Account scope reviewed", + }, + "cloud_copy": { + "receipt_id": "e" * 64, + "lineage_fingerprint": "f" * 64, + "provider": "icloud", + "destination_account_scope": "unknown", + "destination": "/Users/example/iCloud/DiskSage Archive/Video 1.mov", + "copied_at_ms": 5, + "copy_verification_method": "copied-by-disk-sage", + "local_copy_verified": True, + "provider_write_executed": False, + "provider_sync_confirmed": False, + "sync_evidence_record_id": None, + "sync_evidence_kind": None, + "sync_evidence_id": None, + "sync_confirmed_at_ms": None, + "remote_object_id": None, + "remote_revision": None, + "remote_location_bound": None, + }, + } + payload.update(overrides) + return payload + + +def test_valid_envelope_keeps_graph_projection_deterministic(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + + assert envelope.schema_kind == "disksage.file-lineage" + assert len(envelope.ontology_relations) == 2 + assert ontology_predicates(envelope) == ["https://disksage.app/ontology#archivedTo"] + assert len(canonical_envelope_sha256(envelope)) == 64 + assert canonical_envelope_json(envelope).encode("utf-8") + + +@pytest.mark.parametrize("schema_version", [1, 2, 3]) +def test_accepts_disk_sage_file_lineage_schema_versions(schema_version: int): + envelope = FileLineageEnvelope.model_validate( + _envelope(schema_version=schema_version) + ) + + assert envelope.schema_version == schema_version + + +def test_legacy_v2_without_new_ontology_projection_is_migrated(): + payload = _envelope(schema_version=2) + payload.pop("ontology_class") + payload.pop("ontology_relations") + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.ontology_class.endswith("#Unknown") + assert envelope.ontology_relations == [] + + +def test_v3_requires_ontology_projection(): + payload = _envelope(schema_version=3) + payload.pop("ontology_class") + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_v2_preserves_attributed_copy_approval_fields(): + payload = _envelope( + schema_version=2, + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "copy_approval_id": "1" * 64, + "copy_approval_action": "copy-only", + "copy_approved_at_ms": 6, + "copy_approved_by": "human:local:test", + "copy_approval_rationale": "approved after metadata review", + }, + ) + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.copy_approval_action == "copy-only" + assert envelope.cloud_copy.copy_approval_id == "1" * 64 + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("source_relative_path", "/absolute/path.mov"), + ("source_relative_path", "../escape.mov"), + ("source_relative_path", "DaVinci\\Video 1.mov"), + ("source_relative_path", "C:/Video 1.mov"), + ("source_filename", "different.mov"), + ], +) +def test_path_binding_rejects_unsafe_or_mismatched_values(field: str, value: str): + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(_envelope(**{field: value})) + + +def test_provider_write_claim_is_rejected_even_when_copy_is_verified(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "provider_write_executed": True, + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_provider_api_copy_method_is_preserved_without_naruon_write_authority(): + payload = _envelope( + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "copy_verification_method": "copied-by-provider-api", + } + ) + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.copy_verification_method == "copied-by-provider-api" + assert envelope.cloud_copy.provider_write_executed is False + + +def test_provider_sync_state_preserves_pending_upload_without_eviction_claim(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "provider_sync_state": "pending-upload", + } + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.provider_sync_state == "pending-upload" + assert envelope.cloud_copy.provider_sync_confirmed is False + + +@pytest.mark.parametrize("provider_sync_state", ["pending-upload", "not-local-current"]) +def test_provider_sync_incomplete_states_remain_unconfirmed(provider_sync_state: str): + payload = _envelope( + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "provider_sync_state": provider_sync_state, + } + ) + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.provider_sync_state == provider_sync_state + assert envelope.cloud_copy.provider_sync_confirmed is False + + +def test_provider_sync_timeliness_projection_is_preserved_without_eviction_authority(): + payload = _envelope( + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "provider_sync_state": "unknown", + "sync_timeliness": "pending", + "sync_pending_age_ms": 123, + "sync_overdue_after_ms": 456, + "sync_reason_codes": ["provider-sync-confirmation-pending"], + } + ) + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.sync_timeliness == "pending" + assert envelope.cloud_copy.sync_pending_age_ms == 123 + assert envelope.cloud_copy.sync_overdue_after_ms == 456 + assert envelope.cloud_copy.sync_reason_codes == [ + "provider-sync-confirmation-pending" + ] + assert envelope.cloud_copy.provider_sync_confirmed is False + + +def test_complete_timeliness_requires_confirmed_complete_provider_state(): + payload = _envelope( + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "provider_sync_confirmed": True, + "provider_sync_state": "complete", + "sync_evidence_record_id": "1" * 64, + "sync_evidence_kind": "provider-native-status", + "sync_evidence_id": "icloud-uploaded-flag", + "sync_confirmed_at_ms": 7, + "sync_timeliness": "complete", + "sync_pending_age_ms": 0, + "sync_overdue_after_ms": 456, + "sync_reason_codes": [], + } + ) + + envelope = FileLineageEnvelope.model_validate(payload) + + assert envelope.cloud_copy.sync_timeliness == "complete" + + +def test_filename_date_is_only_auxiliary_when_embedded_metadata_exists(): + payload = _envelope( + production_time={ + "selected_value_ms": 1, + "selected_source": "filename:path-token", + "confidence": "low", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + metadata_evidence=[ + { + "field": "filename-date-hint", + "value": "1970-01-01", + "source": "filename:path-token", + "confidence": "low", + }, + { + "field": "production-date", + "value": "1970-01-01", + "source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + }, + ], + ) + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_filename_date_can_be_selected_only_without_higher_precedence_evidence(): + payload = _envelope( + production_time={ + "selected_value_ms": 1, + "selected_source": "filename:path-token", + "confidence": "low", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + metadata_evidence=[ + { + "field": "filename-date-hint", + "value": "1970-01-01", + "source": "filename:path-token", + "confidence": "low", + } + ], + ) + envelope = FileLineageEnvelope.model_validate(payload) + assert envelope.production_time.selected_source == "filename:path-token" + + +def test_filesystem_modified_fallback_uses_modified_evidence_source(): + payload = _envelope( + production_time={ + "selected_value_ms": 1, + "selected_source": "filesystem:modified-fallback", + "confidence": "low", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + metadata_evidence=[ + { + "field": "filesystem-modified-date", + "value": "1970-01-01", + "source": "filesystem:modified", + "confidence": "low", + } + ], + ) + envelope = FileLineageEnvelope.model_validate(payload) + assert envelope.production_time.selected_source == "filesystem:modified-fallback" + + +def test_provider_sync_confirmation_requires_evidence(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "provider_sync_confirmed": True, + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_complete_provider_sync_state_requires_confirmation(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "provider_sync_state": "complete", + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_copy_approval_evidence_is_all_or_none_and_versioned(): + partial = _envelope() + partial["cloud_copy"] = { + **partial["cloud_copy"], # type: ignore[arg-type] + "copy_approval_id": "1" * 64, + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(partial) + + v1_complete = _envelope( + cloud_copy={ + **_envelope()["cloud_copy"], # type: ignore[arg-type] + "copy_approval_id": "1" * 64, + "copy_approval_action": "copy-only", + "copy_approved_at_ms": 6, + "copy_approved_by": "human:local:test", + "copy_approval_rationale": "approved", + } + ) + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(v1_complete) + + +def test_confirmed_provider_evidence_rejects_blank_identifiers(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "provider_sync_confirmed": True, + "sync_evidence_record_id": " " * 2, + "sync_evidence_kind": "provider-native-status", + "sync_evidence_id": "evidence-id", + "sync_confirmed_at_ms": 7, + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_location_bound_remote_proof_rejects_blank_object_id(): + payload = _envelope() + payload["cloud_copy"] = { + **payload["cloud_copy"], # type: ignore[arg-type] + "remote_location_bound": True, + "remote_object_id": " ", + } + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_unknown_fields_are_rejected_at_the_handoff_boundary(): + payload = _envelope(untrusted_private_value="must-not-persist") + with pytest.raises(ValidationError): + FileLineageEnvelope.model_validate(payload) + + +def test_naruon_exposes_the_scoped_lineage_resource(): + from main import app + + assert "/api/disksage/file-lineage" in app.openapi()["paths"] diff --git a/backend/tests/test_disksage_file_lineage_api.py b/backend/tests/test_disksage_file_lineage_api.py new file mode 100644 index 000000000..3005252dd --- /dev/null +++ b/backend/tests/test_disksage_file_lineage_api.py @@ -0,0 +1,261 @@ +import datetime +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError, StatementError + +from api.auth import AuthContext +from api.disksage import ingest_file_lineage, list_file_lineage +from core.runtime_secrets import EncryptionKeyMissingError +from services.disksage_file_lineage import ( + FileLineageEnvelope, + canonical_envelope_sha256, +) + + +def _envelope() -> dict[str, object]: + return { + "schema_version": 1, + "schema_kind": "disksage.file-lineage", + "source_kind": "file", + "archive_kind": "media", + "source_filename": "Video 1.mov", + "source_relative_path": "DaVinci Resolve/Video 1.mov", + "source_context": "DaVinci Resolve", + "ontology_class": "https://disksage.app/ontology#Media", + "ontology_relations": [], + "raw_content_sha256": "a" * 64, + "raw_content_blake3": "b" * 64, + "bytes": 160085038, + "production_time": { + "selected_value_ms": 1, + "selected_source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + "filesystem_time": {"created_at_ms": 2, "modified_at_ms": 3}, + "metadata_evidence": [ + { + "field": "production-date", + "value": "1970-01-01", + "source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + } + ], + "content_authors": [], + "content_context": [], + "review": { + "candidate_fingerprint": "c" * 64, + "review_fingerprint": "d" * 64, + "requires_review": False, + "reason_codes": [], + }, + "cloud_copy": { + "receipt_id": "e" * 64, + "lineage_fingerprint": "f" * 64, + "provider": "icloud", + "destination_account_scope": "unknown", + "destination": "/Users/example/iCloud/Video 1.mov", + "copied_at_ms": 5, + "copy_verification_method": "copied-by-disk-sage", + "local_copy_verified": True, + "provider_write_executed": False, + "provider_sync_confirmed": False, + }, + } + + +class _Result: + def __init__(self, record=None, records=()): + self._record = record + self._records = tuple(records) + + def scalar_one_or_none(self): + return self._record + + def scalars(self): + return self + + def all(self): + return list(self._records) + + +class _Session: + def __init__(self, results, *, commit_error=None): + self._results = iter(results) + self.commit_error = commit_error + self.statements = [] + self.rollback_count = 0 + self.commit_count = 0 + + async def execute(self, statement): + self.statements.append(statement) + return next(self._results) + + def add(self, _record): + pass + + async def commit(self): + self.commit_count += 1 + if self.commit_error is not None: + raise self.commit_error + + async def refresh(self, _record): + pass + + async def rollback(self): + self.rollback_count += 1 + + +def _auth() -> AuthContext: + return AuthContext( + user_id="user-1", + role="member", + organization_id="org-1", + group_ids=(), + workspace_id="workspace-1", + ) + + +def _record(**overrides): + values = { + "lineage_record_uid": "disksage_lineage_1", + "organization_id": "org-1", + "lineage_fingerprint": "f" * 64, + "schema_version": 1, + "source_kind": "file", + "archive_kind": "media", + "raw_content_sha256": "a" * 64, + "raw_content_blake3": "b" * 64, + "content_bytes": 1, + "ontology_class": "https://disksage.app/ontology#Media", + "ontology_relation_count": 0, + "ontology_predicates": [], + "provider_name": "icloud", + "provider_sync_confirmed": False, + "provider_sync_state": "pending-upload", + "created_at": datetime.datetime.now(datetime.timezone.utc), + "envelope_sha256": "c" * 64, + } + values.update(overrides) + return SimpleNamespace(**values) + + +@pytest.mark.asyncio +async def test_replayed_lineage_is_idempotent_with_workspace_and_org_scope(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + record = _record(envelope_sha256=canonical_envelope_sha256(envelope)) + session = _Session([_Result(record)]) + + result = await ingest_file_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert result.lineage_record_uid == record.lineage_record_uid + assert session.commit_count == 0 + statement = str(session.statements[0]) + assert "organization_id" in statement + assert "workspace_id" in statement + + +@pytest.mark.asyncio +async def test_conflicting_lineage_fingerprint_returns_conflict(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + session = _Session([_Result(_record())]) + + with pytest.raises(HTTPException) as error: + await ingest_file_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert error.value.status_code == 409 + + +@pytest.mark.asyncio +async def test_missing_encryption_key_rolls_back_as_service_unavailable(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + session = _Session( + [_Result(None)], + commit_error=EncryptionKeyMissingError("missing active key"), + ) + + with pytest.raises(HTTPException) as error: + await ingest_file_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert error.value.status_code == 503 + assert session.rollback_count == 1 + + +@pytest.mark.asyncio +async def test_wrapped_missing_encryption_key_rolls_back_as_service_unavailable(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + session = _Session( + [_Result(None)], + commit_error=StatementError( + "encrypted bind failed", + None, + None, + EncryptionKeyMissingError("missing active key"), + ), + ) + + with pytest.raises(HTTPException) as error: + await ingest_file_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert error.value.status_code == 503 + assert session.rollback_count == 1 + + +@pytest.mark.asyncio +async def test_racing_fingerprint_in_another_organization_returns_conflict(): + envelope = FileLineageEnvelope.model_validate(_envelope()) + record = _record( + envelope_sha256=canonical_envelope_sha256(envelope), + organization_id="org-2", + ) + session = _Session( + [_Result(None), _Result(record)], + commit_error=IntegrityError("unique violation", None, Exception("duplicate")), + ) + + with pytest.raises(HTTPException) as error: + await ingest_file_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert error.value.status_code == 409 + assert "different organization" in error.value.detail + assert session.rollback_count == 1 + + +@pytest.mark.asyncio +async def test_list_lineage_query_is_tenant_scoped(): + session = _Session([_Result(records=[_record()])]) + + result = await list_file_lineage(limit=50, auth_context=_auth(), db=session) + + assert len(result) == 1 + statement = str(session.statements[0]) + assert "organization_id" in statement + assert "workspace_id" in statement + assert "user_id" in statement diff --git a/backend/tests/test_disksage_file_lineage_postgres.py b/backend/tests/test_disksage_file_lineage_postgres.py new file mode 100644 index 000000000..f526552ff --- /dev/null +++ b/backend/tests/test_disksage_file_lineage_postgres.py @@ -0,0 +1,130 @@ +import uuid + +import asyncpg +import pytest +from cryptography.fernet import Fernet +from pydantic import SecretStr +from sqlalchemy import delete, text +from sqlalchemy.exc import OperationalError +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from api.auth import AuthContext +from api.disksage import ingest_file_lineage, list_file_lineage +from core.config import settings +from db.models import Base, DiskSageFileLineageRecord +from services.disksage_file_lineage import FileLineageEnvelope + + +def _envelope() -> dict[str, object]: + return { + "schema_version": 1, + "schema_kind": "disksage.file-lineage", + "source_kind": "file", + "archive_kind": "media", + "source_filename": "Video 1.mov", + "source_relative_path": "DaVinci Resolve/Video 1.mov", + "source_context": "DaVinci Resolve", + "ontology_class": "https://disksage.app/ontology#Media", + "ontology_relations": [], + "raw_content_sha256": "a" * 64, + "raw_content_blake3": "b" * 64, + "bytes": 160085038, + "production_time": { + "selected_value_ms": 1, + "selected_source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + "evidence_precedence": [ + "embedded_metadata", + "explicit_filename_date", + "filesystem_created_at", + "filesystem_modified_at", + ], + }, + "filesystem_time": {"created_at_ms": 2, "modified_at_ms": 3}, + "metadata_evidence": [ + { + "field": "production-date", + "value": "1970-01-01", + "source": "embedded:exiftool:MediaCreateDate", + "confidence": "high", + } + ], + "content_authors": [], + "content_context": [], + "review": { + "candidate_fingerprint": "c" * 64, + "review_fingerprint": "d" * 64, + "requires_review": False, + "reason_codes": [], + }, + "cloud_copy": { + "receipt_id": "e" * 64, + "lineage_fingerprint": "f" * 64, + "provider": "icloud", + "destination_account_scope": "unknown", + "destination": "/Users/example/iCloud/Video 1.mov", + "copied_at_ms": 5, + "copy_verification_method": "copied-by-disk-sage", + "local_copy_verified": True, + "provider_write_executed": False, + "provider_sync_confirmed": False, + }, + } + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_file_lineage_ingest_and_list_postgres_smoke(): + database_url = getattr(settings, "DATABASE_URL", None) + if not database_url: + pytest.skip("PostgreSQL smoke path unavailable: DATABASE_URL is not set") + + scope = uuid.uuid4().hex + auth_context = AuthContext( + user_id=f"disksage_smoke_user_{scope}", + role="member", + organization_id=f"disksage_smoke_org_{scope}", + group_ids=(), + workspace_id=f"disksage_smoke_workspace_{scope}", + ) + envelope = FileLineageEnvelope.model_validate(_envelope()) + engine = create_async_engine(database_url, echo=False) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + previous_key = settings.ENCRYPTION_KEY + settings.ENCRYPTION_KEY = SecretStr(Fernet.generate_key().decode("ascii")) + record_uid = None + try: + try: + async with engine.begin() as connection: + await connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + await connection.run_sync(Base.metadata.create_all) + except (OperationalError, asyncpg.PostgresError, OSError): + pytest.skip("PostgreSQL smoke schema is unavailable") + + async with session_factory() as session: + summary = await ingest_file_lineage( + envelope=envelope, + auth_context=auth_context, + db=session, + ) + record_uid = summary.lineage_record_uid + + async with session_factory() as session: + summaries = await list_file_lineage( + limit=50, + auth_context=auth_context, + db=session, + ) + assert [item.lineage_record_uid for item in summaries] == [record_uid] + assert summaries[0].provider_sync_state == "unknown" + finally: + settings.ENCRYPTION_KEY = previous_key + if record_uid is not None: + async with session_factory() as session: + await session.execute( + delete(DiskSageFileLineageRecord).where( + DiskSageFileLineageRecord.lineage_record_uid == record_uid + ) + ) + await session.commit() + await engine.dispose() diff --git a/backend/tests/test_disksage_organization_lineage.py b/backend/tests/test_disksage_organization_lineage.py new file mode 100644 index 000000000..c4425b9db --- /dev/null +++ b/backend/tests/test_disksage_organization_lineage.py @@ -0,0 +1,177 @@ +import datetime +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException +from pydantic import ValidationError +from sqlalchemy.exc import IntegrityError + +from api.auth import AuthContext +from api.disksage import ingest_organization_lineage, list_organization_lineage +from services.disksage_organization_lineage import ( + OrganizationLineageBatch, + canonical_batch_sha256, +) + + +def _batch() -> dict[str, object]: + return { + "schema": "disksage.organization-lineage-batch", + "version": 1, + "generated_at_ms": 1_000, + "complete": True, + "batch_fingerprint_sha256": "a" * 64, + "items": [ + { + "lineage_fingerprint": "b" * 64, + "source_size": 42, + "source_mtime_ms": 123, + "production_time_ms": 456, + "production_time_source": "embedded:exiftool:MediaCreateDate", + "production_time_confidence": "high", + "ontology_class": "https://disksage.app/ontology#Media", + "destination_relation": "targetFolder", + "action": "move", + } + ], + } + + +class _Result: + def __init__(self, record=None, records=()): + self._record = record + self._records = tuple(records) + + def scalar_one_or_none(self): + return self._record + + def scalars(self): + return self + + def all(self): + return list(self._records) + + +class _Session: + def __init__(self, results, *, commit_error=None): + self._results = iter(results) + self.commit_error = commit_error + self.statements = [] + self.commit_count = 0 + self.rollback_count = 0 + + async def execute(self, statement): + self.statements.append(statement) + return next(self._results) + + def add(self, _record): + pass + + async def commit(self): + self.commit_count += 1 + if self.commit_error is not None: + raise self.commit_error + + async def refresh(self, _record): + pass + + async def rollback(self): + self.rollback_count += 1 + + +def _auth() -> AuthContext: + return AuthContext( + user_id="user-1", + role="member", + organization_id="org-1", + group_ids=(), + workspace_id="workspace-1", + ) + + +def _record(**overrides): + batch = OrganizationLineageBatch.model_validate(_batch()) + values = { + "organization_lineage_record_uid": "disksage_org_lineage_1", + "organization_id": "org-1", + "batch_fingerprint_sha256": batch.batch_fingerprint_sha256, + "envelope_sha256": canonical_batch_sha256(batch), + "schema_version": 1, + "item_count": 1, + "ontology_classes": ["https://disksage.app/ontology#Media"], + "created_at": datetime.datetime.now(datetime.timezone.utc), + } + values.update(overrides) + return SimpleNamespace(**values) + + +def test_contract_is_path_free_and_rejects_unknown_fields(): + envelope = OrganizationLineageBatch.model_validate(_batch()) + assert envelope.items[0].ontology_class.endswith("#Media") + unsafe = _batch() + unsafe["source_path"] = "/private/source/secret.mov" + with pytest.raises(ValidationError): + OrganizationLineageBatch.model_validate(unsafe) + + +def test_contract_rejects_duplicate_lineage_fingerprints(): + unsafe = _batch() + unsafe["items"] = [unsafe["items"][0], unsafe["items"][0]] + with pytest.raises(ValidationError): + OrganizationLineageBatch.model_validate(unsafe) + + +@pytest.mark.asyncio +async def test_racing_batch_in_another_organization_returns_conflict(): + envelope = OrganizationLineageBatch.model_validate(_batch()) + record = _record( + organization_id="org-2", + envelope_sha256=canonical_batch_sha256(envelope), + ) + session = _Session( + [_Result(None), _Result(record)], + commit_error=IntegrityError("unique violation", None, Exception("duplicate")), + ) + + with pytest.raises(HTTPException) as error: + await ingest_organization_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert error.value.status_code == 409 + assert "different organization" in error.value.detail + assert session.rollback_count == 1 + assert "organization_id" not in str(session.statements[1]).split("WHERE", 1)[1] + + +@pytest.mark.asyncio +async def test_replayed_batch_is_idempotent_and_workspace_scoped(): + envelope = OrganizationLineageBatch.model_validate(_batch()) + record = _record() + session = _Session([_Result(record)]) + + result = await ingest_organization_lineage( + envelope=envelope, + auth_context=_auth(), + db=session, + ) + + assert result.organization_lineage_record_uid == record.organization_lineage_record_uid + assert session.commit_count == 0 + statement = str(session.statements[0]) + assert "workspace_id" in statement + assert "user_id" in statement + assert "organization_id" in statement + + +@pytest.mark.asyncio +async def test_list_returns_redacted_summaries_only(): + session = _Session([_Result(records=[_record()])]) + + result = await list_organization_lineage(limit=50, auth_context=_auth(), db=session) + + assert len(result) == 1 + assert result[0].ontology_classes == ["https://disksage.app/ontology#Media"] + assert "envelope_json_encrypted" not in result[0].model_dump() diff --git a/docs/architecture/disksage-file-lineage-handoff.md b/docs/architecture/disksage-file-lineage-handoff.md new file mode 100644 index 000000000..6291d6170 --- /dev/null +++ b/docs/architecture/disksage-file-lineage-handoff.md @@ -0,0 +1,70 @@ +# DiskSage file-lineage handoff + +**Status**: Accepted +**Date**: 2026-08-13 + +## Context + +DiskSage can prove a local copy, metadata precedence, review decision, and +provider synchronization evidence. A File Provider placeholder is not itself +proof that a provider API write executed. Naruon already persists content and +knowledge graphs, but the current cloud-copy handoff is only a readiness +summary and cannot represent a general file's provenance or ontology edges. + +## Decision + +Naruon accepts `disksage.file-lineage` versions 1, 2, and 3 through +`POST /api/disksage/file-lineage`. The request boundary is strict and rejects +unknown fields, unsafe relative paths, unverified copies, and +`provider_write_executed=true` claims. `copy_verification_method` may be +`copied-by-disk-sage`, `copied-by-provider-api`, or `adopted-existing`; the +provider-API variant records DiskSage's authenticated write evidence while +Naruon remains a catalog projection and never becomes the write authority. The +complete envelope is encrypted at +rest and scoped by authenticated user/workspace. `GET /api/disksage/file-lineage` +returns only `lineage_record_uid`, `lineage_fingerprint`, `schema_version`, +`source_kind`, `archive_kind`, content hashes, `content_bytes`, ontology class, +predicate projection, `provider_name`, provider-native sync state, sync status, +and `created_at`; it does not expose local paths or raw metadata values. A state +such as `pending-upload` is retained as an incomplete provider proof and never +authorizes source eviction. + +Production-time validation is repeated at the Naruon trust boundary with the +canonical order `embedded_metadata` > `explicit_filename_date` > +`filesystem_created_at` > `filesystem_modified_at`. Filename date tokens such +as `2026-04-28` or `251210` are auxiliary evidence; they cannot override an +embedded production date, and a non-embedded selection must be low confidence. + +The payload keeps explicit file → archive destination → provider/account and +review relations. These relation edges follow the same entity/provenance +separation as PROV-O; DiskSage's local ontology remains the domain vocabulary. +The path-free semantic catalog candidate batch is the handoff to +semantic-data-portal. The durable table has an explicit DBML projection at +`docs/architecture/erd/disksage-file-lineage.dbml`; pg-erd-cloud can convert it +to the same snapshot shape as a live database without granting it mutation +authority. The projection was validated against pg-erd-cloud's DBML converter; +default and simple index preservation is tracked in +[pg-erd-cloud#931](https://github.com/ContextualWisdomLab/pg-erd-cloud/pull/931). +The Alembic model remains the runtime source of truth. + +Local ontology organization uses a separate path-free contract: +`disksage.organization-lineage-batch` version 1 through +`POST /api/disksage/organization-lineage`. It records only immutable lineage +fingerprints, file size/mtime, metadata-first production time, ontology class, +the `targetFolder` relation, and the planned `move` action. Absolute paths, +names, content, provider credentials, and OAuth tokens are never part of this +payload. The endpoint stores the batch encrypted and returns only a scoped +summary; it is an audit/catalog handoff and never executes a move. + +## Consequences + +- Naruon can index provenance and ontology predicates without authorizing source + deletion or inventing provider writes. +- The encrypted envelope is not directly queryable for graph search; a later + scoped projection can be added when a catalog consumer exists. +- DiskSage remains the authority for copy, hash, review, and provider evidence. + +## References + +- [W3C PROV-O](https://www.w3.org/TR/prov-o/) +- [W3C OWL 2 overview](https://www.w3.org/TR/owl2-overview/) diff --git a/docs/architecture/erd/disksage-file-lineage.dbml b/docs/architecture/erd/disksage-file-lineage.dbml new file mode 100644 index 000000000..ce879990c --- /dev/null +++ b/docs/architecture/erd/disksage-file-lineage.dbml @@ -0,0 +1,52 @@ +// Design projection for pg-erd-cloud. Runtime truth remains the Alembic model. +// Source paths and metadata values stay inside envelope_json_encrypted. +Table public.disksage_file_lineage_records { + lineage_record_uid varchar(96) [pk] + user_id varchar [not null] + organization_id varchar + workspace_id varchar [not null] + lineage_fingerprint varchar(64) [not null] + envelope_sha256 varchar(64) [not null] + schema_version integer [not null] + schema_kind varchar(96) [not null] + source_kind varchar(64) [not null] + archive_kind varchar(64) [not null] + raw_content_sha256 varchar(64) [not null] + raw_content_blake3 varchar(64) [not null] + content_bytes bigint [not null] + ontology_class varchar(256) [not null] + ontology_relation_count integer [not null] + ontology_predicates json [not null] + provider_name varchar(32) [not null] + provider_sync_confirmed boolean [not null] + provider_sync_state varchar(32) [not null, default: 'unknown'] + envelope_json_encrypted text [not null] + created_at timestamptz [not null] + + indexes { + (user_id, workspace_id, lineage_fingerprint) [unique] + (user_id, workspace_id, created_at) + (workspace_id, ontology_class) + } +} + +// Path-free local organization handoff. The encrypted payload contains no +// queryable source/destination paths or file names. +Table public.disksage_organization_lineage_records { + organization_lineage_record_uid varchar(96) [pk] + user_id varchar [not null] + organization_id varchar + workspace_id varchar [not null] + batch_fingerprint_sha256 varchar(64) [not null] + envelope_sha256 varchar(64) [not null] + schema_version integer [not null] + item_count integer [not null] + ontology_classes json [not null] + envelope_json_encrypted text [not null] + created_at timestamptz [not null] + + indexes { + (user_id, workspace_id, batch_fingerprint_sha256) [unique] + (user_id, workspace_id, created_at) + } +}