"""Managed archive routes: real authorization, preview and extraction pipeline. Only external blob persistence, directory membership and audit delivery are isolated; source/version/share checks and destination writes use SQLite. """ from __future__ import annotations import hashlib import io import tarfile import unittest import zipfile from types import SimpleNamespace from unittest.mock import patch import pyzipper from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from govoplan_access.backend.db.models import Account, Group, User from govoplan_core.auth import get_api_principal from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base from govoplan_core.db.session import get_session from govoplan_files.backend.db.models import ( FileAsset, FileBlob, FileConnectorPolicy, FileShare, FileVersion, ) from govoplan_files.backend.routes.managed_archives import router from govoplan_files.backend.storage.common import FileStorageError, utcnow from govoplan_files.backend.storage.files import create_file_asset def archive_bytes(entries=None, *, password=None, tar=False): entries = entries or {"folder/one.txt": b"one", "other/two.txt": b"two"} output = io.BytesIO() if tar: with tarfile.open(fileobj=output, mode="w:gz") as archive: for name, data in entries.items(): member = tarfile.TarInfo(name) member.size = len(data) archive.addfile(member, io.BytesIO(data)) else: archive_type = pyzipper.AESZipFile if password else zipfile.ZipFile options = {"encryption": pyzipper.WZ_AES} if password else {} with archive_type( output, "w", compression=zipfile.ZIP_DEFLATED, **options ) as archive: if password: archive.setpassword(password.encode()) for name, data in entries.items(): archive.writestr(name, data) return output.getvalue() class ManagedArchiveTests(unittest.TestCase): def setUp(self): self.engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all( self.engine, tables=[ model.__table__ for model in ( Account, User, Group, ChangeSequenceEntry, FileBlob, FileAsset, FileVersion, FileShare, FileConnectorPolicy, ) ], ) self.session = Session(self.engine, expire_on_commit=False) self.addCleanup(self.engine.dispose) self.addCleanup(self.session.close) self.objects = {} self.scopes = {"files:file:read", "files:file:download", "files:file:upload"} self.principal = SimpleNamespace( tenant_id="tenant-1", user=SimpleNamespace(id="user-1"), has=lambda scope: scope in self.scopes, ) self.settings = SimpleNamespace( file_upload_zip_max_bytes=10 * 1024 * 1024, file_archive_max_entries=10_000, file_archive_max_expanded_bytes=20 * 1024 * 1024, file_archive_max_expansion_ratio=100, file_archive_preview_ttl_seconds=1800, file_upload_max_bytes=1024 * 1024, ) for module in ("uploads", "managed_archives"): self.enterContext( patch(f"govoplan_files.backend.routes.{module}.settings", self.settings) ) self.enterContext( patch( "govoplan_files.backend.storage.files.user_group_ids", return_value=[] ) ) self.enterContext( patch( "govoplan_files.backend.storage.files._get_or_create_blob", side_effect=self.store_blob, ) ) self.enterContext( patch( "govoplan_files.backend.storage.files.get_storage_backend", return_value=SimpleNamespace( name="test", get_bytes=lambda key: self.objects[key] ), ) ) self.audit = self.enterContext( patch("govoplan_files.backend.route_support.audit_from_principal") ) self.enterContext( patch( "govoplan_files.backend.route_support.asset_is_audit_relevant", return_value=False, ) ) self.enterContext( patch( "govoplan_files.backend.route_support._sent_campaign_asset_ids", return_value=set(), ) ) self.source = self.seed_source(archive_bytes()) app = FastAPI() app.include_router(router, prefix="/api/v1") app.dependency_overrides[get_api_principal] = lambda: self.principal app.dependency_overrides[get_session] = lambda: self.session self.client = self.enterContext(TestClient(app)) def store_blob(self, session, *, tenant_id, data, content_type=None, **kwargs): checksum = hashlib.sha256(data).hexdigest() blob = FileBlob( tenant_id=tenant_id, storage_backend="test", storage_key=f"objects/{checksum}", checksum_sha256=checksum, size_bytes=len(data), content_type=content_type, ref_count=1, ) session.add(blob) session.flush() self.objects[blob.storage_key] = data return blob def seed_source(self, data, filename="source.zip"): stored = create_file_asset( self.session, tenant_id="tenant-1", owner_type="user", owner_id="user-1", user_id="user-1", filename=filename, data=data, folder="archives", ) self.session.commit() return stored def preview(self, **changes): return self.client.post( f"/api/v1/files/{self.source.asset.id}/archive-preview", json={ "source_version_id": self.source.version.id, "path": "extracted", **changes, }, ) def confirm(self, token, **changes): return self.client.post( f"/api/v1/files/{self.source.asset.id}/archive-confirm", json={ "source_version_id": self.source.version.id, "path": "extracted", "preview_token": token, "selected_paths": ["folder/one.txt"], **changes, }, ) def token(self): response = self.preview() self.assertEqual(200, response.status_code, response.text) return response.json()["preview_token"] def assert_no_extraction(self): self.assertEqual(1, self.session.query(FileAsset).count()) self.assertIsNone(self.source.asset.deleted_at) self.assertEqual(self.source.version.id, self.source.asset.current_version_id) self.audit.assert_not_called() def test_preview_is_read_only_and_confirmation_extracts_only_selected_members(self): token = self.token() self.assert_no_extraction() original_data = self.objects[self.source.blob.storage_key] response = self.confirm(token) self.assertEqual(200, response.status_code, response.text) files = response.json()["files"] self.assertEqual( ["extracted/folder/one.txt"], [file["display_path"] for file in files] ) self.assertEqual(2, self.session.query(FileAsset).count()) self.assertEqual(original_data, self.objects[self.source.blob.storage_key]) self.assertEqual(self.source.version.id, self.source.asset.current_version_id) self.assertIsNone(self.source.asset.deleted_at) self.assertEqual( self.source.asset.id, files[0]["source_provenance"]["metadata"]["archive_source_file_id"], ) self.assertEqual( self.source.version.id, files[0]["source_provenance"]["metadata"]["archive_source_version_id"], ) self.audit.assert_called_once() def test_all_three_permissions_are_required(self): token = self.token() for scope in tuple(self.scopes): with self.subTest(scope=scope): self.scopes.remove(scope) self.assertEqual(403, self.preview().status_code) self.assertEqual(403, self.confirm(token).status_code) self.scopes.add(scope) self.assert_no_extraction() def test_changed_source_version_is_rejected_before_read_or_write(self): token = self.token() self.source.asset.current_version_id = "replacement-version" self.session.commit() self.assertIn("version changed", self.confirm(token).text) self.assertEqual(400, self.preview().status_code) self.assertEqual(1, self.session.query(FileAsset).count()) def test_mismatched_version_or_blob_tenant_is_rejected(self): self.source.version.file_asset_id = "other-file" self.session.commit() self.assertIn("does not belong", self.preview().text) self.source.version.file_asset_id = self.source.asset.id self.source.blob.tenant_id = "other-tenant" self.session.commit() self.assertIn("does not belong", self.preview().text) self.assert_no_extraction() def test_expired_preview_cannot_extract(self): with patch("cryptography.fernet.time.time", return_value=1): token = self.token() response = self.confirm(token) self.assertEqual(400, response.status_code, response.text) self.assertIn("expired", response.text) self.assert_no_extraction() def test_source_refresh_during_verified_read_cannot_change_preview_or_extract(self): token = self.token() replacement_version = SimpleNamespace(id="refreshed-version") with patch( "govoplan_files.backend.routes.managed_archives.read_asset_bytes", return_value=( self.objects[self.source.blob.storage_key], replacement_version, self.source.blob, ), ): preview = self.preview() confirmed = self.confirm(token) for response in (preview, confirmed): self.assertEqual(400, response.status_code, response.text) self.assertIn("version changed during the verified read", response.text) self.assert_no_extraction() def test_revoked_shared_access_cannot_reuse_preview(self): self.source.asset.owner_user_id = "other-user" share = FileShare( tenant_id="tenant-1", file_asset_id=self.source.asset.id, target_type="user", target_id="user-1", permission="read", ) self.session.add(share) self.session.commit() token = self.token() share.revoked_at = utcnow() self.session.commit() response = self.confirm(token) self.assertEqual(400, response.status_code, response.text) self.assertIn("No access", response.text) self.assert_no_extraction() def test_deleted_or_cross_tenant_sources_are_rejected(self): token = self.token() self.source.asset.deleted_at = utcnow() self.session.commit() self.assertIn("File not found", self.confirm(token).text) self.source.asset.deleted_at = None self.source.asset.tenant_id = "other-tenant" self.session.commit() self.assertIn("File not found", self.confirm(token).text) self.assert_no_extraction() def test_confirmation_is_bound_to_destination_actor_version_and_file(self): token = self.token() for change in ( {"path": "elsewhere"}, {"owner_id": "other-user"}, {"source_version_id": "wrong"}, ): with self.subTest(change=change): response = self.confirm(token, **change) self.assertEqual(400, response.status_code, response.text) self.assertIn("does not match", response.text) self.principal.user.id = "other-user" self.assertIn("does not match", self.confirm(token).text) self.principal.user.id = "user-1" self.assertEqual(400, self.confirm("tampered-token").status_code) response = self.client.post( "/api/v1/files/other-file/archive-confirm", json={ "source_version_id": self.source.version.id, "path": "extracted", "preview_token": token, "selected_paths": ["folder"], }, ) self.assertIn("does not match", response.text) self.assert_no_extraction() def test_destination_ownership_is_checked_during_preview(self): response = self.preview(owner_id="other-user") self.assertEqual(400, response.status_code, response.text) self.assertIn("No access to this user file space", response.text) self.assert_no_extraction() def test_connector_policy_is_rechecked_at_confirmation(self): self.source.asset.metadata_ = { "source_provenance": { "source_type": "webdav", "provider": "webdav", "external_id": "archive", } } self.session.commit() token = self.token() self.session.add( FileConnectorPolicy( tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", policy={"deny": {"providers": ["webdav"]}}, ) ) self.session.commit() response = self.confirm(token) self.assertEqual(403, response.status_code, response.text) self.assertEqual(403, self.preview().status_code) self.assert_no_extraction() def test_source_collision_rolls_back_all_destination_writes(self): self.source = self.seed_source( archive_bytes( {"a-first.txt": b"first", "collision.zip": b"cannot replace source"} ), "collision.zip", ) response = self.preview(path="archives") self.assertEqual(200, response.status_code, response.text) result = self.confirm( response.json()["preview_token"], path="archives", selected_paths=["a-first.txt", "collision.zip"], ) self.assertEqual(400, result.status_code, result.text) self.assertIn("already exists", result.text) self.assertEqual(2, self.session.query(FileAsset).count()) self.assertEqual(self.source.version.id, self.source.asset.current_version_id) self.assertIsNone(self.source.asset.deleted_at) def test_encrypted_zip_uses_the_same_password_and_selection_flow(self): self.source = self.seed_source( archive_bytes(password="correct horse"), "encrypted.zip" ) preview = self.preview() self.assertTrue(preview.json()["requires_password"]) self.assertFalse(preview.json()["password_verified"]) token = preview.json()["preview_token"] self.assertEqual(400, self.confirm(token).status_code) self.assertEqual(400, self.preview(password="wrong").status_code) verified = self.preview(password="correct horse") self.assertTrue(verified.json()["password_verified"]) response = self.confirm( verified.json()["preview_token"], password="correct horse" ) self.assertEqual(200, response.status_code, response.text) self.assertNotIn("correct horse", response.text) self.assertNotIn("correct horse", str(self.audit.call_args_list)) def test_tar_reuses_the_same_confirmation_flow(self): self.source = self.seed_source(archive_bytes(tar=True), "source.tar.gz") preview = self.preview() self.assertEqual("tar.gz", preview.json()["archive_format"]) self.assertEqual(200, self.confirm(preview.json()["preview_token"]).status_code) def test_unsafe_members_and_size_limits_fail_before_extraction(self): self.settings.file_upload_zip_max_bytes = 1 self.assertIn("size limit", self.preview().text) self.settings.file_upload_zip_max_bytes = 10 * 1024 * 1024 self.source = self.seed_source( archive_bytes({"../escape": b"no"}), "unsafe.zip" ) self.assertIn("Unsafe archive member", self.preview().text) self.audit.assert_not_called() def test_quarantined_or_protected_sources_use_verified_read_path(self): self.source.blob.quarantined_at = utcnow() self.session.commit() self.assertIn("quarantin", self.preview().text.lower()) self.source.blob.quarantined_at = None self.source.blob.encryption_envelope_id = "protected-envelope" self.session.commit() with patch( "govoplan_files.backend.storage.content_protection.unprotect_blob_content", side_effect=FileStorageError("Decryption is unavailable"), ) as decrypt: response = self.preview() self.assertEqual(400, response.status_code, response.text) self.assertIn("Decryption is unavailable", response.text) decrypt.assert_called_once() self.assert_no_extraction() if __name__ == "__main__": unittest.main()