Add campaign resource access explanations

src/govoplan_campaign/backend/capabilities.py

@@ -4,6 +4,7 @@ from collections.abc import Iterable, Mapping
 
 from sqlalchemy import and_, or_
 
+from govoplan_core.core.access import AccessDecisionProvenance, PrincipalRef
 from govoplan_core.core.campaigns import (
     CampaignAccessProvider,
     CampaignDeliveryTaskProvider,
@@ -13,10 +14,14 @@ from govoplan_core.core.campaigns import (
     CampaignPolicyContextProvider,
     CampaignRetentionProvider,
 )
+from govoplan_core.security.module_permissions import scopes_grant_compatible
 
 from govoplan_campaign.backend.db.models import Campaign, CampaignShare
 
 
+READ_ACTIONS = {"campaigns:campaign:read", "campaign:read"}
+
+
 class CampaignMailPolicyContextService(CampaignMailPolicyContextProvider):
     def get_campaign_mail_policy_context(
         self,
@@ -105,11 +110,122 @@ class CampaignAccessService(CampaignAccessProvider):
         )
         return share is not None
 
+    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 not in {"campaign", "campaign_object", "campaigns:campaign"}:
+            return ()
+        campaign = session.get(Campaign, resource_id)  # type: ignore[attr-defined]
+        if campaign is None or (principal.tenant_id and campaign.tenant_id != principal.tenant_id):
+            return (
+                AccessDecisionProvenance(
+                    kind="resource",
+                    id=resource_id,
+                    tenant_id=principal.tenant_id,
+                    source="campaigns.not_found",
+                    details={"resource_type": "campaign", "found": False},
+                ),
+            )
+        items = [
+            AccessDecisionProvenance(
+                kind="resource",
+                id=campaign.id,
+                label=campaign.name,
+                tenant_id=campaign.tenant_id,
+                source="campaigns.campaign",
+                details={
+                    "resource_type": "campaign",
+                    "external_id": campaign.external_id,
+                    "status": campaign.status,
+                },
+            )
+        ]
+        items.extend(_owner_provenance(campaign, principal))
+        items.extend(_tenant_admin_provenance(principal))
+        permission_values = {"read", "write"} if action in READ_ACTIONS else {"write"}
+        shares = (
+            session.query(CampaignShare)  # type: ignore[attr-defined]
+            .filter(
+                CampaignShare.tenant_id == campaign.tenant_id,
+                CampaignShare.campaign_id == campaign.id,
+                CampaignShare.revoked_at.is_(None),
+                CampaignShare.permission.in_(sorted(permission_values)),
+                or_(
+                    (CampaignShare.target_type == "user") & (CampaignShare.target_id == principal.membership_id),
+                    (CampaignShare.target_type == "group") & (CampaignShare.target_id.in_(sorted(principal.group_ids))),
+                ),
+            )
+            .order_by(CampaignShare.target_type.asc(), CampaignShare.target_id.asc())
+            .all()
+        )
+        for share in shares:
+            items.append(
+                AccessDecisionProvenance(
+                    kind="share",
+                    id=share.id,
+                    label=share.permission,
+                    tenant_id=share.tenant_id,
+                    source="campaigns.share",
+                    details={
+                        "target_type": share.target_type,
+                        "target_id": share.target_id,
+                        "permission": share.permission,
+                    },
+                )
+            )
+        return tuple(items)
+
 
 def access_capability(context: object) -> CampaignAccessService:
     return CampaignAccessService()
 
 
+def _owner_provenance(campaign: Campaign, principal: PrincipalRef) -> tuple[AccessDecisionProvenance, ...]:
+    if campaign.owner_user_id and campaign.owner_user_id == principal.membership_id:
+        return (
+            AccessDecisionProvenance(
+                kind="owner",
+                id=campaign.owner_user_id,
+                tenant_id=campaign.tenant_id,
+                source="campaigns.owner",
+                details={"owner_type": "user"},
+            ),
+        )
+    if campaign.owner_group_id and campaign.owner_group_id in principal.group_ids:
+        return (
+            AccessDecisionProvenance(
+                kind="owner",
+                id=campaign.owner_group_id,
+                tenant_id=campaign.tenant_id,
+                source="campaigns.owner",
+                details={"owner_type": "group"},
+            ),
+        )
+    return ()
+
+
+def _tenant_admin_provenance(principal: PrincipalRef) -> tuple[AccessDecisionProvenance, ...]:
+    if not scopes_grant_compatible(principal.scopes, "tenant:*"):
+        return ()
+    return (
+        AccessDecisionProvenance(
+            kind="policy",
+            id="tenant:*",
+            label="tenant:*",
+            tenant_id=principal.tenant_id,
+            source="campaigns.tenant_admin",
+            details={"grant": "tenant_admin"},
+        ),
+    )
+
+
 class CampaignPolicyContextService(CampaignPolicyContextProvider):
     def get_campaign_policy_context(
         self,