391 lines
15 KiB
Python
391 lines
15 KiB
Python
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.files import (
|
|
ManagedArtifactRef,
|
|
ManagedArtifactStore,
|
|
ManagedArtifactWriteRequest,
|
|
)
|
|
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,
|
|
managed_match_payloads,
|
|
prepared_campaign_snapshot,
|
|
public_attachment_summary_payload,
|
|
share_assets_with_campaign,
|
|
)
|
|
from govoplan_files.backend.storage.campaign_usage import record_campaign_attachment_uses_for_jobs
|
|
from govoplan_files.backend.storage.files import (
|
|
create_file_asset,
|
|
current_version_and_blob,
|
|
sync_file_asset_from_source,
|
|
)
|
|
from govoplan_files.backend.storage.paths import normalize_folder
|
|
from govoplan_files.backend.storage.share_state import effective_file_share_clause
|
|
|
|
|
|
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:
|
|
prepared_campaign_snapshot = staticmethod(prepared_campaign_snapshot)
|
|
managed_match_payloads = staticmethod(managed_match_payloads)
|
|
public_attachment_summary_payload = staticmethod(public_attachment_summary_payload)
|
|
annotate_built_messages_with_managed_files = staticmethod(annotate_built_messages_with_managed_files)
|
|
record_campaign_attachment_uses_for_jobs = staticmethod(record_campaign_attachment_uses_for_jobs)
|
|
current_version_and_blob = staticmethod(current_version_and_blob)
|
|
share_assets_with_campaign = staticmethod(share_assets_with_campaign)
|
|
|
|
@staticmethod
|
|
def mark_job_attachment_uses_sent(session: Any, job: Any) -> None:
|
|
from govoplan_files.backend.storage.campaign_usage import mark_job_attachment_uses_sent
|
|
|
|
mark_job_attachment_uses_sent(session, job)
|
|
|
|
|
|
def campaign_capability(context: ModuleContext) -> FilesCampaignCapability:
|
|
configure_runtime(registry=context.registry, settings=context.settings)
|
|
return FilesCampaignCapability()
|
|
|
|
|
|
class FilesArtifactStore(ManagedArtifactStore):
|
|
def store_artifact(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ManagedArtifactWriteRequest,
|
|
) -> ManagedArtifactRef:
|
|
if not hasattr(session, "query") or not hasattr(session, "flush"):
|
|
raise TypeError("Files artifact storage requires a SQLAlchemy session.")
|
|
if not hasattr(principal, "has") or not principal.has("files:file:upload"):
|
|
raise PermissionError("Managed artifact storage requires files:file:upload.")
|
|
user = getattr(principal, "user", None)
|
|
user_id = str(getattr(user, "id", "") or "")
|
|
tenant_id = str(getattr(principal, "tenant_id", "") or "")
|
|
if not user_id or not tenant_id:
|
|
raise PermissionError("Managed artifact storage requires a tenant user principal.")
|
|
metadata = dict(request.metadata)
|
|
if request.idempotency_key:
|
|
metadata["source_provenance"] = {
|
|
"source_type": "generated_artifact",
|
|
"connector_id": "files.artifact_store",
|
|
"provider": str(metadata.get("producer_module") or "platform"),
|
|
"external_id": request.idempotency_key,
|
|
"revision": str(metadata.get("output_sha256") or "") or None,
|
|
}
|
|
stored, _action, _previous_version_id = sync_file_asset_from_source(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
owner_type="user",
|
|
owner_id=user_id,
|
|
user_id=user_id,
|
|
filename=request.filename,
|
|
data=request.payload,
|
|
metadata=metadata,
|
|
folder=request.folder,
|
|
content_type=request.content_type,
|
|
conflict_strategy="rename",
|
|
is_admin=principal.has("files:file:admin"),
|
|
)
|
|
else:
|
|
stored = create_file_asset(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
owner_type="user",
|
|
owner_id=user_id,
|
|
user_id=user_id,
|
|
filename=request.filename,
|
|
data=request.payload,
|
|
folder=request.folder,
|
|
content_type=request.content_type,
|
|
description=request.description,
|
|
metadata=metadata,
|
|
conflict_strategy="rename",
|
|
is_admin=principal.has("files:file:admin"),
|
|
)
|
|
return ManagedArtifactRef(
|
|
file_asset_id=stored.asset.id,
|
|
file_version_id=stored.version.id,
|
|
filename=stored.version.filename_at_upload,
|
|
display_path=stored.asset.display_path,
|
|
content_type=stored.version.content_type or request.content_type,
|
|
size_bytes=stored.version.size_bytes,
|
|
sha256=stored.version.checksum_sha256,
|
|
provenance={
|
|
"module": "files",
|
|
"owner_type": stored.asset.owner_type,
|
|
"managed": True,
|
|
},
|
|
)
|
|
|
|
|
|
def artifact_store_capability(context: ModuleContext) -> FilesArtifactStore:
|
|
configure_runtime(registry=context.registry, settings=context.settings)
|
|
return FilesArtifactStore()
|
|
|
|
|
|
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,
|
|
effective_file_share_clause(),
|
|
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
|