diff --git a/src/govoplan_files/backend/capabilities.py b/src/govoplan_files/backend/capabilities.py
index 609d920..5e79516 100644
--- a/src/govoplan_files/backend/capabilities.py
+++ b/src/govoplan_files/backend/capabilities.py
@@ -1,8 +1,16 @@
from __future__ import annotations
+import base64
+import binascii
from typing import Any
+from sqlalchemy import or_
+
+from govoplan_core.core.access import AccessDecisionProvenance, PrincipalRef
+from govoplan_core.core.files import FileAccessProvider
from govoplan_core.core.modules import ModuleContext
+from govoplan_core.security.module_permissions import scopes_grant_compatible
+from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare
from govoplan_files.backend.runtime import configure_runtime
from govoplan_files.backend.storage.campaign_attachments import (
annotate_built_messages_with_managed_files,
@@ -12,6 +20,21 @@ from govoplan_files.backend.storage.campaign_attachments import (
)
from govoplan_files.backend.storage.campaign_usage import record_campaign_attachment_uses_for_jobs
from govoplan_files.backend.storage.files import current_version_and_blob
+from govoplan_files.backend.storage.paths import normalize_folder
+
+
+VIRTUAL_FOLDER_RESOURCE_PREFIX = "virtual-folder:v1"
+
+WRITE_ACTIONS = {
+ "files:file:upload",
+ "files:file:organize",
+ "files:file:share",
+ "files:file:delete",
+ "files:upload",
+ "files:organize",
+ "files:share",
+ "files:delete",
+}
class FilesCampaignCapability:
@@ -32,3 +55,247 @@ class FilesCampaignCapability:
def campaign_capability(context: ModuleContext) -> FilesCampaignCapability:
configure_runtime(registry=context.registry, settings=context.settings)
return FilesCampaignCapability()
+
+
+class FilesAccessService(FileAccessProvider):
+ def explain_resource_provenance(
+ self,
+ session: object,
+ principal: PrincipalRef,
+ *,
+ resource_type: str,
+ resource_id: str,
+ action: str,
+ ) -> tuple[AccessDecisionProvenance, ...]:
+ normalized_type = resource_type.lower().strip()
+ if normalized_type in {"file", "file_asset", "files:file"}:
+ return self._explain_file(session, principal, resource_id=resource_id, action=action)
+ if normalized_type in {"folder", "file_folder", "files:folder"}:
+ return self._explain_folder(session, principal, resource_id=resource_id, action=action)
+ return ()
+
+ def _explain_file(
+ self,
+ session: object,
+ principal: PrincipalRef,
+ *,
+ resource_id: str,
+ action: str,
+ ) -> tuple[AccessDecisionProvenance, ...]:
+ asset = session.get(FileAsset, resource_id) # type: ignore[attr-defined]
+ if asset is None or (principal.tenant_id and asset.tenant_id != principal.tenant_id):
+ return (_missing_resource("file", resource_id, principal.tenant_id, source="files.not_found"),)
+ items = [_file_resource(asset)]
+ items.extend(_owner_provenance(asset.owner_type, _owner_id(asset.owner_type, asset.owner_user_id, asset.owner_group_id), principal, source="files.owner"))
+ items.extend(_admin_provenance(principal, "files:file:admin", source="files.admin_scope"))
+ permission_values = {"write", "manage"} if _requires_write_share(action) else {"read", "write", "manage"}
+ shares = (
+ session.query(FileShare) # type: ignore[attr-defined]
+ .filter(
+ FileShare.tenant_id == asset.tenant_id,
+ FileShare.file_asset_id == asset.id,
+ FileShare.revoked_at.is_(None),
+ FileShare.permission.in_(sorted(permission_values)),
+ or_(
+ (FileShare.target_type == "user") & (FileShare.target_id == principal.membership_id),
+ (FileShare.target_type == "group") & (FileShare.target_id.in_(sorted(principal.group_ids))),
+ (FileShare.target_type == "tenant") & (FileShare.target_id == asset.tenant_id),
+ ),
+ )
+ .order_by(FileShare.target_type.asc(), FileShare.target_id.asc())
+ .all()
+ )
+ for share in shares:
+ items.append(_share_provenance(share, source="files.share"))
+ return tuple(items)
+
+ def _explain_folder(
+ self,
+ session: object,
+ principal: PrincipalRef,
+ *,
+ resource_id: str,
+ action: str,
+ ) -> tuple[AccessDecisionProvenance, ...]:
+ del action
+ folder = session.get(FileFolder, resource_id) # type: ignore[attr-defined]
+ if folder is None:
+ return self._explain_virtual_folder(session, principal, resource_id=resource_id)
+ if principal.tenant_id and folder.tenant_id != principal.tenant_id:
+ return (_missing_resource("folder", resource_id, principal.tenant_id, source="files.not_found"),)
+ return tuple([
+ AccessDecisionProvenance(
+ kind="resource",
+ id=folder.id,
+ label=folder.path,
+ tenant_id=folder.tenant_id,
+ source="files.folder",
+ details={
+ "resource_type": "folder",
+ "path": folder.path,
+ "owner_type": folder.owner_type,
+ "deleted": folder.deleted_at is not None,
+ },
+ ),
+ *_owner_provenance(folder.owner_type, _owner_id(folder.owner_type, folder.owner_user_id, folder.owner_group_id), principal, source="files.owner"),
+ *_admin_provenance(principal, "files:file:admin", source="files.admin_scope"),
+ ])
+
+ def _explain_virtual_folder(
+ self,
+ session: object,
+ principal: PrincipalRef,
+ *,
+ resource_id: str,
+ ) -> tuple[AccessDecisionProvenance, ...]:
+ folder_ref = parse_virtual_folder_resource_id(resource_id)
+ if folder_ref is None:
+ return (_missing_resource("folder", resource_id, principal.tenant_id, source="files.not_found"),)
+ tenant_id, owner_type, owner_id, path = folder_ref
+ if principal.tenant_id and tenant_id != principal.tenant_id:
+ return (_missing_resource("folder", resource_id, principal.tenant_id, source="files.not_found"),)
+ child_prefix = f"{path}/"
+ owner_filter = FileAsset.owner_user_id == owner_id if owner_type == "user" else FileAsset.owner_group_id == owner_id
+ child = (
+ session.query(FileAsset.id) # type: ignore[attr-defined]
+ .filter(
+ FileAsset.tenant_id == tenant_id,
+ FileAsset.owner_type == owner_type,
+ owner_filter,
+ FileAsset.deleted_at.is_(None),
+ FileAsset.display_path.like(f"{child_prefix}%"),
+ )
+ .first()
+ )
+ if child is None:
+ return (_missing_resource("folder", resource_id, principal.tenant_id, source="files.not_found"),)
+ return tuple([
+ AccessDecisionProvenance(
+ kind="resource",
+ id=resource_id,
+ label=path,
+ tenant_id=tenant_id,
+ source="files.virtual_folder",
+ details={
+ "resource_type": "folder",
+ "path": path,
+ "owner_type": owner_type,
+ "owner_id": owner_id,
+ "virtual": True,
+ "deleted": False,
+ },
+ ),
+ *_owner_provenance(owner_type, owner_id, principal, source="files.owner"),
+ *_admin_provenance(principal, "files:file:admin", source="files.admin_scope"),
+ ])
+
+
+def access_capability(context: ModuleContext) -> FilesAccessService:
+ configure_runtime(registry=context.registry, settings=context.settings)
+ return FilesAccessService()
+
+
+def virtual_folder_resource_id(*, tenant_id: str, owner_type: str, owner_id: str, path: str) -> str:
+ normalized_path = normalize_folder(path)
+ encoded_path = base64.urlsafe_b64encode(normalized_path.encode("utf-8")).decode("ascii").rstrip("=")
+ return f"{VIRTUAL_FOLDER_RESOURCE_PREFIX}:{tenant_id}:{owner_type}:{owner_id}:{encoded_path}"
+
+
+def parse_virtual_folder_resource_id(resource_id: str) -> tuple[str, str, str, str] | None:
+ parts = resource_id.split(":", 5)
+ if len(parts) != 6 or f"{parts[0]}:{parts[1]}" != VIRTUAL_FOLDER_RESOURCE_PREFIX:
+ return None
+ _, _, tenant_id, owner_type, owner_id, encoded_path = parts
+ if owner_type not in {"user", "group"} or not tenant_id or not owner_id or not encoded_path:
+ return None
+ padding = "=" * (-len(encoded_path) % 4)
+ try:
+ decoded_path = base64.urlsafe_b64decode(f"{encoded_path}{padding}").decode("utf-8")
+ path = normalize_folder(decoded_path)
+ except (binascii.Error, UnicodeDecodeError, ValueError):
+ return None
+ if not path:
+ return None
+ return tenant_id, owner_type, owner_id, path
+
+
+def _file_resource(asset: FileAsset) -> AccessDecisionProvenance:
+ return AccessDecisionProvenance(
+ kind="resource",
+ id=asset.id,
+ label=asset.filename,
+ tenant_id=asset.tenant_id,
+ source="files.file",
+ details={
+ "resource_type": "file",
+ "display_path": asset.display_path,
+ "owner_type": asset.owner_type,
+ "deleted": asset.deleted_at is not None,
+ },
+ )
+
+
+def _missing_resource(resource_type: str, resource_id: str, tenant_id: str | None, *, source: str) -> AccessDecisionProvenance:
+ return AccessDecisionProvenance(
+ kind="resource",
+ id=resource_id,
+ tenant_id=tenant_id,
+ source=source,
+ details={"resource_type": resource_type, "found": False},
+ )
+
+
+def _owner_id(owner_type: str, owner_user_id: str | None, owner_group_id: str | None) -> str | None:
+ return owner_user_id if owner_type == "user" else owner_group_id
+
+
+def _owner_provenance(owner_type: str, owner_id: str | None, principal: PrincipalRef, *, source: str) -> tuple[AccessDecisionProvenance, ...]:
+ if owner_id is None:
+ return ()
+ matches_user = owner_type == "user" and owner_id == principal.membership_id
+ matches_group = owner_type == "group" and owner_id in principal.group_ids
+ if not (matches_user or matches_group):
+ return ()
+ return (
+ AccessDecisionProvenance(
+ kind="owner",
+ id=owner_id,
+ tenant_id=principal.tenant_id,
+ source=source,
+ details={"owner_type": owner_type},
+ ),
+ )
+
+
+def _admin_provenance(principal: PrincipalRef, required_scope: str, *, source: str) -> tuple[AccessDecisionProvenance, ...]:
+ if not scopes_grant_compatible(principal.scopes, required_scope):
+ return ()
+ return (
+ AccessDecisionProvenance(
+ kind="policy",
+ id=required_scope,
+ label=required_scope,
+ tenant_id=principal.tenant_id,
+ source=source,
+ details={"grant": "tenant_admin"},
+ ),
+ )
+
+
+def _share_provenance(share: FileShare, *, source: str) -> AccessDecisionProvenance:
+ return AccessDecisionProvenance(
+ kind="share",
+ id=share.id,
+ label=share.permission,
+ tenant_id=share.tenant_id,
+ source=source,
+ details={
+ "target_type": share.target_type,
+ "target_id": share.target_id,
+ "permission": share.permission,
+ },
+ )
+
+
+def _requires_write_share(action: str) -> bool:
+ return action in WRITE_ACTIONS
diff --git a/src/govoplan_files/backend/manifest.py b/src/govoplan_files/backend/manifest.py
index 31b5140..bce1972 100644
--- a/src/govoplan_files/backend/manifest.py
+++ b/src/govoplan_files/backend/manifest.py
@@ -3,6 +3,7 @@ from __future__ import annotations
from pathlib import Path
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
+from govoplan_core.core.files import CAPABILITY_FILES_ACCESS
from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
from govoplan_core.core.modules import (
FrontendModule,
@@ -115,6 +116,7 @@ manifest = ModuleManifest(
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
optional_dependencies=("campaigns",),
provides_interfaces=(
+ ModuleInterfaceProvider(name="files.access", version="0.1.6"),
ModuleInterfaceProvider(name="files.campaign_attachments", version="0.1.6"),
),
requires_interfaces=(
@@ -172,6 +174,7 @@ manifest = ModuleManifest(
),
),
capability_factories={
+ CAPABILITY_FILES_ACCESS: lambda context: __import__("govoplan_files.backend.capabilities", fromlist=["access_capability"]).access_capability(context),
"files.campaign_attachments": lambda context: __import__("govoplan_files.backend.capabilities", fromlist=["campaign_capability"]).campaign_capability(context),
},
)
diff --git a/tests/test_access_provider.py b/tests/test_access_provider.py
new file mode 100644
index 0000000..e52fb0e
--- /dev/null
+++ b/tests/test_access_provider.py
@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.db.base import Base
+from govoplan_files.backend.capabilities import FilesAccessService, virtual_folder_resource_id
+from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare
+
+
+TENANT_ID = "tenant-1"
+USER_ID = "user-1"
+OTHER_USER_ID = "user-2"
+GROUP_ID = "group-1"
+
+
+class FilesAccessProviderTests(unittest.TestCase):
+ def test_file_access_provider_explains_owner_share_admin_and_missing_resources(self) -> None:
+ session = _session()
+ _seed_access_subjects(session)
+ owned = FileAsset(id="file-owned", tenant_id=TENANT_ID, owner_type="user", owner_user_id=USER_ID, display_path="owned.pdf", filename="owned.pdf")
+ shared = FileAsset(id="file-shared", tenant_id=TENANT_ID, owner_type="user", owner_user_id=OTHER_USER_ID, display_path="shared.pdf", filename="shared.pdf")
+ session.add_all([
+ owned,
+ shared,
+ FileShare(id="share-group", tenant_id=TENANT_ID, file_asset_id=shared.id, target_type="group", target_id=GROUP_ID, permission="read"),
+ ])
+ session.commit()
+
+ service = FilesAccessService()
+ owned_items = service.explain_resource_provenance(session, _principal(), resource_type="file", resource_id=owned.id, action="files:file:read")
+ shared_items = service.explain_resource_provenance(session, _principal(group_ids={GROUP_ID}), resource_type="file", resource_id=shared.id, action="files:file:read")
+ admin_items = service.explain_resource_provenance(session, _principal(scopes={"files:file:admin"}), resource_type="file", resource_id=shared.id, action="files:file:read")
+ missing_items = service.explain_resource_provenance(session, _principal(), resource_type="file", resource_id="missing-file", action="files:file:read")
+
+ self.assertTrue(any(item.kind == "resource" and item.source == "files.file" and item.id == owned.id for item in owned_items))
+ self.assertTrue(any(item.kind == "owner" and item.id == USER_ID for item in owned_items))
+ self.assertTrue(any(item.kind == "share" and item.id == "share-group" for item in shared_items))
+ self.assertTrue(any(item.kind == "policy" and item.id == "files:file:admin" for item in admin_items))
+ self.assertEqual("files.not_found", missing_items[0].source)
+ self.assertIs(missing_items[0].details["found"], False)
+
+ def test_file_access_provider_explains_persisted_and_virtual_folders(self) -> None:
+ session = _session()
+ _seed_access_subjects(session)
+ persisted = FileFolder(id="folder-persisted", tenant_id=TENANT_ID, owner_type="group", owner_group_id=GROUP_ID, path="records")
+ virtual_child = FileAsset(
+ id="file-in-virtual-folder",
+ tenant_id=TENANT_ID,
+ owner_type="group",
+ owner_group_id=GROUP_ID,
+ display_path="inferred/sub/file.pdf",
+ filename="file.pdf",
+ )
+ session.add_all([persisted, virtual_child])
+ session.commit()
+
+ service = FilesAccessService()
+ principal = _principal(group_ids={GROUP_ID})
+ persisted_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=persisted.id, action="files:file:read")
+ virtual_id = virtual_folder_resource_id(tenant_id=TENANT_ID, owner_type="group", owner_id=GROUP_ID, path="inferred/sub")
+ virtual_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=virtual_id, action="files:file:read")
+ missing_virtual_id = virtual_folder_resource_id(tenant_id=TENANT_ID, owner_type="group", owner_id=GROUP_ID, path="inferred/missing")
+ missing_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=missing_virtual_id, action="files:file:read")
+
+ self.assertTrue(any(item.kind == "resource" and item.source == "files.folder" and item.id == persisted.id for item in persisted_items))
+ self.assertTrue(any(item.kind == "owner" and item.id == GROUP_ID for item in persisted_items))
+ self.assertTrue(any(item.kind == "resource" and item.source == "files.virtual_folder" and item.id == virtual_id for item in virtual_items))
+ self.assertTrue(any(item.kind == "owner" and item.id == GROUP_ID for item in virtual_items))
+ self.assertEqual("files.not_found", missing_items[0].source)
+
+
+def _session():
+ engine = create_engine("sqlite:///:memory:", future=True)
+ Base.metadata.create_all(bind=engine, tables=[Account.__table__, User.__table__, Group.__table__, FileAsset.__table__, FileFolder.__table__, FileShare.__table__])
+ return sessionmaker(bind=engine, future=True)()
+
+
+def _seed_access_subjects(session) -> None:
+ session.add_all([
+ Account(id="account-1", email="one@example.test", normalized_email="one@example.test"),
+ Account(id="account-2", email="two@example.test", normalized_email="two@example.test"),
+ User(id=USER_ID, tenant_id=TENANT_ID, account_id="account-1", email="one@example.test"),
+ User(id=OTHER_USER_ID, tenant_id=TENANT_ID, account_id="account-2", email="two@example.test"),
+ Group(id=GROUP_ID, tenant_id=TENANT_ID, slug="group", name="Group"),
+ ])
+ session.commit()
+
+
+def _principal(*, scopes: set[str] | None = None, group_ids: set[str] | None = None) -> PrincipalRef:
+ return PrincipalRef(
+ account_id="account-1",
+ membership_id=USER_ID,
+ tenant_id=TENANT_ID,
+ scopes=frozenset(scopes or {"files:file:read"}),
+ group_ids=frozenset(group_ids or set()),
+ )
diff --git a/webui/src/api/files.ts b/webui/src/api/files.ts
index 18062cd..8db6750 100644
--- a/webui/src/api/files.ts
+++ b/webui/src/api/files.ts
@@ -559,6 +559,55 @@ export function bulkDeleteFiles(settings: ApiSettings, fileIds: string[]): Promi
export type FileBulkShareResponse = {shares: FileShare[];shared_count: number;};
+export type AccessExplanationUser = {
+ id: string;
+ account_id?: string | null;
+ email?: string | null;
+ display_name?: string | null;
+};
+
+export type AccessDecisionProvenanceItem = {
+ kind: string;
+ id?: string | null;
+ label?: string | null;
+ tenant_id?: string | null;
+ source?: string | null;
+ details?: Record
{userLabel}
{resourceLabel}
{explanation.action}
{explanation.provenance.length}
i18n:govoplan-files.no_access_evidence_was_returned.84a21e4e
: +{item.id}>}
+ {item.label}
} + {Object.keys(item.details ?? {}).length > 0 &&{formatProvenanceDetails(item.details ?? {})}
} +