Repository: localstack/localstack
Pull request: StepFunctions: Multi-accounts compatibility
## Motivation

This PR introduces multi-account support to the legacy StepFunctions provider. It also makes multi-account friendly fixes to the work-in-progress StepFunctions v2 provider.

cc: @MEPalma 

## Implementation

The legacy provider uses [StepFunctions Local](https://docs.aws.amazon.com/step-functions/latest/dg/sfn-local-jar.html) which itself does not support namespacing. So we take an approach similar to Kinesis. Each account+region combo will run its own instance of SFLocal. Of course this is not optimal but should fulfill simple usecases. 

However, SFLocal does not respect the `--account-id` argument and overwrites ARNs like Lambda Functions during invocations. So I guess the way forward is to only support true multi-accounts in the new SF provider and keep this limitation in the legacy one. The multi-region parameter is maintained.

Criteria for merging: all stepfunctions tests (legacy as well as the **ci/circleci: itest-sfn-v2-provider** job)  pass when `AWS_ACCESS_KEY_ID` is set to a non-default account ID or when `AWS_REGION_NAME` is set to non-`us-east-1` region.

**79ec46e61e52940f2be5d2377b4933a77e5c2b61 must be reverted before merge**

## Related

Closes https://github.com/localstack/localstack/issues/8205

Problem statement:
bug: Stepfunctions creates resources in the fallback account
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

Stepfunctions v2 provider creates resources in the fallback account `000000000000` and not the account made by the original requester.

### Expected Behavior

_No response_

### How are you starting LocalStack?

With a docker-compose file

### Steps To Reproduce

#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)

Start with `PROVIDER_OVERRIDE_STEPFUNCTIONS=v2`

#### Client commands (e.g., AWS SDK code snippet, or sequence of "awslocal" commands)

```sh
$ AWS_ACCESS_KEY_ID=111111111111 awslocal iam create-role \
   --role-name example-role \
   --assume-role-policy-document '{"Version": "2012-10-17", "Statement": {"Action": "sts:AssumeRole", "Effect": "Allow", "Principal": {"Service": "states.amazonaws.com"}}}'

$ AWS_ACCESS_KEY_ID=111111111111 awslocal iam create-policy \
   --policy-name example-policy \
   --policy-document '{"Version": "2012-10-17", "Statement": {"Action": "sns:createTopic", "Effect": "Allow", "Resource": "*"}}'

$ AWS_ACCESS_KEY_ID=111111111111 awslocal iam attach-role-policy \
   --role-name example-role \
   --policy-arn arn:aws:iam::111111111111:policy/example-policy

$ AWS_ACCESS_KEY_ID=111111111111 awslocal stepfunctions create-state-machine \
   --name example \
   --definition '{"StartAt": "CreateTopicTask", "States": {"CreateTopicTask": {"End": true, "Type": "Task", "Resource": "arn:aws:states:::aws-sdk:sns:createTopic", "Parameters": {"Name.$": "$.Name"}}}}' \
   --role-arn arn:aws:iam::111111111111:role/example-role

$ AWS_ACCESS_KEY_ID=111111111111 awslocal stepfunctions start-execution \
   --state-machine-arn arn:aws:states:us-east-1:111111111111:stateMachine:example \
   --input '{"Name": "example-topic"}'

# When execution finishes, it would be expected that the topic would be created in the the original account

$ AWS_ACCESS_KEY_ID=111111111111 awslocal sns list-topics                                                                                                                                  
{
    "Topics": []
}

# but in fact it is created in the fallback account

$ AWS_ACCESS_KEY_ID=000000000000 awslocal sns list-topics
{
    "Topics": [
        {
            "TopicArn": "arn:aws:sns:us-east-1:000000000000:example-topic"
        }
    ]
}
```

### Environment

```markdown
- OS: Ubuntu 22.04
- LocalStack: latest
```


### Anything else?

_No response_

Hints from issue discussion:

Commit message: Maintain backward compatibility

Patch to review:
diff --git a/localstack/constants.py b/localstack/constants.py
index 8abca4bf08009..547da354d4cf7 100644
--- a/localstack/constants.py
+++ b/localstack/constants.py
@@ -153,10 +153,10 @@
 # Credentials used in the test suite
 # These can be overridden if the tests are being run against AWS
 # If a structured access key ID is used, it must correspond to the account ID
-TEST_AWS_ACCOUNT_ID = os.getenv("TEST_AWS_ACCOUNT_ID") or DEFAULT_AWS_ACCOUNT_ID
-TEST_AWS_ACCESS_KEY_ID = os.getenv("TEST_AWS_ACCESS_KEY_ID") or "test"
+TEST_AWS_ACCOUNT_ID = os.getenv("TEST_AWS_ACCOUNT_ID") or "000000000001"
+TEST_AWS_ACCESS_KEY_ID = os.getenv("TEST_AWS_ACCESS_KEY_ID") or "000000000001"
 TEST_AWS_SECRET_ACCESS_KEY = os.getenv("TEST_AWS_SECRET_ACCESS_KEY") or "test"
-TEST_AWS_REGION_NAME = os.getenv("TEST_AWS_REGION") or "us-east-1"
+TEST_AWS_REGION_NAME = os.getenv("TEST_AWS_REGION") or "us-west-1"
 
 # Additional credentials used in the test suite (when running cross-account tests)
 SECONDARY_TEST_AWS_ACCOUNT_ID = os.getenv("SECONDARY_TEST_AWS_ACCOUNT_ID") or "000000000002"
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/lambda_eval_utils.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/lambda_eval_utils.py
index 1dd03cb568da4..6439b1783cbd3 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/lambda_eval_utils.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/lambda_eval_utils.py
@@ -2,12 +2,10 @@
 from json import JSONDecodeError
 from typing import Any, Final, Optional
 
-from botocore.config import Config
-
 from localstack.aws.api.lambda_ import InvocationResponse
-from localstack.aws.connect import connect_externally_to
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.collections import select_from_typed_dict
 from localstack.utils.run import to_str
 from localstack.utils.strings import to_bytes
@@ -23,7 +21,7 @@ def __init__(self, function_error: Optional[str], payload: str):
 
 
 def exec_lambda_function(env: Environment, parameters: dict) -> None:
-    lambda_client = connect_externally_to(config=Config(parameter_validation=False)).lambda_
+    lambda_client = get_boto_client(env, "lambda")
     invocation_resp: InvocationResponse = lambda_client.invoke(**parameters)
 
     func_error: Optional[str] = invocation_resp.get("FunctionError")
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py
index f790e8cd03e7c..c8cd46b96ce1e 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py
@@ -5,7 +5,6 @@
 from typing import Final, Optional
 
 from localstack.services.stepfunctions.asl.component.component import Component
-from localstack.utils.aws import aws_stack
 
 
 class ResourceCondition(str):
@@ -85,10 +84,12 @@ def __init__(self, resource_arn: ResourceARN):
         self.account = resource_arn.account
 
     @staticmethod
-    def from_resource_arn(arn: str) -> Resource:
+    def from_resource_arn(account_id: str, region_name: str, arn: str) -> Resource:
         resource_arn = ResourceARN.from_arn(arn)
+        if not resource_arn.account:
+            resource_arn.account = account_id
         if not resource_arn.region:
-            resource_arn.region = aws_stack.get_region()
+            resource_arn.region = region_name
         match resource_arn.service, resource_arn.task_type:
             case "lambda", "function":
                 return LambdaResource(resource_arn=resource_arn)
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py
index 148d8aba7bceb..9c009f5a95fcb 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_aws_sdk.py
@@ -1,8 +1,6 @@
-from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
-from localstack.aws.connect import connect_externally_to
 from localstack.aws.protocol.service_router import get_service_catalog
 from localstack.services.stepfunctions.asl.component.common.error_name.failure_event import (
     FailureEvent,
@@ -19,6 +17,7 @@
 from localstack.services.stepfunctions.asl.component.state.state_props import StateProps
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.common import camel_to_snake_case
 
 
@@ -105,9 +104,7 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
         return super()._from_error(env=env, ex=ex)
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
-        api_client = connect_externally_to.get_client(
-            service_name=self._normalised_api_name, config=Config(parameter_validation=False)
-        )
+        api_client = get_boto_client(env, self._normalised_api_name)
         response = getattr(api_client, self._normalised_api_action)(**parameters) or dict()
         if response:
             response.pop("ResponseMetadata", None)
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_dynamodb.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_dynamodb.py
index 48d8273b938b9..a2f0309a299bd 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_dynamodb.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_dynamodb.py
@@ -1,10 +1,8 @@
 from typing import Final, Optional
 
-from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
-from localstack.aws.connect import connect_to
 from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
     CustomErrorName,
 )
@@ -16,6 +14,7 @@
 )
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.strings import camel_to_snake_case
 
 
@@ -124,8 +123,7 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
         api_action = camel_to_snake_case(self.resource.api_action)
-
-        dynamodb_client = connect_to(config=Config(parameter_validation=False)).dynamodb
+        dynamodb_client = get_boto_client(env, "dynamodb")
         response = getattr(dynamodb_client, api_action)(**parameters)
         response.pop("ResponseMetadata", None)
         env.stack.append(response)
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sfn.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sfn.py
index bc93a9f31ed71..a139a1ac7feaf 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sfn.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sfn.py
@@ -1,7 +1,6 @@
 import json
 from typing import Any, Final, Optional
 
-from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import (
@@ -10,7 +9,6 @@
     HistoryEventType,
     TaskFailedEventDetails,
 )
-from localstack.aws.connect import connect_externally_to
 from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
     CustomErrorName,
 )
@@ -30,6 +28,7 @@
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
 from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.collections import select_from_typed_dict
 from localstack.utils.strings import camel_to_snake_case
 
@@ -100,10 +99,6 @@ def _apply_normalisation(lookup_keys, dictionary):
         lower_to_normalise_key = _build_lower_to_key_dict(keys)
         _apply_normalisation(lower_to_normalise_key, response)
 
-    @staticmethod
-    def _get_sfn_client():
-        return connect_externally_to(config=Config(parameter_validation=False)).stepfunctions
-
     def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
         if isinstance(ex, ClientError):
             error_code = ex.response["Error"]["Code"]
@@ -166,15 +161,14 @@ def _replace_with_json_if_str(key: str) -> None:
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
         api_action = camel_to_snake_case(self.resource.api_action)
-        sfn_client = self._get_sfn_client()
+        sfn_client = get_boto_client(env, "stepfunctions")
         response = getattr(sfn_client, api_action)(**parameters)
         response.pop("ResponseMetadata", None)
         self._normalise_botocore_response(self.resource.api_action, response)
         env.stack.append(response)
 
     def _sync_to_start_machine(self, env: Environment, sync2_response: bool) -> None:
-        sfn_client = self._get_sfn_client()
-
+        sfn_client = get_boto_client(env, "stepfunctions")
         submission_output: dict = env.stack.pop()
         execution_arn: str = submission_output["ExecutionArn"]
 
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sns.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sns.py
index 3f5b79741fe08..3d8d59f0c4aa5 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sns.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sns.py
@@ -1,10 +1,8 @@
 from typing import Final, Optional
 
-from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
-from localstack.aws.connect import connect_externally_to
 from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
     CustomErrorName,
 )
@@ -16,6 +14,7 @@
 )
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.strings import camel_to_snake_case
 
 
@@ -72,7 +71,7 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
         api_action = camel_to_snake_case(self.resource.api_action)
-        sns_client = connect_externally_to(config=Config(parameter_validation=False)).sns
+        sns_client = get_boto_client(env, "sns")
         response = getattr(sns_client, api_action)(**parameters)
         response.pop("ResponseMetadata", None)
         env.stack.append(response)
diff --git a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sqs.py b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sqs.py
index 9c207afc24b90..ae9dd10860c39 100644
--- a/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sqs.py
+++ b/localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_sqs.py
@@ -1,10 +1,8 @@
 from typing import Final, Optional
 
-from botocore.config import Config
 from botocore.exceptions import ClientError
 
 from localstack.aws.api.stepfunctions import HistoryEventType, TaskFailedEventDetails
-from localstack.aws.connect import connect_externally_to
 from localstack.services.stepfunctions.asl.component.common.error_name.custom_error_name import (
     CustomErrorName,
 )
@@ -17,6 +15,7 @@
 from localstack.services.stepfunctions.asl.eval.environment import Environment
 from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
 from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
+from localstack.services.stepfunctions.backend.utils import get_boto_client
 from localstack.utils.strings import camel_to_snake_case
 
 
@@ -65,7 +64,7 @@ def _eval_service_task(self, env: Environment, parameters: dict) -> None:
                 parameters["MessageBody"] = to_json_str(message_body)
 
         api_action = camel_to_snake_case(self.resource.api_action)
-        sqs_client = connect_externally_to(config=Config(parameter_validation=False)).sqs
+        sqs_client = get_boto_client(env, "sqs")
         response = getattr(sqs_client, api_action)(**parameters)
         response.pop("ResponseMetadata", None)
         env.stack.append(response)
diff --git a/localstack/services/stepfunctions/asl/eval/environment.py b/localstack/services/stepfunctions/asl/eval/environment.py
index 9bcc001c0c8d5..fe4915223cddf 100644
--- a/localstack/services/stepfunctions/asl/eval/environment.py
+++ b/localstack/services/stepfunctions/asl/eval/environment.py
@@ -26,8 +26,14 @@
 
 
 class Environment:
-    def __init__(self, context_object_init: ContextObjectInitData):
+    def __init__(
+        self, account_id: str, region_name: str, context_object_init: ContextObjectInitData
+    ):
         super(Environment, self).__init__()
+
+        self.account_id = account_id
+        self.region_name = region_name
+
         self._state_mutex = threading.RLock()
         self._program_state: Optional[ProgramState] = None
         self.program_state_event = threading.Event()
@@ -54,7 +60,11 @@ def as_frame_of(cls, env: Environment):
             Execution=env.context_object_manager.context_object["Execution"],
             StateMachine=env.context_object_manager.context_object["StateMachine"],
         )
-        frame = cls(context_object_init=context_object_init)
+        frame = cls(
+            account_id=env.account_id,
+            region_name=env.region_name,
+            context_object_init=context_object_init,
+        )
         frame._is_frame = True
         frame.event_history = env.event_history
         frame.callback_pool_manager = env.callback_pool_manager
diff --git a/localstack/services/stepfunctions/asl/parse/asl_parser.py b/localstack/services/stepfunctions/asl/parse/asl_parser.py
index 260c326523c59..0d21557ef5f99 100644
--- a/localstack/services/stepfunctions/asl/parse/asl_parser.py
+++ b/localstack/services/stepfunctions/asl/parse/asl_parser.py
@@ -11,13 +11,13 @@
 
 class AmazonStateLanguageParser(abc.ABC):
     @staticmethod
-    def parse(src: str) -> Program:
+    def parse(account_id: str, region_name: str, src: str) -> Program:
         input_stream = InputStream(src)
         lexer = ASLLexer(input_stream)
         stream = CommonTokenStream(lexer)
         parser = ASLParser(stream)
         parser._errHandler = antlr4.BailErrorStrategy()
         tree = parser.program_decl()
-        preprocessor = Preprocessor()
+        preprocessor = Preprocessor(account_id, region_name)
         program = preprocessor.visit(tree)
         return program
diff --git a/localstack/services/stepfunctions/asl/parse/preprocessor.py b/localstack/services/stepfunctions/asl/parse/preprocessor.py
index 909c9a144ffc5..53db7856a94eb 100644
--- a/localstack/services/stepfunctions/asl/parse/preprocessor.py
+++ b/localstack/services/stepfunctions/asl/parse/preprocessor.py
@@ -191,6 +191,10 @@
 
 
 class Preprocessor(ASLParserVisitor):
+    def __init__(self, account_id: str, region_name: str):
+        self.account_id = account_id
+        self.region_name = region_name
+
     @staticmethod
     def _inner_string_of(parse_tree: ParseTree) -> Optional[str]:
         if Antlr4Utils.is_terminal(parse_tree, ASLLexer.NULL):
@@ -231,7 +235,7 @@ def visitState_type(self, ctx: ASLParser.State_typeContext) -> StateType:
 
     def visitResource_decl(self, ctx: ASLParser.Resource_declContext) -> Resource:
         inner_str = self._inner_string_of(parse_tree=ctx.keyword_or_string())
-        return Resource.from_resource_arn(inner_str)
+        return Resource.from_resource_arn(self.account_id, self.region_name, inner_str)
 
     def visitEnd_decl(self, ctx: ASLParser.End_declContext) -> End:
         bool_child: ParseTree = ctx.children[-1]
diff --git a/localstack/services/stepfunctions/backend/execution.py b/localstack/services/stepfunctions/backend/execution.py
index 338915d3f60bc..f667a94eaece2 100644
--- a/localstack/services/stepfunctions/backend/execution.py
+++ b/localstack/services/stepfunctions/backend/execution.py
@@ -80,6 +80,10 @@ def terminated(self) -> None:
     name: Final[str]
     role_arn: Final[Arn]
     exec_arn: Final[Arn]
+
+    account_id: str
+    region_name: str
+
     state_machine: Final[StateMachineInstance]
     start_date: Final[Timestamp]
     input_data: Final[Optional[dict]]
@@ -102,6 +106,8 @@ def __init__(
         name: str,
         role_arn: Arn,
         exec_arn: Arn,
+        account_id: str,
+        region_name: str,
         state_machine: StateMachineInstance,
         start_date: Timestamp,
         input_data: Optional[dict] = None,
@@ -110,6 +116,8 @@ def __init__(
         self.name = name
         self.role_arn = role_arn
         self.exec_arn = exec_arn
+        self.account_id = account_id
+        self.region_name = region_name
         self.state_machine = state_machine
         self.start_date = start_date
         self.input_data = input_data
@@ -122,7 +130,9 @@ def __init__(
         self.exec_worker = None
         self.error = None
         self.cause = None
-        self._events_client = connect_to().events
+        self._events_client = connect_to(
+            aws_access_key_id=self.account_id, region_name=self.region_name
+        ).events
 
     def to_start_output(self) -> StartExecutionOutput:
         return StartExecutionOutput(executionArn=self.exec_arn, startDate=self.start_date)
@@ -200,6 +210,8 @@ def start(self) -> None:
             raise InvalidName()  # TODO.
 
         self.exec_worker = ExecutionWorker(
+            account_id=self.account_id,
+            region_name=self.region_name,
             role_arn=self.role_arn,
             definition=self.state_machine.definition,
             input_data=self.input_data,
diff --git a/localstack/services/stepfunctions/backend/execution_worker.py b/localstack/services/stepfunctions/backend/execution_worker.py
index 7e326a0dbab5b..0569aa6132a7d 100644
--- a/localstack/services/stepfunctions/backend/execution_worker.py
+++ b/localstack/services/stepfunctions/backend/execution_worker.py
@@ -31,12 +31,16 @@ class ExecutionWorker:
 
     def __init__(
         self,
+        account_id: str,
+        region_name: str,
         role_arn: Arn,
         definition: Definition,
         input_data: Optional[dict],
         context_object_init: ContextObjectInitData,
         exec_comm: ExecutionWorkerComm,
     ):
+        self.account_id = account_id
+        self.region_name = region_name
         self.role_arn = role_arn
         self.definition = definition
         self.input_data = input_data
@@ -45,8 +49,14 @@ def __init__(
         self.exec_comm = exec_comm
 
     def _execution_logic(self):
-        program: Program = AmazonStateLanguageParser.parse(self.definition)
-        self.env = Environment(context_object_init=self._context_object_init)
+        program: Program = AmazonStateLanguageParser.parse(
+            self.account_id, self.region_name, self.definition
+        )
+        self.env = Environment(
+            account_id=self.account_id,
+            region_name=self.region_name,
+            context_object_init=self._context_object_init,
+        )
         self.env.inp = copy.deepcopy(
             self.input_data
         )  # The program will mutate the input_data, which is otherwise constant in regard to the execution value.
diff --git a/localstack/services/stepfunctions/backend/utils.py b/localstack/services/stepfunctions/backend/utils.py
new file mode 100644
index 0000000000000..d89817f8c71a6
--- /dev/null
+++ b/localstack/services/stepfunctions/backend/utils.py
@@ -0,0 +1,13 @@
+from botocore.config import Config
+
+from localstack.aws.connect import connect_to
+from localstack.services.stepfunctions.asl.eval.environment import Environment
+
+
+def get_boto_client(env: Environment, service: str):
+    return connect_to.get_client(
+        aws_access_key_id=env.account_id,
+        region_name=env.region_name,
+        service_name=service,
+        config=Config(parameter_validation=False),
+    )
diff --git a/localstack/services/stepfunctions/provider.py b/localstack/services/stepfunctions/provider.py
index 059acb5b9fe4c..260143cdddc86 100644
--- a/localstack/services/stepfunctions/provider.py
+++ b/localstack/services/stepfunctions/provider.py
@@ -3,6 +3,7 @@
 import threading
 
 from localstack import config
+from localstack.aws.accounts import get_aws_account_id
 from localstack.aws.api import RequestContext, handler
 from localstack.aws.api.stepfunctions import (
     CreateStateMachineInput,
@@ -16,12 +17,9 @@
 from localstack.aws.forwarder import get_request_forwarder_http
 from localstack.constants import LOCALHOST
 from localstack.services.plugins import ServiceLifecycleHook
-from localstack.services.stepfunctions.stepfunctions_starter import (
-    start_stepfunctions,
-    stop_stepfunctions,
-    wait_for_stepfunctions,
-)
+from localstack.services.stepfunctions.stepfunctions_starter import StepFunctionsServerManager
 from localstack.state import AssetDirectory, StateVisitor
+from localstack.utils.aws import aws_stack
 
 # lock to avoid concurrency issues when creating state machines in parallel (required for StepFunctions-Local)
 CREATION_LOCK = threading.RLock()
@@ -30,33 +28,29 @@
 
 
 class StepFunctionsProvider(StepfunctionsApi, ServiceLifecycleHook):
+    server_manager = StepFunctionsServerManager()
+
     def __init__(self):
         self.forward_request = get_request_forwarder_http(self.get_forward_url)
 
     def get_forward_url(self) -> str:
         """Return the URL of the backend StepFunctions server to forward requests to"""
-        return f"http://{LOCALHOST}:{config.LOCAL_PORT_STEPFUNCTIONS}"
+        account_id = get_aws_account_id()
+        region_name = aws_stack.get_region()
+        server = self.server_manager.get_server_for_account_region(account_id, region_name)
+        return f"http://{LOCALHOST}:{server.port}"
 
     def accept_state_visitor(self, visitor: StateVisitor):
         visitor.visit(AssetDirectory(self.service, os.path.join(config.dirs.data, self.service)))
 
-    def on_before_start(self):
-        start_stepfunctions()
-        wait_for_stepfunctions()
-
-    def on_before_state_reset(self):
-        stop_stepfunctions()
-
     def on_before_state_load(self):
-        stop_stepfunctions()
+        self.server_manager.shutdown_all()
 
-    def on_after_state_reset(self):
-        start_stepfunctions()
-        wait_for_stepfunctions()
+    def on_before_state_reset(self):
+        self.server_manager.shutdown_all()
 
-    def on_after_state_load(self):
-        start_stepfunctions()
-        wait_for_stepfunctions()
+    def on_before_stop(self):
+        self.server_manager.shutdown_all()
 
     def create_state_machine(
         self, context: RequestContext, request: CreateStateMachineInput
diff --git a/localstack/services/stepfunctions/provider_v2.py b/localstack/services/stepfunctions/provider_v2.py
index 0328f31b02bf8..e649db89930cd 100644
--- a/localstack/services/stepfunctions/provider_v2.py
+++ b/localstack/services/stepfunctions/provider_v2.py
@@ -148,11 +148,11 @@ def _revision_by_name(
         return None
 
     @staticmethod
-    def _validate_definition(definition: str):
+    def _validate_definition(account_id: str, region_name: str, definition: str):
         # Validate
         # TODO: pass through static analyser.
         try:
-            AmazonStateLanguageParser.parse(definition)
+            AmazonStateLanguageParser.parse(account_id, region_name, definition)
         except Exception as ex:
             # TODO: add message from static analyser, this just helps the user debug issues in the derivation.
             raise InvalidDefinition(f"Error '{str(ex)}' in definition '{definition}'.")
@@ -182,7 +182,11 @@ def create_state_machine(
             )
 
         state_machine_definition: str = request["definition"]
-        StepFunctionsProvider._validate_definition(definition=state_machine_definition)
+        StepFunctionsProvider._validate_definition(
+            account_id=context.account_id,
+            region_name=context.region,
+            definition=state_machine_definition,
+        )
 
         name: Optional[Name] = request["name"]
         arn = aws_stack_state_machine_arn(
@@ -351,6 +355,8 @@ def start_execution(
             name=exec_name,
             role_arn=state_machine_clone.role_arn,
             exec_arn=exec_arn,
+            account_id=context.account_id,
+            region_name=context.region,
             state_machine=state_machine_clone,
             start_date=datetime.datetime.now(),
             input_data=input_data,
@@ -496,7 +502,9 @@ def update_state_machine(
             )
 
         if definition is not None:
-            self._validate_definition(definition=definition)
+            self._validate_definition(
+                account_id=context.account_id, region_name=context.region, definition=definition
+            )
 
         revision_id = state_machine.create_revision(definition=definition, role_arn=role_arn)
 
diff --git a/localstack/services/stepfunctions/stepfunctions_starter.py b/localstack/services/stepfunctions/stepfunctions_starter.py
index 456f3d55d72d1..c9f064be76ab8 100644
--- a/localstack/services/stepfunctions/stepfunctions_starter.py
+++ b/localstack/services/stepfunctions/stepfunctions_starter.py
@@ -1,125 +1,151 @@
 import logging
-import subprocess
+import threading
+from typing import Any, Dict
 
 from localstack import config
-from localstack.aws.accounts import get_aws_account_id
-from localstack.aws.connect import connect_to
-from localstack.services.infra import do_run, log_startup_message
 from localstack.services.stepfunctions.packages import stepfunctions_local_package
 from localstack.utils.aws import aws_stack
-from localstack.utils.common import wait_for_port_open
-from localstack.utils.net import wait_for_port_closed
-from localstack.utils.run import ShellCommandThread, wait_for_process_to_be_killed
-from localstack.utils.sync import retry
+from localstack.utils.net import get_free_tcp_port, port_can_be_bound
+from localstack.utils.run import ShellCommandThread
+from localstack.utils.serving import Server
+from localstack.utils.threads import TMP_THREADS, FuncThread
 
 LOG = logging.getLogger(__name__)
 
 # max heap size allocated for the Java process
 MAX_HEAP_SIZE = "256m"
 
-# todo: will be replaced with plugin mechanism
-PROCESS_THREAD: ShellCommandThread | subprocess.Popen | None = None
-
-
-# TODO: pass env more explicitly
-def get_command(backend_port):
-    install_dir_stepfunctions = stepfunctions_local_package.get_installed_dir()
-    cmd = (
-        "cd %s; PORT=%s java "
-        "-javaagent:aspectjweaver-1.9.7.jar "
-        "-Dorg.aspectj.weaver.loadtime.configuration=META-INF/aop.xml "
-        "-Dcom.amazonaws.sdk.disableCertChecking -Xmx%s "
-        "-jar StepFunctionsLocal.jar --aws-account %s"
-    ) % (
-        install_dir_stepfunctions,
-        backend_port,
-        MAX_HEAP_SIZE,
-        get_aws_account_id(),
-    )
-    if config.STEPFUNCTIONS_LAMBDA_ENDPOINT.lower() != "default":
-        lambda_endpoint = config.STEPFUNCTIONS_LAMBDA_ENDPOINT or aws_stack.get_local_service_url(
-            "lambda"
+
+class StepFunctionsServer(Server):
+    def __init__(
+        self, port: int, account_id: str, region_name: str, host: str = "localhost"
+    ) -> None:
+        self.account_id = account_id
+        self.region_name = region_name
+        super().__init__(port, host)
+
+    def do_start_thread(self) -> FuncThread:
+        cmd = self.generate_shell_command()
+        env_vars = self.generate_env_vars()
+        cwd = stepfunctions_local_package.get_installed_dir()
+        LOG.debug("Starting StepFunctions process %s with env vars %s", cmd, env_vars)
+        t = ShellCommandThread(
+            cmd,
+            strip_color=True,
+            env_vars=env_vars,
+            log_listener=self._log_listener,
+            name="stepfunctions",
+            cwd=cwd,
         )
-        cmd += f" --lambda-endpoint {lambda_endpoint}"
-    # add service endpoint flags
-    services = [
-        "athena",
-        "batch",
-        "dynamodb",
-        "ecs",
-        "eks",
-        "events",
-        "glue",
-        "sagemaker",
-        "sns",
-        "sqs",
-        "stepfunctions",
-    ]
-    for service in services:
-        flag = f"--{service}-endpoint"
-        if service == "stepfunctions":
-            flag = "--step-functions-endpoint"
-        elif service == "events":
-            flag = "--eventbridge-endpoint"
-        elif service in ["athena", "eks"]:
-            flag = f"--step-functions-{service}"
-        endpoint = aws_stack.get_local_service_url(service)
-        cmd += f" {flag} {endpoint}"
-
-    return cmd
-
-
-def start_stepfunctions(asynchronous: bool = True, persistence_path: str | None = None):
-    # TODO: introduce Server abstraction for StepFunctions process
-    global PROCESS_THREAD
-    backend_port = config.LOCAL_PORT_STEPFUNCTIONS
-    stepfunctions_local_package.install()
-    cmd = get_command(backend_port)
-    log_startup_message("StepFunctions")
-    # TODO: change ports in stepfunctions.jar, then update here
-    PROCESS_THREAD = do_run(
-        cmd,
-        asynchronous,
-        strip_color=True,
-        env_vars={
+        TMP_THREADS.append(t)
+        t.start()
+        return t
+
+    def generate_env_vars(self) -> Dict[str, Any]:
+        return {
             "EDGE_PORT": config.EDGE_PORT_HTTP or config.EDGE_PORT,
             "EDGE_PORT_HTTP": config.EDGE_PORT_HTTP or config.EDGE_PORT,
-            "DATA_DIR": persistence_path or config.dirs.data,
-        },
-    )
-    return PROCESS_THREAD
-
-
-def wait_for_stepfunctions():
-    retry(check_stepfunctions, sleep=0.5, retries=15)
-
-
-def stop_stepfunctions():
-    if not PROCESS_THREAD or not PROCESS_THREAD.process:
-        return
-    LOG.debug("Restarting StepFunctions process ...")
-
-    pid = PROCESS_THREAD.process.pid
-    PROCESS_THREAD.stop()
-    wait_for_port_closed(config.LOCAL_PORT_STEPFUNCTIONS, sleep_time=0.5, retries=15)
-    try:
-        # TODO: currently failing in CI (potentially due to a defunct process) - need to investigate!
-        wait_for_process_to_be_killed(pid, sleep=0.3, retries=10)
-    except Exception as e:
-        LOG.warning("StepFunctions process not properly terminated: %s", e)
-
-
-def check_stepfunctions(expect_shutdown: bool = False, print_error: bool = False) -> None:
-    out = None
-    try:
-        wait_for_port_open(config.LOCAL_PORT_STEPFUNCTIONS, sleep_time=2)
-        endpoint_url = f"http://127.0.0.1:{config.LOCAL_PORT_STEPFUNCTIONS}"
-        out = connect_to(endpoint_url=endpoint_url).stepfunctions.list_state_machines()
-    except Exception:
-        if print_error:
-            LOG.exception("StepFunctions health check failed")
-
-    if expect_shutdown:
-        assert out is None
-    else:
-        assert out and isinstance(out.get("stateMachines"), list)
+            "DATA_DIR": config.dirs.data,
+            "PORT": self._port,
+        }
+
+    def generate_shell_command(self) -> str:
+        cmd = (
+            f"java "
+            f"-javaagent:aspectjweaver-1.9.7.jar "
+            f"-Dorg.aspectj.weaver.loadtime.configuration=META-INF/aop.xml "
+            f"-Dcom.amazonaws.sdk.disableCertChecking "
+            f"-Xmx{MAX_HEAP_SIZE} "
+            f"-jar StepFunctionsLocal.jar "
+            f"--aws-account {self.account_id} "
+            f"--aws-region {self.region_name} "
+        )
+
+        if config.STEPFUNCTIONS_LAMBDA_ENDPOINT.lower() != "default":
+            lambda_endpoint = (
+                config.STEPFUNCTIONS_LAMBDA_ENDPOINT or aws_stack.get_local_service_url("lambda")
+            )
+            cmd += f" --lambda-endpoint {lambda_endpoint}"
+
+        # add service endpoint flags
+        services = [
+            "athena",
+            "batch",
+            "dynamodb",
+            "ecs",
+            "eks",
+            "events",
+            "glue",
+            "sagemaker",
+            "sns",
+            "sqs",
+            "stepfunctions",
+        ]
+
+        for service in services:
+            flag = f"--{service}-endpoint"
+            if service == "stepfunctions":
+                flag = "--step-functions-endpoint"
+            elif service == "events":
+                flag = "--eventbridge-endpoint"
+            elif service in ["athena", "eks"]:
+                flag = f"--step-functions-{service}"
+            endpoint = aws_stack.get_local_service_url(service)
+            cmd += f" {flag} {endpoint}"
+
+        return cmd
+
+    def _log_listener(self, line, **kwargs):
+        LOG.debug(line.rstrip())
+
+
+class StepFunctionsServerManager:
+    default_startup_timeout = 20
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._servers: dict[tuple[str, str], StepFunctionsServer] = {}
+
+    def get_server_for_account_region(
+        self, account_id: str, region_name: str
+    ) -> StepFunctionsServer:
+        locator = (account_id, region_name)
+
+        if locator in self._servers:
+            return self._servers[locator]
+
+        with self._lock:
+            if locator in self._servers:
+                return self._servers[locator]
+
+            LOG.info("Creating StepFunctions server for %s", locator)
+            self._servers[locator] = self._create_stepfunctions_server(account_id, region_name)
+
+            self._servers[locator].start()
+
+            if not self._servers[locator].wait_is_up(timeout=self.default_startup_timeout):
+                raise TimeoutError("Gave up waiting for StepFunctions server to start up")
+
+            return self._servers[locator]
+
+    def shutdown_all(self):
+        with self._lock:
+            while self._servers:
+                locator, server = self._servers.popitem()
+                LOG.info("Shutting down StepFunctions for %s", locator)
+                server.shutdown()
+
+    def _create_stepfunctions_server(
+        self, account_id: str, region_name: str
+    ) -> StepFunctionsServer:
+        port = config.LOCAL_PORT_STEPFUNCTIONS
+        if not port_can_be_bound(port):
+            port = get_free_tcp_port()
+        stepfunctions_local_package.install()
+
+        server = StepFunctionsServer(
+            port=port,
+            account_id=account_id,
+            region_name=region_name,
+        )
+        return server
diff --git a/localstack/services/stepfunctions/stepfunctions_utils.py b/localstack/services/stepfunctions/stepfunctions_utils.py
index 04af050222842..beca5e20b3192 100644
--- a/localstack/services/stepfunctions/stepfunctions_utils.py
+++ b/localstack/services/stepfunctions/stepfunctions_utils.py
@@ -2,6 +2,7 @@
 from typing import Dict
 
 from localstack.aws.connect import connect_to
+from localstack.utils.aws.arns import parse_arn
 from localstack.utils.common import retry
 
 LOG = logging.getLogger(__name__)
@@ -10,7 +11,11 @@
 def await_sfn_execution_result(execution_arn: str, timeout_secs: int = 60) -> Dict:
     """Wait until the given SFN execution ARN is no longer in RUNNING status, then return execution result."""
 
-    client = connect_to().stepfunctions
+    arn_data = parse_arn(execution_arn)
+
+    client = connect_to(
+        aws_access_key_id=arn_data["account"], region_name=arn_data["region"]
+    ).stepfunctions
 
     def _get_result():
         result = client.describe_execution(executionArn=execution_arn)
diff --git a/tests/aws/services/stepfunctions/legacy/test_stepfunctions_legacy.py b/tests/aws/services/stepfunctions/legacy/test_stepfunctions_legacy.py
index 50e3be0bcc05d..78ffac7888610 100644
--- a/tests/aws/services/stepfunctions/legacy/test_stepfunctions_legacy.py
+++ b/tests/aws/services/stepfunctions/legacy/test_stepfunctions_legacy.py
@@ -4,6 +4,7 @@
 
 import pytest
 
+from localstack.constants import TEST_AWS_REGION_NAME
 from localstack.services.events.provider import TEST_EVENTS_CACHE
 from localstack.services.stepfunctions.stepfunctions_utils import await_sfn_execution_result
 from localstack.testing.aws.util import is_aws_cloud
@@ -173,10 +174,23 @@
 
 LOG = logging.getLogger(__name__)
 
+# The legacy StepFunctions provider does not properly support multi-accounts
+# Although StepFunctions Local has an `--account-id` argument,
+# it does not obey the override especially during Lambda invocations.
+# As such, the tests in this module only run for the following account.
+SF_TEST_AWS_ACCOUNT_ID = "000000000000"
+
+
+@pytest.fixture(scope="module")
+def custom_client(aws_client_factory):
+    return aws_client_factory(
+        region_name=TEST_AWS_REGION_NAME, aws_access_key_id=SF_TEST_AWS_ACCOUNT_ID
+    )
+
 
 @pytest.fixture(scope="module")
-def setup_and_tear_down(aws_client):
-    lambda_client = aws_client.lambda_
+def setup_and_tear_down(custom_client):
+    lambda_client = custom_client.lambda_
 
     zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_ENV), get_content=True)
     zip_file2 = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON_ECHO), get_content=True)
@@ -184,30 +198,35 @@ def setup_and_tear_down(aws_client):
         func_name=TEST_LAMBDA_NAME_1,
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE},
-        client=aws_client.lambda_,
+        client=custom_client.lambda_,
+        s3_client=custom_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_2,
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE_2},
-        client=aws_client.lambda_,
+        client=custom_client.lambda_,
+        s3_client=custom_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_3,
         zip_file=zip_file,
         envvars={"Hello": "Replace Value"},
-        client=aws_client.lambda_,
+        client=custom_client.lambda_,
+        s3_client=custom_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_4,
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE_4},
-        client=aws_client.lambda_,
+        client=custom_client.lambda_,
+        s3_client=custom_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_5,
         zip_file=zip_file2,
-        client=aws_client.lambda_,
+        client=custom_client.lambda_,
+        s3_client=custom_client.s3,
     )
 
     active_waiter = lambda_client.get_waiter("function_active_v2")
@@ -219,17 +238,17 @@ def setup_and_tear_down(aws_client):
 
     yield
 
-    aws_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_1)
-    aws_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_2)
-    aws_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_3)
-    aws_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_4)
-    aws_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_5)
+    custom_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_1)
+    custom_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_2)
+    custom_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_3)
+    custom_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_4)
+    custom_client.lambda_.delete_function(FunctionName=TEST_LAMBDA_NAME_5)
 
 
 @pytest.fixture
-def sfn_execution_role(aws_client):
+def sfn_execution_role(custom_client):
     role_name = f"role-{short_uid()}"
-    result = aws_client.iam.create_role(
+    result = custom_client.iam.create_role(
         RoleName=role_name,
         AssumeRolePolicyDocument='{"Version": "2012-10-17", "Statement": {"Action": "sts:AssumeRole", "Effect": "Allow", "Principal": {"Service": "states.amazonaws.com"}}}',
     )
@@ -282,26 +301,28 @@ def get_machine_arn(sm_name, sfn_client):
 @pytest.mark.usefixtures("setup_and_tear_down")
 class TestStateMachine:
     @markers.aws.needs_fixing
-    def test_create_choice_state_machine(self, aws_client):
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
-        role_arn = arns.role_arn("sfn_role")
+    def test_create_choice_state_machine(self, custom_client):
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
 
         definition = clone(STATE_MACHINE_CHOICE)
-        lambda_arn_4 = arns.lambda_function_arn(TEST_LAMBDA_NAME_4)
+        lambda_arn_4 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_4, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Add"]["Resource"] = lambda_arn_4
         definition = json.dumps(definition)
         sm_name = f"choice-{short_uid()}"
-        aws_client.stepfunctions.create_state_machine(
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # assert that the SM has been created
-        assert_machine_created(state_machines_before, aws_client.stepfunctions)
+        assert_machine_created(state_machines_before, custom_client.stepfunctions)
 
         # run state machine
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
         input = {"x": "1", "y": "2"}
-        result = aws_client.stepfunctions.start_execution(
+        result = custom_client.stepfunctions.start_execution(
             stateMachineArn=sm_arn, input=json.dumps(input)
         )
         assert result.get("executionArn")
@@ -310,141 +331,155 @@ def test_create_choice_state_machine(self, aws_client):
         test_output = {**input, "added": {"Hello": TEST_RESULT_VALUE_4}}
 
         def check_result():
-            result = _get_execution_results(sm_arn, aws_client.stepfunctions)
+            result = _get_execution_results(sm_arn, custom_client.stepfunctions)
             assert test_output == result
 
         # assert that the result is correct
         retry(check_result, sleep=2, retries=10)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, sfn_client=aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, sfn_client=custom_client.stepfunctions)
 
     @markers.aws.needs_fixing
-    def test_create_run_map_state_machine(self, aws_client):
+    def test_create_run_map_state_machine(self, custom_client):
         names = ["Bob", "Meg", "Joe"]
         test_input = [{"map": name} for name in names]
         test_output = [{"Hello": name} for name in names]
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
 
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_MAP)
-        lambda_arn_3 = arns.lambda_function_arn(TEST_LAMBDA_NAME_3)
+        lambda_arn_3 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_3, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["ExampleMapState"]["Iterator"]["States"]["CallLambda"][
             "Resource"
         ] = lambda_arn_3
         definition = json.dumps(definition)
         sm_name = f"map-{short_uid()}"
-        aws_client.stepfunctions.create_state_machine(
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # assert that the SM has been created
-        assert_machine_created(state_machines_before, aws_client.stepfunctions)
+        assert_machine_created(state_machines_before, custom_client.stepfunctions)
 
         # run state machine
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
-        result = aws_client.stepfunctions.start_execution(
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
+        result = custom_client.stepfunctions.start_execution(
             stateMachineArn=sm_arn, input=json.dumps(test_input)
         )
         assert result.get("executionArn")
 
         def check_invocations():
             # assert that the result is correct
-            result = _get_execution_results(sm_arn, aws_client.stepfunctions)
+            result = _get_execution_results(sm_arn, custom_client.stepfunctions)
             assert test_output == result
 
         # assert that the lambda has been invoked by the SM execution
         retry(check_invocations, sleep=1, retries=10)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, custom_client.stepfunctions)
 
     @markers.aws.needs_fixing
-    def test_create_run_state_machine(self, aws_client):
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+    def test_create_run_state_machine(self, custom_client):
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_BASIC)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_1)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_2)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_1, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_2, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["step1"]["Resource"] = lambda_arn_1
         definition["States"]["step2"]["Resource"] = lambda_arn_2
         definition = json.dumps(definition)
         sm_name = f"basic-{short_uid()}"
-        aws_client.stepfunctions.create_state_machine(
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # assert that the SM has been created
-        assert_machine_created(state_machines_before, aws_client.stepfunctions)
+        assert_machine_created(state_machines_before, custom_client.stepfunctions)
 
         # run state machine
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
-        result = aws_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
+        result = custom_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
         assert result.get("executionArn")
 
         def check_invocations():
             # assert that the result is correct
-            result = _get_execution_results(sm_arn, aws_client.stepfunctions)
+            result = _get_execution_results(sm_arn, custom_client.stepfunctions)
             assert {"Hello": TEST_RESULT_VALUE_2} == result["result_value"]
 
         # assert that the lambda has been invoked by the SM execution
         retry(check_invocations, sleep=0.7, retries=25)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, custom_client.stepfunctions)
 
     @markers.aws.needs_fixing
-    def test_try_catch_state_machine(self, aws_client):
+    def test_try_catch_state_machine(self, custom_client):
         if os.environ.get("AWS_DEFAULT_REGION") != "us-east-1":
             pytest.skip("skipping non us-east-1 temporarily")
 
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_CATCH)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_1)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_2)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_1, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_2, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Start"]["Parameters"]["FunctionName"] = lambda_arn_1
         definition["States"]["ErrorHandler"]["Resource"] = lambda_arn_2
         definition["States"]["Final"]["Resource"] = lambda_arn_2
         definition = json.dumps(definition)
         sm_name = f"catch-{short_uid()}"
-        aws_client.stepfunctions.create_state_machine(
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # run state machine
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
-        result = aws_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
+        result = custom_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
         assert result.get("executionArn")
 
         def check_invocations():
             # assert that the result is correct
-            result = _get_execution_results(sm_arn, aws_client.stepfunctions)
+            result = _get_execution_results(sm_arn, custom_client.stepfunctions)
             assert {"Hello": TEST_RESULT_VALUE_2} == result.get("handled")
 
         # assert that the lambda has been invoked by the SM execution
         retry(check_invocations, sleep=1, retries=10)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, custom_client.stepfunctions)
 
     # TODO: validate against AWS
     @markers.aws.needs_fixing
-    def test_intrinsic_functions(self, aws_client):
+    def test_intrinsic_functions(self, custom_client):
         if os.environ.get("AWS_DEFAULT_REGION") != "us-east-1":
             pytest.skip("skipping non us-east-1 temporarily")
 
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_INTRINSIC_FUNCS)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_5)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_5)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_5, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_5, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         if isinstance(definition["States"]["state1"].get("Parameters"), dict):
             definition["States"]["state1"]["Parameters"]["lambda_params"][
                 "FunctionName"
@@ -452,33 +487,33 @@ def test_intrinsic_functions(self, aws_client):
             definition["States"]["state3"]["Resource"] = lambda_arn_2
         definition = json.dumps(definition)
         sm_name = f"intrinsic-{short_uid()}"
-        aws_client.stepfunctions.create_state_machine(
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # run state machine
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
         input = {}
-        result = aws_client.stepfunctions.start_execution(
+        result = custom_client.stepfunctions.start_execution(
             stateMachineArn=sm_arn, input=json.dumps(input)
         )
         assert result.get("executionArn")
 
         def check_invocations():
             # assert that the result is correct
-            result = _get_execution_results(sm_arn, aws_client.stepfunctions)
+            result = _get_execution_results(sm_arn, custom_client.stepfunctions)
             assert {"payload": {"values": [1, "v2"]}} == result.get("result_value")
 
         # assert that the lambda has been invoked by the SM execution
         retry(check_invocations, sleep=1, retries=10)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, custom_client.stepfunctions)
 
     @markers.aws.needs_fixing
-    def test_events_state_machine(self, aws_client):
-        events = aws_client.events
-        state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+    def test_events_state_machine(self, custom_client):
+        events = custom_client.events
+        state_machines_before = custom_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create event bus
         bus_name = f"bus-{short_uid()}"
@@ -489,15 +524,15 @@ def test_events_state_machine(self, aws_client):
         definition["States"]["step1"]["Parameters"]["Entries"][0]["EventBusName"] = bus_name
         definition = json.dumps(definition)
         sm_name = f"events-{short_uid()}"
-        role_arn = arns.role_arn("sfn_role")
-        aws_client.stepfunctions.create_state_machine(
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
+        custom_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
 
         # run state machine
         events_before = len(TEST_EVENTS_CACHE)
-        sm_arn = get_machine_arn(sm_name, aws_client.stepfunctions)
-        result = aws_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
+        sm_arn = get_machine_arn(sm_name, custom_client.stepfunctions)
+        result = custom_client.stepfunctions.start_execution(stateMachineArn=sm_arn)
         assert result.get("executionArn")
 
         def check_invocations():
@@ -513,11 +548,11 @@ def check_invocations():
         retry(check_invocations, sleep=1, retries=10)
 
         # clean up
-        cleanup(sm_arn, state_machines_before, aws_client.stepfunctions)
+        cleanup(sm_arn, state_machines_before, custom_client.stepfunctions)
         events.delete_event_bus(Name=bus_name)
 
     @markers.aws.needs_fixing
-    def test_create_state_machines_in_parallel(self, cleanups, aws_client):
+    def test_create_state_machines_in_parallel(self, cleanups, custom_client):
         """
         Perform a test that creates a series of state machines in parallel. Without concurrency control, using
         StepFunctions-Local, the following error is pretty consistently reproducible:
@@ -526,29 +561,33 @@ def test_create_state_machines_in_parallel(self, cleanups, aws_client):
         CreateStateMachine operation: Invalid State Machine Definition: ''DUPLICATE_STATE_NAME: Duplicate State name:
         MissingValue at /States/MissingValue', 'DUPLICATE_STATE_NAME: Duplicate State name: Add at /States/Add''
         """
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_CHOICE)
-        lambda_arn_4 = arns.lambda_function_arn(TEST_LAMBDA_NAME_4)
+        lambda_arn_4 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_4, SF_TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Add"]["Resource"] = lambda_arn_4
         definition = json.dumps(definition)
         results = []
 
         def _create_sm(*_):
             sm_name = f"sm-{short_uid()}"
-            result = aws_client.stepfunctions.create_state_machine(
+            result = custom_client.stepfunctions.create_state_machine(
                 name=sm_name, definition=definition, roleArn=role_arn
             )
             assert result["ResponseMetadata"]["HTTPStatusCode"] == 200
             cleanups.append(
-                lambda: aws_client.stepfunctions.delete_state_machine(
+                lambda: custom_client.stepfunctions.delete_state_machine(
                     stateMachineArn=result["stateMachineArn"]
                 )
             )
             results.append(result)
-            aws_client.stepfunctions.describe_state_machine(
+            custom_client.stepfunctions.describe_state_machine(
                 stateMachineArn=result["stateMachineArn"]
             )
-            aws_client.stepfunctions.list_tags_for_resource(resourceArn=result["stateMachineArn"])
+            custom_client.stepfunctions.list_tags_for_resource(
+                resourceArn=result["stateMachineArn"]
+            )
 
         num_machines = 30
         parallelize(_create_sm, list(range(num_machines)), size=2)
@@ -608,10 +647,12 @@ def _create_sm(*_):
 @pytest.mark.parametrize("statemachine_definition", (TEST_STATE_MACHINE_3,))  # TODO: add sync2 test
 @markers.aws.needs_fixing
 def test_multiregion_nested(aws_client_factory, region_name, statemachine_definition):
-    client1 = aws_client_factory(region_name=region_name).stepfunctions
+    client1 = aws_client_factory(
+        region_name=region_name, aws_access_key_id=SF_TEST_AWS_ACCOUNT_ID
+    ).stepfunctions
     # create state machine
     child_machine_name = f"sf-child-{short_uid()}"
-    role = arns.role_arn("sfn_role")
+    role = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, region_name)
     child_machine_result = client1.create_state_machine(
         name=child_machine_name, definition=json.dumps(TEST_STATE_MACHINE), roleArn=role
     )
@@ -619,7 +660,7 @@ def test_multiregion_nested(aws_client_factory, region_name, statemachine_defini
 
     # create parent state machine
     name = f"sf-parent-{short_uid()}"
-    role = arns.role_arn("sfn_role")
+    role = arns.role_arn("sfn_role", SF_TEST_AWS_ACCOUNT_ID, region_name)
     result = client1.create_state_machine(
         name=name,
         definition=json.dumps(statemachine_definition).replace(
@@ -659,10 +700,10 @@ def assert_success():
 
 
 @markers.aws.validated
-def test_default_logging_configuration(create_state_machine, aws_client):
+def test_default_logging_configuration(create_state_machine, custom_client):
     role_name = f"role_name-{short_uid()}"
     try:
-        role_arn = aws_client.iam.create_role(
+        role_arn = custom_client.iam.create_role(
             RoleName=role_name,
             AssumeRolePolicyDocument=json.dumps(STS_ROLE_POLICY_DOC),
         )["Role"]["Arn"]
@@ -674,18 +715,18 @@ def test_default_logging_configuration(create_state_machine, aws_client):
         result = create_state_machine(name=sm_name, definition=definition, roleArn=role_arn)
 
         assert result["ResponseMetadata"]["HTTPStatusCode"] == 200
-        result = aws_client.stepfunctions.describe_state_machine(
+        result = custom_client.stepfunctions.describe_state_machine(
             stateMachineArn=result["stateMachineArn"]
         )
         assert result["ResponseMetadata"]["HTTPStatusCode"] == 200
         assert result["loggingConfiguration"] == {"level": "OFF", "includeExecutionData": False}
     finally:
-        aws_client.iam.delete_role(RoleName=role_name)
+        custom_client.iam.delete_role(RoleName=role_name)
 
 
 @pytest.mark.skip("Does not work against Pro in new pipeline.")
 @markers.aws.needs_fixing
-def test_aws_sdk_task(sfn_execution_role, aws_client):
+def test_aws_sdk_task(sfn_execution_role, custom_client):
     statemachine_definition = {
         "StartAt": "CreateTopicTask",
         "States": {
@@ -703,15 +744,15 @@ def test_aws_sdk_task(sfn_execution_role, aws_client):
     policy_name = f"policy-{short_uid()}"
     topic_name = f"topic-{short_uid()}"
 
-    policy = aws_client.iam.create_policy(
+    policy = custom_client.iam.create_policy(
         PolicyDocument='{"Version": "2012-10-17", "Statement": {"Action": "sns:createTopic", "Effect": "Allow", "Resource": "*"}}',
         PolicyName=policy_name,
     )
-    aws_client.iam.attach_role_policy(
+    custom_client.iam.attach_role_policy(
         RoleName=sfn_execution_role["RoleName"], PolicyArn=policy["Policy"]["Arn"]
     )
 
-    result = aws_client.stepfunctions.create_state_machine(
+    result = custom_client.stepfunctions.create_state_machine(
         name=name,
         definition=json.dumps(statemachine_definition),
         roleArn=sfn_execution_role["Arn"],
@@ -719,13 +760,13 @@ def test_aws_sdk_task(sfn_execution_role, aws_client):
     machine_arn = result["stateMachineArn"]
 
     try:
-        result = aws_client.stepfunctions.list_state_machines()["stateMachines"]
+        result = custom_client.stepfunctions.list_state_machines()["stateMachines"]
         assert len(result) > 0
         assert len([sm for sm in result if sm["name"] == name]) == 1
 
         def assert_execution_success(executionArn: str):
             def _assert_execution_success():
-                status = aws_client.stepfunctions.describe_execution(executionArn=executionArn)[
+                status = custom_client.stepfunctions.describe_execution(executionArn=executionArn)[
                     "status"
                 ]
                 if status == "FAILED":
@@ -739,35 +780,35 @@ def _retry_execution():
             # start state machine execution
             # AWS initially straight up fails until the permissions seem to take effect
             # so we wait until the statemachine is at least running
-            result = aws_client.stepfunctions.start_execution(
+            result = custom_client.stepfunctions.start_execution(
                 stateMachineArn=machine_arn, input='{"Name": "' f"{topic_name}" '"}'
             )
             assert wait_until(assert_execution_success(result["executionArn"]))
-            describe_result = aws_client.stepfunctions.describe_execution(
+            describe_result = custom_client.stepfunctions.describe_execution(
                 executionArn=result["executionArn"]
             )
             output = describe_result["output"]
             assert topic_name in output
-            result = aws_client.stepfunctions.describe_state_machine_for_execution(
+            result = custom_client.stepfunctions.describe_state_machine_for_execution(
                 executionArn=result["executionArn"]
             )
             assert result["stateMachineArn"] == machine_arn
             topic_arn = json.loads(describe_result["output"])["TopicArn"]
-            topics = aws_client.sns.list_topics()
+            topics = custom_client.sns.list_topics()
             assert topic_arn in [t["TopicArn"] for t in topics["Topics"]]
-            aws_client.sns.delete_topic(TopicArn=topic_arn)
+            custom_client.sns.delete_topic(TopicArn=topic_arn)
             return True
 
         assert wait_until(_retry_execution, max_retries=3, strategy="linear", wait=3.0)
 
     finally:
-        aws_client.iam.delete_policy(PolicyArn=policy["Policy"]["Arn"])
-        aws_client.stepfunctions.delete_state_machine(stateMachineArn=machine_arn)
+        custom_client.iam.delete_policy(PolicyArn=policy["Policy"]["Arn"])
+        custom_client.stepfunctions.delete_state_machine(stateMachineArn=machine_arn)
 
 
 @pytest.mark.skip("Does not work against Pro in new pipeline.")
 @markers.aws.needs_fixing
-def test_aws_sdk_task_delete_s3_object(s3_bucket, sfn_execution_role, aws_client):
+def test_aws_sdk_task_delete_s3_object(s3_bucket, sfn_execution_role, custom_client):
     s3_key = "test-key"
     statemachine_definition = {
         "StartAt": "CreateTopicTask",
@@ -782,21 +823,21 @@ def test_aws_sdk_task_delete_s3_object(s3_bucket, sfn_execution_role, aws_client
     }
 
     # create state machine
-    aws_client.s3.put_object(Bucket=s3_bucket, Key=s3_key, Body=b"")
+    custom_client.s3.put_object(Bucket=s3_bucket, Key=s3_key, Body=b"")
     name = f"statemachine-{short_uid()}"
 
-    result = aws_client.stepfunctions.create_state_machine(
+    result = custom_client.stepfunctions.create_state_machine(
         name=name,
         definition=json.dumps(statemachine_definition),
         roleArn=sfn_execution_role["Arn"],
     )
     machine_arn = result["stateMachineArn"]
 
-    result = aws_client.stepfunctions.start_execution(stateMachineArn=machine_arn, input="{}")
+    result = custom_client.stepfunctions.start_execution(stateMachineArn=machine_arn, input="{}")
     execution_arn = result["executionArn"]
 
     await_sfn_execution_result(execution_arn)
 
     with pytest.raises(Exception) as exc:
-        aws_client.s3.head_object(Bucket=s3_bucket, Key=s3_key)
+        custom_client.s3.head_object(Bucket=s3_bucket, Key=s3_key)
     assert "Not Found" in str(exc)
diff --git a/tests/aws/services/stepfunctions/v2/test_stepfunctions_v2.py b/tests/aws/services/stepfunctions/v2/test_stepfunctions_v2.py
index bb29a2f2a0257..942377ab2362a 100644
--- a/tests/aws/services/stepfunctions/v2/test_stepfunctions_v2.py
+++ b/tests/aws/services/stepfunctions/v2/test_stepfunctions_v2.py
@@ -7,6 +7,8 @@
 from localstack.constants import (
     SECONDARY_TEST_AWS_ACCESS_KEY_ID,
     SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
+    TEST_AWS_ACCOUNT_ID,
+    TEST_AWS_REGION_NAME,
 )
 from localstack.services.events.provider import TEST_EVENTS_CACHE
 from localstack.testing.aws.util import is_aws_cloud
@@ -194,29 +196,34 @@ def setup_and_tear_down(aws_client):
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE},
         client=aws_client.lambda_,
+        s3_client=aws_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_2,
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE_2},
         client=aws_client.lambda_,
+        s3_client=aws_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_3,
         zip_file=zip_file,
         envvars={"Hello": "Replace Value"},
         client=aws_client.lambda_,
+        s3_client=aws_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_4,
         zip_file=zip_file,
         envvars={"Hello": TEST_RESULT_VALUE_4},
         client=aws_client.lambda_,
+        s3_client=aws_client.s3,
     )
     testutil.create_lambda_function(
         func_name=TEST_LAMBDA_NAME_5,
         zip_file=zip_file2,
         client=aws_client.lambda_,
+        s3_client=aws_client.s3,
     )
 
     active_waiter = lambda_client.get_waiter("function_active_v2")
@@ -282,10 +289,12 @@ class TestStateMachine:
     @markers.aws.needs_fixing
     def test_create_choice_state_machine(self, aws_client):
         state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
 
         definition = clone(STATE_MACHINE_CHOICE)
-        lambda_arn_4 = arns.lambda_function_arn(TEST_LAMBDA_NAME_4)
+        lambda_arn_4 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_4, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Add"]["Resource"] = lambda_arn_4
         definition = json.dumps(definition)
         sm_name = f"choice-{short_uid()}"
@@ -324,9 +333,11 @@ def test_create_run_map_state_machine(self, aws_client):
         test_output = [{"Hello": name} for name in names]
         state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
 
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_MAP)
-        lambda_arn_3 = arns.lambda_function_arn(TEST_LAMBDA_NAME_3)
+        lambda_arn_3 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_3, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["ExampleMapState"]["ItemProcessor"]["States"]["CallLambda"][
             "Resource"
         ] = lambda_arn_3
@@ -362,10 +373,14 @@ def test_create_run_state_machine(self, aws_client):
         state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_BASIC)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_1)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_2)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_1, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_2, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["step1"]["Resource"] = lambda_arn_1
         definition["States"]["step2"]["Resource"] = lambda_arn_2
         definition = json.dumps(definition)
@@ -398,10 +413,14 @@ def test_try_catch_state_machine(self, aws_client):
         state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_CATCH)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_1)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_2)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_1, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_2, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Start"]["Parameters"]["FunctionName"] = lambda_arn_1
         definition["States"]["ErrorHandler"]["Resource"] = lambda_arn_2
         definition["States"]["Final"]["Resource"] = lambda_arn_2
@@ -432,10 +451,14 @@ def test_intrinsic_functions(self, aws_client):
         state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
 
         # create state machine
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_INTRINSIC_FUNCS)
-        lambda_arn_1 = arns.lambda_function_arn(TEST_LAMBDA_NAME_5)
-        lambda_arn_2 = arns.lambda_function_arn(TEST_LAMBDA_NAME_5)
+        lambda_arn_1 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_5, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
+        lambda_arn_2 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_5, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         if isinstance(definition["States"]["state1"].get("Parameters"), dict):
             definition["States"]["state1"]["Parameters"]["lambda_params"][
                 "FunctionName"
@@ -483,7 +506,7 @@ def test_events_state_machine(self, aws_client):
         definition["States"]["step1"]["Parameters"]["Entries"][0]["EventBusName"] = bus_name
         definition = json.dumps(definition)
         sm_name = f"events-{short_uid()}"
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         aws_client.stepfunctions.create_state_machine(
             name=sm_name, definition=definition, roleArn=role_arn
         )
@@ -520,9 +543,11 @@ def test_create_state_machines_in_parallel(self, cleanups, aws_client):
         CreateStateMachine operation: Invalid State Machine Definition: ''DUPLICATE_STATE_NAME: Duplicate State name:
         MissingValue at /States/MissingValue', 'DUPLICATE_STATE_NAME: Duplicate State name: Add at /States/Add''
         """
-        role_arn = arns.role_arn("sfn_role")
+        role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
         definition = clone(STATE_MACHINE_CHOICE)
-        lambda_arn_4 = arns.lambda_function_arn(TEST_LAMBDA_NAME_4)
+        lambda_arn_4 = arns.lambda_function_arn(
+            TEST_LAMBDA_NAME_4, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
+        )
         definition["States"]["Add"]["Resource"] = lambda_arn_4
         definition = json.dumps(definition)
         results = []
@@ -613,7 +638,7 @@ def test_multiregion_nested(aws_client_factory, region_name, statemachine_defini
     )
     # create state machine
     child_machine_name = f"sf-child-{short_uid()}"
-    role = arns.role_arn("sfn_role")
+    role = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
     child_machine_result = client1.create_state_machine(
         name=child_machine_name, definition=json.dumps(TEST_STATE_MACHINE), roleArn=role
     )
@@ -621,7 +646,7 @@ def test_multiregion_nested(aws_client_factory, region_name, statemachine_defini
 
     # create parent state machine
     name = f"sf-parent-{short_uid()}"
-    role = arns.role_arn("sfn_role")
+    role = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
     result = client1.create_state_machine(
         name=name,
         definition=json.dumps(statemachine_definition).replace(
@@ -783,7 +808,7 @@ def test_run_aws_sdk_secrets_manager(aws_client):
     state_machines_before = aws_client.stepfunctions.list_state_machines()["stateMachines"]
 
     # create state machine
-    role_arn = arns.role_arn("sfn_role")
+    role_arn = arns.role_arn("sfn_role", TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME)
     definition = {
         "StartAt": "StateCreateSecret",
         "States": {


Code review comments:
diff_hunk: @@ -26,8 +26,14 @@
 
 
 class Environment:
-    def __init__(self, context_object_init: ContextObjectInitData):
+    def __init__(
+        self, account_id: str, region_name: str, context_object_init: ContextObjectInitData
+    ):
         super(Environment, self).__init__()
+
path: localstack/services/stepfunctions/asl/eval/environment.py
text: @user1:
I think we should decorate these in an object where the environment can expose runtime values with, like these 2.

diff_hunk: @@ -85,10 +84,12 @@ def __init__(self, resource_arn: ResourceARN):
         self.account = resource_arn.account
 
     @staticmethod
-    def from_resource_arn(arn: str) -> Resource:
+    def from_resource_arn(account_id: str, region_name: str, arn: str) -> Resource:
path: localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/resource.py
text: @user1:
Runtime values are ideally evaluated during execution. So resource should become an EvalComponent, evaluate its region and account in `_eval_body` using the environment, and push to the stack a resource object that includes all the values. 

@author:
Sorry I'm not familiar with the internal workings at all!

Could you point me in the exact direction where I should start?

diff_hunk: @@ -124,8 +123,7 @@ def _from_error(self, env: Environment, ex: Exception) -> FailureEvent:
 
     def _eval_service_task(self, env: Environment, parameters: dict) -> None:
         api_action = camel_to_snake_case(self.resource.api_action)
-
-        dynamodb_client = connect_to(config=Config(parameter_validation=False)).dynamodb
+        dynamodb_client = get_boto_client(env, "dynamodb")
path: localstack/services/stepfunctions/asl/component/state/state_execution/state_task/service/state_task_service_dynamodb.py
text: @user1:
Is this still ensuring the parameters validation is off?

@author:
Yes! Please make sure you use this helper to get the Boto client and not `connect_externally_to`

https://github.com/localstack/localstack/blob/aad9b80492fdc9f7b22f48af0184cb5febf6dd23/localstack/services/stepfunctions/backend/utils.py#L7-L13

diff_hunk: @@ -191,6 +191,10 @@
 
 
 class Preprocessor(ASLParserVisitor):
+    def __init__(self, account_id: str, region_name: str):
path: localstack/services/stepfunctions/asl/parse/preprocessor.py
text: @user1:
Adding the values to the environment is a decent solution, and evaluation components can produce runtime values like region and account during evaluation. So I don't think we should pollute the Preprocessor with runtime values, as conceptually it just builds the evaluation tree.

@author:
When the `Preprocessor` constructs the `Resource` object from ARN, the ARN does not necessarily contain the region and the account ID. So it is necessary that the correct account ID and region from the request context are passed down.

https://github.com/localstack/localstack/blob/aad9b80492fdc9f7b22f48af0184cb5febf6dd23/localstack/services/stepfunctions/asl/parse/preprocessor.py#L238

Using `aws_stack.get_region()` is not an option as it doesn't always have the proper region. That method will likely be removed in the future for this exact reason.

What do you think would be the best way to achieve this?