GovOPlaN/govoplan-files
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add governed Files DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-20 22:40:33 +02:00
parent 6c3cf1c55e
commit ce4eaefbfd
6 changed files with 1481 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_dsar_provider.py
+468
View File
@@ -0,0 +1,468 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from govoplan_access.backend.db.models import Account, Group, User
from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
DataSubjectRequest,
create_data_subject_request,
execute_data_subject_erasure,
plan_data_subject_erasure,
search_data_subject_request,
)
from govoplan_files.backend.db.models import (
FileAsset,
FileBlob,
FileConnectorCredential,
FileConnectorPolicy,
FileConnectorProfile,
FileConnectorSpace,
FileFolder,
FileFormEvidenceGrant,
FileIntegrityFinding,
FileIntegrityScan,
FileShare,
FileVersion,
)
from govoplan_files.backend.dsar_provider import (
FILES_DSAR_CAPABILITY,
FilesDsarProvider,
)
from govoplan_files.backend.manifest import manifest
class _Registry:
def __init__(self, provider: FilesDsarProvider, *, files_active: bool = True) -> None:
self.provider = provider
self.files_active = files_active
def capability_names(self):
return (FILES_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "files"
def tenant_entitlement_resolver(self):
files_active = self.files_active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("files",) if files_active else ()},
)()
return _Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
self._assert_capability(name)
return self.provider
def manifests(self):
return (type("Manifest", (), {"id": "files"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != FILES_DSAR_CAPABILITY:
raise KeyError(name)
class FilesDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(
bind=self.engine,
tables=[
Account.__table__,
User.__table__,
Group.__table__,
ChangeSequenceEntry.__table__,
DataSubjectRequest.__table__,
FileBlob.__table__,
FileAsset.__table__,
FileVersion.__table__,
FileFolder.__table__,
FileFormEvidenceGrant.__table__,
FileShare.__table__,
FileConnectorCredential.__table__,
FileConnectorPolicy.__table__,
FileConnectorProfile.__table__,
FileConnectorSpace.__table__,
FileIntegrityScan.__table__,
FileIntegrityFinding.__table__,
],
)
self.session = sessionmaker(bind=self.engine, future=True)()
self.account = Account(
id="account-1",
email="subject@example.test",
normalized_email="subject@example.test",
display_name="Subject",
password_hash="not-exported",
)
self.user = User(
id="membership-1",
tenant_id="tenant-1",
account_id=self.account.id,
email="subject@example.test",
display_name="Subject",
)
self.blob = FileBlob(
id="blob-1",
tenant_id="tenant-1",
storage_backend="local",
storage_key="private/storage/key-do-not-export",
checksum_sha256="a" * 64,
size_bytes=12,
ref_count=1,
)
self.asset = FileAsset(
id="asset-1",
tenant_id="tenant-1",
owner_type="user",
owner_user_id=self.user.id,
created_by_user_id=self.user.id,
current_version_id="version-1",
display_path="subjects/private.txt",
filename="private.txt",
description="Subject-provided document",
retained_until=datetime.now(timezone.utc) + timedelta(days=30),
lifecycle_reason="Pending proceeding",
metadata_={"password": "metadata-secret-do-not-export"},
)
self.version = FileVersion(
id="version-1",
tenant_id="tenant-1",
file_asset_id=self.asset.id,
blob_id=self.blob.id,
version_number=1,
filename_at_upload=self.asset.filename,
display_path_at_upload=self.asset.display_path,
content_type="text/plain",
size_bytes=12,
checksum_sha256="a" * 64,
created_by_user_id=self.user.id,
)
self.folder = FileFolder(
id="folder-1",
tenant_id="tenant-1",
owner_type="user",
owner_user_id=self.user.id,
path="subjects",
created_by_user_id=self.user.id,
)
self.share = FileShare(
id="share-1",
tenant_id="tenant-1",
file_asset_id=self.asset.id,
target_type="user",
target_id=self.user.id,
permission="read",
created_by_user_id=self.user.id,
)
self.form_evidence = FileFormEvidenceGrant(
id="evidence-1",
tenant_id="tenant-1",
form_instance_id="form-instance-1",
form_definition_id="application",
form_definition_revision="7",
token_sha256="b" * 64,
idempotency_key="evidence-key-1",
request_sha256="c" * 64,
custodian_user_id=self.user.id,
evidence_kind="attachment",
purpose="Submitted application evidence",
status="uploaded",
expires_at=datetime.now(timezone.utc) + timedelta(days=1),
max_size_bytes=1024,
allowed_content_types=["text/plain"],
file_asset_id=self.asset.id,
file_version_id=self.version.id,
metadata_={},
)
self.credential = FileConnectorCredential(
id="credential-1",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
label="Subject-created credential",
provider="s3",
credential_mode="database",
username="subject-user",
password_encrypted="encrypted-password-do-not-export",
token_encrypted="encrypted-token-do-not-export",
password_env="PASSWORD_ENV_DO_NOT_EXPORT",
secret_ref="vault://do-not-export",
created_by_user_id=self.user.id,
updated_by_user_id=self.user.id,
)
self.integrity_scan = FileIntegrityScan(
id="scan-1",
tenant_id="tenant-1",
storage_backend="local",
storage_prefix="private-prefix-do-not-export",
created_by_user_id=self.user.id,
)
tenant_two_asset = FileAsset(
id="asset-tenant-2",
tenant_id="tenant-2",
owner_type="user",
owner_user_id=self.user.id,
display_path="other-tenant.txt",
filename="other-tenant.txt",
)
self.session.add_all(
[
self.account,
self.user,
self.blob,
self.asset,
self.version,
self.folder,
self.share,
self.form_evidence,
self.credential,
self.integrity_scan,
tenant_two_asset,
]
)
self.session.commit()
self.provider = FilesDsarProvider()
self.subject = DsarSubjectRef(membership_id=self.user.id)
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def test_manifest_publishes_protocol_conforming_provider(self) -> None:
provided_names = {item.name for item in manifest.provides_interfaces}
self.assertIn(FILES_DSAR_CAPABILITY, provided_names)
provider = manifest.capability_factories[FILES_DSAR_CAPABILITY](None)
self.assertIsInstance(provider, DsarProvider)
def test_search_is_tenant_scoped_and_excludes_bytes_and_secrets(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self.subject,
)
resource_types = {record.resource_type for record in records}
self.assertTrue(
{
"file_asset",
"file_version",
"file_folder",
"file_share",
"file_form_evidence",
"connector_credential",
"file_integrity_scan",
}.issubset(resource_types)
)
serialized = repr([record.to_dict() for record in records])
self.assertNotIn("asset-tenant-2", serialized)
self.assertNotIn("private/storage/key-do-not-export", serialized)
self.assertNotIn("metadata-secret-do-not-export", serialized)
self.assertNotIn("encrypted-password-do-not-export", serialized)
self.assertNotIn("encrypted-token-do-not-export", serialized)
self.assertNotIn("subject-user", serialized)
self.assertNotIn("PASSWORD_ENV_DO_NOT_EXPORT", serialized)
self.assertNotIn("vault://do-not-export", serialized)
self.assertNotIn("private-prefix-do-not-export", serialized)
def test_conflicting_direct_subject_references_fail_closed(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
membership_id=self.user.id,
external_references={"files.user": "another-membership"},
),
)
self.assertEqual((), records)
def test_plan_classifies_retention_and_manual_review(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self.subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
records=records,
)
kinds = {action.kind for action in actions}
self.assertTrue({"retain", "manual_review", "revoke", "detach"}.issubset(kinds))
asset_retention = next(
action
for action in actions
if action.action_id == "files:retain:file_asset:asset-1"
)
self.assertIn("retained until", asset_retention.rationale)
self.assertTrue(
any(action.action_id == "files:revoke:file_share:share-1" for action in actions)
)
self.assertFalse(
any(action.kind == "delete" and action.executable for action in actions)
)
def test_execution_is_revalidated_and_idempotent(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self.subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
records=records,
)
executable = tuple(action for action in actions if action.executable)
first = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=executable,
request_id="dsar-1",
)
self.assertEqual({"executed"}, {result.status for result in first})
self.assertIsNotNone(self.share.revoked_at)
self.assertIsNone(self.asset.created_by_user_id)
self.assertIsNone(self.version.created_by_user_id)
self.assertIsNone(self.folder.created_by_user_id)
self.assertIsNone(self.credential.created_by_user_id)
self.assertIsNone(self.credential.updated_by_user_id)
repeated = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=executable,
request_id="dsar-1",
)
self.assertEqual({"unchanged"}, {result.status for result in repeated})
def test_execution_blocks_when_reference_changed_after_planning(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self.subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
records=records,
)
detach = next(
action
for action in actions
if action.action_id
== "files:detach:file_version:created_by_user_id:version-1"
)
self.version.created_by_user_id = "replacement-user"
self.session.flush()
result = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=self.subject,
actions=(detach,),
request_id="dsar-2",
)
self.assertEqual("blocked", result[0].status)
self.assertEqual("replacement-user", self.version.created_by_user_id)
def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled(
self,
) -> None:
request = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-FILES-1",
request_kind="access_and_erasure",
subject=self.subject,
purpose="Respond to an authorized privacy request.",
legal_basis="Article 15 and 17 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
registry = _Registry(self.provider)
search_data_subject_request(
self.session,
registry=registry,
row=request,
expected_revision=1,
)
self.assertEqual(
"searched", request.status, request.search_result["provider_runs"]
)
self.assertEqual(["files"], request.coverage["covered_modules"])
self.assertEqual([], request.coverage["modules_without_provider"])
plan_data_subject_erasure(
self.session,
registry=registry,
row=request,
expected_revision=2,
)
executable_ids = [
action["action_id"]
for action in request.erasure_plan["actions"]
if action["executable"]
]
execute_data_subject_erasure(
self.session,
registry=registry,
row=request,
expected_revision=3,
action_ids=executable_ids,
)
self.assertEqual("completed", request.status)
disabled = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-FILES-DISABLED",
request_kind="access",
subject=self.subject,
purpose="Verify disabled-module coverage.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
search_data_subject_request(
self.session,
registry=_Registry(self.provider, files_active=False),
row=disabled,
expected_revision=1,
)
self.assertEqual(0, disabled.search_result["record_count"])
self.assertEqual(
[FILES_DSAR_CAPABILITY],
disabled.coverage["inactive_provider_capabilities"],
)
if __name__ == "__main__":
unittest.main()
tests/test_manifest_documentation.py
+9
View File
@@ -12,6 +12,7 @@ STATIC_TOPIC_IDS = {
"files.workflow.share-managed-files",
"files.workflow.delete-managed-files",
"files.workflow.restore-retain-and-purge",
"files.privacy.data-subject-requests",
"files.governed-connectors-and-provenance",
"files.reference.integrity-recovery-and-fail-closed-transports",
"files.reference.shared-storage-profile",
@@ -130,6 +131,14 @@ class FilesManifestDocumentationTests(unittest.TestCase):
self.assertIn("preview hash", lifecycle.body)
self.assertIn("recovery ledger", lifecycle.body)
privacy = self.topic("files.privacy.data-subject-requests")
self.assertEqual(("admin",), privacy.documentation_types)
self.assertIn("without raw file bytes", privacy.body)
self.assertIn("separately authorized Files purge", privacy.body)
self.assertTrue(
any("membership" in item for item in privacy.metadata["limitations"])
)
def test_admin_topic_covers_policy_redaction_and_atomic_credential_deletion(
self,
) -> None: