fix(stepfunctions-tasks): IAM policy use wildcard for JsonPath or JSO…

…Nata

packages/aws-cdk-lib/aws-stepfunctions-tasks/lib/batch/submit-job.ts

@@ -3,7 +3,7 @@ import * as ec2 from '../../../aws-ec2';
 import * as iam from '../../../aws-iam';
 import * as sfn from '../../../aws-stepfunctions';
 import { Size, Stack, withResolved } from '../../../core';
-import { integrationResourceArn, validatePatternSupported } from '../private/task-utils';
+import { integrationResourceArn, isJsonPathOrJsonataExpression, validatePatternSupported } from '../private/task-utils';
 
 /**
  * The overrides that should be sent to a container.
@@ -311,7 +311,7 @@ export class BatchSubmitJob extends sfn.TaskStateBase {
       // Using the alternative permissions as mentioned here:
       // https://docs.aws.amazon.com/batch/latest/userguide/batch-supported-iam-actions-resources.html
       new iam.PolicyStatement({
-        resources: [
+        resources: isJsonPathOrJsonataExpression(this.props.jobQueueArn) ? ['*'] : [
           Stack.of(this).formatArn({
             service: 'batch',
             resource: 'job-definition',

packages/aws-cdk-lib/aws-stepfunctions-tasks/test/batch/submit-job.test.ts

@@ -5,6 +5,7 @@ import * as ecs from '../../../aws-ecs';
 import * as sfn from '../../../aws-stepfunctions';
 import { BatchSubmitJob } from '../../../aws-stepfunctions-tasks';
 import * as cdk from '../../../core';
+import { Template } from '../../../assertions';
 
 let stack: cdk.Stack;
 let batchJobDefinition: batch.IJobDefinition;
@@ -512,3 +513,80 @@ test('throws if tags has invalid value', () => {
     /Tag value maximum size is 256/,
   );
 });
+
+test('supports passing jobQueueArn as JsonPath or JSONata', () => {
+  // WHEN
+  const task = new BatchSubmitJob(stack, 'Task', {
+    jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+    jobQueueArn: sfn.JsonPath.stringAt('$.jobQueueArn'),
+    jobName: sfn.JsonPath.stringAt('$.jobName'),
+  });
+
+  new sfn.StateMachine(stack, 'ParentStateMachine', {
+    definitionBody: sfn.DefinitionBody.fromChainable(task),
+  });
+
+  // THEN
+  expect(stack.resolve(task.toStateJson())).toEqual({
+    Type: 'Task',
+    Resource: {
+      'Fn::Join': [
+        '',
+        [
+          'arn:',
+          {
+            Ref: 'AWS::Partition',
+          },
+          ':states:::batch:submitJob.sync',
+        ],
+      ],
+    },
+    End: true,
+    Parameters: {
+      'JobDefinition': { Ref: 'JobDefinition24FFE3ED' },
+      'JobName.$': '$.jobName',
+      'JobQueue.$': '$.jobQueueArn'
+    },
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [
+        {
+          Action: 'batch:SubmitJob',
+          Effect: 'Allow',
+          Resource: '*',
+        },
+        {
+          Action: [
+            'events:PutTargets',
+            'events:PutRule',
+            'events:DescribeRule'
+          ],
+          Effect: 'Allow',
+          Resource: {
+            'Fn::Join': [
+              '',
+              [
+                'arn:',
+                {
+                  Ref: 'AWS::Partition',
+                },
+                ':events:',
+                {
+                  Ref: 'AWS::Region',
+                },
+                ':',
+                {
+                  Ref: 'AWS::AccountId',
+                },
+                ':rule/StepFunctionsGetEventsForBatchJobsRule',
+              ],
+            ],
+          },
+        },
+      ],
+      Version: '2012-10-17',
+    },
+  });
+});