feat(stepfunctions-tasks): AWS SDK service integrations (#16746)

Add support for Step Functions' AWS SDK integrations to call
any of the over two hundred AWS services directly from a state
machine.

See https://docs.aws.amazon.com/step-functions/latest/dg/supported-services-awssdk.html
See https://aws.amazon.com/blogs/aws/now-aws-step-functions-supports-200-aws-services-to-enable-easier-workflow-automation/

Closes #16780 

----

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

Author: jogold

packages/@aws-cdk/aws-stepfunctions-tasks/README.md

@@ -33,6 +33,7 @@ This module is part of the [AWS Cloud Development Kit](https://github.com/aws/aw
   - [API Gateway](#api-gateway)
     - [Call REST API Endpoint](#call-rest-api-endpoint)
     - [Call HTTP API Endpoint](#call-http-api-endpoint)
+  - [AWS SDK](#aws-sdk)
   - [Athena](#athena)
     - [StartQueryExecution](#startqueryexecution)
     - [GetQueryExecution](#getqueryexecution)
@@ -205,7 +206,7 @@ const submitJob = new tasks.LambdaInvoke(this, 'Invoke Handler', {
 });
 ```
 
-You can also use [intrinsic functions](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-intrinsic-functions.html) with `JsonPath.stringAt()`. 
+You can also use [intrinsic functions](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-intrinsic-functions.html) with `JsonPath.stringAt()`.
 Here is an example of starting an Athena query that is dynamically created using the task input:
 
 ```ts
@@ -314,6 +315,43 @@ const invokeTask = new tasks.CallApiGatewayHttpApiEndpoint(stack, 'Call HTTP API
 });
 ```
 
+### AWS SDK
+
+Step Functions supports calling [AWS service's API actions](https://docs.aws.amazon.com/step-functions/latest/dg/supported-services-awssdk.html)
+through the service integration pattern.
+
+You can use Step Functions' AWS SDK integrations to call any of the over two hundred AWS services
+directly from your state machine, giving you access to over nine thousand API actions.
+
+```ts
+const getObject = new tasks.CallAwsService(this, 'GetObject', {
+  service: 's3',
+  action: 'getObject',
+  parameters: {
+    Bucket: myBucket.bucketName,
+    Key: sfn.JsonPath.stringAt('$.key')
+  },
+  iamResources: [myBucket.arnForObjects('*')],
+});
+```
+
+Use camelCase for actions and PascalCase for parameter names.
+
+The task automatically adds an IAM statement to the state machine role's policy based on the
+service and action called. The resources for this statement must be specified in `iamResources`.
+
+Use the `iamAction` prop to manually specify the IAM action name in the case where the IAM
+action name does not match with the API service/action name:
+
+```ts
+const listBuckets = new tasks.CallAwsService(this, 'ListBuckets', {
+  service: 's3',
+  action: 'ListBuckets',
+  iamResources: ['*'],
+  iamAction: 's3:ListAllMyBuckets'
+});
+```
+
 ## Athena
 
 Step Functions supports [Athena](https://docs.aws.amazon.com/step-functions/latest/dg/connect-athena.html) through the service integration pattern.

packages/@aws-cdk/aws-stepfunctions-tasks/lib/aws-sdk/call-aws-service.ts

@@ -0,0 +1,96 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import { Token } from '@aws-cdk/core';
+import { Construct } from 'constructs';
+import { integrationResourceArn } from '../private/task-utils';
+
+/**
+ * Properties for calling an AWS service's API action from your
+ * state machine.
+ *
+ * @see https://docs.aws.amazon.com/step-functions/latest/dg/supported-services-awssdk.html
+ */
+export interface CallAwsServiceProps extends sfn.TaskStateBaseProps {
+  /**
+   * The AWS service to call.
+   *
+   * @see https://docs.aws.amazon.com/step-functions/latest/dg/supported-services-awssdk.html
+   */
+  readonly service: string;
+
+  /**
+   * The API action to call.
+   *
+   * Use camelCase.
+   */
+  readonly action: string;
+
+  /**
+   * Parameters for the API action call.
+   *
+   * Use PascalCase for the parameter names.
+   *
+   * @default - no parameters
+   */
+  readonly parameters?: { [key: string]: any };
+
+  /**
+   * The resources for the IAM statement that will be added to the state
+   * machine role's policy to allow the state machine to make the API call.
+   *
+   * By default the action for this IAM statement will be `service:action`.
+   */
+  readonly iamResources: string[];
+
+  /**
+   * The action for the IAM statement that will be added to the state
+   * machine role's policy to allow the state machine to make the API call.
+   *
+   * Use in the case where the IAM action name does not match with the
+   * API service/action name, e.g. `s3:ListBuckets` requires `s3:ListAllMyBuckets`.
+   *
+   * @default - service:action
+   */
+  readonly iamAction?: string;
+}
+
+/**
+ * A StepFunctions task to call an AWS service API
+ */
+export class CallAwsService extends sfn.TaskStateBase {
+  protected readonly taskMetrics?: sfn.TaskMetricsConfig;
+  protected readonly taskPolicies?: iam.PolicyStatement[];
+
+  constructor(scope: Construct, id: string, private readonly props: CallAwsServiceProps) {
+    super(scope, id, props);
+
+    this.taskPolicies = [
+      new iam.PolicyStatement({
+        resources: props.iamResources,
+        // The prefix and the action name are case insensitive
+        // https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_action.html
+        actions: [props.iamAction ?? `${props.service}:${props.action}`],
+      }),
+    ];
+  }
+
+  /**
+   * @internal
+   */
+  protected _renderTask(): any {
+    let service = this.props.service;
+
+    if (!Token.isUnresolved(service)) {
+      service = service.toLowerCase();
+    }
+
+    return {
+      Resource: integrationResourceArn(
+        'aws-sdk',
+        `${service}:${this.props.action}`,
+        this.props.integrationPattern,
+      ),
+      Parameters: sfn.FieldUtils.renderObject(this.props.parameters) ?? {}, // Parameters is required for aws-sdk
+    };
+  }
+}

packages/@aws-cdk/aws-stepfunctions-tasks/lib/index.ts

@@ -47,3 +47,4 @@ export * from './databrew/start-job-run';
 export * from './eks/call';
 export * from './apigateway';
 export * from './eventbridge/put-events';
+export * from './aws-sdk/call-aws-service';

packages/@aws-cdk/aws-stepfunctions-tasks/lib/private/task-utils.ts

@@ -26,7 +26,7 @@ const resourceArnSuffix: Record<IntegrationPattern, string> = {
   [IntegrationPattern.WAIT_FOR_TASK_TOKEN]: '.waitForTaskToken',
 };
 
-export function integrationResourceArn(service: string, api: string, integrationPattern: IntegrationPattern): string {
+export function integrationResourceArn(service: string, api: string, integrationPattern?: IntegrationPattern): string {
   if (!service || !api) {
     throw new Error("Both 'service' and 'api' must be provided to build the resource ARN.");
   }

packages/@aws-cdk/aws-stepfunctions-tasks/test/aws-sdk/call-aws-service.test.ts

@@ -0,0 +1,148 @@
+import '@aws-cdk/assert-internal/jest';
+import * as sfn from '@aws-cdk/aws-stepfunctions';
+import * as cdk from '@aws-cdk/core';
+import * as tasks from '../../lib';
+
+let stack: cdk.Stack;
+
+beforeEach(() => {
+  // GIVEN
+  stack = new cdk.Stack();
+});
+
+test('CallAwsService task', () => {
+  // WHEN
+  const task = new tasks.CallAwsService(stack, 'GetObject', {
+    service: 's3',
+    action: 'getObject',
+    parameters: {
+      Bucket: 'my-bucket',
+      Key: sfn.JsonPath.stringAt('$.key'),
+    },
+    iamResources: ['*'],
+  });
+
+  new sfn.StateMachine(stack, 'StateMachine', {
+    definition: task,
+  });
+
+  // THEN
+  expect(stack.resolve(task.toStateJson())).toEqual({
+    Type: 'Task',
+    Resource: {
+      'Fn::Join': [
+        '',
+        [
+          'arn:',
+          {
+            Ref: 'AWS::Partition',
+          },
+          ':states:::aws-sdk:s3:getObject',
+        ],
+      ],
+    },
+    End: true,
+    Parameters: {
+      'Bucket': 'my-bucket',
+      'Key.$': '$.key',
+    },
+  });
+
+  expect(stack).toHaveResource('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [
+        {
+          Action: 's3:getObject',
+          Effect: 'Allow',
+          Resource: '*',
+        },
+      ],
+      Version: '2012-10-17',
+    },
+  });
+});
+
+test('with custom IAM action', () => {
+  // WHEN
+  const task = new tasks.CallAwsService(stack, 'ListBuckets', {
+    service: 's3',
+    action: 'listBuckets',
+    iamResources: ['*'],
+    iamAction: 's3:ListAllMyBuckets',
+  });
+
+  new sfn.StateMachine(stack, 'StateMachine', {
+    definition: task,
+  });
+
+  // THEN
+  expect(stack.resolve(task.toStateJson())).toEqual({
+    Type: 'Task',
+    Resource: {
+      'Fn::Join': [
+        '',
+        [
+          'arn:',
+          {
+            Ref: 'AWS::Partition',
+          },
+          ':states:::aws-sdk:s3:listBuckets',
+        ],
+      ],
+    },
+    End: true,
+    Parameters: {},
+  });
+
+  expect(stack).toHaveResource('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [
+        {
+          Action: 's3:ListAllMyBuckets',
+          Effect: 'Allow',
+          Resource: '*',
+        },
+      ],
+      Version: '2012-10-17',
+    },
+  });
+});
+
+test('with unresolved tokens', () => {
+  // WHEN
+  const task = new tasks.CallAwsService(stack, 'ListBuckets', {
+    service: new cdk.CfnParameter(stack, 'Service').valueAsString,
+    action: new cdk.CfnParameter(stack, 'Action').valueAsString,
+    iamResources: ['*'],
+  });
+
+  new sfn.StateMachine(stack, 'StateMachine', {
+    definition: task,
+  });
+
+  // THEN
+  expect(stack.resolve(task.toStateJson())).toEqual({
+    Type: 'Task',
+    Resource: {
+      'Fn::Join': [
+        '',
+        [
+          'arn:',
+          {
+            Ref: 'AWS::Partition',
+          },
+          ':states:::aws-sdk:',
+          {
+            Ref: 'Service',
+          },
+          ':',
+          {
+            Ref: 'Action',
+          },
+        ],
+      ],
+    },
+    End: true,
+    Parameters: {},
+  });
+});