feat(stepfunctions): add check for state machine encryption at rest (#11538)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 - `entra_conditional_access_policy_explicitly_targets_azure_devops` check for M365 provider, verifying at least one enabled Conditional Access policy explicitly includes the Azure DevOps cloud application instead of relying on a broad "All cloud apps" policy [(#11182)](https://github.com/prowler-cloud/prowler/pull/11182)
 - `entra_conditional_access_policy_no_exclusion_gaps` check for M365 provider, verifying every user, group, role, or application excluded from an enabled Conditional Access policy stays in scope of another enabled policy [(#11577)](https://github.com/prowler-cloud/prowler/pull/11577)
+- `stepfunctions_statemachine_encrypted_with_cmk` check for AWS provider, verifying that each Step Functions state machine uses a customer-managed KMS key for encryption at rest rather than the default AWS-owned key [(#11538)](https://github.com/prowler-cloud/prowler/pull/11538)
 
 ---
 

prowler/providers/aws/services/stepfunctions/stepfunctions_statemachine_encrypted_with_cmk/stepfunctions_statemachine_encrypted_with_cmk.metadata.json

@@ -0,0 +1,43 @@
+{
+  "Provider": "aws",
+  "CheckID": "stepfunctions_statemachine_encrypted_with_cmk",
+  "CheckTitle": "Step Functions state machine is encrypted at rest with a customer-managed KMS key",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices",
+    "Software and Configuration Checks/Industry and Regulatory Standards/AWS Foundational Security Best Practices",
+    "Software and Configuration Checks/Industry and Regulatory Standards/NIST 800-53 Controls (USA)"
+  ],
+  "ServiceName": "stepfunctions",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "AwsStepFunctionStateMachine",
+  "ResourceGroup": "serverless",
+  "Description": "**AWS Step Functions state machines** store execution history and input/output data passed between workflow states. This check verifies that each state machine uses a **customer-managed KMS key** (`CUSTOMER_MANAGED_KMS_KEY`) for encryption at rest rather than the default AWS-owned key.",
+  "Risk": "Without a customer-managed KMS key, execution history containing **sensitive input/output data** is protected only by an AWS-owned key you cannot control, rotate, or revoke. This limits **auditability** via CloudTrail, prevents independent access revocation, and weakens **confidentiality** for regulated workloads.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/step-functions/latest/dg/encryption-at-rest.html",
+    "https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#customer-cmk"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws stepfunctions update-state-machine --state-machine-arn <state-machine-arn> --encryption-configuration '{\"kmsKeyId\": \"<kms-key-id>\", \"type\": \"CUSTOMER_MANAGED_KMS_KEY\", \"kmsDataKeyReusePeriodSeconds\": 300}'",
+      "NativeIaC": "```yaml\nResources:\n  <example_resource_name>:\n    Type: AWS::StepFunctions::StateMachine\n    Properties:\n      RoleArn: arn:aws:iam::<account-id>:role/<example_role_name>\n      DefinitionString: |\n        {\"StartAt\":\"Pass\",\"States\":{\"Pass\":{\"Type\":\"Pass\",\"End\":true}}}\n      EncryptionConfiguration:\n        KmsKeyId: arn:aws:kms:<region>:<account-id>:key/<key-id>  # Critical: customer-managed KMS key\n        Type: CUSTOMER_MANAGED_KMS_KEY                             # Critical: must be CUSTOMER_MANAGED_KMS_KEY\n        KmsDataKeyReusePeriodSeconds: 300\n```",
+      "Other": "1. Open AWS Console > Step Functions > State machines\n2. Select the state machine and click Edit\n3. Under Encryption, select Customer managed key\n4. Choose an existing KMS key or create a new one\n5. Save changes",
+      "Terraform": "```hcl\nresource \"aws_sfn_state_machine\" \"<example_resource_name>\" {\n  name       = \"<example_resource_name>\"\n  role_arn   = \"arn:aws:iam::<account-id>:role/<example_role_name>\"\n  definition = jsonencode({ StartAt = \"Pass\", States = { Pass = { Type = \"Pass\", End = true } } })\n\n  encryption_configuration {\n    kms_key_id                        = \"arn:aws:kms:<region>:<account-id>:key/<key-id>\"  # Critical: customer-managed KMS key\n    type                              = \"CUSTOMER_MANAGED_KMS_KEY\"                         # Critical: must be CUSTOMER_MANAGED_KMS_KEY\n    kms_data_key_reuse_period_seconds = 300\n  }\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Configure each Step Functions state machine to use a **customer-managed KMS key** for encryption at rest. Assign a least-privilege key policy, enable **automatic key rotation**, and grant the execution role `kms:GenerateDataKey` and `kms:Decrypt`. Monitor key usage via CloudTrail.",
+      "Url": "https://hub.prowler.com/check/stepfunctions_statemachine_encrypted_with_cmk"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "stepfunctions_statemachine_logging_enabled"
+  ],
+  "Notes": ""
+}

prowler/providers/aws/services/stepfunctions/stepfunctions_statemachine_encrypted_with_cmk/stepfunctions_statemachine_encrypted_with_cmk.py

@@ -0,0 +1,50 @@
+from typing import List
+
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.stepfunctions.stepfunctions_client import (
+    stepfunctions_client,
+)
+from prowler.providers.aws.services.stepfunctions.stepfunctions_service import (
+    EncryptionType,
+)
+
+
+class stepfunctions_statemachine_encrypted_with_cmk(Check):
+    """Ensure Step Functions state machines are encrypted at rest with a customer-managed KMS key.
+
+    This check evaluates whether each AWS Step Functions state machine uses a
+    customer-managed KMS key (CUSTOMER_MANAGED_KMS_KEY) for encryption at rest rather
+    than the default AWS-owned key (AWS_OWNED_KEY).
+
+    - PASS: The state machine encryption_configuration type is CUSTOMER_MANAGED_KMS_KEY.
+    - FAIL: The state machine has no encryption_configuration or its type is AWS_OWNED_KEY.
+    """
+
+    def execute(self) -> List[Check_Report_AWS]:
+        """Execute the Step Functions state machine encryption at rest check.
+
+        Iterates over all Step Functions state machines and generates a report
+        indicating whether each state machine uses a customer-managed KMS key
+        for encryption at rest.
+
+        Returns:
+            List[Check_Report_AWS]: A list of report objects with the results of the check.
+        """
+        findings = []
+        for state_machine in stepfunctions_client.state_machines.values():
+            report = Check_Report_AWS(metadata=self.metadata(), resource=state_machine)
+
+            if (
+                state_machine.encryption_configuration
+                and state_machine.encryption_configuration.type
+                == EncryptionType.CUSTOMER_MANAGED_KMS_KEY
+            ):
+                report.status = "PASS"
+                report.status_extended = f"Step Functions state machine {state_machine.name} is encrypted at rest with a customer-managed KMS key."
+            else:
+                report.status = "FAIL"
+                report.status_extended = f"Step Functions state machine {state_machine.name} is not encrypted at rest with a customer-managed KMS key."
+
+            findings.append(report)
+
+        return findings

tests/providers/aws/services/stepfunctions/stepfunctions_statemachine_encrypted_with_cmk/stepfunctions_statemachine_encrypted_with_cmk_test.py

@@ -0,0 +1,142 @@
+from datetime import datetime
+from unittest.mock import patch
+
+import pytest
+from moto import mock_aws
+
+from prowler.providers.aws.services.stepfunctions.stepfunctions_service import (
+    EncryptionConfiguration,
+    EncryptionType,
+    StateMachine,
+    StepFunctions,
+)
+from tests.providers.aws.utils import set_mocked_aws_provider
+
+AWS_REGION_EU_WEST_1 = "eu-west-1"
+STATE_MACHINE_ID = "state-machine-12345"
+STATE_MACHINE_ARN = f"arn:aws:states:{AWS_REGION_EU_WEST_1}:123456789012:stateMachine:{STATE_MACHINE_ID}"
+KMS_KEY_ARN = "arn:aws:kms:eu-west-1:123456789012:key/some-key-id"
+
+
+def create_state_machine(name, encryption_configuration):
+    """Create a mock StateMachine instance for use in tests.
+
+    Args:
+        name (str): The display name of the state machine.
+        encryption_configuration (Optional[EncryptionConfiguration]): The encryption
+            configuration to assign to the state machine, or None.
+
+    Returns:
+        StateMachine: A StateMachine instance pre-populated with test constants.
+    """
+    return StateMachine(
+        id=STATE_MACHINE_ID,
+        arn=STATE_MACHINE_ARN,
+        name=name,
+        region=AWS_REGION_EU_WEST_1,
+        encryption_configuration=encryption_configuration,
+        tags=[],
+        status="ACTIVE",
+        definition="{}",
+        role_arn="arn:aws:iam::123456789012:role/step-functions-role",
+        type="STANDARD",
+        creation_date=datetime.now(),
+    )
+
+
+@pytest.mark.parametrize(
+    "state_machines, expected_count, expected_status, expected_status_extended",
+    [
+        # No state machines , no findings
+        ({}, 0, None, None),
+        # AWS-owned key (default) , FAIL
+        (
+            {
+                STATE_MACHINE_ARN: create_state_machine(
+                    "TestStateMachine",
+                    EncryptionConfiguration(
+                        type=EncryptionType.AWS_OWNED_KEY,
+                        kms_key_id=None,
+                        kms_data_key_reuse_period_seconds=None,
+                    ),
+                )
+            },
+            1,
+            "FAIL",
+            "Step Functions state machine TestStateMachine is not encrypted at rest with a customer-managed KMS key.",
+        ),
+        # No encryption configuration (None) , FAIL
+        (
+            {
+                STATE_MACHINE_ARN: create_state_machine(
+                    "TestStateMachine",
+                    None,
+                )
+            },
+            1,
+            "FAIL",
+            "Step Functions state machine TestStateMachine is not encrypted at rest with a customer-managed KMS key.",
+        ),
+        # Customer-managed KMS key , PASS
+        (
+            {
+                STATE_MACHINE_ARN: create_state_machine(
+                    "TestStateMachine",
+                    EncryptionConfiguration(
+                        type=EncryptionType.CUSTOMER_MANAGED_KMS_KEY,
+                        kms_key_id=KMS_KEY_ARN,
+                        kms_data_key_reuse_period_seconds=300,
+                    ),
+                )
+            },
+            1,
+            "PASS",
+            "Step Functions state machine TestStateMachine is encrypted at rest with a customer-managed KMS key.",
+        ),
+    ],
+)
+@mock_aws(config={"stepfunctions": {"execute_state_machine": True}})
+def test_stepfunctions_statemachine_encrypted_with_cmk(
+    state_machines,
+    expected_count,
+    expected_status,
+    expected_status_extended,
+):
+    """Test stepfunctions_statemachine_encrypted_with_cmk check across multiple scenarios.
+
+    Parametrized test cases cover:
+    - No state machines present (empty findings).
+    - State machine using the default AWS-owned key (FAIL).
+    - State machine with no encryption configuration set (FAIL).
+    - State machine using a customer-managed KMS key (PASS).
+
+    Args:
+        state_machines (dict): Mapping of ARN to StateMachine used to mock the service client.
+        expected_count (int): Expected number of findings returned by the check.
+        expected_status (Optional[str]): Expected status of the finding, or None if no findings.
+        expected_status_extended (Optional[str]): Expected status_extended message, or None.
+    """
+    mocked_aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+    stepfunctions_client = StepFunctions(mocked_aws_provider)
+    stepfunctions_client.state_machines = state_machines
+
+    with patch(
+        "prowler.providers.aws.services.stepfunctions.stepfunctions_statemachine_encrypted_with_cmk.stepfunctions_statemachine_encrypted_with_cmk.stepfunctions_client",
+        new=stepfunctions_client,
+    ):
+        from prowler.providers.aws.services.stepfunctions.stepfunctions_statemachine_encrypted_with_cmk.stepfunctions_statemachine_encrypted_with_cmk import (
+            stepfunctions_statemachine_encrypted_with_cmk,
+        )
+
+        check = stepfunctions_statemachine_encrypted_with_cmk()
+        result = check.execute()
+
+        assert len(result) == expected_count
+
+        if expected_count == 1:
+            assert result[0].status == expected_status
+            assert result[0].status_extended == expected_status_extended
+            assert result[0].resource_id == STATE_MACHINE_ID
+            assert result[0].resource_arn == STATE_MACHINE_ARN
+            assert result[0].region == AWS_REGION_EU_WEST_1
+            assert result[0].resource == state_machines[STATE_MACHINE_ARN]