fix(stepfunctions-tasks): scope down BatchSubmitJob permissions to specific job definition

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions-tasks/test/batch/integ.submit-job.js.snapshot/aws-stepfunctions-integ.template.json

@@ -669,24 +669,7 @@
          ]
         },
         {
-         "Fn::Join": [
-          "",
-          [
-           "arn:",
-           {
-            "Ref": "AWS::Partition"
-           },
-           ":batch:",
-           {
-            "Ref": "AWS::Region"
-           },
-           ":",
-           {
-            "Ref": "AWS::AccountId"
-           },
-           ":job-definition/*"
-          ]
-         ]
+         "Ref": "JobDefinition24FFE3ED"
         }
        ]
       },

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

@@ -308,17 +308,11 @@ export class BatchSubmitJob extends sfn.TaskStateBase {
   }
 
   private configurePolicyStatements(): iam.PolicyStatement[] {
+    const useWildcard = isJsonPathOrJsonataExpression(this.props.jobQueueArn) || isJsonPathOrJsonataExpression(this.props.jobDefinitionArn);
     return [
-      // Resource level access control for job-definition requires revision which batch does not support yet
-      // 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: isJsonPathOrJsonataExpression(this.props.jobQueueArn) ? ['*'] : [
-          Stack.of(this).formatArn({
-            service: 'batch',
-            resource: 'job-definition',
-            resourceName: '*',
-          }),
+        resources: useWildcard ? ['*'] : [
+          this.props.jobDefinitionArn,
           this.props.jobQueueArn,
         ],
         actions: ['batch:SubmitJob'],

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

@@ -1,5 +1,5 @@
 import * as path from 'path';
-import { Template } from '../../../assertions';
+import { Match, Template } from '../../../assertions';
 import * as batch from '../../../aws-batch';
 import * as ec2 from '../../../aws-ec2';
 import * as ecs from '../../../aws-ecs';
@@ -590,3 +590,29 @@ test('supports passing jobQueueArn as JsonPath or JSONata', () => {
     },
   });
 });
+
+test('scopes down permissions to specific job definition and queue ARNs', () => {
+  const task = new BatchSubmitJob(stack, 'ScopedTask', {
+    jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
+    jobName: 'MyJob',
+    jobQueueArn: batchJobQueue.jobQueueArn,
+  });
+
+  new sfn.StateMachine(stack, 'SM2', {
+    definition: task,
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: Match.arrayWith([
+        Match.objectLike({
+          Action: 'batch:SubmitJob',
+          Effect: 'Allow',
+          Resource: Match.arrayWith([
+            { Ref: 'JobDefinition24FFE3ED' },
+          ]),
+        }),
+      ]),
+    },
+  });
+});