"""DMS permissions — API views. Endpoints --------- ACL Policies: GET /dms/acl/policies/ list all policies (tenant) POST /dms/acl/policies/ create policy GET /dms/acl/policies/{policy_id}/ retrieve policy PATCH /dms/acl/policies/{policy_id}/ update policy DELETE /dms/acl/policies/{policy_id}/ soft-delete policy ACL Rules (nested under policy): GET /dms/acl/policies/{policy_id}/rules/ list rules POST /dms/acl/policies/{policy_id}/rules/ add rule DELETE /dms/acl/policies/{policy_id}/rules/{rule_id}/ remove rule Document Permissions (nested under document): GET /dms/documents/{doc_id}/acl/ list direct permissions POST /dms/documents/{doc_id}/acl/ grant permission GET /dms/documents/{doc_id}/acl/{perm_id}/ retrieve permission DELETE /dms/documents/{doc_id}/acl/{perm_id}/ revoke permission GET /dms/documents/{doc_id}/acl/resolve/ resolve effective permissions Share Links (nested under document): GET /dms/documents/{doc_id}/share-links/ list share links POST /dms/documents/{doc_id}/share-links/ create share link GET /dms/documents/{doc_id}/share-links/{link_id}/ retrieve share link DELETE /dms/documents/{doc_id}/share-links/{link_id}/ revoke+delete share link POST /dms/documents/{doc_id}/share-links/{link_id}/revoke/ revoke (keep record) Public endpoint (no auth): POST /dms/share/{token}/use/ validate & consume share link """ from __future__ import annotations from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.dms.common.exceptions import AssetNotFound from simorgh.apps.dms.documents import queries as doc_queries from simorgh.apps.dms.permissions import queries, services from simorgh.apps.dms.permissions.iam_permissions import ( PERM_ACL_MANAGE, PERM_ACL_VIEW, PERM_SHARE_MANAGE, PERM_SHARE_VIEW, ) from simorgh.apps.dms.permissions.serializers import ( ACLPolicyCreateInputSerializer, ACLPolicySerializer, ACLPolicyUpdateInputSerializer, ACLRuleCreateInputSerializer, ACLRuleSerializer, DocumentPermissionGrantInputSerializer, DocumentPermissionSerializer, ResolvePermissionsInputSerializer, ShareLinkCreateInputSerializer, ShareLinkSerializer, ShareLinkUseInputSerializer, ) from simorgh.apps.dms.permissions.services import ACLError, ShareLinkError # --------------------------------------------------------------------------- # Shared helpers (same pattern as other DMS bounded contexts) # --------------------------------------------------------------------------- def _tenant(request: Request): tenant = getattr(request, "tenant", None) if tenant is None: from rest_framework.exceptions import PermissionDenied raise PermissionDenied("Tenant required.") return tenant def _org_node_id(request: Request) -> int: from simorgh.apps.memberships.models import Membership, MembershipStatus t = _tenant(request) m = ( Membership.objects.filter( tenant=t, users=request.user, status=MembershipStatus.ACTIVE, ) .order_by("id") .first() ) if m is not None: return m.organization_node_id if getattr(request.user, "is_superuser", False): from simorgh.apps.organizations.models import OrganizationNode root = OrganizationNode.objects.filter(tenant=t).order_by("id").first() if root is not None: return root.pk from rest_framework.exceptions import PermissionDenied raise PermissionDenied("No active membership for this tenant.") def _require_perm(request: Request, codename: str) -> None: from rest_framework.exceptions import PermissionDenied from simorgh.core.context import current_request_context ctx = current_request_context() if ctx.is_superuser: return if codename not in (ctx.permissions or set()): raise PermissionDenied(f"Permission required: {codename}") def _get_document(request: Request, doc_id: str): """Resolve a Document from URL param, scoped to current tenant.""" try: return doc_queries.get_document(_tenant(request).pk, doc_id) except AssetNotFound: from rest_framework.exceptions import NotFound raise NotFound(f"Document {doc_id!r} not found.") # --------------------------------------------------------------------------- # ACL Policy list / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def policy_list(request: Request) -> Response: """GET — list all active policies for tenant. POST — create a new policy.""" if request.method == "GET": _require_perm(request, PERM_ACL_VIEW) subject_type = request.query_params.get("subject_type") qs = queries.list_all_policies( _tenant(request).pk, subject_type=subject_type, include_inactive="include_inactive" in request.query_params, ) return Response(ACLPolicySerializer(qs, many=True).data) # POST _require_perm(request, PERM_ACL_MANAGE) ser = ACLPolicyCreateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: policy = services.create_policy( tenant_id=_tenant(request).pk, organization_node_id=_org_node_id(request), subject_type=d["subject_type"], subject_id=d["subject_id"], name=d["name"], description=d.get("description", ""), inheritable=d.get("inheritable", True), ) except ACLError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ACLPolicySerializer(policy).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # ACL Policy detail / update / delete # --------------------------------------------------------------------------- @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def policy_detail(request: Request, policy_id: str) -> Response: """GET — retrieve policy. PATCH — update. DELETE — soft-delete.""" try: policy = queries.get_policy(_tenant(request).pk, policy_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_ACL_VIEW) return Response(ACLPolicySerializer(policy).data) if request.method == "PATCH": _require_perm(request, PERM_ACL_MANAGE) ser = ACLPolicyUpdateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data policy = services.update_policy( policy, name=d.get("name"), description=d.get("description"), inheritable=d.get("inheritable"), is_active=d.get("is_active"), ) return Response(ACLPolicySerializer(policy).data) # DELETE _require_perm(request, PERM_ACL_MANAGE) services.delete_policy(policy) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # ACL Rule list / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def rule_list(request: Request, policy_id: str) -> Response: """GET — list rules for a policy. POST — add a rule.""" try: policy = queries.get_policy(_tenant(request).pk, policy_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_ACL_VIEW) qs = queries.list_rules_for_policy(policy) return Response(ACLRuleSerializer(qs, many=True).data) # POST _require_perm(request, PERM_ACL_MANAGE) ser = ACLRuleCreateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: rule = services.add_rule( policy, principal_type=d["principal_type"], principal_id=d.get("principal_id", ""), action=d["action"], effect=d.get("effect", "allow"), priority=d.get("priority", 0), ) except ACLError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ACLRuleSerializer(rule).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # ACL Rule delete # --------------------------------------------------------------------------- @api_view(["DELETE"]) @permission_classes([IsAuthenticated]) def rule_detail(request: Request, policy_id: str, rule_id: str) -> Response: """DELETE — remove a rule from a policy.""" _require_perm(request, PERM_ACL_MANAGE) try: policy = queries.get_policy(_tenant(request).pk, policy_id) rule = queries.get_rule(policy, rule_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) services.remove_rule(rule) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # DocumentPermission list / create / resolve # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def doc_permission_list(request: Request, doc_id: str) -> Response: """GET — list direct permissions. POST — grant permission.""" doc = _get_document(request, doc_id) if request.method == "GET": _require_perm(request, PERM_ACL_VIEW) qs = queries.list_document_permissions(doc) return Response(DocumentPermissionSerializer(qs, many=True).data) # POST _require_perm(request, PERM_ACL_MANAGE) ser = DocumentPermissionGrantInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: perm = services.grant_document_permission( document=doc, tenant_id=_tenant(request).pk, organization_node_id=_org_node_id(request), principal_type=d["principal_type"], principal_id=d.get("principal_id", ""), action=d["action"], effect=d.get("effect", "allow"), granted_by=request.user, expires_at=d.get("expires_at"), inheritable=d.get("inheritable", False), notes=d.get("notes", ""), ) except ACLError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(DocumentPermissionSerializer(perm).data, status=status.HTTP_201_CREATED) @api_view(["GET", "DELETE"]) @permission_classes([IsAuthenticated]) def doc_permission_detail(request: Request, doc_id: str, perm_id: str) -> Response: """GET — retrieve a permission. DELETE — revoke it.""" doc = _get_document(request, doc_id) try: perm = queries.get_document_permission(doc, perm_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_ACL_VIEW) return Response(DocumentPermissionSerializer(perm).data) # DELETE _require_perm(request, PERM_ACL_MANAGE) services.revoke_document_permission(perm) return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def doc_permission_resolve(request: Request, doc_id: str) -> Response: """Resolve effective ACL actions for the current user on this document.""" _require_perm(request, PERM_ACL_VIEW) doc = _get_document(request, doc_id) ser = ResolvePermissionsInputSerializer(data=request.query_params) ser.is_valid(raise_exception=True) d = ser.validated_data allowed = services.resolve_document_permissions( doc, user_id=str(request.user.pk), role_slugs=d.get("role_slugs") or [], org_node_ids=d.get("org_node_ids") or [], ) return Response({"allowed_actions": sorted(allowed)}) # --------------------------------------------------------------------------- # ShareLink list / create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def share_link_list(request: Request, doc_id: str) -> Response: """GET — list share links for a document. POST — create share link.""" doc = _get_document(request, doc_id) if request.method == "GET": _require_perm(request, PERM_SHARE_VIEW) qs = queries.list_share_links_for_document(doc) return Response(ShareLinkSerializer(qs, many=True).data) # POST _require_perm(request, PERM_SHARE_MANAGE) ser = ShareLinkCreateInputSerializer(data=request.data) ser.is_valid(raise_exception=True) d = ser.validated_data try: link = services.create_share_link( document=doc, tenant_id=_tenant(request).pk, organization_node_id=_org_node_id(request), created_by=request.user, label=d.get("label", ""), expires_at=d.get("expires_at"), max_uses=d.get("max_uses"), allow_download=d.get("allow_download", True), allow_preview=d.get("allow_preview", True), notes=d.get("notes", ""), password=d.get("password"), ) except ACLError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ShareLinkSerializer(link).data, status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # ShareLink detail / revoke / delete # --------------------------------------------------------------------------- @api_view(["GET", "DELETE"]) @permission_classes([IsAuthenticated]) def share_link_detail(request: Request, doc_id: str, link_id: str) -> Response: """GET — retrieve share link. DELETE — soft-delete.""" doc = _get_document(request, doc_id) try: link = queries.get_share_link(_tenant(request).pk, link_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if link.document_id != doc.pk: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, PERM_SHARE_VIEW) return Response(ShareLinkSerializer(link).data) # DELETE _require_perm(request, PERM_SHARE_MANAGE) link.is_deleted = True link.save(update_fields=["is_deleted"]) return Response(status=status.HTTP_204_NO_CONTENT) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def share_link_revoke(request: Request, doc_id: str, link_id: str) -> Response: """Deactivate a share link (keeps the record; just sets is_active=False).""" _require_perm(request, PERM_SHARE_MANAGE) doc = _get_document(request, doc_id) try: link = queries.get_share_link(_tenant(request).pk, link_id) except AssetNotFound: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if link.document_id != doc.pk: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) link = services.revoke_share_link(link) return Response(ShareLinkSerializer(link).data) # --------------------------------------------------------------------------- # Public share link use (no authentication required) # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([AllowAny]) def share_link_use(request: Request, token: str) -> Response: """Validate and consume a share link by token. No authentication required — this is the public entry point. Returns the link's allowed capabilities so the frontend knows what to render (preview / download buttons). A 404 is returned for any invalid/unknown/deleted token to prevent token enumeration. """ link = queries.get_share_link_by_token(token) if link is None: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) ser = ShareLinkUseInputSerializer(data=request.data) ser.is_valid(raise_exception=True) password = ser.validated_data.get("password") try: link = services.use_share_link(link, password=password or None) except ShareLinkError as exc: return Response({"detail": str(exc)}, status=status.HTTP_403_FORBIDDEN) return Response( { "document_id": str(link.document.public_id), "allow_download": link.allow_download, "allow_preview": link.allow_preview, "use_count": link.use_count, } )