Add file resource access explanations
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user