From fa013ea24bce503ad9f66641bd1b348b54ecc5c1 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 29 Jul 2026 17:48:36 +0200 Subject: [PATCH] feat: preserve automation actor provenance --- src/govoplan_dataflow/backend/schemas.py | 11 +++++++ src/govoplan_dataflow/backend/triggers.py | 16 ++++++++-- tests/test_triggers.py | 37 ++++++++++++++++++++--- 3 files changed, 57 insertions(+), 7 deletions(-) diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index 01d9780..b90075f 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -429,6 +429,16 @@ class DataflowTriggerDeliveryListResponse(BaseModel): deliveries: list[DataflowTriggerDeliveryResponse] +class DataflowEventActor(BaseModel): + type: str = Field( + min_length=1, + max_length=100, + pattern=r"^[A-Za-z0-9_.:-]+$", + ) + id: str | None = Field(default=None, max_length=255) + label: str | None = Field(default=None, max_length=300) + + class DataflowEventDeliveryRequest(BaseModel): event_id: str = Field(min_length=1, max_length=128) type: str = Field( @@ -445,6 +455,7 @@ class DataflowEventDeliveryRequest(BaseModel): occurred_at: datetime correlation_id: str | None = Field(default=None, max_length=128) causation_id: str | None = Field(default=None, max_length=128) + actor: DataflowEventActor | None = None classification: Literal[ "public", "internal", diff --git a/src/govoplan_dataflow/backend/triggers.py b/src/govoplan_dataflow/backend/triggers.py index bd90493..112421b 100644 --- a/src/govoplan_dataflow/backend/triggers.py +++ b/src/govoplan_dataflow/backend/triggers.py @@ -746,9 +746,10 @@ def _dispatch_delivery( now=now, error="Access automation-principal capability is unavailable.", ) + event = delivery.event_ or {} resolution = provider.resolve_automation_principal( session, - request=AutomationPrincipalRequest( + request=AutomationPrincipalRequest.delegated_user( tenant_id=trigger.tenant_id, account_id=trigger.authorization_account_id, membership_id=trigger.authorization_membership_id, @@ -760,6 +761,8 @@ def _dispatch_delivery( "delivery_ref": ( f"dataflow-trigger-delivery:{delivery.id}" ), + "event_actor": event.get("actor"), + "operator_override": event.get("operator_override"), }, ), ) @@ -783,7 +786,6 @@ def _dispatch_delivery( action="automate", ) publication = _publication_target(trigger) - event = delivery.event_ or {} run, _ = start_pipeline_run( session, tenant_id=trigger.tenant_id, @@ -813,6 +815,11 @@ def _dispatch_delivery( causation_id=_optional_text(event.get("causation_id")), scheduled_for=delivery.scheduled_for, requested_by=trigger.authorization_account_id, + metadata={ + "authorization_provenance": dict( + resolution.provenance + ) + }, ), ), ) @@ -968,6 +975,11 @@ class SqlDataflowTriggerDispatcher: occurred_at=event.occurred_at, correlation_id=event.correlation_id, causation_id=event.causation_id, + actor=( + event.actor.to_dict() + if event.actor is not None + else None + ), classification=event.classification, ), ) diff --git a/tests/test_triggers.py b/tests/test_triggers.py index db91ca2..bbd5da0 100644 --- a/tests/test_triggers.py +++ b/tests/test_triggers.py @@ -97,15 +97,20 @@ class CurrentPrincipalProvider: def __init__(self, actor: ApiPrincipal, *, allowed: bool = True) -> None: self.actor = actor self.allowed = allowed + self.last_request = None def resolve_automation_principal(self, _session, *, request): + self.last_request = request return AutomationPrincipalResolution( allowed=self.allowed, principal=self.actor if self.allowed else None, reason=None if self.allowed else "Owner authorization revoked.", granted_scopes=tuple(request.grant_scopes) if self.allowed else (), missing_scopes=() if self.allowed else tuple(request.grant_scopes), - provenance={"status": "current" if self.allowed else "revoked"}, + provenance={ + "status": "current" if self.allowed else "revoked", + "event_actor": request.context.get("event_actor"), + }, ) @@ -116,12 +121,13 @@ class Registry: *, automation_allowed: bool = True, ) -> None: + self.automation_provider = CurrentPrincipalProvider( + actor, + allowed=automation_allowed, + ) self.capabilities = { POLICY_CAPABILITY: AllowDefinitionPolicy(), - AUTOMATION_CAPABILITY: CurrentPrincipalProvider( - actor, - allowed=automation_allowed, - ), + AUTOMATION_CAPABILITY: self.automation_provider, } def has_capability(self, name: str) -> bool: @@ -322,6 +328,11 @@ class DataflowTriggerTests(unittest.TestCase): module_id="files", payload={"folder_id": "incoming", "file_id": "file-1"}, occurred_at=utcnow(), + actor={ + "type": "user", + "id": "account-2", + "label": "Uploader", + }, ) self.assertEqual( @@ -355,6 +366,22 @@ class DataflowTriggerTests(unittest.TestCase): self.assertEqual(1, result.succeeded) self.assertEqual("event", run.invocation_kind) self.assertEqual("event-1", run.request_["invocation"]["event_id"]) + self.assertEqual( + { + "type": "user", + "id": "account-2", + "label": "Uploader", + }, + self.registry.automation_provider.last_request.context[ + "event_actor" + ], + ) + self.assertEqual( + "account-2", + run.request_["invocation"]["metadata"][ + "authorization_provenance" + ]["event_actor"]["id"], + ) def test_revoked_owner_blocks_delivery_before_execution(self) -> None: trigger = self._event_trigger()