We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: StateMachine yaml Definition is not supported (DefinitionString works)
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

when describing using cloudformation YAML with Definition, LocalStack fails to create StateMachine. Switching to YAML with DefinitionString works as expected. 

Examples taken from here
https://docs.aws.amazon.com/step-functions/latest/dg/development-options.html#development-options-format

Attaching working example 
[localstack_stepfunctions_and_serverless.zip](https://github.com/localstack/localstack/files/7791134/localstack_stepfunctions_and_serverless.z




### Expected Behavior

YAML with Definition is much more preferable in a complex state machine description with bunch of substitutions, references and parameters

### How are you starting LocalStack?

With the `localstack` script

### Steps To Reproduce

#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)

    localstack --debug start -d

#### Client commands (e.g., AWS SDK code snippet, or sequence of "awslocal" commands)

   awslocal s3 mb s3://s4-echo-serverless-deployment-share-local
sls deploy --stage local


### Environment

```markdown
- OS: OSX Monterey
- LocalStack: latest
```


### Anything else?

Error when using YAML with Definition
[error.txt](https://github.com/localstack/localstack/files/7791149/error.txt)

Success when using YAML with DefinitionString
[success.txt](https://github.com/localstack/localstack/files/7791151/success.txt)



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py`
Content:
```
1 # LocalStack Resource Provider Scaffolding v2
2 from __future__ import annotations
3 
4 import re
5 from pathlib import Path
6 from typing import Optional, TypedDict
7 
8 import localstack.services.cloudformation.provider_utils as util
9 from localstack.services.cloudformation.resource_provider import (
10     LOG,
11     OperationStatus,
12     ProgressEvent,
13     ResourceProvider,
14     ResourceRequest,
15 )
16 from localstack.utils.strings import to_str
17 
18 
19 class StepFunctionsStateMachineProperties(TypedDict):
20     RoleArn: Optional[str]
21     Arn: Optional[str]
22     Definition: Optional[dict]
23     DefinitionS3Location: Optional[S3Location]
24     DefinitionString: Optional[str]
25     DefinitionSubstitutions: Optional[dict]
26     LoggingConfiguration: Optional[LoggingConfiguration]
27     Name: Optional[str]
28     StateMachineName: Optional[str]
29     StateMachineRevisionId: Optional[str]
30     StateMachineType: Optional[str]
31     Tags: Optional[list[TagsEntry]]
32     TracingConfiguration: Optional[TracingConfiguration]
33 
34 
35 class CloudWatchLogsLogGroup(TypedDict):
36     LogGroupArn: Optional[str]
37 
38 
39 class LogDestination(TypedDict):
40     CloudWatchLogsLogGroup: Optional[CloudWatchLogsLogGroup]
41 
42 
43 class LoggingConfiguration(TypedDict):
44     Destinations: Optional[list[LogDestination]]
45     IncludeExecutionData: Optional[bool]
46     Level: Optional[str]
47 
48 
49 class TracingConfiguration(TypedDict):
50     Enabled: Optional[bool]
51 
52 
53 class S3Location(TypedDict):
54     Bucket: Optional[str]
55     Key: Optional[str]
56     Version: Optional[str]
57 
58 
59 class TagsEntry(TypedDict):
60     Key: Optional[str]
61     Value: Optional[str]
62 
63 
64 REPEATED_INVOCATION = "repeated_invocation"
65 
66 
67 class StepFunctionsStateMachineProvider(ResourceProvider[StepFunctionsStateMachineProperties]):
68     TYPE = "AWS::StepFunctions::StateMachine"  # Autogenerated. Don't change
69     SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
70 
71     def create(
72         self,
73         request: ResourceRequest[StepFunctionsStateMachineProperties],
74     ) -> ProgressEvent[StepFunctionsStateMachineProperties]:
75         """
76         Create a new resource.
77 
78         Primary identifier fields:
79           - /properties/Arn
80 
81         Required properties:
82           - RoleArn
83 
84         Create-only properties:
85           - /properties/StateMachineName
86           - /properties/StateMachineType
87 
88         Read-only properties:
89           - /properties/Arn
90           - /properties/Name
91           - /properties/StateMachineRevisionId
92 
93         IAM permissions required:
94           - states:CreateStateMachine
95           - iam:PassRole
96           - s3:GetObject
97 
98         """
99         model = request.desired_state
100         step_function = request.aws_client_factory.stepfunctions
101 
102         if not model.get("StateMachineName"):
103             model["StateMachineName"] = util.generate_default_name(
104                 stack_name=request.stack_name, logical_resource_id=request.logical_resource_id
105             )
106 
107         params = {
108             "name": model.get("StateMachineName"),
109             "roleArn": model.get("RoleArn"),
110             "type": model.get("StateMachineType", "STANDARD"),
111         }
112 
113         # get definition
114         s3_client = request.aws_client_factory.s3
115 
116         definition_str = self._get_definition(model, s3_client)
117 
118         params["definition"] = definition_str
119 
120         response = step_function.create_state_machine(**params)
121 
122         model["Arn"] = response["stateMachineArn"]
123         model["Name"] = model["StateMachineName"]
124 
125         return ProgressEvent(
126             status=OperationStatus.SUCCESS,
127             resource_model=model,
128             custom_context=request.custom_context,
129         )
130 
131     def _get_definition(self, model, s3_client):
132         definition_str = model.get("DefinitionString")
133         s3_location = model.get("DefinitionS3Location")
134         if not definition_str and s3_location:
135             # TODO: currently not covered by tests - add a test to mimick the behavior of "sam deploy ..."
136 
137             LOG.debug("Fetching state machine definition from S3: %s", s3_location)
138             result = s3_client.get_object(Bucket=s3_location["Bucket"], Key=s3_location["Key"])
139             definition_str = to_str(result["Body"].read())
140         substitutions = model.get("DefinitionSubstitutions")
141         if substitutions is not None:
142             definition_str = _apply_substitutions(definition_str, substitutions)
143         return definition_str
144 
145     def read(
146         self,
147         request: ResourceRequest[StepFunctionsStateMachineProperties],
148     ) -> ProgressEvent[StepFunctionsStateMachineProperties]:
149         """
150         Fetch resource information
151 
152         IAM permissions required:
153           - states:DescribeStateMachine
154           - states:ListTagsForResource
155         """
156         raise NotImplementedError
157 
158     def delete(
159         self,
160         request: ResourceRequest[StepFunctionsStateMachineProperties],
161     ) -> ProgressEvent[StepFunctionsStateMachineProperties]:
162         """
163         Delete a resource
164 
165         IAM permissions required:
166           - states:DeleteStateMachine
167           - states:DescribeStateMachine
168         """
169         model = request.desired_state
170         step_function = request.aws_client_factory.stepfunctions
171 
172         step_function.delete_state_machine(stateMachineArn=model["Arn"])
173 
174         return ProgressEvent(
175             status=OperationStatus.SUCCESS,
176             resource_model=model,
177             custom_context=request.custom_context,
178         )
179 
180     def update(
181         self,
182         request: ResourceRequest[StepFunctionsStateMachineProperties],
183     ) -> ProgressEvent[StepFunctionsStateMachineProperties]:
184         """
185         Update a resource
186 
187         IAM permissions required:
188           - states:UpdateStateMachine
189           - states:TagResource
190           - states:UntagResource
191           - states:ListTagsForResource
192           - iam:PassRole
193         """
194         model = request.desired_state
195         step_function = request.aws_client_factory.stepfunctions
196 
197         if not model.get("Arn"):
198             model["Arn"] = request.previous_state["Arn"]
199 
200         params = {
201             "stateMachineArn": model["Arn"],
202             "definition": model["DefinitionString"],
203         }
204 
205         step_function.update_state_machine(**params)
206 
207         return ProgressEvent(
208             status=OperationStatus.SUCCESS,
209             resource_model=model,
210             custom_context=request.custom_context,
211         )
212 
213 
214 def _apply_substitutions(definition: str, substitutions: dict[str, str]) -> str:
215     substitution_regex = re.compile("\\${[a-zA-Z0-9_]+}")  # might be a bit too strict in some cases
216     tokens = substitution_regex.findall(definition)
217     result = definition
218     for token in tokens:
219         raw_token = token[2:-1]  # strip ${ and }
220         if raw_token not in substitutions.keys():
221             raise
222         result = result.replace(token, substitutions[raw_token])
223 
224     return result
225
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py b/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py
--- a/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py
+++ b/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py
@@ -1,6 +1,7 @@
 # LocalStack Resource Provider Scaffolding v2
 from __future__ import annotations
 
+import json
 import re
 from pathlib import Path
 from typing import Optional, TypedDict
@@ -129,14 +130,20 @@
         )
 
     def _get_definition(self, model, s3_client):
-        definition_str = model.get("DefinitionString")
-        s3_location = model.get("DefinitionS3Location")
-        if not definition_str and s3_location:
+        if "DefinitionString" in model:
+            definition_str = model.get("DefinitionString")
+        elif "DefinitionS3Location" in model:
             # TODO: currently not covered by tests - add a test to mimick the behavior of "sam deploy ..."
-
+            s3_location = model.get("DefinitionS3Location")
             LOG.debug("Fetching state machine definition from S3: %s", s3_location)
             result = s3_client.get_object(Bucket=s3_location["Bucket"], Key=s3_location["Key"])
             definition_str = to_str(result["Body"].read())
+        elif "Definition" in model:
+            definition = model.get("Definition")
+            definition_str = json.dumps(definition)
+        else:
+            definition_str = None
+
         substitutions = model.get("DefinitionSubstitutions")
         if substitutions is not None:
             definition_str = _apply_substitutions(definition_str, substitutions)

{"golden_diff": "diff --git a/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py b/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py\n--- a/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py\n+++ b/localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py\n@@ -1,6 +1,7 @@\n # LocalStack Resource Provider Scaffolding v2\n from __future__ import annotations\n \n+import json\n import re\n from pathlib import Path\n from typing import Optional, TypedDict\n@@ -129,14 +130,20 @@\n         )\n \n     def _get_definition(self, model, s3_client):\n-        definition_str = model.get(\"DefinitionString\")\n-        s3_location = model.get(\"DefinitionS3Location\")\n-        if not definition_str and s3_location:\n+        if \"DefinitionString\" in model:\n+            definition_str = model.get(\"DefinitionString\")\n+        elif \"DefinitionS3Location\" in model:\n             # TODO: currently not covered by tests - add a test to mimick the behavior of \"sam deploy ...\"\n-\n+            s3_location = model.get(\"DefinitionS3Location\")\n             LOG.debug(\"Fetching state machine definition from S3: %s\", s3_location)\n             result = s3_client.get_object(Bucket=s3_location[\"Bucket\"], Key=s3_location[\"Key\"])\n             definition_str = to_str(result[\"Body\"].read())\n+        elif \"Definition\" in model:\n+            definition = model.get(\"Definition\")\n+            definition_str = json.dumps(definition)\n+        else:\n+            definition_str = None\n+\n         substitutions = model.get(\"DefinitionSubstitutions\")\n         if substitutions is not None:\n             definition_str = _apply_substitutions(definition_str, substitutions)\n", "issue": "bug: StateMachine yaml Definition is not supported (DefinitionString works)\n### Is there an existing issue for this?\n\n- [X] I have searched the existing issues\n\n### Current Behavior\n\nwhen describing using cloudformation YAML with Definition, LocalStack fails to create StateMachine. Switching to YAML with DefinitionString works as expected. \r\n\r\nExamples taken from here\r\nhttps://docs.aws.amazon.com/step-functions/latest/dg/development-options.html#development-options-format\r\n\r\nAttaching working example \r\n[localstack_stepfunctions_and_serverless.zip](https://github.com/localstack/localstack/files/7791134/localstack_stepfunctions_and_serverless.z\r\n\r\n\r\n\n\n### Expected Behavior\n\nYAML with Definition is much more preferable in a complex state machine description with bunch of substitutions, references and parameters\n\n### How are you starting LocalStack?\n\nWith the `localstack` script\n\n### Steps To Reproduce\n\n#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)\r\n\r\n    localstack --debug start -d\r\n\r\n#### Client commands (e.g., AWS SDK code snippet, or sequence of \"awslocal\" commands)\r\n\r\n   awslocal s3 mb s3://s4-echo-serverless-deployment-share-local\r\nsls deploy --stage local\r\n\n\n### Environment\n\n```markdown\n- OS: OSX Monterey\r\n- LocalStack: latest\n```\n\n\n### Anything else?\n\nError when using YAML with Definition\r\n[error.txt](https://github.com/localstack/localstack/files/7791149/error.txt)\r\n\r\nSuccess when using YAML with DefinitionString\r\n[success.txt](https://github.com/localstack/localstack/files/7791151/success.txt)\r\n\r\n\n", "before_files": [{"content": "# LocalStack Resource Provider Scaffolding v2\nfrom __future__ import annotations\n\nimport re\nfrom pathlib import Path\nfrom typing import Optional, TypedDict\n\nimport localstack.services.cloudformation.provider_utils as util\nfrom localstack.services.cloudformation.resource_provider import (\n    LOG,\n    OperationStatus,\n    ProgressEvent,\n    ResourceProvider,\n    ResourceRequest,\n)\nfrom localstack.utils.strings import to_str\n\n\nclass StepFunctionsStateMachineProperties(TypedDict):\n    RoleArn: Optional[str]\n    Arn: Optional[str]\n    Definition: Optional[dict]\n    DefinitionS3Location: Optional[S3Location]\n    DefinitionString: Optional[str]\n    DefinitionSubstitutions: Optional[dict]\n    LoggingConfiguration: Optional[LoggingConfiguration]\n    Name: Optional[str]\n    StateMachineName: Optional[str]\n    StateMachineRevisionId: Optional[str]\n    StateMachineType: Optional[str]\n    Tags: Optional[list[TagsEntry]]\n    TracingConfiguration: Optional[TracingConfiguration]\n\n\nclass CloudWatchLogsLogGroup(TypedDict):\n    LogGroupArn: Optional[str]\n\n\nclass LogDestination(TypedDict):\n    CloudWatchLogsLogGroup: Optional[CloudWatchLogsLogGroup]\n\n\nclass LoggingConfiguration(TypedDict):\n    Destinations: Optional[list[LogDestination]]\n    IncludeExecutionData: Optional[bool]\n    Level: Optional[str]\n\n\nclass TracingConfiguration(TypedDict):\n    Enabled: Optional[bool]\n\n\nclass S3Location(TypedDict):\n    Bucket: Optional[str]\n    Key: Optional[str]\n    Version: Optional[str]\n\n\nclass TagsEntry(TypedDict):\n    Key: Optional[str]\n    Value: Optional[str]\n\n\nREPEATED_INVOCATION = \"repeated_invocation\"\n\n\nclass StepFunctionsStateMachineProvider(ResourceProvider[StepFunctionsStateMachineProperties]):\n    TYPE = \"AWS::StepFunctions::StateMachine\"  # Autogenerated. Don't change\n    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change\n\n    def create(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Create a new resource.\n\n        Primary identifier fields:\n          - /properties/Arn\n\n        Required properties:\n          - RoleArn\n\n        Create-only properties:\n          - /properties/StateMachineName\n          - /properties/StateMachineType\n\n        Read-only properties:\n          - /properties/Arn\n          - /properties/Name\n          - /properties/StateMachineRevisionId\n\n        IAM permissions required:\n          - states:CreateStateMachine\n          - iam:PassRole\n          - s3:GetObject\n\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        if not model.get(\"StateMachineName\"):\n            model[\"StateMachineName\"] = util.generate_default_name(\n                stack_name=request.stack_name, logical_resource_id=request.logical_resource_id\n            )\n\n        params = {\n            \"name\": model.get(\"StateMachineName\"),\n            \"roleArn\": model.get(\"RoleArn\"),\n            \"type\": model.get(\"StateMachineType\", \"STANDARD\"),\n        }\n\n        # get definition\n        s3_client = request.aws_client_factory.s3\n\n        definition_str = self._get_definition(model, s3_client)\n\n        params[\"definition\"] = definition_str\n\n        response = step_function.create_state_machine(**params)\n\n        model[\"Arn\"] = response[\"stateMachineArn\"]\n        model[\"Name\"] = model[\"StateMachineName\"]\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n    def _get_definition(self, model, s3_client):\n        definition_str = model.get(\"DefinitionString\")\n        s3_location = model.get(\"DefinitionS3Location\")\n        if not definition_str and s3_location:\n            # TODO: currently not covered by tests - add a test to mimick the behavior of \"sam deploy ...\"\n\n            LOG.debug(\"Fetching state machine definition from S3: %s\", s3_location)\n            result = s3_client.get_object(Bucket=s3_location[\"Bucket\"], Key=s3_location[\"Key\"])\n            definition_str = to_str(result[\"Body\"].read())\n        substitutions = model.get(\"DefinitionSubstitutions\")\n        if substitutions is not None:\n            definition_str = _apply_substitutions(definition_str, substitutions)\n        return definition_str\n\n    def read(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Fetch resource information\n\n        IAM permissions required:\n          - states:DescribeStateMachine\n          - states:ListTagsForResource\n        \"\"\"\n        raise NotImplementedError\n\n    def delete(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Delete a resource\n\n        IAM permissions required:\n          - states:DeleteStateMachine\n          - states:DescribeStateMachine\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        step_function.delete_state_machine(stateMachineArn=model[\"Arn\"])\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n    def update(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Update a resource\n\n        IAM permissions required:\n          - states:UpdateStateMachine\n          - states:TagResource\n          - states:UntagResource\n          - states:ListTagsForResource\n          - iam:PassRole\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        if not model.get(\"Arn\"):\n            model[\"Arn\"] = request.previous_state[\"Arn\"]\n\n        params = {\n            \"stateMachineArn\": model[\"Arn\"],\n            \"definition\": model[\"DefinitionString\"],\n        }\n\n        step_function.update_state_machine(**params)\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n\ndef _apply_substitutions(definition: str, substitutions: dict[str, str]) -> str:\n    substitution_regex = re.compile(\"\\\\${[a-zA-Z0-9_]+}\")  # might be a bit too strict in some cases\n    tokens = substitution_regex.findall(definition)\n    result = definition\n    for token in tokens:\n        raw_token = token[2:-1]  # strip ${ and }\n        if raw_token not in substitutions.keys():\n            raise\n        result = result.replace(token, substitutions[raw_token])\n\n    return result\n", "path": "localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py"}], "after_files": [{"content": "# LocalStack Resource Provider Scaffolding v2\nfrom __future__ import annotations\n\nimport json\nimport re\nfrom pathlib import Path\nfrom typing import Optional, TypedDict\n\nimport localstack.services.cloudformation.provider_utils as util\nfrom localstack.services.cloudformation.resource_provider import (\n    LOG,\n    OperationStatus,\n    ProgressEvent,\n    ResourceProvider,\n    ResourceRequest,\n)\nfrom localstack.utils.strings import to_str\n\n\nclass StepFunctionsStateMachineProperties(TypedDict):\n    RoleArn: Optional[str]\n    Arn: Optional[str]\n    Definition: Optional[dict]\n    DefinitionS3Location: Optional[S3Location]\n    DefinitionString: Optional[str]\n    DefinitionSubstitutions: Optional[dict]\n    LoggingConfiguration: Optional[LoggingConfiguration]\n    Name: Optional[str]\n    StateMachineName: Optional[str]\n    StateMachineRevisionId: Optional[str]\n    StateMachineType: Optional[str]\n    Tags: Optional[list[TagsEntry]]\n    TracingConfiguration: Optional[TracingConfiguration]\n\n\nclass CloudWatchLogsLogGroup(TypedDict):\n    LogGroupArn: Optional[str]\n\n\nclass LogDestination(TypedDict):\n    CloudWatchLogsLogGroup: Optional[CloudWatchLogsLogGroup]\n\n\nclass LoggingConfiguration(TypedDict):\n    Destinations: Optional[list[LogDestination]]\n    IncludeExecutionData: Optional[bool]\n    Level: Optional[str]\n\n\nclass TracingConfiguration(TypedDict):\n    Enabled: Optional[bool]\n\n\nclass S3Location(TypedDict):\n    Bucket: Optional[str]\n    Key: Optional[str]\n    Version: Optional[str]\n\n\nclass TagsEntry(TypedDict):\n    Key: Optional[str]\n    Value: Optional[str]\n\n\nREPEATED_INVOCATION = \"repeated_invocation\"\n\n\nclass StepFunctionsStateMachineProvider(ResourceProvider[StepFunctionsStateMachineProperties]):\n    TYPE = \"AWS::StepFunctions::StateMachine\"  # Autogenerated. Don't change\n    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change\n\n    def create(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Create a new resource.\n\n        Primary identifier fields:\n          - /properties/Arn\n\n        Required properties:\n          - RoleArn\n\n        Create-only properties:\n          - /properties/StateMachineName\n          - /properties/StateMachineType\n\n        Read-only properties:\n          - /properties/Arn\n          - /properties/Name\n          - /properties/StateMachineRevisionId\n\n        IAM permissions required:\n          - states:CreateStateMachine\n          - iam:PassRole\n          - s3:GetObject\n\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        if not model.get(\"StateMachineName\"):\n            model[\"StateMachineName\"] = util.generate_default_name(\n                stack_name=request.stack_name, logical_resource_id=request.logical_resource_id\n            )\n\n        params = {\n            \"name\": model.get(\"StateMachineName\"),\n            \"roleArn\": model.get(\"RoleArn\"),\n            \"type\": model.get(\"StateMachineType\", \"STANDARD\"),\n        }\n\n        # get definition\n        s3_client = request.aws_client_factory.s3\n\n        definition_str = self._get_definition(model, s3_client)\n\n        params[\"definition\"] = definition_str\n\n        response = step_function.create_state_machine(**params)\n\n        model[\"Arn\"] = response[\"stateMachineArn\"]\n        model[\"Name\"] = model[\"StateMachineName\"]\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n    def _get_definition(self, model, s3_client):\n        if \"DefinitionString\" in model:\n            definition_str = model.get(\"DefinitionString\")\n        elif \"DefinitionS3Location\" in model:\n            # TODO: currently not covered by tests - add a test to mimick the behavior of \"sam deploy ...\"\n            s3_location = model.get(\"DefinitionS3Location\")\n            LOG.debug(\"Fetching state machine definition from S3: %s\", s3_location)\n            result = s3_client.get_object(Bucket=s3_location[\"Bucket\"], Key=s3_location[\"Key\"])\n            definition_str = to_str(result[\"Body\"].read())\n        elif \"Definition\" in model:\n            definition = model.get(\"Definition\")\n            definition_str = json.dumps(definition)\n        else:\n            definition_str = None\n\n        substitutions = model.get(\"DefinitionSubstitutions\")\n        if substitutions is not None:\n            definition_str = _apply_substitutions(definition_str, substitutions)\n        return definition_str\n\n    def read(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Fetch resource information\n\n        IAM permissions required:\n          - states:DescribeStateMachine\n          - states:ListTagsForResource\n        \"\"\"\n        raise NotImplementedError\n\n    def delete(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Delete a resource\n\n        IAM permissions required:\n          - states:DeleteStateMachine\n          - states:DescribeStateMachine\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        step_function.delete_state_machine(stateMachineArn=model[\"Arn\"])\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n    def update(\n        self,\n        request: ResourceRequest[StepFunctionsStateMachineProperties],\n    ) -> ProgressEvent[StepFunctionsStateMachineProperties]:\n        \"\"\"\n        Update a resource\n\n        IAM permissions required:\n          - states:UpdateStateMachine\n          - states:TagResource\n          - states:UntagResource\n          - states:ListTagsForResource\n          - iam:PassRole\n        \"\"\"\n        model = request.desired_state\n        step_function = request.aws_client_factory.stepfunctions\n\n        if not model.get(\"Arn\"):\n            model[\"Arn\"] = request.previous_state[\"Arn\"]\n\n        params = {\n            \"stateMachineArn\": model[\"Arn\"],\n            \"definition\": model[\"DefinitionString\"],\n        }\n\n        step_function.update_state_machine(**params)\n\n        return ProgressEvent(\n            status=OperationStatus.SUCCESS,\n            resource_model=model,\n            custom_context=request.custom_context,\n        )\n\n\ndef _apply_substitutions(definition: str, substitutions: dict[str, str]) -> str:\n    substitution_regex = re.compile(\"\\\\${[a-zA-Z0-9_]+}\")  # might be a bit too strict in some cases\n    tokens = substitution_regex.findall(definition)\n    result = definition\n    for token in tokens:\n        raw_token = token[2:-1]  # strip ${ and }\n        if raw_token not in substitutions.keys():\n            raise\n        result = result.replace(token, substitutions[raw_token])\n\n    return result\n", "path": "localstack/services/stepfunctions/resource_providers/aws_stepfunctions_statemachine.py"}]}