From aaccf327b626304ac0391c25e37683afe4867346 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:31:09 +0000 Subject: [PATCH 1/3] Bind workflow service operations to trusted host authority --- CHANGELOG.md | 12 + .../workflow-service-calls-architecture.md | 48 ++ src/Providers/WorkflowServiceProvider.php | 6 + .../WorkflowServiceOperationAuthorizer.php | 26 + ...ServiceOperationAuthorizationException.php | 15 + src/V2/Support/DefaultServiceControlPlane.php | 17 +- src/V2/Support/DefaultWorkflowTaskBridge.php | 88 +++- ...DenyWorkflowServiceOperationAuthorizer.php | 18 + .../Support/RunServiceOperationAuthorizer.php | 23 + src/V2/Support/ServiceCallPrincipal.php | 4 +- src/V2/Support/WorkflowExecutor.php | 28 +- .../WorkflowServiceOperationAuthority.php | 138 +++++ tests/Support/NexusCallerDeliveryScenario.php | 37 ++ ...WorkflowTaskBridgeServiceOperationTest.php | 22 +- .../WorkflowExecutorServiceOperationTest.php | 10 + .../WorkflowServiceOperationAuthorityTest.php | 471 ++++++++++++++++++ 16 files changed, 913 insertions(+), 50 deletions(-) create mode 100644 src/V2/Contracts/WorkflowServiceOperationAuthorizer.php create mode 100644 src/V2/Exceptions/ServiceOperationAuthorizationException.php create mode 100644 src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php create mode 100644 src/V2/Support/RunServiceOperationAuthorizer.php create mode 100644 src/V2/Support/WorkflowServiceOperationAuthority.php create mode 100644 tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index b8f7185f3..0e8792048 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +- Require a trusted host authorizer for workflow service operations. Refuse + worker-supplied principal overrides, foreign caller namespaces and service + call IDs or idempotency keys owned by another run, instance or operation. + Preserve explicit embedded null namespaces instead of using ambient config. +- Service-operation hosts must bind `WorkflowServiceOperationAuthorizer` before + upgrading. The default refuses service operations; explicitly trusted embedded + hosts can opt into `RunServiceOperationAuthorizer` for same-namespace calls + with an unprivileged workflow principal. Authenticated servers supply their + own current principal and authorize requested target scope. + ## 2.13.1 - 2026-10-24 - Cancel the live task when a retried activity receives cooperative cancellation. diff --git a/docs/architecture/workflow-service-calls-architecture.md b/docs/architecture/workflow-service-calls-architecture.md index 9c94198b5..773daf32d 100644 --- a/docs/architecture/workflow-service-calls-architecture.md +++ b/docs/architecture/workflow-service-calls-architecture.md @@ -352,6 +352,54 @@ of the five reasons, and a Cancelled call MUST carry `ServiceCallFailureReason::Cancellation` even though its terminal status is Cancelled rather than Failed. +## Workflow operation authority + +Workflow service operations require a host binding for +`Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer`. The default binding +refuses them with `service_operation_authority_invalid`. This applies to embedded +execution and external worker completions, including retained-claim prefixes. +Upgrade the host binding before accepting service-operation work. + +An authorizer receives the locked caller run and normalized command intent. +It must validate the requested target scope against trusted host/run context, +then return a `ServiceCallPrincipal` derived from that context. It must not +reconstruct authority from the command's principal fields or metadata. It is an +authorization check, so it must not reserve admission budgets or perform +external effects; a transaction retry or preflight may invoke it again. + +An embedded host that explicitly trusts its own workflow code to invoke +services within each run's namespace can opt into the provided workload policy: + +```php +$app->bind( + \Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer::class, + \Workflow\V2\Support\RunServiceOperationAuthorizer::class, +); +``` + +That policy permits only the run's namespace. Its principal is +`workflow:`, method `workflow`, with no roles, tenant or claims. It does +not satisfy a catalog policy that requires privileged roles. A server with +authenticated workers instead binds an authorizer that checks its current +authentication/namespace policy and returns the authenticated worker principal. +Authorized cross-namespace routing remains possible through that trusted host +binding; a worker's requested `namespace` alone grants no permission. + +Caller namespace and workflow run/instance identity always come from the locked +run. A mismatched non-null `caller_namespace` or non-null `principal_*` override +is refused before applying sibling commands. Unknown fields do not grant +authority. Explicit null run namespaces stay null even when the process has a +configured default namespace. + +An explicit `service_call_id` or an existing idempotency-key match must bind to +the same caller namespace, run and instance, the authorized target namespace, +and the exact endpoint/service/operation names. Existing rows are checked under +locks before replay or dispatch. A missing or foreign explicit call, or a +foreign key match, refuses the worker turn with `service_call_not_owned` and +leaves the lease and durable rows unchanged. New default idempotency keys are +checked by the same rule. Server adapters retain their current-policy replay +checks before handing the call to Core. + ## Observability surface The observability surface for a cross-namespace service call MUST be diff --git a/src/Providers/WorkflowServiceProvider.php b/src/Providers/WorkflowServiceProvider.php index d20f4a748..9efa531fb 100644 --- a/src/Providers/WorkflowServiceProvider.php +++ b/src/Providers/WorkflowServiceProvider.php @@ -43,6 +43,7 @@ use Workflow\V2\Contracts\ServiceBoundaryPolicy; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowLink; @@ -63,6 +64,7 @@ use Workflow\V2\Support\DefaultServiceControlPlane; use Workflow\V2\Support\DefaultWorkflowControlPlane; use Workflow\V2\Support\DefaultWorkflowTaskBridge; +use Workflow\V2\Support\DenyWorkflowServiceOperationAuthorizer; use Workflow\V2\Support\ExecutionTracing; use Workflow\V2\Support\HistoryProjectionMaintenanceFallback; use Workflow\V2\Support\InMemoryTaskFairnessState; @@ -132,6 +134,10 @@ static function ($app): HistoryProjectionMaintenanceRole { ); $this->app->singletonIf(ServiceControlPlane::class, DefaultServiceControlPlane::class); + $this->app->singletonIf( + WorkflowServiceOperationAuthorizer::class, + DenyWorkflowServiceOperationAuthorizer::class + ); $this->app->singletonIf(SchedulerRole::class, DefaultSchedulerRole::class); diff --git a/src/V2/Contracts/WorkflowServiceOperationAuthorizer.php b/src/V2/Contracts/WorkflowServiceOperationAuthorizer.php new file mode 100644 index 000000000..4219b7c66 --- /dev/null +++ b/src/V2/Contracts/WorkflowServiceOperationAuthorizer.php @@ -0,0 +1,26 @@ + $command + */ + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal; +} diff --git a/src/V2/Exceptions/ServiceOperationAuthorizationException.php b/src/V2/Exceptions/ServiceOperationAuthorizationException.php new file mode 100644 index 000000000..63b6c8462 --- /dev/null +++ b/src/V2/Exceptions/ServiceOperationAuthorizationException.php @@ -0,0 +1,15 @@ +newCall(); $principal = $this->principal($options); - $callerNamespace = $this->namespaceOption($options, 'caller_namespace') - ?? ($call instanceof WorkflowServiceCall ? $this->stringFrom($call->caller_namespace) : null) - ?? $this->configuredNamespace(); + $callerNamespace = array_key_exists('caller_namespace', $options) + ? $this->namespaceOption($options, 'caller_namespace') + : (($call instanceof WorkflowServiceCall ? $this->stringFrom($call->caller_namespace) : null) + ?? $this->configuredNamespace()); if ($call->accepted_at === null) { $this->stampPending( @@ -467,9 +468,13 @@ private function emptyCallShape(): array */ private function targetNamespace(array $options): ?string { - return $this->namespaceOption($options, 'target_namespace') - ?? $this->namespaceOption($options, 'namespace') - ?? $this->configuredNamespace(); + foreach (['target_namespace', 'namespace'] as $field) { + if (array_key_exists($field, $options)) { + return $this->namespaceOption($options, $field); + } + } + + return $this->configuredNamespace(); } /** diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php index 69e3e6594..45e1e7b6b 100644 --- a/src/V2/Support/DefaultWorkflowTaskBridge.php +++ b/src/V2/Support/DefaultWorkflowTaskBridge.php @@ -39,6 +39,7 @@ use Workflow\V2\Enums\UpdateStatus; use Workflow\V2\Exceptions\ActivityTimeoutException; use Workflow\V2\Exceptions\HistoryEventShapeMismatchException; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Models\ActivityAttempt; use Workflow\V2\Models\ActivityExecution; use Workflow\V2\Models\WorkflowChildCall; @@ -1234,7 +1235,10 @@ public function complete(string $taskId, array $commands): array ]; } - return DB::transaction(function () use ($taskId, $parsed): array { + return self::serviceOperationTransaction(fn (): array => DB::transaction(function () use ( + $taskId, + $parsed + ): array { /** @var WorkflowTask|null $task */ $task = ConfiguredV2Models::query('task_model', WorkflowTask::class) ->lockForUpdate() @@ -1334,6 +1338,17 @@ public function complete(string $taskId, array $commands): array $sequence = WorkflowStepHistory::nextDurableCommandSequence($run); $createdTaskIds = []; + $serviceError = $this->serviceOperationAuthorityError($run, $parsed['non_terminal'], $sequence); + if ($serviceError !== null) { + return [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => $run->id, + 'run_status' => $run->status->value, + 'created_task_ids' => [], + 'reason' => $serviceError, + ]; + } $scopeError = $this->validateCancellationScopeMembership($run, $parsed['non_terminal'], $sequence); if ($scopeError !== null) { return [ @@ -1427,7 +1442,14 @@ public function complete(string $taskId, array $commands): array 'created_task_ids' => $createdTaskIds, 'reason' => null, ]; - }); + }), static fn (string $reason): array => [ + 'completed' => false, + 'task_id' => $taskId, + 'workflow_run_id' => null, + 'run_status' => null, + 'created_task_ids' => [], + 'reason' => $reason, + ]); } /** @@ -1534,7 +1556,7 @@ private function checkpointCommands( return $refused('task_not_found'); } - return DB::transaction(function () use ( + return self::serviceOperationTransaction(fn (): array => DB::transaction(function () use ( $snapshot, $taskId, $leaseOwner, @@ -1571,6 +1593,10 @@ private function checkpointCommands( || ($run->run_deadline_at !== null && now()->gte($run->run_deadline_at))) { return $refused('run_deadline_expired'); } + $serviceError = $this->serviceOperationAuthorityError($run, $parsed['non_terminal'], $startSequence); + if ($serviceError !== null) { + return $refused($serviceError); + } $payload = is_array($task->payload) ? $task->payload : []; $receiptKey = $scopePrefix ? 'portable_scope_checkpoint' : ($group ? 'portable_local_group_checkpoint' : 'portable_local_checkpoint'); @@ -1732,7 +1758,7 @@ private function checkpointCommands( 'duplicate' => false, 'reason' => null, ]; - }, 5); + }, 5), $refused); } /** @@ -3522,17 +3548,13 @@ private function applyStartServiceOperation( $payloadCodec = is_string($command['payload_codec'] ?? null) && $command['payload_codec'] !== '' ? $command['payload_codec'] : ($run->payload_codec ?? CodecRegistry::defaultCodec()); + $options = $this->serviceOperationControlPlaneOptions($run, $command, $sequence, $payloadCodec); $namespace = is_string($run->namespace) ? $run->namespace : null; $requestPayload = isset($command['request_payload']) && is_string($command['request_payload']) ? ExternalPayloads::externalizeForNamespace($command['request_payload'], $payloadCodec, $namespace) : null; $surface = $this->serviceControlPlane() - ->execute( - $command['endpoint_name'], - $command['service_name'], - $command['operation_name'], - $this->serviceOperationControlPlaneOptions($run, $command, $sequence, $payloadCodec), - ); + ->execute($command['endpoint_name'], $command['service_name'], $command['operation_name'], $options); $eventType = self::serviceOperationEventTypeForSurface($surface); $event = WorkflowHistoryEvent::record($run, $eventType, $this->serviceOperationEventPayload( @@ -3580,6 +3602,7 @@ private function serviceOperationControlPlaneOptions( int $sequence, string $payloadCodec, ): array { + $command['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); $metadata = is_array($command['metadata'] ?? null) ? $command['metadata'] : []; $metadata = [ 'caller_sdk_language' => self::metadataString($metadata, 'caller_sdk_language') ?? 'workflow-php', @@ -3588,8 +3611,7 @@ private function serviceOperationControlPlaneOptions( 'workflow_sequence' => $sequence, ] + $metadata; - return array_filter([ - 'namespace' => self::nonEmptyString($command['namespace'] ?? null) ?? $run->namespace, + return WorkflowServiceOperationAuthority::options($run, $command) + array_filter([ 'arguments' => $this->serviceOperationRequestArguments($command, $payloadCodec), 'payload_blob' => self::nonEmptyString($command['request_payload'] ?? null), 'payload_codec' => $payloadCodec, @@ -3601,7 +3623,6 @@ private function serviceOperationControlPlaneOptions( 'wait_timeout_seconds' => is_int($command['wait_timeout_seconds'] ?? null) ? $command['wait_timeout_seconds'] : null, - 'caller_namespace' => self::nonEmptyString($command['caller_namespace'] ?? null) ?? $run->namespace, 'caller_workflow_instance_id' => $run->workflow_instance_id, 'caller_workflow_run_id' => $run->id, 'target_workflow_instance_id' => self::nonEmptyString($command['target_workflow_instance_id'] ?? null), @@ -3617,14 +3638,45 @@ private function serviceOperationControlPlaneOptions( 'duplicate_start_policy' => self::nonEmptyString($command['duplicate_start_policy'] ?? null), 'metadata' => $metadata, 'request_payload_reference' => self::nonEmptyString($command['request_payload_reference'] ?? null), - 'principal_subject' => self::nonEmptyString($command['principal_subject'] ?? null), - 'principal_method' => self::nonEmptyString($command['principal_method'] ?? null), - 'principal_roles' => is_array($command['principal_roles'] ?? null) ? $command['principal_roles'] : null, - 'principal_tenant' => self::nonEmptyString($command['principal_tenant'] ?? null), - 'principal_claims' => is_array($command['principal_claims'] ?? null) ? $command['principal_claims'] : null, ], static fn (mixed $value): bool => $value !== null); } + /** + * @param list $commands + */ + private function serviceOperationAuthorityError(WorkflowRun $run, array $commands, int $sequence): ?string + { + try { + foreach ($commands as $command) { + if ($command['type'] === 'start_service_operation') { + $command['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); + WorkflowServiceOperationAuthority::options($run, $command); + } + if ($command['type'] !== 'cancel_selection_operation') { + ++$sequence; + } + } + } catch (ServiceOperationAuthorizationException $exception) { + return $exception->getMessage(); + } + + return null; + } + + /** + * @param callable(): array $transaction + * @param callable(string): array $refused + * @return array + */ + private static function serviceOperationTransaction(callable $transaction, callable $refused): array + { + try { + return $transaction(); + } catch (ServiceOperationAuthorizationException $exception) { + return $refused($exception->getMessage()); + } + } + /** * @param array $command */ diff --git a/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php b/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php new file mode 100644 index 000000000..b69438204 --- /dev/null +++ b/src/V2/Support/DenyWorkflowServiceOperationAuthorizer.php @@ -0,0 +1,18 @@ +namespace; + if ($target !== $run->namespace) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } +} diff --git a/src/V2/Support/ServiceCallPrincipal.php b/src/V2/Support/ServiceCallPrincipal.php index 4e2ef5e6c..9987006ed 100644 --- a/src/V2/Support/ServiceCallPrincipal.php +++ b/src/V2/Support/ServiceCallPrincipal.php @@ -11,8 +11,8 @@ * the audit row so accepted *and* rejected calls share the same * audit shape. This object is intentionally narrow — only the fields * that survive into the durable audit row — so it can be reconstructed - * cleanly from server-side `Principal`, from worker-supplied identity - * headers, or from a synthetic "system" caller. + * cleanly from a server-authenticated `Principal` or an explicitly + * authorized embedded workload. Worker command fields are not identity. * * Privacy boundary: this DTO carries identity fields (subject, method, * roles, tenant, optional opaque claims). Payload material — request diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php index 5a5ab5882..4490c2797 100644 --- a/src/V2/Support/WorkflowExecutor.php +++ b/src/V2/Support/WorkflowExecutor.php @@ -5969,14 +5969,16 @@ private function recordServiceOperationEvent( : CodecRegistry::defaultCodec()); $namespace = is_string($run->namespace) ? $run->namespace : null; $serializedRequest = Serializer::serializeWithCodec($payloadCodec, $call->requestPayload); + $options = $this->serviceOperationControlPlaneOptions( + $run, + $sequence, + $call, + $payloadCodec, + $serializedRequest + ); $storedRequest = ExternalPayloads::externalizeForNamespace($serializedRequest, $payloadCodec, $namespace); $surface = $this->serviceControlPlane() - ->execute( - $call->endpointName, - $call->serviceName, - $call->operationName, - $this->serviceOperationControlPlaneOptions($run, $sequence, $call, $payloadCodec, $serializedRequest), - ); + ->execute($call->endpointName, $call->serviceName, $call->operationName, $options); $eventType = self::serviceOperationEventTypeForSurface($surface); $event = WorkflowHistoryEvent::record( @@ -6001,6 +6003,7 @@ private function serviceOperationControlPlaneOptions( string $serializedRequest, ): array { $options = $call->options?->toCommandOptions() ?? []; + $options['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence); $metadata = is_array($options['metadata'] ?? null) ? $options['metadata'] : []; $metadata = [ 'caller_sdk_language' => 'workflow-php', @@ -6009,8 +6012,11 @@ private function serviceOperationControlPlaneOptions( 'workflow_sequence' => $sequence, ] + $metadata; - return array_filter([ - 'namespace' => self::nonEmptyString($options['namespace'] ?? null) ?? $run->namespace, + return WorkflowServiceOperationAuthority::options($run, $options + [ + 'endpoint_name' => $call->endpointName, + 'service_name' => $call->serviceName, + 'operation_name' => $call->operationName, + ]) + array_filter([ 'arguments' => $call->requestPayload, 'payload_blob' => $serializedRequest, 'payload_codec' => $payloadCodec, @@ -6022,7 +6028,6 @@ private function serviceOperationControlPlaneOptions( 'wait_timeout_seconds' => is_int($options['wait_timeout_seconds'] ?? null) ? $options['wait_timeout_seconds'] : null, - 'caller_namespace' => self::nonEmptyString($options['caller_namespace'] ?? null) ?? $run->namespace, 'caller_workflow_instance_id' => $run->workflow_instance_id, 'caller_workflow_run_id' => $run->id, 'target_workflow_instance_id' => self::nonEmptyString($options['target_workflow_instance_id'] ?? null), @@ -6038,11 +6043,6 @@ private function serviceOperationControlPlaneOptions( 'duplicate_start_policy' => self::nonEmptyString($options['duplicate_start_policy'] ?? null), 'metadata' => $metadata, 'request_payload_reference' => self::nonEmptyString($options['request_payload_reference'] ?? null), - 'principal_subject' => self::nonEmptyString($options['principal_subject'] ?? null), - 'principal_method' => self::nonEmptyString($options['principal_method'] ?? null), - 'principal_roles' => is_array($options['principal_roles'] ?? null) ? $options['principal_roles'] : null, - 'principal_tenant' => self::nonEmptyString($options['principal_tenant'] ?? null), - 'principal_claims' => is_array($options['principal_claims'] ?? null) ? $options['principal_claims'] : null, ], static fn (mixed $value): bool => $value !== null); } diff --git a/src/V2/Support/WorkflowServiceOperationAuthority.php b/src/V2/Support/WorkflowServiceOperationAuthority.php new file mode 100644 index 000000000..715b2db50 --- /dev/null +++ b/src/V2/Support/WorkflowServiceOperationAuthority.php @@ -0,0 +1,138 @@ + $command + * @return array + */ + public static function options(WorkflowRun $run, array $command): array + { + foreach (['principal', 'principal_subject', 'principal_method', 'principal_roles', + 'principal_tenant', 'principal_claims'] as $field) { + if (array_key_exists($field, $command)) { + throw new ServiceOperationAuthorizationException(); + } + } + if (isset($command['caller_namespace']) && $command['caller_namespace'] !== $run->namespace) { + throw new ServiceOperationAuthorizationException(); + } + + $principal = app(WorkflowServiceOperationAuthorizer::class)->authorize($run, $command); + $namespace = $command['namespace'] ?? $run->namespace; + self::checkReference($run, $command, $namespace); + + // Preserve explicit null namespaces: null is the embedded namespace, + // not permission to use ambient configuration or any existing call. + return [ + 'namespace' => $namespace, + 'caller_namespace' => $run->namespace, + 'caller_workflow_instance_id' => $run->workflow_instance_id, + 'caller_workflow_run_id' => $run->id, + 'principal' => $principal, + ]; + } + + /** + * @param array $command + */ + private static function checkReference(WorkflowRun $run, array $command, ?string $namespace): void + { + $callId = $command['service_call_id'] ?? null; + $key = $command['idempotency_key'] ?? null; + if ($callId === null && $key === null) { + return; + } + + $query = ConfiguredV2Models::query('service_call_model', WorkflowServiceCall::class); + if ($callId !== null) { + $query->whereKey($callId); + } else { + $query->where('target_namespace', $namespace) + ->where('endpoint_name', strtolower(trim($command['endpoint_name']))) + ->where('service_name', strtolower(trim($command['service_name']))) + ->where('operation_name', strtolower(trim($command['operation_name']))) + ->where('idempotency_key', $key) + ->oldest('created_at') + ->oldest('id'); + } + /** @var WorkflowServiceCall|null $call */ + $call = $query->lockForUpdate() + ->first(); + if ($call === null && $callId === null) { + return; + } + if ($call === null || $call->namespace !== $namespace || $call->target_namespace !== $namespace + || $call->caller_namespace !== $run->namespace + || $call->caller_workflow_instance_id !== $run->workflow_instance_id + || $call->caller_workflow_run_id !== $run->id + || $call->endpoint_name !== strtolower(trim($command['endpoint_name'])) + || $call->service_name !== strtolower(trim($command['service_name'])) + || $call->operation_name !== strtolower(trim($command['operation_name']))) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + self::checkCatalog($call, $namespace); + } + + private static function checkCatalog(WorkflowServiceCall $call, ?string $namespace): void + { + $resolutionFailure = $call->status === ServiceCallStatus::Failed->value + && $call->outcome === ServiceCallOutcome::RejectedNotFound + && ($call->outcome_metadata['failure_reason'] ?? null) === 'resolution_failure'; + foreach ([ + [ + 'service_endpoint_model', WorkflowServiceEndpoint::class, $call->workflow_service_endpoint_id, + [ + 'endpoint_name' => $call->endpoint_name, + ]], + [ + 'service_model', WorkflowService::class, $call->workflow_service_id, + [ + 'workflow_service_endpoint_id' => $call->workflow_service_endpoint_id, + 'service_name' => $call->service_name, + ]], + [ + 'service_operation_model', WorkflowServiceOperation::class, $call->workflow_service_operation_id, + [ + 'workflow_service_endpoint_id' => $call->workflow_service_endpoint_id, + 'workflow_service_id' => $call->workflow_service_id, + 'operation_name' => $call->operation_name, + ]], + ] as [$role, $model, $id, $attributes]) { + // A retained resolution failure may have no catalog component. + // Known components still have to be this call's scoped objects. + if ($id === 'unresolved' && $resolutionFailure) { + continue; + } + $row = ConfiguredV2Models::query($role, $model)->where('namespace', $namespace) + ->lockForUpdate() + ->find($id); + if ($row === null) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + foreach ([ + 'id' => $id, + 'namespace' => $namespace, + ] + $attributes as $field => $expected) { + if ($row->getAttribute($field) !== $expected) { + throw new ServiceOperationAuthorizationException('service_call_not_owned'); + } + } + } + } +} diff --git a/tests/Support/NexusCallerDeliveryScenario.php b/tests/Support/NexusCallerDeliveryScenario.php index 5aea820a1..d8278dc34 100644 --- a/tests/Support/NexusCallerDeliveryScenario.php +++ b/tests/Support/NexusCallerDeliveryScenario.php @@ -10,10 +10,12 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; @@ -24,6 +26,7 @@ use Workflow\V2\Models\WorkflowServiceOperation; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\DefaultServiceControlPlane; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\TaskWatchdog; use Workflow\V2\Workflow; @@ -37,6 +40,7 @@ trait NexusCallerDeliveryScenario protected function assertNexusCallerSdkAttribution(array $fixture): void { $previous = $this->app->make(ServiceControlPlane::class); + $previousAuthority = $this->installNexusServiceAuthority(); $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -96,6 +100,9 @@ protected function assertNexusCallerSdkAttribution(array $fixture): void 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, ]); $this->app->instance(ServiceControlPlane::class, $previous); + if ($previousAuthority !== null) { + $this->app->instance(WorkflowServiceOperationAuthorizer::class, $previousAuthority); + } } } @@ -113,6 +120,7 @@ protected function assertWaitingNexusCallerReceivesResult( ): void { $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); $previousControlPlane = $this->app->make(ServiceControlPlane::class); + $previousAuthority = $this->installNexusServiceAuthority(); $previousNow = Carbon::getTestNow(); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -243,6 +251,35 @@ protected function assertWaitingNexusCallerReceivesResult( 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, ]); $this->app->instance(ServiceControlPlane::class, $previousControlPlane); + if ($previousAuthority !== null) { + $this->app->instance(WorkflowServiceOperationAuthorizer::class, $previousAuthority); + } } } + + private function installNexusServiceAuthority(): ?object + { + // The retained consumer also runs against the pre-authorizer base. + if (! interface_exists(WorkflowServiceOperationAuthorizer::class)) { + return null; + } + $previous = $this->app->make(WorkflowServiceOperationAuthorizer::class); + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + $target = $command['namespace'] ?? $run->namespace; + $operation = [$command['endpoint_name'], $command['service_name'], $command['operation_name']]; + $allowed = ($target === 'caller' && $operation === ['billing', 'invoices', 'create']) + || ($target === 'service' && $operation === ['php-service', 'greeter', 'greet']); + if ($run->namespace !== 'caller' || ! $allowed) { + throw new ServiceOperationAuthorizationException(); + } + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); + return $previous; + } } diff --git a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php index 7dc2e92bf..d5173165a 100644 --- a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php +++ b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php @@ -13,6 +13,7 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Attributes\Type; use Workflow\V2\Contracts\ServiceControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -23,6 +24,8 @@ use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Support\RunServiceOperationAuthorizer; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\WorkerCompatibilityFleet; use Workflow\V2\Workflow; use Workflow\V2\WorkflowStub; @@ -38,6 +41,7 @@ final class DefaultWorkflowTaskBridgeServiceOperationTest extends TestCase protected function setUp(): void { parent::setUp(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); Carbon::setTestNow('2026-10-08T02:00:00Z'); config([ 'workflows.v2.task_dispatch_mode' => 'poll', @@ -304,9 +308,8 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad ]; $payload = Serializer::serialize($arguments); $options = [ - 'namespace' => 'billing', - 'caller_namespace' => 'finance', - 'service_call_id' => 'requested-service-call', + 'namespace' => 'caller', + 'caller_namespace' => 'caller', 'idempotency_key' => 'invoice-request-42', 'mode_override' => ' SYNC ', 'wait_for' => ' COMPLETED ', @@ -327,13 +330,6 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad ], 'duplicate_start_policy' => 'return_existing_active', 'request_payload_reference' => 'invoice-request-reference', - 'principal_subject' => 'user:billing-clerk', - 'principal_method' => 'token', - 'principal_roles' => ['invoice:create', 'invoice:read'], - 'principal_tenant' => 'customer-a', - 'principal_claims' => [ - 'approved' => false, - ], 'metadata' => [ 'caller_sdk_language' => 'sdk-python', 'caller_workflow_instance_id' => 'forged-owner', @@ -368,7 +364,13 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad 'service_sdk_language' => 'rust', 'trace_id' => 'trace-42', ]; + $this->assertInstanceOf(ServiceCallPrincipal::class, $forwarded['principal']); + $forwarded['principal'] = $forwarded['principal']->toAuditArray(); $expected += [ + 'principal' => (new ServiceCallPrincipal( + subject: 'workflow:' . $this->run->id, + method: 'workflow' + ))->toAuditArray(), 'arguments' => $arguments, 'payload_blob' => $payload, 'payload_codec' => 'avro', diff --git a/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php b/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php index 843b1f784..f4b03591f 100644 --- a/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php +++ b/tests/Unit/V2/WorkflowExecutorServiceOperationTest.php @@ -15,6 +15,7 @@ use Workflow\Serializers\Serializer; use Workflow\V2\Attributes\Type; use Workflow\V2\Contracts\ServiceControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Exceptions\RestoredWorkflowException; @@ -26,7 +27,9 @@ use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\DefaultWorkflowTaskBridge; +use Workflow\V2\Support\RunServiceOperationAuthorizer; use Workflow\V2\Support\RunTimelineProjector; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceOperationOptions; use Workflow\V2\Support\ServiceOperationResult; use Workflow\V2\Support\WorkerCompatibilityFleet; @@ -39,6 +42,7 @@ final class WorkflowExecutorServiceOperationTest extends TestCase protected function setUp(): void { parent::setUp(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); Carbon::setTestNow('2026-10-08 17:10:00'); Queue::fake(); config() @@ -107,6 +111,10 @@ public function testPublicExecutionRecordsTheBoundaryOutcomeAndReplaysWithoutCal $this->assertCount(1, $service->executions); $request = Serializer::serializeWithCodec('avro', ExecutorServiceOperationWorkflow::REQUEST); $expectedOptions = [ + 'principal' => (new ServiceCallPrincipal( + subject: 'workflow:' . $run->id, + method: 'workflow' + ))->toAuditArray(), 'namespace' => 'service-caller', 'arguments' => ExecutorServiceOperationWorkflow::REQUEST, 'payload_blob' => $request, @@ -128,6 +136,8 @@ public function testPublicExecutionRecordsTheBoundaryOutcomeAndReplaysWithoutCal 'wait_for' => 'accepted', ]; } + $this->assertInstanceOf(ServiceCallPrincipal::class, $service->executions[0][3]['principal']); + $service->executions[0][3]['principal'] = $service->executions[0][3]['principal']->toAuditArray(); $this->assertSame( $this->ordered(['billing', 'invoices', 'create', $expectedOptions]), $this->ordered($service->executions[0]) diff --git a/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php b/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php new file mode 100644 index 000000000..af47acd99 --- /dev/null +++ b/tests/Unit/V2/WorkflowServiceOperationAuthorityTest.php @@ -0,0 +1,471 @@ +app->instance(WorkflowServiceOperationAuthorizer::class, new RunServiceOperationAuthorizer()); + Carbon::setTestNow('2026-10-25T21:00:00Z'); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + 'workflows.v2.compatibility.current' => 'authority-test', + 'workflows.v2.compatibility.supported' => ['authority-test'], + 'workflows.v2.namespace' => null, + ]); + WorkerCompatibilityFleet::clear(); + Queue::fake(); + $workflow = WorkflowStub::make(ServiceAuthorityWorkflow::class, 'authority-owner', 'caller'); + $this->assertTrue($workflow->attemptStart()->accepted()); + $this->run = WorkflowRun::query()->findOrFail($workflow->runId()); + $this->task = WorkflowTask::query()->where('workflow_run_id', $this->run->id)->sole(); + $this->bridge = $this->app->make(WorkflowTaskBridge::class); + $this->assertTrue($this->bridge->claimStatus($this->task->id, 'authority-worker')['claimed']); + } + + protected function tearDown(): void + { + Carbon::setTestNow(); + WorkerCompatibilityFleet::clear(); + parent::tearDown(); + } + + /** + * @param array $options + */ + #[DataProvider('forgedAuthority')] + public function testWorkerAuthorityOverridesRefuseTheWholeCompletion(array $options): void + { + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + $options, + ]); + + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_operation_authority_invalid', $result['reason']); + $this->assertSame($before, $this->records()); + } + + /** + * @return iterable}> + */ + public static function forgedAuthority(): iterable + { + yield 'foreign namespace' => [[ + 'namespace' => 'foreign', + ]]; + yield 'foreign caller' => [[ + 'caller_namespace' => 'foreign', + ]]; + yield 'principal subject' => [[ + 'principal_subject' => 'administrator', + ]]; + yield 'principal method' => [[ + 'principal_method' => 'system', + ]]; + yield 'principal roles' => [[ + 'principal_roles' => ['admin'], + ]]; + yield 'principal tenant' => [[ + 'principal_tenant' => 'foreign', + ]]; + yield 'principal claims' => [[ + 'principal_claims' => [ + 'administrator' => true, + ], + ]]; + } + + #[DataProvider('foreignCalls')] + public function testForeignCallReferencesRefuseBeforeAnyCommandWrites( + string $status, + string $mismatch, + bool $idempotency, + ): void { + $call = $this->serviceCall($status); + $call->forceFill(match ($mismatch) { + 'run' => [ + 'caller_workflow_run_id' => 'another-run', + ], + 'instance' => [ + 'caller_workflow_instance_id' => 'another-owner', + ], + 'caller' => [ + 'caller_namespace' => 'another-caller', + ], + 'catalog' => [ + 'operation_name' => 'another-operation', + ], + 'namespace' => [ + 'namespace' => 'foreign', + 'target_namespace' => 'foreign', + ], + 'endpoint id' => [ + 'workflow_service_endpoint_id' => 'foreign-endpoint', + ], + 'service id' => [ + 'workflow_service_id' => 'foreign-service', + ], + 'operation id' => [ + 'workflow_service_operation_id' => 'foreign-operation', + ], + })->save(); + $options = $idempotency ? [ + 'idempotency_key' => $call->idempotency_key, + ] + : [ + 'service_call_id' => $call->id, + ]; + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + $options, + ]); + + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + /** + * @return iterable + */ + public static function foreignCalls(): iterable + { + foreach (['pending', 'started', 'completed'] as $status) { + foreach ([ + 'run', + 'instance', + 'caller', + 'catalog', + 'namespace', + 'endpoint id', + 'service id', + 'operation id', + ] as $mismatch) { + yield $status . ' ' . $mismatch => [$status, $mismatch, false]; + } + foreach (['run', 'instance', 'caller'] as $mismatch) { + yield $status . ' idempotency ' . $mismatch => [$status, $mismatch, true]; + } + } + } + + #[DataProvider('ownedCalls')] + public function testOwnedCallReplayRemainsUsable(string $status, bool $idempotency): void + { + $call = $this->serviceCall($status); + $result = $this->bridge->complete($this->task->id, [ + $this->command() + ($idempotency + ? [ + 'idempotency_key' => $call->idempotency_key, + ] : [ + 'service_call_id' => $call->id, + ])]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame($this->run->id, $call->fresh()->caller_workflow_run_id); + } + + /** + * @return iterable + */ + public static function ownedCalls(): iterable + { + foreach (['started', 'completed'] as $status) { + yield $status . ' by id' => [$status, false]; + yield $status . ' by key' => [$status, true]; + } + } + + public function testNewCallsUseTheRunWorkloadPrincipal(): void + { + $result = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $call = WorkflowServiceCall::query()->sole(); + $this->assertSame('caller', $call->namespace); + $this->assertSame('caller', $call->caller_namespace); + $this->assertSame('workflow:' . $this->run->id, $call->caller_principal_subject); + $this->assertSame('workflow', $call->caller_principal_method); + $this->assertNull($call->caller_principal_roles); + $this->assertNull($call->caller_principal_tenant); + $this->assertNull($call->caller_principal_claims); + } + + public function testMissingTrustedContextRefusesBeforeAnyCommandWrites(): void + { + $this->app->forgetInstance(WorkflowServiceOperationAuthorizer::class); + $this->assertInstanceOf( + DenyWorkflowServiceOperationAuthorizer::class, + $this->app->make(WorkflowServiceOperationAuthorizer::class) + ); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command(), + ]); + $this->assertFalse($result['completed']); + $this->assertSame('service_operation_authority_invalid', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testNullRunNamespaceDoesNotUseAmbientConfiguration(): void + { + $this->run->instance->forceFill([ + 'namespace' => null, + ])->save(); + $this->run->forceFill([ + 'namespace' => null, + ])->save(); + $this->task->forceFill([ + 'namespace' => null, + ])->save(); + config([ + 'workflows.v2.namespace' => 'ambient-namespace', + ]); + + $result = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $call = WorkflowServiceCall::query()->sole(); + $this->assertNull($call->namespace); + $this->assertNull($call->target_namespace); + $this->assertNull($call->caller_namespace); + $this->assertSame($this->run->id, $call->caller_workflow_run_id); + } + + public function testDefaultIdempotencyKeyCannotReplayAnotherRunsCall(): void + { + $call = $this->serviceCall('completed'); + $call->forceFill([ + 'caller_workflow_run_id' => 'another-run', + 'idempotency_key' => 'workflow-service-operation:' . $this->run->workflow_instance_id . ':' . $this->run->id . ':2', + ])->save(); + $before = $this->records(); + $result = $this->bridge->complete($this->task->id, [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command(), + ]); + $this->assertFalse($result['completed']); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testRetainedClaimPrefixRefusesForeignCallBeforeWritingItsSibling(): void + { + $call = $this->serviceCall('completed'); + $call->forceFill([ + 'caller_workflow_instance_id' => 'another-instance', + ])->save(); + $before = $this->records(); + $result = $this->app->make(\Workflow\V2\Contracts\PreparedLocalActivityTaskBridge::class) + ->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $this->task->fresh() +->attempt_count, + 'authority-checkpoint', + 1, + [ + [ + 'type' => 'start_timer', + 'delay_seconds' => 60, + ], + $this->command() + [ + 'service_call_id' => $call->id, + ], + ], + '1.20' + ); + $this->assertFalse($result['checkpointed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame('service_call_not_owned', $result['reason']); + $this->assertSame($before, $this->records()); + } + + public function testDuplicatePrefixStillRequiresCurrentTrustedAuthority(): void + { + $call = $this->serviceCall('completed'); + $bridge = $this->app->make(\Workflow\V2\Contracts\PreparedLocalActivityTaskBridge::class); + $commands = [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]; + $attempt = $this->task->fresh() +->attempt_count; + $first = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertTrue($first['checkpointed'], json_encode($first, JSON_THROW_ON_ERROR)); + $before = $this->records(); + $this->app->instance(WorkflowServiceOperationAuthorizer::class, new DenyWorkflowServiceOperationAuthorizer()); + $duplicate = $bridge->checkpointLocalActivityPrefix( + $this->task->id, + 'authority-worker', + $attempt, + 'authority-checkpoint', + 1, + $commands, + '1.20' + ); + $this->assertFalse($duplicate['checkpointed'], json_encode($duplicate, JSON_THROW_ON_ERROR)); + $this->assertSame('service_operation_authority_invalid', $duplicate['reason']); + $this->assertSame($before, $this->records()); + } + + public function testOwnedResolutionFailureCanBeReplayed(): void + { + $first = $this->bridge->complete($this->task->id, [$this->command()]); + $this->assertTrue($first['completed']); + $call = WorkflowServiceCall::query()->sole(); + $before = $call->getRawOriginal(); + $this->assertSame('failed', $call->status); + $task = WorkflowTask::query()->whereKey($first['created_task_ids'][0])->sole(); + $this->assertTrue($this->bridge->claimStatus($task->id, 'authority-worker')['claimed']); + $replay = $this->bridge->complete($task->id, [ + $this->command() + [ + 'service_call_id' => $call->id, + ]]); + $this->assertTrue($replay['completed'], json_encode($replay, JSON_THROW_ON_ERROR)); + $this->assertSame(1, WorkflowServiceCall::query()->count()); + $this->assertSame($before, $call->fresh()->getRawOriginal()); + } + + /** + * @return array + */ + private function command(): array + { + return [ + 'type' => 'start_service_operation', + 'endpoint_name' => 'billing', + 'service_name' => 'invoices', + 'operation_name' => 'create', + ]; + } + + private function serviceCall(string $status): WorkflowServiceCall + { + $endpoint = WorkflowServiceEndpoint::query()->create([ + 'namespace' => 'caller', + 'endpoint_name' => 'billing', + ]); + $service = WorkflowService::query()->create([ + 'namespace' => 'caller', + 'workflow_service_endpoint_id' => $endpoint->id, + 'service_name' => 'invoices', + ]); + $operation = WorkflowServiceOperation::query()->create([ + 'namespace' => 'caller', + 'workflow_service_endpoint_id' => $endpoint->id, + 'workflow_service_id' => $service->id, + 'operation_name' => 'create', + 'operation_mode' => 'async', + 'handler_binding_kind' => ServiceCallBindingKind::WorkflowRun->value, + 'handler_target_reference' => 'service-authority-workflow', + ]); + return WorkflowServiceCall::query()->create([ + 'namespace' => 'caller', + 'target_namespace' => 'caller', + 'caller_namespace' => 'caller', + 'caller_workflow_instance_id' => $this->run->workflow_instance_id, + 'caller_workflow_run_id' => $this->run->id, + 'workflow_service_endpoint_id' => $endpoint->id, + 'workflow_service_id' => $service->id, + 'workflow_service_operation_id' => $operation->id, + 'endpoint_name' => 'billing', + 'service_name' => 'invoices', + 'operation_name' => 'create', + 'operation_mode' => 'async', + 'status' => $status, + 'outcome' => $status === 'completed' ? ServiceCallOutcome::Completed->value + : ServiceCallOutcome::Accepted->value, + 'resolved_binding_kind' => ServiceCallBindingKind::WorkflowRun->value, + 'idempotency_key' => 'shared-idempotency-key', + 'accepted_at' => $status === ServiceCallStatus::Pending->value ? null : now(), + 'started_at' => $status === ServiceCallStatus::Started->value ? now() : null, + 'completed_at' => $status === ServiceCallStatus::Completed->value ? now() : null, + ]); + } + + /** + * @return array>> + */ + private function records(): array + { + $records = []; + foreach ([WorkflowInstance::class, WorkflowRun::class, WorkflowTask::class, WorkflowCommand::class, + WorkflowHistoryEvent::class, WorkflowRunSummary::class, WorkflowServiceCall::class, + WorkflowServiceEndpoint::class, WorkflowService::class, WorkflowServiceOperation::class, + WorkflowTimer::class] as $model) { + $records[$model] = $model::query()->orderBy('id')->get()->map( + static fn (Model $row): array => $row->getRawOriginal(), + )->all(); + } + return $records; + } +} + +#[Type('service-authority-workflow')] +final class ServiceAuthorityWorkflow extends Workflow +{ + public function handle(): string + { + return 'done'; + } +} From d902242ae737b871e24e889979dc5cac78082fdb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:36:06 +0000 Subject: [PATCH 2/3] Retain service authority refusal in the replay corpus --- .../V2EmbeddedReplayRegressionCorpusTest.php | 6 +++ ...flow-service-operation-host-authority.json | 48 +++++++++++++++++++ tests/Support/NexusCallerDeliveryScenario.php | 41 ++++++++++++++++ 3 files changed, 95 insertions(+) create mode 100644 tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php index b2e788402..ccb65115a 100644 --- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php +++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php @@ -330,6 +330,12 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void continue; } + if ($fixture['id'] === 'workflow-service-operation-host-authority') { + $this->clearWorkflowState(); + $this->assertNexusWorkerCannotSupplyAuthority($fixture); + continue; + } + if ($fixture['id'] === 'nexus-waiting-caller-terminal-delivery') { $this->clearWorkflowState(); $this->assertWaitingNexusCallerReceivesResult( diff --git a/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json new file mode 100644 index 000000000..4f01b6bbf --- /dev/null +++ b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-host-authority.json @@ -0,0 +1,48 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "workflow-service-operation-host-authority", + "protocol_version": "1.19", + "bindings": ["php"], + "consumers": ["workflow-fiber-runner", "workflow-executor"], + "workflow": { + "type": "Tests\\Fixtures\\V2\\TestNexusWaitingCallerWorkflow", + "arguments": ["world"], + "payload_codec": "avro" + }, + "history": [ + {"sequence": 1, "event_type": "WorkflowStarted", "payload": {}}, + { + "sequence": 2, + "event_type": "ServiceCallStarted", + "payload": { + "sequence": 1, + "service_call_id": "authority-recorded-call", + "endpoint_name": "php-service", + "service_name": "greeter", + "operation_name": "greet", + "status": "started", + "operation_mode": "sync", + "wait_for": "completed", + "service_call": {"service_call_id": "authority-recorded-call", "status": "started"} + } + } + ], + "authority": { + "commands": [ + {"type": "start_timer", "delay_seconds": 60}, + { + "type": "start_service_operation", + "endpoint_name": "billing", + "service_name": "invoices", + "operation_name": "create", + "namespace": "foreign", + "caller_namespace": "foreign", + "principal_subject": "administrator", + "principal_roles": ["admin"] + } + ], + "reason": "service_operation_authority_invalid" + }, + "expected": {"completed": false, "result": null, "commands": []} +} diff --git a/tests/Support/NexusCallerDeliveryScenario.php b/tests/Support/NexusCallerDeliveryScenario.php index d8278dc34..390fcd3f7 100644 --- a/tests/Support/NexusCallerDeliveryScenario.php +++ b/tests/Support/NexusCallerDeliveryScenario.php @@ -17,14 +17,17 @@ use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Jobs\RunWorkflowTask; +use Workflow\V2\Models\WorkflowCommand; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; use Workflow\V2\Models\WorkflowRun; +use Workflow\V2\Models\WorkflowRunSummary; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowServiceEndpoint; use Workflow\V2\Models\WorkflowServiceOperation; use Workflow\V2\Models\WorkflowTask; +use Workflow\V2\Models\WorkflowTimer; use Workflow\V2\Support\DefaultServiceControlPlane; use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\WorkflowFiberRunner; @@ -34,6 +37,44 @@ trait NexusCallerDeliveryScenario { + /** + * @param array $fixture + */ + protected function assertNexusWorkerCannotSupplyAuthority(array $fixture): void + { + $previousDispatchMode = config('workflows.v2.task_dispatch_mode'); + config([ + 'workflows.v2.task_dispatch_mode' => 'poll', + ]); + try { + $workflow = WorkflowStub::make($fixture['workflow']['type'], 'authority-corpus-caller', 'caller'); + $this->assertTrue($workflow->attemptStart(...$fixture['workflow']['arguments'])->accepted()); + $task = WorkflowTask::query()->where('workflow_run_id', $workflow->runId())->sole(); + $bridge = $this->app->make(WorkflowTaskBridge::class); + $this->assertTrue($bridge->claimStatus($task->id, 'authority-corpus-worker')['claimed']); + $models = [WorkflowInstance::class, WorkflowRun::class, WorkflowTask::class, WorkflowTimer::class, + WorkflowCommand::class, WorkflowHistoryEvent::class, WorkflowRunSummary::class, WorkflowServiceCall::class]; + $snapshot = static function () use ($models): array { + $rows = []; + foreach ($models as $model) { + $rows[$model] = $model::query()->orderBy('id')->get()->map( + static fn ($row): array => $row->getRawOriginal(), + )->all(); + } + return $rows; + }; + $before = $snapshot(); + $result = $bridge->complete($task->id, $fixture['authority']['commands']); + $this->assertFalse($result['completed'], json_encode($result, JSON_THROW_ON_ERROR)); + $this->assertSame($fixture['authority']['reason'], $result['reason']); + $this->assertSame($before, $snapshot()); + } finally { + config([ + 'workflows.v2.task_dispatch_mode' => $previousDispatchMode, + ]); + } + } + /** * @param array $fixture */ From fce3ac4b4448fc6b3835d338725a0d37d29d2af3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 00:54:29 +0000 Subject: [PATCH 3/3] Bind trusted hosts in service operation tests --- tests/Unit/V2/LinkedServiceCallCloserTest.php | 23 +++++++++++++++++++ .../Unit/V2/NexusCancellationDeliveryTest.php | 19 +++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/tests/Unit/V2/LinkedServiceCallCloserTest.php b/tests/Unit/V2/LinkedServiceCallCloserTest.php index 173906cbd..efd4cef15 100644 --- a/tests/Unit/V2/LinkedServiceCallCloserTest.php +++ b/tests/Unit/V2/LinkedServiceCallCloserTest.php @@ -15,6 +15,7 @@ use Workflow\V2\Contracts\ExternalPayloadStoragePolicy; use Workflow\V2\Contracts\ServiceControlPlane; use Workflow\V2\Contracts\WorkflowControlPlane; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; @@ -22,6 +23,7 @@ use Workflow\V2\Enums\ServiceCallOutcome; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Jobs\RunWorkflowTask; use Workflow\V2\Models\WorkflowHistoryEvent; use Workflow\V2\Models\WorkflowInstance; @@ -36,6 +38,7 @@ use Workflow\V2\Support\ExternalPayloadStorage; use Workflow\V2\Support\LinkedServiceCallCloser; use Workflow\V2\Support\ServiceCallDeadline; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceCallView; use Workflow\V2\Support\ServiceCallWorkflowDelivery; use Workflow\V2\Support\ServiceOperationOptions; @@ -456,6 +459,7 @@ public function testEmbeddedCallerWaitsAndThenReplaysTheActualServiceResult(): v 'workflow_run_id' => $service->id, ]); $this->registerWorkflowService(); + $this->registerTrustedServiceOperationHost(); $this->app->instance(ServiceControlPlane::class, new DefaultServiceControlPlane($control)); $workflow = WorkflowStub::make(NexusWaitingCallerWorkflow::class, 'waiting-caller', 'caller'); $this->assertTrue($workflow->attemptStart()->accepted()); @@ -997,6 +1001,25 @@ private function registerWorkflowService(): void ]); } + private function registerTrustedServiceOperationHost(): void + { + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + if ($run->namespace !== 'caller' || ($command['namespace'] ?? null) !== 'service' + || [$command['endpoint_name'], $command['service_name'], $command['operation_name']] + !== ['php-service', 'greeter', 'greet']) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); + } + private function createRun(string $namespace = 'service'): WorkflowRun { $instance = WorkflowInstance::query()->create([ diff --git a/tests/Unit/V2/NexusCancellationDeliveryTest.php b/tests/Unit/V2/NexusCancellationDeliveryTest.php index e5e5aa11a..794f3a22b 100644 --- a/tests/Unit/V2/NexusCancellationDeliveryTest.php +++ b/tests/Unit/V2/NexusCancellationDeliveryTest.php @@ -11,20 +11,24 @@ use Tests\Fixtures\V2\TestNexusWaitingCallerWorkflow; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer; use Workflow\V2\Contracts\WorkflowTaskBridge; use Workflow\V2\Enums\HistoryEventType; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Enums\TaskStatus; use Workflow\V2\Enums\TaskType; +use Workflow\V2\Exceptions\ServiceOperationAuthorizationException; use Workflow\V2\Exceptions\WorkflowCancellationRequestedException; use Workflow\V2\Jobs\RunTimerTask; use Workflow\V2\Jobs\RunWorkflowTask; +use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowService; use Workflow\V2\Models\WorkflowServiceCall; use Workflow\V2\Models\WorkflowServiceEndpoint; use Workflow\V2\Models\WorkflowServiceOperation; use Workflow\V2\Models\WorkflowTask; use Workflow\V2\Support\CooperativeCancellationDelivery; +use Workflow\V2\Support\ServiceCallPrincipal; use Workflow\V2\Support\ServiceOperationCancellation; use Workflow\V2\Support\WorkflowFiberRunner; use Workflow\V2\WorkflowStub; @@ -40,6 +44,21 @@ protected function setUp(): void 'workflows.v2.compatibility.current' => 'build-a', 'workflows.v2.compatibility.supported' => ['build-a'], ]); + $this->app->instance( + WorkflowServiceOperationAuthorizer::class, + new class() implements WorkflowServiceOperationAuthorizer { + public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal + { + if ($run->namespace !== 'caller' || ($command['namespace'] ?? null) !== 'service' + || [$command['endpoint_name'], $command['service_name'], $command['operation_name']] + !== ['php-service', 'greeter', 'greet']) { + throw new ServiceOperationAuthorizationException(); + } + + return new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'); + } + } + ); Carbon::setTestNow('2026-10-12T08:00:00Z'); }