feat(campaign): explain governed child access

This commit is contained in:
2026-08-20 05:38:00 +02:00
parent 112ef9dc31
commit 5c27527725
5 changed files with 531 additions and 12 deletions
+27 -6
View File
@@ -16,6 +16,11 @@ and they must identify that inheritance explicitly.
- Persisted validation issue, version-bound review decision, and attachment-policy override
- SMTP, IMAP append, Postbox, and printable-output attempts
- Message action, message-action attempt, and job reconciliation decision
- Campaign share and Core-owned Campaign ownership-transfer record
- Independently user-owned recipient import mapping profile
- Saved recipient import execution, identified by `<version UUID>:<import UUID>`
- Persisted validation, build, execution-snapshot, and review evidence, identified
by `<version UUID>:<artifact kind>`
All persisted child IDs are random UUIDs. Embedded build/review children use a
version UUID plus a random job UUID, so callers cannot enumerate a recipient
@@ -35,6 +40,10 @@ snapshots, diagnostic text, or reconciliation notes.
| Transport or worker diagnostics | Campaign read/owner/share | `campaigns:diagnostic:read` |
| Exported delivery evidence | Campaign read/owner/share | `campaigns:report:export` |
| Reconciliation decision | Campaign read/owner/share | Campaign reconcile and diagnostic read |
| Share or ownership transfer | Campaign governance | Campaign share, transfer-participant, group-acceptance, or recovery authority; content access remains a separate decision |
| Import mapping profile | Independent user owner | `campaigns:recipient:import`; no Campaign share is inherited |
| Import execution | Campaign read/owner/share | Recipient read and import authority |
| Persisted protocol artifact | Campaign read/owner/share | Recipient, review, report, diagnostic, or export authority appropriate to the artifact |
Postbox, Mail/IMAP, and printable attempts keep bounded Campaign-owned evidence
after provider acceptance. Their explanation therefore remains available when
@@ -42,13 +51,19 @@ an optional provider module is later disabled. A missing attempt reports only
the optional owner and `unavailable_or_hidden`; it does not distinguish absence
from hidden data.
## Remaining owner boundaries
## Optional and unsupported owner boundaries
1. Campaign shares and ownership-transfer records
2. Import mapping profiles and import executions
3. Reusable Campaign templates and template revisions when the template
library becomes persistent
4. Export packages and protocol/report artifacts
Reusable templates and template revisions are independently governed by the
optional Templates module; Campaign never treats a Campaign share as a template
grant. Durable export packages are independently governed by the optional
Reporting module. Asking the Campaign provider to explain either class therefore
fails closed with `independently_governed_by_optional_module` and
`unavailable_or_hidden`. The response does not reveal whether the optional
module is absent, the object does not exist, or the caller cannot see it.
Campaign reports generated on demand remain non-persisted, version-bound
resources. Their explanation names the report kind and its Campaign/version
parent, and keeps report read, export, and diagnostic permissions distinct.
Each child explanation must include:
@@ -64,3 +79,9 @@ Delivery attempts, review decisions, reports, and exports can contain more
sensitive evidence than the Campaign summary. Their read and diagnostic/export
permissions therefore remain independently enforceable even when the parent
Campaign is readable.
Import explanations include only the stable import identity, source type,
opaque source identity, source revision, and whether additional provenance was
recorded. They never return imported rows, filenames, column mappings, or source
metadata. Mapping-profile explanations expose only non-reversible header
fingerprints and shape information; headers and mappings remain hidden.
@@ -19,6 +19,7 @@ from govoplan_core.core.campaigns import (
from govoplan_core.core.ownership import (
OwnershipActionDecision,
OwnershipSubjectRef,
OwnershipTransfer,
OwnershipTransferError,
)
from govoplan_core.security.module_permissions import scopes_grant_compatible
@@ -35,6 +36,7 @@ from govoplan_campaign.backend.db.models import (
ImapAppendAttempt,
PostboxDeliveryAttempt,
PrintOutputAttempt,
RecipientImportMappingProfile,
SendAttempt,
)
@@ -114,6 +116,36 @@ CAMPAIGN_RECONCILIATION_RESOURCE_TYPES = {
"campaign_reconciliation_decision",
"campaigns:reconciliation_decision",
}
CAMPAIGN_SHARE_RESOURCE_TYPES = {"campaign_share", "campaigns:share"}
CAMPAIGN_OWNERSHIP_TRANSFER_RESOURCE_TYPES = {
"campaign_ownership_transfer",
"campaigns:ownership_transfer",
}
CAMPAIGN_IMPORT_MAPPING_PROFILE_RESOURCE_TYPES = {
"campaign_import_mapping_profile",
"campaigns:import_mapping_profile",
}
CAMPAIGN_IMPORT_EXECUTION_RESOURCE_TYPES = {
"campaign_import_execution",
"campaigns:import_execution",
}
CAMPAIGN_PROTOCOL_ARTIFACT_RESOURCE_TYPES = {
"campaign_protocol_artifact",
"campaigns:protocol_artifact",
}
CAMPAIGN_TEMPLATE_RESOURCE_TYPES = {
"campaign_template",
"campaigns:template",
"campaign_template_revision",
"campaigns:template_revision",
}
CAMPAIGN_EXPORT_PACKAGE_RESOURCE_TYPES = {
"campaign_export_package",
"campaigns:export_package",
}
CAMPAIGN_PROTOCOL_ARTIFACT_KINDS = frozenset(
{"validation", "build", "execution_snapshot", "review"}
)
RECIPIENT_ACCESS_REQUIREMENTS = (
"campaigns:campaign:read",
@@ -136,6 +168,8 @@ RECONCILIATION_ACCESS_REQUIREMENTS = (
"campaigns:campaign:reconcile",
"campaigns:diagnostic:read",
)
GOVERNANCE_ACCESS_REQUIREMENTS = ("campaigns:campaign:share",)
IMPORT_PROFILE_ACCESS_REQUIREMENTS = ("campaigns:recipient:import",)
def campaign_report_resource_id(
@@ -156,6 +190,19 @@ def campaign_version_child_resource_id(*, version_id: str, child_id: str) -> str
return f"{version_id}:{child_id}"
def campaign_import_execution_resource_id(*, version_id: str, import_id: str) -> str:
"""Return the stable typed reference for one saved recipient import."""
return campaign_version_child_resource_id(version_id=version_id, child_id=import_id)
def campaign_protocol_artifact_resource_id(*, version_id: str, artifact_kind: str) -> str:
"""Return the stable reference for one persisted version evidence class."""
kind = artifact_kind.strip().lower()
if kind not in CAMPAIGN_PROTOCOL_ARTIFACT_KINDS:
raise ValueError("Campaign protocol artifact kind is not supported")
return campaign_version_child_resource_id(version_id=version_id, child_id=kind)
def _campaign_report_reference(
resource_id: str,
) -> tuple[str, str, str] | None:
@@ -172,6 +219,51 @@ def _version_child_reference(resource_id: str) -> tuple[str, str] | None:
return parts[0], parts[1]
def _saved_import(version: CampaignVersion, import_id: str) -> Mapping[str, object] | None:
raw_json = version.raw_json if isinstance(version.raw_json, Mapping) else {}
entries = raw_json.get("entries")
imports = entries.get("imports") if isinstance(entries, Mapping) else ()
for item in imports if isinstance(imports, list) else ():
if isinstance(item, Mapping) and str(item.get("id") or "") == import_id:
return item
return None
def _protocol_artifact_details(version: CampaignVersion, artifact_kind: str) -> Mapping[str, object] | None:
if artifact_kind == "validation":
if not isinstance(version.validation_summary, Mapping):
return None
return {"artifact_kind": artifact_kind, "persisted": True, "artifact_content_disclosed": False}
if artifact_kind == "build":
if not isinstance(version.build_summary, Mapping):
return None
return {"artifact_kind": artifact_kind, "persisted": True, "artifact_content_disclosed": False}
if artifact_kind == "execution_snapshot":
if not version.execution_snapshot_hash:
return None
return {
"artifact_kind": artifact_kind,
"persisted": True,
"snapshot_hash": version.execution_snapshot_hash,
"recorded_at": _iso_value(version.execution_snapshot_at),
"artifact_content_disclosed": False,
}
if artifact_kind == "review":
editor_state = version.editor_state if isinstance(version.editor_state, Mapping) else {}
review_state = editor_state.get("review_send")
if not isinstance(review_state, Mapping):
return None
decisions = review_state.get("issue_decisions")
return {
"artifact_kind": artifact_kind,
"persisted": True,
"decision_count": len(decisions) if isinstance(decisions, list) else 0,
"reason_content_disclosed": False,
"artifact_content_disclosed": False,
}
return None
def _iso_value(value: object | None) -> str | None:
isoformat = getattr(value, "isoformat", None)
return str(isoformat()) if callable(isoformat) else None
@@ -408,8 +500,130 @@ class CampaignAccessService(CampaignAccessProvider):
normalized_type = resource_type.lower().strip()
child_item: AccessDecisionProvenance | None = None
campaign: Campaign | None
if normalized_type in CAMPAIGN_TEMPLATE_RESOURCE_TYPES:
return _missing_resource_provenance(
principal,
resource_type=normalized_type,
resource_id=resource_id,
reason="independently_governed_by_optional_module",
optional_module="templates",
)
if normalized_type in CAMPAIGN_EXPORT_PACKAGE_RESOURCE_TYPES:
return _missing_resource_provenance(
principal,
resource_type=normalized_type,
resource_id=resource_id,
reason="independently_governed_by_optional_module",
optional_module="reporting",
)
if normalized_type in CAMPAIGN_IMPORT_MAPPING_PROFILE_RESOURCE_TYPES:
profile = session.get(RecipientImportMappingProfile, resource_id) # type: ignore[attr-defined]
if profile is None or (principal.tenant_id and profile.tenant_id != principal.tenant_id):
return _missing_resource_provenance(
principal,
resource_type="campaign_import_mapping_profile",
resource_id=resource_id,
)
items = [
AccessDecisionProvenance(
kind="resource",
id=profile.id,
label="Recipient import mapping profile",
tenant_id=profile.tenant_id,
source="campaigns.import_mapping_profile",
details={
"resource_type": "campaign_import_mapping_profile",
"authorization_mode": "independently_owned",
"permission_actions": list(IMPORT_PROFILE_ACCESS_REQUIREMENTS),
"column_count": profile.column_count,
"ordered_header_fingerprint": profile.ordered_header_fingerprint,
"unordered_header_fingerprint": profile.unordered_header_fingerprint,
"headers_disclosed": False,
"mapping_disclosed": False,
"owner_identity_disclosed": False,
},
)
]
if profile.owner_user_id == principal.membership_id:
items.append(
AccessDecisionProvenance(
kind="owner",
id=profile.owner_user_id,
tenant_id=profile.tenant_id,
source="campaigns.import_mapping_profile_owner",
details={"owner_type": "user"},
)
)
items.extend(_tenant_admin_provenance(principal))
return tuple(items)
if normalized_type in CAMPAIGN_RESOURCE_TYPES:
campaign = session.get(Campaign, resource_id) # type: ignore[attr-defined]
elif normalized_type in CAMPAIGN_SHARE_RESOURCE_TYPES:
share = session.get(CampaignShare, resource_id) # type: ignore[attr-defined]
if share is None:
return _missing_resource_provenance(principal, resource_type="campaign_share", resource_id=resource_id)
campaign = session.get(Campaign, share.campaign_id) # type: ignore[attr-defined]
child_item = _child_provenance(
principal,
resource_id=share.id,
source="campaigns.share_record",
label="Campaign share",
campaign=campaign,
version_id=None,
details={
"resource_type": "campaign_share",
"state": "revoked" if share.revoked_at else "active",
"permission": share.permission,
"target_type": share.target_type,
"target_identity_disclosed": False,
"authorization_mode": "governance_visibility_separate_from_content_access",
"content_access_effect": "none" if share.revoked_at else share.permission,
},
required_actions=GOVERNANCE_ACCESS_REQUIREMENTS,
)
elif normalized_type in CAMPAIGN_OWNERSHIP_TRANSFER_RESOURCE_TYPES:
transfer = session.get(OwnershipTransfer, resource_id) # type: ignore[attr-defined]
if transfer is None or transfer.resource_module != "campaigns" or transfer.resource_type != "campaign":
return _missing_resource_provenance(
principal,
resource_type="campaign_ownership_transfer",
resource_id=resource_id,
)
campaign = session.get(Campaign, transfer.resource_id) # type: ignore[attr-defined]
if campaign is None or transfer.tenant_id != campaign.tenant_id:
return _missing_resource_provenance(
principal,
resource_type="campaign_ownership_transfer",
resource_id=resource_id,
)
child_item = _child_provenance(
principal,
resource_id=transfer.id,
source="campaigns.ownership_transfer",
label="Campaign ownership transfer",
campaign=campaign,
version_id=None,
details={
"resource_type": "campaign_ownership_transfer",
"kind": transfer.kind,
"status": transfer.status,
"revision": transfer.revision,
"current_owner_type": transfer.current_owner_type,
"target_owner_type": transfer.target_owner_type,
"expires_at": _iso_value(transfer.expires_at),
"completed_at": _iso_value(transfer.completed_at),
"participant_identities_disclosed": False,
"reason_recorded": bool(transfer.reason),
"reason_disclosed": False,
"authorization_mode": "governance_visibility_separate_from_content_access",
"permission_classes": {
"governance": ["campaigns:campaign:share"],
"group_acceptance": ["campaigns:ownership:accept_group"],
"recovery": ["campaigns:ownership:recover"],
},
},
required_actions=GOVERNANCE_ACCESS_REQUIREMENTS,
)
elif normalized_type in CAMPAIGN_VERSION_RESOURCE_TYPES:
version = session.get(CampaignVersion, resource_id) # type: ignore[attr-defined]
if version is None:
@@ -436,6 +650,111 @@ class CampaignAccessService(CampaignAccessProvider):
},
},
)
elif normalized_type in CAMPAIGN_IMPORT_EXECUTION_RESOURCE_TYPES:
reference = _version_child_reference(resource_id)
if reference is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_import_execution",
resource_id=resource_id,
reason="invalid_reference",
)
version_id, import_id = reference
version = session.get(CampaignVersion, version_id) # type: ignore[attr-defined]
if version is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_import_execution",
resource_id=resource_id,
)
campaign = session.get(Campaign, version.campaign_id) # type: ignore[attr-defined]
saved_import = _saved_import(version, import_id)
if saved_import is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_import_execution",
resource_id=resource_id,
reason="import_not_recorded",
)
child_item = _child_provenance(
principal,
resource_id=resource_id,
source="campaigns.import_execution",
label="Saved recipient import",
campaign=campaign,
version_id=version.id,
details={
"resource_type": "campaign_import_execution",
"import_id": import_id,
"imported_at": saved_import.get("imported_at"),
"mode": saved_import.get("mode"),
"source_type": saved_import.get("source_type"),
"source_id": saved_import.get("source_id"),
"source_revision": saved_import.get("source_revision"),
"source_provenance_recorded": bool(saved_import.get("source_provenance")),
"source_rows_disclosed": False,
"filename_disclosed": False,
"mapping_disclosed": False,
},
required_actions=(*RECIPIENT_ACCESS_REQUIREMENTS, "campaigns:recipient:import"),
)
elif normalized_type in CAMPAIGN_PROTOCOL_ARTIFACT_RESOURCE_TYPES:
reference = _version_child_reference(resource_id)
if reference is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_protocol_artifact",
resource_id=resource_id,
reason="invalid_reference",
)
version_id, artifact_kind = reference
if artifact_kind not in CAMPAIGN_PROTOCOL_ARTIFACT_KINDS:
return _missing_resource_provenance(
principal,
resource_type="campaign_protocol_artifact",
resource_id=resource_id,
reason="unsupported_artifact_kind",
)
version = session.get(CampaignVersion, version_id) # type: ignore[attr-defined]
if version is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_protocol_artifact",
resource_id=resource_id,
)
campaign = session.get(Campaign, version.campaign_id) # type: ignore[attr-defined]
artifact_details = _protocol_artifact_details(version, artifact_kind)
if artifact_details is None:
return _missing_resource_provenance(
principal,
resource_type="campaign_protocol_artifact",
resource_id=resource_id,
reason="artifact_not_recorded",
)
required_actions = (
RECIPIENT_ACCESS_REQUIREMENTS
if artifact_kind == "execution_snapshot"
else REVIEW_EVIDENCE_ACCESS_REQUIREMENTS
)
child_item = _child_provenance(
principal,
resource_id=resource_id,
source="campaigns.protocol_artifact",
label=f"Campaign {artifact_kind.replace('_', ' ')} evidence",
campaign=campaign,
version_id=version.id,
details={
"resource_type": "campaign_protocol_artifact",
"permission_classes": {
"read": ["campaigns:campaign:read"],
"report": ["campaigns:report:read"],
"diagnostic": ["campaigns:diagnostic:read"],
"export": ["campaigns:report:export"],
},
**artifact_details,
},
required_actions=required_actions,
)
elif normalized_type in CAMPAIGN_RECIPIENT_SNAPSHOT_RESOURCE_TYPES:
version = session.get(CampaignVersion, resource_id) # type: ignore[attr-defined]
if version is None:
@@ -883,6 +1202,12 @@ class CampaignAccessService(CampaignAccessProvider):
"campaign_version_id": version.id,
"report_kind": report_kind,
"persisted": False,
"authorization_mode": "inherited_and_further_restricted",
"permission_classes": {
"read": ["campaigns:report:read"],
"export": ["campaigns:report:export"],
"diagnostic": ["campaigns:diagnostic:read"],
},
"authorization_inherited_from": {
"resource_type": "campaign",
"resource_id": campaign.id,
+17 -4
View File
@@ -656,7 +656,7 @@ manifest = ModuleManifest(
DocumentationTopic(
id="campaigns.access.child-evidence",
title="Explain access to Campaign child evidence",
summary="Trace recipient, attachment, review, delivery, and reconciliation access without disclosing the protected payload.",
summary="Trace governance, import, recipient, attachment, protocol, delivery, and reconciliation access without disclosing the protected payload.",
body=(
"Campaign child explanations first identify the parent Campaign and immutable version, then state whether owner, "
"group, share, or tenant administration provides the inherited boundary. Recipient evidence additionally requires "
@@ -665,7 +665,11 @@ manifest = ModuleManifest(
"and random job UUID. Missing, cross-tenant, and stale references return non-disclosing provenance. Explanations expose "
"bounded state and hashes only, never recipient addresses, source rows, filenames, storage locators, transport responses, "
"worker claims, provider targets, or operator notes. Persisted Mail, Postbox, and printable attempt evidence remains "
"explainable after an optional provider is disabled; an absent child reports only unavailable-or-hidden."
"explainable after an optional provider is disabled; an absent child reports only unavailable-or-hidden. Campaign share "
"and ownership-transfer records describe governance visibility separately from the content grant. Saved imports expose "
"only their stable source identity and revision, while independently user-owned mapping profiles never inherit a Campaign "
"share. Persisted validation, build, snapshot, and review artifacts use typed version references. Reusable templates and "
"durable export packages stay with their optional owning modules and fail closed when Campaign is asked to explain them."
),
layer="available",
documentation_types=("admin",),
@@ -693,7 +697,7 @@ manifest = ModuleManifest(
translations={
"de": {
"title": "Zugriff auf untergeordnete Campaign-Nachweise erklaeren",
"summary": "Zugriff auf Empfaenger-, Anlagen-, Pruef-, Zustell- und Abgleichnachweise ohne Offenlegung der geschuetzten Inhalte nachvollziehen.",
"summary": "Zugriff auf Governance-, Import-, Empfaenger-, Anlagen-, Protokoll-, Zustell- und Abgleichnachweise ohne Offenlegung der geschuetzten Inhalte nachvollziehen.",
"body": (
"Zugriffserklaerungen fuer untergeordnete Campaign-Nachweise nennen zuerst die uebergeordnete Campaign und "
"die unveraenderliche Version. Danach zeigen sie, ob Eigentum, Gruppe, Freigabe oder Mandantenadministration "
@@ -703,12 +707,21 @@ manifest = ModuleManifest(
"oder veraltete Verweise liefern keine geschuetzten Daten. Adressen, Quellzeilen, Dateinamen, Speicherorte, Transportantworten, "
"Worker-Claims, Anbieterziele und Bediennotizen werden nie offengelegt. Dauerhafte Mail-, Postbox- und Drucknachweise bleiben "
"auch nach Deaktivierung eines optionalen Anbieters erklaerbar; ein fehlender Nachweis meldet nur nicht verfuegbar oder verborgen."
" Campaign-Freigaben und Eigentumsuebertragungen trennen Governance-Sichtbarkeit vom Inhaltszugriff. Gespeicherte Importe "
"nennen nur stabile Quellidentitaet und Revision; benutzereigene Zuordnungsprofile erben keine Campaign-Freigabe. "
"Validierungs-, Build-, Snapshot- und Pruefnachweise verwenden typisierte Versionsverweise. Wiederverwendbare Vorlagen "
"und dauerhafte Exportpakete bleiben bei ihren optionalen Eigentuemer-Modulen und werden sonst geschlossen behandelt."
),
}
},
metadata={
"kind": "reference",
"help_contexts": ["campaign.access", "campaign.report", "campaign.operator-queue"],
"help_contexts": [
"campaign.access",
"campaign.import",
"campaign.report",
"campaign.operator-queue",
],
},
),
DocumentationTopic(
+142 -2
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
import json
import unittest
from unittest.mock import patch
@@ -12,6 +12,8 @@ from govoplan_access.backend.db.models import Account, Group, User
from govoplan_campaign.backend.capabilities import (
CampaignAccessService,
CampaignOwnershipService,
campaign_import_execution_resource_id,
campaign_protocol_artifact_resource_id,
campaign_report_resource_id,
campaign_version_child_resource_id,
)
@@ -28,11 +30,12 @@ from govoplan_campaign.backend.db.models import (
ImapAppendAttempt,
PostboxDeliveryAttempt,
PrintOutputAttempt,
RecipientImportMappingProfile,
SendAttempt,
)
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.ownership import OwnershipSubjectRef, OwnershipTransferError
from govoplan_core.core.ownership import OwnershipSubjectRef, OwnershipTransfer, OwnershipTransferError
from govoplan_core.db.base import Base
@@ -565,6 +568,141 @@ class CampaignAccessProviderTests(unittest.TestCase):
self.assertEqual("unavailable_or_hidden", unavailable[0].details["evidence_availability"])
self.assertNotIn("hidden postbox address", json.dumps(unavailable[0].details))
def test_governance_import_and_protocol_explanation_matrix(self) -> None:
session = _session()
self.addCleanup(_close_session, session)
_seed_access_subjects(session)
now = datetime.now(UTC)
campaign = Campaign(
id="campaign-governed-children", tenant_id=TENANT_ID,
owner_user_id=USER_ID, external_id="governed-children", name="Governed children",
)
share = CampaignShare(
id="share-governed-child", tenant_id=TENANT_ID, campaign_id=campaign.id,
target_type="user", target_id=OTHER_USER_ID, permission="read",
)
version = CampaignVersion(
id="version-governed-children", campaign_id=campaign.id, version_number=2,
raw_json={"entries": {"inline": [], "imports": [{
"id": "import-stable-id", "imported_at": "2026-08-20T08:00:00+00:00",
"mode": "replace", "source_type": "addresses", "source_id": "address-source-42",
"source_revision": "revision-7",
"source_provenance": {"secret_source_row": "do-not-disclose-row"},
"filename": "hidden-import.csv",
"mapping": [{"header": "Secret column", "field_name": "secret"}],
}]}},
validation_summary={"secret_issue": "do-not-disclose-validation"},
build_summary={"secret_recipient": "do-not-disclose-build"},
execution_snapshot={"rows": ["do-not-disclose-snapshot"]},
execution_snapshot_hash="a" * 64, execution_snapshot_at=now,
editor_state={"review_send": {"issue_decisions": [
{"reason": "do-not-disclose-review-reason"}
]}},
)
profile = RecipientImportMappingProfile(
id="mapping-profile-governed-child", tenant_id=TENANT_ID, owner_user_id=USER_ID,
name="Hidden profile name", column_count=2,
headers=["Secret A", "Secret B"], normalized_headers=["secret a", "secret b"],
ordered_header_fingerprint="b" * 64, unordered_header_fingerprint="c" * 64,
delimiter=";", header_rows=1, quoted=True, value_separators=",;|",
mappings=[{"header": "Secret A", "field": "hidden"}],
)
transfer = OwnershipTransfer(
id="transfer-governed-child", tenant_id=TENANT_ID,
resource_module="campaigns", resource_type="campaign", resource_id=campaign.id,
kind="transfer", status="pending_target",
current_owner_type="user", current_owner_id=USER_ID,
target_owner_type="user", target_owner_id=OTHER_USER_ID,
initiated_by_type="user", initiated_by_id=USER_ID,
reason="do-not-disclose-transfer-reason", required_approvals=1,
approvals=[], decisions=[], idempotency_key="transfer-idempotency",
canonical_request_hash="d" * 64, expires_at=now + timedelta(days=1),
revision=3, metadata_={"secret": "do-not-disclose-transfer-metadata"},
)
session.add_all([campaign, share, version, profile, transfer])
session.commit()
service = CampaignAccessService()
principal = _principal(scopes={
"campaigns:campaign:read", "campaigns:campaign:share",
"campaigns:recipient:read", "campaigns:recipient:import",
"campaigns:campaign:review", "campaigns:diagnostic:read",
"campaigns:report:read",
})
import_id = campaign_import_execution_resource_id(
version_id=version.id, import_id="import-stable-id",
)
protocol_ids = [
campaign_protocol_artifact_resource_id(version_id=version.id, artifact_kind=kind)
for kind in ("validation", "build", "execution_snapshot", "review")
]
cases = [
("campaign_share", share.id, "campaigns.share_record"),
("campaign_ownership_transfer", transfer.id, "campaigns.ownership_transfer"),
("campaign_import_execution", import_id, "campaigns.import_execution"),
*(("campaign_protocol_artifact", item, "campaigns.protocol_artifact") for item in protocol_ids),
]
hidden_values = (
OTHER_USER_ID, "do-not-disclose-row", "hidden-import.csv", "Secret column",
"do-not-disclose-validation", "do-not-disclose-build", "do-not-disclose-snapshot",
"do-not-disclose-review-reason", "do-not-disclose-transfer-reason",
"do-not-disclose-transfer-metadata",
)
for resource_type, resource_id, source in cases:
items = service.explain_resource_provenance(
session, principal, resource_type=resource_type, resource_id=resource_id,
action="campaigns:diagnostic:read",
)
child = next(item for item in items if item.source == source)
self.assertEqual(campaign.id, child.details["campaign_id"])
serialized = json.dumps(child.details, sort_keys=True)
for hidden_value in hidden_values:
self.assertNotIn(hidden_value, serialized)
import_item = next(item for item in service.explain_resource_provenance(
session, principal, resource_type="campaign_import_execution",
resource_id=import_id, action="campaigns:recipient:import",
) if item.source == "campaigns.import_execution")
self.assertEqual("address-source-42", import_item.details["source_id"])
self.assertEqual("revision-7", import_item.details["source_revision"])
self.assertFalse(import_item.details["source_rows_disclosed"])
profile_items = service.explain_resource_provenance(
session, principal, resource_type="campaign_import_mapping_profile",
resource_id=profile.id, action="campaigns:recipient:import",
)
profile_item = next(item for item in profile_items if item.source == "campaigns.import_mapping_profile")
self.assertEqual("independently_owned", profile_item.details["authorization_mode"])
self.assertTrue(any(item.kind == "owner" for item in profile_items))
self.assertNotIn("Secret A", json.dumps(profile_item.details))
self.assertNotIn("Hidden profile name", json.dumps(profile_item.to_dict()))
for resource_type, optional_module in (
("campaign_template", "templates"),
("campaign_template_revision", "templates"),
("campaign_export_package", "reporting"),
):
unavailable = service.explain_resource_provenance(
session, principal, resource_type=resource_type,
resource_id="opaque-optional-resource", action="campaigns:diagnostic:read",
)[0]
self.assertEqual("campaigns.not_found", unavailable.source)
self.assertEqual(optional_module, unavailable.details["optional_module"])
self.assertEqual("independently_governed_by_optional_module", unavailable.details["reason"])
self.assertEqual("unavailable_or_hidden", unavailable.details["evidence_availability"])
unsupported = service.explain_resource_provenance(
session, principal, resource_type="campaign_protocol_artifact",
resource_id=f"{version.id}:unsupported", action="campaigns:diagnostic:read",
)
self.assertEqual("unsupported_artifact_kind", unsupported[0].details["reason"])
cross_tenant = service.explain_resource_provenance(
session, _principal(tenant_id="tenant-2"),
resource_type="campaign_import_mapping_profile", resource_id=profile.id,
action="campaigns:recipient:import",
)
self.assertEqual("campaigns.not_found", cross_tenant[0].source)
def test_campaign_ownership_provider_requires_group_acceptance_authority(self) -> None:
session = _session()
self.addCleanup(_close_session, session)
@@ -791,6 +929,8 @@ def _session():
ImapAppendAttempt.__table__,
PostboxDeliveryAttempt.__table__,
PrintOutputAttempt.__table__,
RecipientImportMappingProfile.__table__,
OwnershipTransfer.__table__,
ChangeSequenceEntry.__table__,
],
)
+20
View File
@@ -105,6 +105,26 @@ def test_attachment_reuse_workflow_documents_policy_and_evidence() -> None:
]
def test_access_evidence_reference_documents_independent_owner_boundaries() -> None:
from govoplan_campaign.backend.manifest import get_manifest
topic = next(
item for item in get_manifest().documentation
if item.id == "campaigns.access.child-evidence"
)
rendered = f"{topic.summary}\n{topic.body}".lower()
for expected in (
"governance visibility",
"mapping profiles",
"validation",
"reusable templates",
"export packages",
"fail closed",
):
assert expected in rendered
assert "campaign.import" in topic.metadata["help_contexts"]
def test_runtime_documentation_is_user_only_and_requires_a_campaign_task() -> None:
assert _topics({"docs:documentation:read"}) == ()
assert _topics({"campaigns:campaign:read"}, documentation_type="admin") == ()