fix(step-functions): arn is not valid across partitions (#22314)

There were a number of arns with hardcoded formatting in the stepfunctions-tasks module that would cause policies to be invalid across partitions. Updated these to use our standard `Stack.formatArn` functionality.

----

### All Submissions:

* [ ] Have you followed the guidelines in our [Contributing guide?](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md)

### Adding new Unconventional Dependencies:

* [ ] This PR adds new unconventional dependencies following the process described [here](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md/#adding-new-unconventional-dependencies)

### New Features

* [ ] Have you added the new feature to an [integration test](https://github.com/aws/aws-cdk/blob/main/INTEGRATION_TESTS.md)?
	* [ ] Did you use `yarn integ` to deploy the infrastructure and generate the snapshot (i.e. `yarn integ` without `--dry-run`)?

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

packages/@aws-cdk/aws-stepfunctions-tasks/lib/athena/start-query-execution.ts

@@ -106,7 +106,15 @@ export class AthenaStartQueryExecution extends sfn.TaskStateBase {
           's3:ListBucketMultipartUploads',
           's3:ListMultipartUploadParts',
           's3:PutObject'],
-        resources: [this.props.resultConfiguration?.outputLocation?.bucketName ? `arn:aws:s3:::${this.props.resultConfiguration?.outputLocation?.bucketName}/${this.props.resultConfiguration?.outputLocation?.objectKey}/*` : '*'], // Need S3 location where data is stored or Athena throws an Unable to verify/create output bucket https://docs.aws.amazon.com/athena/latest/ug/security-iam-athena.html
+        resources: [
+          this.props.resultConfiguration?.outputLocation?.bucketName
+            ? cdk.Stack.of(this).formatArn({
+              service: 's3',
+              resource: this.props.resultConfiguration?.outputLocation?.bucketName,
+              resourceName: this.props.resultConfiguration?.outputLocation?.objectKey,
+            })
+            : '*',
+        ],
       }),
     );
 

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/start-job-run.ts

@@ -280,7 +280,10 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     jobExecutionRole.addToPrincipalPolicy(
       new iam.PolicyStatement({
         resources: [
-          'arn:aws:logs:*:*:*',
+          cdk.Stack.of(this).formatArn({
+            service: 'logs',
+            resource: '*',
+          }),
         ],
         actions: [
           'logs:DescribeLogGroups',
@@ -301,7 +304,10 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     this.role.addToPrincipalPolicy(
       new iam.PolicyStatement({
         resources: [
-          'arn:aws:logs:*:*:*',
+          cdk.Stack.of(this).formatArn({
+            service: 'logs',
+            resource: '*',
+          }),
         ],
         actions: [
           'logs:DescribeLogGroups',

packages/@aws-cdk/aws-stepfunctions-tasks/test/athena/start-query-execution.test.ts

@@ -1,3 +1,4 @@
+import { Template, Match } from '@aws-cdk/assertions';
 import * as sfn from '@aws-cdk/aws-stepfunctions';
 import * as cdk from '@aws-cdk/core';
 import { AthenaStartQueryExecution, EncryptionOption } from '../../lib/athena/start-query-execution';
@@ -174,4 +175,62 @@ describe('Start Query Execution', () => {
     // THEN
     expect(stack.resolve(task.toStateJson())).not.toHaveProperty('Parameters.QueryExecutionContext');
   });
+
+  test('bucket arn is formatted as expected in generated policy', () => {
+    // GIVEN
+    const stack = new cdk.Stack();
+
+    // WHEN
+    const task = new AthenaStartQueryExecution(stack, 'Query', {
+      queryString: 'CREATE DATABASE database',
+      clientRequestToken: 'unique-client-request-token',
+      resultConfiguration: {
+        outputLocation: {
+          bucketName: 'query-results-bucket',
+          objectKey: 'folder',
+        },
+      },
+    });
+
+    new sfn.StateMachine(stack, 'StateMachine', {
+      definition: task,
+    });
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+      PolicyDocument: Match.objectLike({
+        Statement: Match.arrayWith([
+          {
+            Action: [
+              's3:AbortMultipartUpload',
+              's3:ListBucketMultipartUploads',
+              's3:ListMultipartUploadParts',
+              's3:PutObject',
+            ],
+            Effect: 'Allow',
+            Resource: {
+              'Fn::Join': [
+                '',
+                [
+                  'arn:',
+                  {
+                    Ref: 'AWS::Partition',
+                  },
+                  ':s3:',
+                  {
+                    Ref: 'AWS::Region',
+                  },
+                  ':',
+                  {
+                    Ref: 'AWS::AccountId',
+                  },
+                  ':query-results-bucket/folder',
+                ],
+              ],
+            },
+          },
+        ]),
+      }),
+    });
+  });
 });