832 lines
31 KiB
Python
832 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db.models import Account, Group, User
|
|
from govoplan_campaign.backend.capabilities import (
|
|
CampaignAccessService,
|
|
CampaignOwnershipService,
|
|
campaign_report_resource_id,
|
|
campaign_version_child_resource_id,
|
|
)
|
|
from govoplan_campaign.backend.db.models import (
|
|
AttachmentBlob,
|
|
AttachmentInstance,
|
|
Campaign,
|
|
CampaignIssue,
|
|
CampaignJob,
|
|
CampaignMessageAction,
|
|
CampaignMessageActionAttempt,
|
|
CampaignShare,
|
|
CampaignVersion,
|
|
ImapAppendAttempt,
|
|
PostboxDeliveryAttempt,
|
|
PrintOutputAttempt,
|
|
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.db.base import Base
|
|
|
|
|
|
TENANT_ID = "tenant-1"
|
|
USER_ID = "user-1"
|
|
OTHER_USER_ID = "user-2"
|
|
GROUP_ID = "group-1"
|
|
|
|
|
|
class CampaignAccessProviderTests(unittest.TestCase):
|
|
def test_campaign_access_provider_explains_owner_share_admin_and_missing_resources(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
owned = Campaign(id="campaign-owned", tenant_id=TENANT_ID, owner_user_id=USER_ID, external_id="owned", name="Owned campaign")
|
|
shared = Campaign(id="campaign-shared", tenant_id=TENANT_ID, owner_user_id=OTHER_USER_ID, external_id="shared", name="Shared campaign")
|
|
session.add_all([
|
|
owned,
|
|
shared,
|
|
CampaignShare(id="share-group", tenant_id=TENANT_ID, campaign_id=shared.id, target_type="group", target_id=GROUP_ID, permission="read"),
|
|
])
|
|
session.commit()
|
|
|
|
service = CampaignAccessService()
|
|
owned_items = service.explain_resource_provenance(session, _principal(), resource_type="campaign", resource_id=owned.id, action="campaigns:campaign:read")
|
|
shared_items = service.explain_resource_provenance(session, _principal(group_ids={GROUP_ID}), resource_type="campaign", resource_id=shared.id, action="campaigns:campaign:read")
|
|
admin_items = service.explain_resource_provenance(session, _principal(scopes={"tenant:*"}), resource_type="campaign", resource_id=shared.id, action="campaigns:campaign:read")
|
|
missing_items = service.explain_resource_provenance(session, _principal(), resource_type="campaign", resource_id="missing-campaign", action="campaigns:campaign:read")
|
|
|
|
self.assertTrue(any(item.kind == "resource" and item.source == "campaigns.campaign" and item.id == owned.id for item in owned_items))
|
|
self.assertTrue(any(item.kind == "owner" and item.id == USER_ID for item in owned_items))
|
|
self.assertTrue(any(item.kind == "share" and item.id == "share-group" for item in shared_items))
|
|
self.assertTrue(any(item.kind == "policy" and item.id == "tenant:*" for item in admin_items))
|
|
self.assertEqual("campaigns.not_found", missing_items[0].source)
|
|
self.assertIs(missing_items[0].details["found"], False)
|
|
|
|
def test_campaign_access_provider_requires_write_share_for_write_actions(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(id="campaign-write", tenant_id=TENANT_ID, owner_user_id=OTHER_USER_ID, external_id="write", name="Write campaign")
|
|
session.add_all([
|
|
campaign,
|
|
CampaignShare(id="share-read", tenant_id=TENANT_ID, campaign_id=campaign.id, target_type="group", target_id=GROUP_ID, permission="read"),
|
|
CampaignShare(id="share-write", tenant_id=TENANT_ID, campaign_id=campaign.id, target_type="user", target_id=USER_ID, permission="write"),
|
|
])
|
|
session.commit()
|
|
|
|
service = CampaignAccessService()
|
|
items = service.explain_resource_provenance(
|
|
session,
|
|
_principal(group_ids={GROUP_ID}),
|
|
resource_type="campaign",
|
|
resource_id=campaign.id,
|
|
action="campaigns:campaign:update",
|
|
)
|
|
|
|
self.assertTrue(any(item.kind == "share" and item.id == "share-write" for item in items))
|
|
self.assertFalse(any(item.kind == "share" and item.id == "share-read" for item in items))
|
|
|
|
def test_campaign_children_explain_their_parent_campaign_access(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-child",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=OTHER_USER_ID,
|
|
external_id="child",
|
|
name="Child resources",
|
|
)
|
|
version = CampaignVersion(
|
|
id="version-child",
|
|
campaign_id=campaign.id,
|
|
version_number=1,
|
|
raw_json={},
|
|
)
|
|
job = CampaignJob(
|
|
id="job-child",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
campaign_version_id=version.id,
|
|
entry_index=1,
|
|
recipient_email="recipient@example.test",
|
|
)
|
|
session.add_all(
|
|
[
|
|
campaign,
|
|
version,
|
|
job,
|
|
CampaignShare(
|
|
id="share-child",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
target_type="group",
|
|
target_id=GROUP_ID,
|
|
permission="read",
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
|
|
service = CampaignAccessService()
|
|
principal = _principal(group_ids={GROUP_ID})
|
|
version_items = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_version",
|
|
resource_id=version.id,
|
|
action="campaigns:version:read",
|
|
)
|
|
job_items = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_delivery_job",
|
|
resource_id=job.id,
|
|
action="campaigns:delivery_job:read",
|
|
)
|
|
report_id = campaign_report_resource_id(
|
|
campaign_id=campaign.id,
|
|
version_id=version.id,
|
|
report_kind="delivery",
|
|
)
|
|
report_items = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_report",
|
|
resource_id=report_id,
|
|
action="campaigns:report:read",
|
|
)
|
|
|
|
for items, source in (
|
|
(version_items, "campaigns.version"),
|
|
(job_items, "campaigns.delivery_job"),
|
|
(report_items, "campaigns.report"),
|
|
):
|
|
child = next(item for item in items if item.source == source)
|
|
self.assertEqual(
|
|
child.details["authorization_inherited_from"],
|
|
{
|
|
"resource_type": "campaign",
|
|
"resource_id": campaign.id,
|
|
},
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
item.source == "campaigns.campaign"
|
|
and item.id == campaign.id
|
|
for item in items
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
any(item.kind == "share" and item.id == "share-child" for item in items)
|
|
)
|
|
|
|
report = next(item for item in report_items if item.source == "campaigns.report")
|
|
self.assertEqual(report.details["campaign_version_id"], version.id)
|
|
self.assertEqual(report.details["report_kind"], "delivery")
|
|
self.assertFalse(report.details["persisted"])
|
|
|
|
def test_recipient_attachment_and_review_children_are_bounded_and_versioned(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=OTHER_USER_ID,
|
|
external_id="sensitive",
|
|
name="Sensitive child evidence",
|
|
)
|
|
version = CampaignVersion(
|
|
id="version-sensitive",
|
|
campaign_id=campaign.id,
|
|
version_number=4,
|
|
raw_json={},
|
|
execution_snapshot={"secret_source_rows": ["do-not-disclose"]},
|
|
execution_snapshot_hash="a" * 64,
|
|
)
|
|
job = CampaignJob(
|
|
id="job-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
campaign_version_id=version.id,
|
|
entry_index=7,
|
|
recipient_email="hidden-recipient@example.test",
|
|
execution_input_sha256="b" * 64,
|
|
resolved_attachments=[{"filename": "hidden-file.pdf", "storage_key": "secret/key"}],
|
|
issues_snapshot=[{
|
|
"code": "attachment_override_required",
|
|
"source": "attachments.policy",
|
|
"behavior": "ask",
|
|
"message": "hidden validation detail",
|
|
}],
|
|
)
|
|
version.editor_state = {
|
|
"review_send": {
|
|
"issue_decisions": [{
|
|
"job_id": job.id,
|
|
"review_key": "opaque-review-key",
|
|
"decision": "accept",
|
|
"reason": "hidden operator reason",
|
|
"decided_at": "2026-08-19T10:00:00+00:00",
|
|
"issue_fingerprint": "c" * 64,
|
|
}]
|
|
}
|
|
}
|
|
issue = CampaignIssue(
|
|
id="issue-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
campaign_version_id=version.id,
|
|
job_id=job.id,
|
|
severity="warning",
|
|
code="attachment_override_required",
|
|
message="hidden validation detail",
|
|
source="hidden/source/path",
|
|
behavior="ask",
|
|
)
|
|
blob = AttachmentBlob(
|
|
id="blob-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
sha256="d" * 64,
|
|
size_bytes=123,
|
|
storage_bucket="secret-bucket",
|
|
storage_key="secret/object",
|
|
)
|
|
attachment = AttachmentInstance(
|
|
id="attachment-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=OTHER_USER_ID,
|
|
campaign_id=campaign.id,
|
|
blob_id=blob.id,
|
|
filename="hidden-file.pdf",
|
|
)
|
|
session.add_all([
|
|
campaign,
|
|
version,
|
|
job,
|
|
issue,
|
|
blob,
|
|
attachment,
|
|
CampaignShare(
|
|
id="share-sensitive",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
target_type="group",
|
|
target_id=GROUP_ID,
|
|
permission="read",
|
|
),
|
|
])
|
|
session.commit()
|
|
|
|
service = CampaignAccessService()
|
|
principal = _principal(
|
|
scopes={
|
|
"campaigns:campaign:read",
|
|
"campaigns:recipient:read",
|
|
"campaigns:campaign:review",
|
|
"campaigns:diagnostic:read",
|
|
},
|
|
group_ids={GROUP_ID},
|
|
)
|
|
version_child_id = campaign_version_child_resource_id(
|
|
version_id=version.id,
|
|
child_id=job.id,
|
|
)
|
|
cases = (
|
|
("campaign_recipient_source_snapshot", version.id, "campaigns.recipient_source_snapshot"),
|
|
("campaign_recipient", version_child_id, "campaigns.recipient"),
|
|
("campaign_attachment_binding", attachment.id, "campaigns.attachment_binding"),
|
|
("campaign_attachment_resolution", version_child_id, "campaigns.attachment_resolution"),
|
|
("campaign_validation_issue", issue.id, "campaigns.validation_issue"),
|
|
("campaign_review_decision", version_child_id, "campaigns.review_decision"),
|
|
("campaign_attachment_override", version_child_id, "campaigns.attachment_override"),
|
|
)
|
|
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"])
|
|
self.assertEqual(
|
|
{"resource_type": "campaign", "resource_id": campaign.id},
|
|
child.details["authorization_inherited_from"],
|
|
)
|
|
self.assertEqual("inherited_and_further_restricted", child.details["authorization_mode"])
|
|
self.assertTrue(any(item.id == "share-sensitive" for item in items))
|
|
serialized = json.dumps(child.details, sort_keys=True)
|
|
for hidden in (
|
|
"hidden-recipient@example.test",
|
|
"hidden-file.pdf",
|
|
"secret/object",
|
|
"secret/key",
|
|
"hidden validation detail",
|
|
"hidden operator reason",
|
|
"hidden/source/path",
|
|
"do-not-disclose",
|
|
):
|
|
self.assertNotIn(hidden, serialized)
|
|
|
|
denied_items = service.explain_resource_provenance(
|
|
session,
|
|
_principal(scopes=set()),
|
|
resource_type="campaign_recipient",
|
|
resource_id=version_child_id,
|
|
action="campaigns:recipient:read",
|
|
)
|
|
denied_child = next(item for item in denied_items if item.source == "campaigns.recipient")
|
|
self.assertIn("campaigns:recipient:read", denied_child.details["permission_actions"])
|
|
self.assertFalse(any(item.kind in {"owner", "share", "policy"} for item in denied_items))
|
|
|
|
missing = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_recipient",
|
|
resource_id=f"{version.id}:missing-job",
|
|
action="campaigns:recipient:read",
|
|
)
|
|
self.assertEqual("campaigns.not_found", missing[0].source)
|
|
self.assertNotIn("hidden-recipient", json.dumps(missing[0].details))
|
|
|
|
stale = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_review_decision",
|
|
resource_id=f"stale-version:{job.id}",
|
|
action="campaigns:diagnostic:read",
|
|
)
|
|
self.assertEqual("stale_version_reference", stale[0].details["reason"])
|
|
|
|
hidden = service.explain_resource_provenance(
|
|
session,
|
|
_principal(tenant_id="tenant-2"),
|
|
resource_type="campaign_validation_issue",
|
|
resource_id=issue.id,
|
|
action="campaigns:diagnostic:read",
|
|
)
|
|
self.assertEqual("campaigns.not_found", hidden[0].source)
|
|
self.assertNotIn(issue.code, json.dumps(hidden[0].details))
|
|
|
|
def test_delivery_evidence_explanations_separate_permissions_and_hide_transport_data(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-delivery-evidence",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=USER_ID,
|
|
external_id="delivery-evidence",
|
|
name="Delivery evidence",
|
|
)
|
|
version = CampaignVersion(
|
|
id="version-delivery-evidence",
|
|
campaign_id=campaign.id,
|
|
version_number=1,
|
|
raw_json={},
|
|
)
|
|
job = CampaignJob(
|
|
id="job-delivery-evidence",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
campaign_version_id=version.id,
|
|
entry_index=1,
|
|
recipient_email="hidden-delivery-recipient@example.test",
|
|
send_status="sent",
|
|
postbox_status="accepted",
|
|
imap_status="appended",
|
|
last_error="hidden reconciliation note",
|
|
)
|
|
send_attempt = SendAttempt(
|
|
id="send-attempt-evidence",
|
|
job_id=job.id,
|
|
attempt_number=1,
|
|
status="reconciled_not_sent",
|
|
claim_token="hidden-claim-token",
|
|
smtp_response="hidden-smtp-response",
|
|
)
|
|
imap_attempt = ImapAppendAttempt(
|
|
id="imap-attempt-evidence",
|
|
job_id=job.id,
|
|
attempt_number=1,
|
|
status="reconciled_imap_appended",
|
|
folder="Hidden/Sent",
|
|
error_message="hidden imap note",
|
|
)
|
|
postbox_attempt = PostboxDeliveryAttempt(
|
|
id="postbox-attempt-evidence",
|
|
tenant_id=TENANT_ID,
|
|
job_id=job.id,
|
|
target_key="opaque-target",
|
|
target_index=0,
|
|
attempt_number=1,
|
|
idempotency_key="opaque-postbox-key",
|
|
status="accepted",
|
|
target_snapshot={"address": "hidden postbox address"},
|
|
evidence={
|
|
"operator_reconciliation": {
|
|
"decision": "accepted",
|
|
"note": "hidden postbox note",
|
|
}
|
|
},
|
|
)
|
|
print_attempt = PrintOutputAttempt(
|
|
id="print-attempt-evidence",
|
|
tenant_id=TENANT_ID,
|
|
job_id=job.id,
|
|
attempt_number=1,
|
|
idempotency_key="opaque-print-key",
|
|
status="accepted",
|
|
render_id="hidden-render-id",
|
|
evidence={"storage_key": "hidden print locator"},
|
|
)
|
|
message_action = CampaignMessageAction(
|
|
id="message-action-evidence",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
campaign_version_id=version.id,
|
|
job_id=job.id,
|
|
kind="retry",
|
|
idempotency_key="opaque-action-key",
|
|
canonical_request_hash="e" * 64,
|
|
reason="hidden action reason",
|
|
message_sha256="f" * 64,
|
|
recipient_manifest_sha256="1" * 64,
|
|
recipient_count=1,
|
|
prior_send_status="failed",
|
|
prior_attempt_count=1,
|
|
status="completed",
|
|
)
|
|
action_attempt = CampaignMessageActionAttempt(
|
|
id="message-action-attempt-evidence",
|
|
action_id=message_action.id,
|
|
attempt_number=1,
|
|
status="completed",
|
|
started_at=datetime.now(UTC),
|
|
diagnostic_summary="hidden diagnostic summary",
|
|
)
|
|
session.add_all([
|
|
campaign,
|
|
version,
|
|
job,
|
|
send_attempt,
|
|
imap_attempt,
|
|
postbox_attempt,
|
|
print_attempt,
|
|
message_action,
|
|
action_attempt,
|
|
])
|
|
session.commit()
|
|
|
|
service = CampaignAccessService()
|
|
principal = _principal(scopes={
|
|
"campaigns:campaign:read",
|
|
"campaigns:report:read",
|
|
"campaigns:report:export",
|
|
"campaigns:diagnostic:read",
|
|
"campaigns:campaign:reconcile",
|
|
})
|
|
cases = (
|
|
("campaign_send_attempt", send_attempt.id, "campaigns.send_attempt"),
|
|
("campaign_imap_append_attempt", imap_attempt.id, "campaigns.imap_append_attempt"),
|
|
("campaign_postbox_attempt", postbox_attempt.id, "campaigns.postbox_attempt"),
|
|
("campaign_print_attempt", print_attempt.id, "campaigns.print_attempt"),
|
|
("campaign_message_action", message_action.id, "campaigns.message_action"),
|
|
("campaign_message_action_attempt", action_attempt.id, "campaigns.message_action_attempt"),
|
|
)
|
|
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"])
|
|
self.assertEqual(version.id, child.details["campaign_version_id"])
|
|
self.assertEqual(job.id, child.details["job_id"])
|
|
self.assertEqual(
|
|
{
|
|
"read": ["campaigns:campaign:read"],
|
|
"report": ["campaigns:report:read"],
|
|
"diagnostic": ["campaigns:diagnostic:read"],
|
|
"export": ["campaigns:report:export"],
|
|
},
|
|
child.details["permission_classes"],
|
|
)
|
|
serialized = json.dumps(child.details, sort_keys=True)
|
|
for hidden_value in (
|
|
"hidden-delivery-recipient@example.test",
|
|
"hidden-claim-token",
|
|
"hidden-smtp-response",
|
|
"Hidden/Sent",
|
|
"hidden imap note",
|
|
"hidden postbox address",
|
|
"hidden postbox note",
|
|
"hidden-render-id",
|
|
"hidden print locator",
|
|
"hidden action reason",
|
|
"hidden diagnostic summary",
|
|
"hidden reconciliation note",
|
|
):
|
|
self.assertNotIn(hidden_value, serialized)
|
|
|
|
reconciliation = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_reconciliation_decision",
|
|
resource_id=job.id,
|
|
action="campaigns:campaign:reconcile",
|
|
)
|
|
decision = next(item for item in reconciliation if item.source == "campaigns.reconciliation_decision")
|
|
self.assertEqual(["smtp", "imap", "postbox"], decision.details["channels"])
|
|
self.assertTrue(decision.details["evidence_note_recorded"])
|
|
self.assertFalse(decision.details["evidence_note_disclosed"])
|
|
|
|
unavailable = service.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type="campaign_postbox_attempt",
|
|
resource_id="missing-postbox-attempt",
|
|
action="campaigns:diagnostic:read",
|
|
)
|
|
self.assertEqual("postbox", unavailable[0].details["optional_module"])
|
|
self.assertEqual("unavailable_or_hidden", unavailable[0].details["evidence_availability"])
|
|
self.assertNotIn("hidden postbox address", json.dumps(unavailable[0].details))
|
|
|
|
def test_campaign_ownership_provider_requires_group_acceptance_authority(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-ownership",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=USER_ID,
|
|
external_id="ownership",
|
|
name="Ownership",
|
|
)
|
|
session.add(campaign)
|
|
session.commit()
|
|
|
|
service = CampaignOwnershipService()
|
|
current_owner = OwnershipSubjectRef(type="user", id=USER_ID)
|
|
target_group = OwnershipSubjectRef(type="group", id=GROUP_ID)
|
|
ordinary_member = OwnershipSubjectRef(
|
|
type="user",
|
|
id=OTHER_USER_ID,
|
|
group_ids=frozenset({GROUP_ID}),
|
|
)
|
|
group_manager = OwnershipSubjectRef(
|
|
type="user",
|
|
id=OTHER_USER_ID,
|
|
group_ids=frozenset({GROUP_ID}),
|
|
scopes=frozenset({"campaigns:ownership:accept_group"}),
|
|
)
|
|
|
|
denied = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="accept_group_transfer",
|
|
actor=ordinary_member,
|
|
current_owner=current_owner,
|
|
target_owner=target_group,
|
|
)
|
|
allowed = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="accept_group_transfer",
|
|
actor=group_manager,
|
|
current_owner=current_owner,
|
|
target_owner=target_group,
|
|
)
|
|
|
|
self.assertFalse(denied.allowed)
|
|
self.assertTrue(allowed.allowed)
|
|
|
|
@patch(
|
|
"govoplan_campaign.backend.capabilities._valid_campaign_owner_target",
|
|
return_value=None,
|
|
)
|
|
def test_campaign_ownership_request_requires_existing_read_access(
|
|
self,
|
|
_target_validation,
|
|
) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-request",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=USER_ID,
|
|
external_id="request",
|
|
name="Request",
|
|
)
|
|
session.add(campaign)
|
|
session.commit()
|
|
service = CampaignOwnershipService()
|
|
requester = OwnershipSubjectRef(
|
|
type="user",
|
|
id=OTHER_USER_ID,
|
|
scopes=frozenset({"campaigns:campaign:read"}),
|
|
)
|
|
current_owner = OwnershipSubjectRef(type="user", id=USER_ID)
|
|
|
|
denied = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="request_ownership",
|
|
actor=requester,
|
|
current_owner=current_owner,
|
|
target_owner=requester,
|
|
)
|
|
session.add(
|
|
CampaignShare(
|
|
id="share-requester",
|
|
tenant_id=TENANT_ID,
|
|
campaign_id=campaign.id,
|
|
target_type="user",
|
|
target_id=OTHER_USER_ID,
|
|
permission="read",
|
|
)
|
|
)
|
|
session.commit()
|
|
allowed = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="request_ownership",
|
|
actor=requester,
|
|
current_owner=current_owner,
|
|
target_owner=requester,
|
|
)
|
|
|
|
self.assertFalse(denied.allowed)
|
|
self.assertTrue(allowed.allowed)
|
|
|
|
@patch(
|
|
"govoplan_campaign.backend.capabilities._valid_campaign_owner_target",
|
|
return_value=None,
|
|
)
|
|
def test_campaign_owner_proposal_requires_share_authority(
|
|
self,
|
|
_target_validation,
|
|
) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-proposal",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=USER_ID,
|
|
external_id="proposal",
|
|
name="Proposal",
|
|
)
|
|
session.add(campaign)
|
|
session.commit()
|
|
service = CampaignOwnershipService()
|
|
current_owner = OwnershipSubjectRef(type="user", id=USER_ID)
|
|
target_owner = OwnershipSubjectRef(type="user", id=OTHER_USER_ID)
|
|
|
|
denied = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="propose_transfer",
|
|
actor=current_owner,
|
|
current_owner=current_owner,
|
|
target_owner=target_owner,
|
|
)
|
|
allowed = service.authorize_ownership_action(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
action="propose_transfer",
|
|
actor=OwnershipSubjectRef(
|
|
type="user",
|
|
id=USER_ID,
|
|
scopes=frozenset({"campaigns:campaign:share"}),
|
|
),
|
|
current_owner=current_owner,
|
|
target_owner=target_owner,
|
|
)
|
|
|
|
self.assertFalse(denied.allowed)
|
|
self.assertTrue(allowed.allowed)
|
|
|
|
def test_campaign_ownership_provider_applies_only_against_expected_owner(self) -> None:
|
|
session = _session()
|
|
self.addCleanup(_close_session, session)
|
|
_seed_access_subjects(session)
|
|
campaign = Campaign(
|
|
id="campaign-owner-apply",
|
|
tenant_id=TENANT_ID,
|
|
owner_user_id=USER_ID,
|
|
external_id="owner-apply",
|
|
name="Owner apply",
|
|
)
|
|
session.add(campaign)
|
|
session.commit()
|
|
|
|
service = CampaignOwnershipService()
|
|
service.apply_owner(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
expected_owner=OwnershipSubjectRef(type="user", id=USER_ID),
|
|
target_owner=OwnershipSubjectRef(type="group", id=GROUP_ID),
|
|
actor=OwnershipSubjectRef(type="user", id=OTHER_USER_ID),
|
|
reason=None,
|
|
)
|
|
session.flush()
|
|
self.assertIsNone(campaign.owner_user_id)
|
|
self.assertEqual(campaign.owner_group_id, GROUP_ID)
|
|
|
|
with self.assertRaisesRegex(
|
|
OwnershipTransferError,
|
|
"changed while the transfer was pending",
|
|
):
|
|
service.apply_owner(
|
|
session,
|
|
tenant_id=TENANT_ID,
|
|
resource_id=campaign.id,
|
|
expected_owner=OwnershipSubjectRef(type="user", id=USER_ID),
|
|
target_owner=OwnershipSubjectRef(type="user", id=OTHER_USER_ID),
|
|
actor=OwnershipSubjectRef(type="user", id=USER_ID),
|
|
reason=None,
|
|
)
|
|
|
|
|
|
def _session():
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(
|
|
bind=engine,
|
|
tables=[
|
|
Account.__table__,
|
|
User.__table__,
|
|
Group.__table__,
|
|
Campaign.__table__,
|
|
CampaignShare.__table__,
|
|
CampaignVersion.__table__,
|
|
CampaignJob.__table__,
|
|
CampaignIssue.__table__,
|
|
AttachmentBlob.__table__,
|
|
AttachmentInstance.__table__,
|
|
SendAttempt.__table__,
|
|
CampaignMessageAction.__table__,
|
|
CampaignMessageActionAttempt.__table__,
|
|
ImapAppendAttempt.__table__,
|
|
PostboxDeliveryAttempt.__table__,
|
|
PrintOutputAttempt.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
],
|
|
)
|
|
return sessionmaker(bind=engine, future=True)()
|
|
|
|
|
|
def _close_session(session) -> None:
|
|
engine = session.get_bind()
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def _seed_access_subjects(session) -> None:
|
|
session.add_all([
|
|
Account(id="account-1", email="one@example.test", normalized_email="one@example.test"),
|
|
Account(id="account-2", email="two@example.test", normalized_email="two@example.test"),
|
|
User(id=USER_ID, tenant_id=TENANT_ID, account_id="account-1", email="one@example.test"),
|
|
User(id=OTHER_USER_ID, tenant_id=TENANT_ID, account_id="account-2", email="two@example.test"),
|
|
Group(id=GROUP_ID, tenant_id=TENANT_ID, slug="group", name="Group"),
|
|
])
|
|
session.commit()
|
|
|
|
|
|
def _principal(
|
|
*,
|
|
scopes: set[str] | None = None,
|
|
group_ids: set[str] | None = None,
|
|
tenant_id: str = TENANT_ID,
|
|
) -> PrincipalRef:
|
|
return PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id=USER_ID,
|
|
tenant_id=tenant_id,
|
|
scopes=frozenset(
|
|
{"campaigns:campaign:read"} if scopes is None else scopes
|
|
),
|
|
group_ids=frozenset(group_ids or set()),
|
|
)
|