80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.access import (
|
|
CAPABILITY_ACCESS_DIRECTORY,
|
|
AccessDirectory,
|
|
GroupRef,
|
|
)
|
|
from govoplan_core.core.runtime import get_registry
|
|
from govoplan_files.backend.storage.common import FileStorageError
|
|
|
|
|
|
def _access_directory() -> AccessDirectory:
|
|
registry = get_registry()
|
|
if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_ACCESS_DIRECTORY):
|
|
raise FileStorageError("Access directory capability is not configured")
|
|
capability = registry.require_capability(CAPABILITY_ACCESS_DIRECTORY)
|
|
if not isinstance(capability, AccessDirectory):
|
|
raise FileStorageError("Access directory capability is invalid")
|
|
return capability
|
|
|
|
|
|
def group_refs_for_ids(*, tenant_id: str, group_ids: list[str]) -> list[GroupRef]:
|
|
if not group_ids:
|
|
return []
|
|
groups = _access_directory().get_groups(group_ids)
|
|
return sorted(
|
|
[group for group in groups.values() if group.tenant_id == tenant_id],
|
|
key=lambda group: group.name.casefold(),
|
|
)
|
|
|
|
|
|
def user_group_ids(session: Session, *, tenant_id: str, user_id: str, include_admin_groups: bool = False) -> list[str]:
|
|
del session
|
|
directory = _access_directory()
|
|
if include_admin_groups:
|
|
return [group.id for group in directory.groups_for_tenant(tenant_id)]
|
|
return [group.id for group in directory.groups_for_user(user_id, tenant_id=tenant_id)]
|
|
|
|
|
|
def ensure_group_access(session: Session, *, tenant_id: str, group_id: str, user_id: str, is_admin: bool = False) -> None:
|
|
group = _access_directory().get_group(group_id)
|
|
if not group or group.tenant_id != tenant_id:
|
|
raise FileStorageError("Group not found")
|
|
if is_admin:
|
|
return
|
|
if group_id not in user_group_ids(session, tenant_id=tenant_id, user_id=user_id):
|
|
raise FileStorageError("No access to this group file space")
|
|
|
|
|
|
def ensure_owner_access(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, user_id: str, is_admin: bool = False) -> None:
|
|
owner_type = owner_type.lower().strip()
|
|
if owner_type == "user":
|
|
if owner_id != user_id and not is_admin:
|
|
raise FileStorageError("No access to this user file space")
|
|
return
|
|
if owner_type == "group":
|
|
ensure_group_access(session, tenant_id=tenant_id, group_id=owner_id, user_id=user_id, is_admin=is_admin)
|
|
return
|
|
raise FileStorageError("Files must be owned by a user or group")
|
|
|
|
|
|
def ensure_share_target_exists(*, tenant_id: str, target_type: str, target_id: str) -> None:
|
|
target_type = target_type.lower().strip()
|
|
if target_type == "user":
|
|
user = _access_directory().get_user(target_id)
|
|
if not user or user.tenant_id != tenant_id or user.status != "active":
|
|
raise FileStorageError("User not found")
|
|
return
|
|
if target_type == "group":
|
|
group = _access_directory().get_group(target_id)
|
|
if not group or group.tenant_id != tenant_id or group.status != "active":
|
|
raise FileStorageError("Group not found")
|
|
return
|
|
if target_type == "tenant":
|
|
if target_id != tenant_id:
|
|
raise FileStorageError("Tenant not found")
|
|
return
|