From 976911c8bbf271cda4fe11e05e5bdc8ece00fb22 Mon Sep 17 00:00:00 2001 From: Waldemar Hummer Date: Sat, 7 Feb 2026 19:20:45 +0100 Subject: [PATCH 1/2] Add proxy support and tests for Lambda service Add Lambda resource matching in the proxy handler (FunctionName/ARN-based) and comprehensive integration tests covering basic proxying, read-only mode, and resource name pattern matching. Co-Authored-By: Claude Opus 4.6 --- aws-proxy/AGENTS.md | 2 +- .../aws_proxy/server/aws_request_forwarder.py | 17 ++ aws-proxy/tests/proxy/test_lambda.py | 280 ++++++++++++++++++ 3 files changed, 298 insertions(+), 1 deletion(-) create mode 100644 aws-proxy/tests/proxy/test_lambda.py diff --git a/aws-proxy/AGENTS.md b/aws-proxy/AGENTS.md index a69f5d74..bd2d8078 100644 --- a/aws-proxy/AGENTS.md +++ b/aws-proxy/AGENTS.md @@ -13,7 +13,7 @@ You are an AI agent tasked with adding additional functionality or test coverage * You can make modifications to files (no need to prompt for confirmation) * You can delete existing files if needed, but only after user confirmation * You can call different `make` targets (e.g., `make test`) in this repo (no need to prompt for confirmation) -* For each new file created or existing file modified, add a header comment to the file, something like `# Note/disclosure: This file has been (partially or fully) generated by an AI agent.` +* For each new file created or existing file modified, add a header comment to the file, something like `# Note: This file has been (partially or fully) generated by an AI agent.` * The proxy tests are executed against real AWS and may incur some costs, so rather than executing the entire test suite or entire modules, focus the testing on individual test functions within a module only. * Before claiming success, always double-check against real AWS (via `aws` CLI commands) that everything has been cleaned up and there are no leftover resources from the proxy tests. * Never add any `print(..)` statements to the code - use a logger to report any status to the user, if required. diff --git a/aws-proxy/aws_proxy/server/aws_request_forwarder.py b/aws-proxy/aws_proxy/server/aws_request_forwarder.py index 785f8181..4bd06371 100644 --- a/aws-proxy/aws_proxy/server/aws_request_forwarder.py +++ b/aws-proxy/aws_proxy/server/aws_request_forwarder.py @@ -151,6 +151,17 @@ def _request_matches_resource( return False # For metric operations without alarm names, check if pattern is generic return bool(re.match(resource_name_pattern, ".*")) + if service_name == "lambda": + # Lambda function ARN format: arn:aws:lambda:{region}:{account}:function:{name} + function_name = context.service_request.get("FunctionName") or "" + if function_name: + if ":function:" not in function_name: + function_arn = f"arn:aws:lambda:{context.region}:{context.account_id}:function:{function_name}" + else: + function_arn = function_name + return bool(re.match(resource_name_pattern, function_arn)) + # For operations without FunctionName (e.g., ListFunctions), check if pattern is generic + return bool(re.match(resource_name_pattern, ".*")) if service_name == "logs": # CloudWatch Logs ARN format: arn:aws:logs:{region}:{account}:log-group:{name}:* log_group_name = context.service_request.get("logGroupName") or "" @@ -266,6 +277,12 @@ def _is_read_request(self, context: RequestContext) -> bool: "PartiQLSelect", }: return True + if context.service.service_name == "lambda" and operation_name in { + "Invoke", + "InvokeAsync", + "InvokeWithResponseStream", + }: + return True if context.service.service_name == "appsync" and operation_name in { "EvaluateCode", "EvaluateMappingTemplate", diff --git a/aws-proxy/tests/proxy/test_lambda.py b/aws-proxy/tests/proxy/test_lambda.py new file mode 100644 index 00000000..6dc27110 --- /dev/null +++ b/aws-proxy/tests/proxy/test_lambda.py @@ -0,0 +1,280 @@ +# Note: This file has been (partially or fully) generated by an AI agent. +import io +import json +import logging +import zipfile + +import boto3 +import pytest +from botocore.exceptions import ClientError +from localstack.aws.connect import connect_to +from localstack.utils.strings import short_uid +from localstack.utils.sync import retry + +from aws_proxy.shared.models import ProxyConfig + +LOG = logging.getLogger(__name__) + +LAMBDA_RUNTIME = "python3.12" +LAMBDA_HANDLER = "index.handler" +LAMBDA_HANDLER_CODE = """ +import json +def handler(event, context): + return {"statusCode": 200, "body": json.dumps({"message": "hello", "input": event})} +""" + + +def _create_lambda_zip() -> bytes: + """Create an in-memory ZIP deployment package with a simple handler.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr("index.py", LAMBDA_HANDLER_CODE) + return buf.getvalue() + + +@pytest.fixture(scope="module") +def lambda_execution_role(): + """Create a basic IAM role for Lambda execution in real AWS, shared across tests in this module.""" + iam_client = boto3.client("iam") + role_name = f"test-lambda-proxy-role-{short_uid()}" + trust_policy = json.dumps( + { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"Service": "lambda.amazonaws.com"}, + "Action": "sts:AssumeRole", + } + ], + } + ) + role = iam_client.create_role( + RoleName=role_name, + AssumeRolePolicyDocument=trust_policy, + Description="Test role for Lambda proxy tests", + ) + role_arn = role["Role"]["Arn"] + + # Wait for the role to be usable by Lambda (IAM eventual consistency) + def _wait_for_role(): + iam_client.get_role(RoleName=role_name) + + retry(_wait_for_role, retries=10, sleep=2) + + yield role_arn + + # cleanup + try: + iam_client.delete_role(RoleName=role_name) + except Exception as e: + LOG.warning("Failed to clean up IAM role %s: %s", role_name, e) + + +def _create_lambda_function( + lambda_client_aws, function_name, role_arn, cleanups, env_vars=None +): + """Helper to create a Lambda function in real AWS and register cleanup.""" + kwargs = { + "FunctionName": function_name, + "Runtime": LAMBDA_RUNTIME, + "Role": role_arn, + "Handler": LAMBDA_HANDLER, + "Code": {"ZipFile": _create_lambda_zip()}, + "Timeout": 30, + } + if env_vars: + kwargs["Environment"] = {"Variables": env_vars} + + # Retry create_function to handle IAM role propagation delay + def _create(): + lambda_client_aws.create_function(**kwargs) + + retry(_create, retries=15, sleep=5) + cleanups.append( + lambda: lambda_client_aws.delete_function(FunctionName=function_name) + ) + + # Wait for function to become Active + def _wait_active(): + config = lambda_client_aws.get_function_configuration( + FunctionName=function_name + ) + if config["State"] != "Active": + raise AssertionError( + f"Function {function_name} not active yet: {config['State']}" + ) + + retry(_wait_active, retries=30, sleep=2) + + +def test_lambda_requests(start_aws_proxy, cleanups, lambda_execution_role): + """Test basic Lambda proxy: create in AWS, describe/invoke via LocalStack proxy.""" + function_name_aws = f"test-fn-aws-{short_uid()}" + function_name_local = f"test-fn-local-{short_uid()}" + + # start proxy - only forwarding requests for function name matching the AWS function + config = ProxyConfig( + services={"lambda": {"resources": f".*:function:{function_name_aws}"}} + ) + start_aws_proxy(config) + + # create clients + region_name = "us-east-1" + lambda_client = connect_to(region_name=region_name).lambda_ + lambda_client_aws = boto3.client("lambda", region_name=region_name) + + # create function in real AWS + _create_lambda_function( + lambda_client_aws, function_name_aws, lambda_execution_role, cleanups + ) + + # assert that local call for GetFunction is proxied and returns AWS data + fn_local = lambda_client.get_function(FunctionName=function_name_aws) + fn_aws = lambda_client_aws.get_function(FunctionName=function_name_aws) + assert fn_local["Configuration"]["FunctionName"] == function_name_aws + assert ( + fn_local["Configuration"]["FunctionArn"] + == fn_aws["Configuration"]["FunctionArn"] + ) + assert fn_local["Configuration"]["Runtime"] == LAMBDA_RUNTIME + + # assert that GetFunctionConfiguration is also proxied + config_local = lambda_client.get_function_configuration( + FunctionName=function_name_aws + ) + config_aws = lambda_client_aws.get_function_configuration( + FunctionName=function_name_aws + ) + assert config_local["FunctionArn"] == config_aws["FunctionArn"] + assert config_local["Handler"] == LAMBDA_HANDLER + + # invoke function through proxy and verify it executes on real AWS + response_local = lambda_client.invoke( + FunctionName=function_name_aws, + Payload=json.dumps({"key": "value"}), + ) + payload_local = json.loads(response_local["Payload"].read()) + assert payload_local["statusCode"] == 200 + body = json.loads(payload_local["body"]) + assert body["message"] == "hello" + assert body["input"] == {"key": "value"} + + # invoke via AWS client directly and compare + response_aws = lambda_client_aws.invoke( + FunctionName=function_name_aws, + Payload=json.dumps({"key": "value"}), + ) + payload_aws = json.loads(response_aws["Payload"].read()) + assert payload_aws["statusCode"] == 200 + + # negative test: a non-matching function should NOT exist in AWS + with pytest.raises(ClientError) as ctx: + lambda_client_aws.get_function(FunctionName=function_name_local) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" + + +def test_lambda_read_only(start_aws_proxy, cleanups, lambda_execution_role): + """Test Lambda proxy in read-only mode: reads proxied, writes/invokes blocked.""" + function_name = f"test-fn-ro-{short_uid()}" + + # start proxy in read-only mode with wildcard resources + config = ProxyConfig(services={"lambda": {"resources": ".*", "read_only": True}}) + start_aws_proxy(config) + + # create clients + region_name = "us-east-1" + lambda_client = connect_to(region_name=region_name).lambda_ + lambda_client_aws = boto3.client("lambda", region_name=region_name) + + # create function in real AWS (direct, not through proxy) + _create_lambda_function( + lambda_client_aws, function_name, lambda_execution_role, cleanups + ) + + # read operations should be proxied + fn_local = lambda_client.get_function(FunctionName=function_name) + fn_aws = lambda_client_aws.get_function(FunctionName=function_name) + assert ( + fn_local["Configuration"]["FunctionArn"] + == fn_aws["Configuration"]["FunctionArn"] + ) + + config_local = lambda_client.get_function_configuration(FunctionName=function_name) + assert config_local["FunctionArn"] == fn_aws["Configuration"]["FunctionArn"] + + # ListFunctions should also be proxied (read operation) + def _list_contains_function(): + functions = lambda_client.list_functions()["Functions"] + func_names = [f["FunctionName"] for f in functions] + assert function_name in func_names + + retry(_list_contains_function, retries=5, sleep=2) + + # Invoke is classified as a read operation for proxy purposes + # (it executes the function but doesn't modify it) + response = lambda_client.invoke( + FunctionName=function_name, + Payload=json.dumps({"key": "value"}), + ) + payload = json.loads(response["Payload"].read()) + assert payload["statusCode"] == 200 + body = json.loads(payload["body"]) + assert body["message"] == "hello" + + # UpdateFunctionConfiguration is a write operation - should be blocked + with pytest.raises(ClientError) as ctx: + lambda_client.update_function_configuration( + FunctionName=function_name, + Description="updated via proxy - should not work", + ) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" + + +def test_lambda_resource_name_matching( + start_aws_proxy, cleanups, lambda_execution_role +): + """Test that only functions matching the resource pattern are proxied.""" + fn_match = f"proxy-fn-{short_uid()}" + fn_nomatch = f"local-fn-{short_uid()}" + + # start proxy - only forwarding requests for functions starting with "proxy-fn-" + config = ProxyConfig(services={"lambda": {"resources": ".*:function:proxy-fn-.*"}}) + start_aws_proxy(config) + + # create clients + region_name = "us-east-1" + lambda_client = connect_to(region_name=region_name).lambda_ + lambda_client_aws = boto3.client("lambda", region_name=region_name) + + # create matching function in real AWS + _create_lambda_function( + lambda_client_aws, fn_match, lambda_execution_role, cleanups + ) + + # matching function should be accessible through proxy + fn_local = lambda_client.get_function(FunctionName=fn_match) + fn_aws = lambda_client_aws.get_function(FunctionName=fn_match) + assert ( + fn_local["Configuration"]["FunctionArn"] + == fn_aws["Configuration"]["FunctionArn"] + ) + + # invoke matching function through proxy + response = lambda_client.invoke( + FunctionName=fn_match, + Payload=json.dumps({"test": True}), + ) + payload = json.loads(response["Payload"].read()) + assert payload["statusCode"] == 200 + + # non-matching function should NOT exist in AWS (negative test) + with pytest.raises(ClientError) as ctx: + lambda_client_aws.get_function(FunctionName=fn_nomatch) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" + + # GetFunction for non-matching name through local client should NOT be proxied + # (goes to LocalStack which doesn't have it either) + with pytest.raises(ClientError) as ctx: + lambda_client.get_function(FunctionName=fn_nomatch) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" From 63211db9a83f38334be51d936fe7ef74637949f1 Mon Sep 17 00:00:00 2001 From: Waldemar Hummer Date: Sat, 21 Mar 2026 20:53:54 +0100 Subject: [PATCH 2/2] introduce invoke mode for Lambda proxy; block invocations in read_only Lambda Invoke operations have side-effects and should not be treated as read operations. Removes them from _is_read_request and introduces a new 'invoke' config flag that explicitly allows invocations alongside read_only mode. Updates tests and README accordingly. Co-Authored-By: Claude Sonnet 4.6 --- aws-proxy/README.md | 3 ++ .../aws_proxy/server/aws_request_forwarder.py | 29 ++++++++--- aws-proxy/aws_proxy/shared/models.py | 3 ++ aws-proxy/tests/proxy/test_lambda.py | 50 +++++++++++++++++-- 4 files changed, 74 insertions(+), 11 deletions(-) diff --git a/aws-proxy/README.md b/aws-proxy/README.md index abe83c01..4f048d52 100644 --- a/aws-proxy/README.md +++ b/aws-proxy/README.md @@ -85,6 +85,9 @@ services: - 'Put.*' # optionally, specify that only read requests should be allowed (Get*/List*/Describe*, etc) read_only: false + # optionally, allow invoke/execute operations (e.g., Lambda invocations) alongside read_only mode. + # execute operations have side-effects and are deliberately excluded from read_only by default. + execute: false ``` Store the configuration above to a file named `proxy_config.yml`, then we can start up the proxy via: diff --git a/aws-proxy/aws_proxy/server/aws_request_forwarder.py b/aws-proxy/aws_proxy/server/aws_request_forwarder.py index 4bd06371..fc57e84e 100644 --- a/aws-proxy/aws_proxy/server/aws_request_forwarder.py +++ b/aws-proxy/aws_proxy/server/aws_request_forwarder.py @@ -88,8 +88,12 @@ def select_proxy(self, context: RequestContext) -> Optional[ProxyInstance]: # check if only read requests should be forwarded read_only = service_config.get("read_only") - if read_only and not self._is_read_request(context): - return + if read_only: + allow_execute = service_config.get("execute") + if not self._is_read_request(context) and not ( + allow_execute and self._is_execute_request(context) + ): + return # check if any operation name pattern matches operation_names = ensure_list(service_config.get("operations", [])) @@ -277,12 +281,6 @@ def _is_read_request(self, context: RequestContext) -> bool: "PartiQLSelect", }: return True - if context.service.service_name == "lambda" and operation_name in { - "Invoke", - "InvokeAsync", - "InvokeWithResponseStream", - }: - return True if context.service.service_name == "appsync" and operation_name in { "EvaluateCode", "EvaluateMappingTemplate", @@ -303,6 +301,21 @@ def _is_read_request(self, context: RequestContext) -> bool: # TODO: add more rules return False + def _is_execute_request(self, context: RequestContext) -> bool: + """ + Function to determine whether a request is an invoke/execute request. + Invoke operations have side-effects and are not considered read operations. + They can be explicitly allowed alongside read_only mode via the 'execute' config flag. + """ + operation_name = context.service_operation.operation + if context.service.service_name == "lambda" and operation_name in { + "Invoke", + "InvokeAsync", + "InvokeWithResponseStream", + }: + return True + return False + def _extract_region_from_domain(self, context: RequestContext): """ If the request domain name contains a valid region name (e.g., "us-east-2.cognito.localhost.localstack.cloud"), diff --git a/aws-proxy/aws_proxy/shared/models.py b/aws-proxy/aws_proxy/shared/models.py index 3f776271..6cc96832 100644 --- a/aws-proxy/aws_proxy/shared/models.py +++ b/aws-proxy/aws_proxy/shared/models.py @@ -11,6 +11,9 @@ class ProxyServiceConfig(TypedDict, total=False): operations: List[str] # whether only read requests should be forwarded read_only: bool + # whether invoke/execute operations (e.g., Lambda invocations) should be forwarded + # (only relevant when read_only is True, since execute has side-effects and is not a read operation) + execute: bool class ProxyConfig(TypedDict, total=False): diff --git a/aws-proxy/tests/proxy/test_lambda.py b/aws-proxy/tests/proxy/test_lambda.py index 6dc27110..c45dad9b 100644 --- a/aws-proxy/tests/proxy/test_lambda.py +++ b/aws-proxy/tests/proxy/test_lambda.py @@ -211,8 +211,52 @@ def _list_contains_function(): retry(_list_contains_function, retries=5, sleep=2) - # Invoke is classified as a read operation for proxy purposes - # (it executes the function but doesn't modify it) + # Invoke has side-effects and is NOT a read operation - should be blocked in read_only mode + with pytest.raises(ClientError) as ctx: + lambda_client.invoke( + FunctionName=function_name, + Payload=json.dumps({"key": "value"}), + ) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" + + # UpdateFunctionConfiguration is a write operation - should be blocked + with pytest.raises(ClientError) as ctx: + lambda_client.update_function_configuration( + FunctionName=function_name, + Description="updated via proxy - should not work", + ) + assert ctx.value.response["Error"]["Code"] == "ResourceNotFoundException" + + +def test_lambda_invoke_mode(start_aws_proxy, cleanups, lambda_execution_role): + """Test Lambda proxy with read_only + execute mode: reads and invocations proxied, writes blocked.""" + function_name = f"test-fn-inv-{short_uid()}" + + # start proxy with read_only + invoke flags + config = ProxyConfig( + services={"lambda": {"resources": ".*", "read_only": True, "execute": True}} + ) + start_aws_proxy(config) + + # create clients + region_name = "us-east-1" + lambda_client = connect_to(region_name=region_name).lambda_ + lambda_client_aws = boto3.client("lambda", region_name=region_name) + + # create function in real AWS (direct, not through proxy) + _create_lambda_function( + lambda_client_aws, function_name, lambda_execution_role, cleanups + ) + + # read operations should be proxied + fn_local = lambda_client.get_function(FunctionName=function_name) + fn_aws = lambda_client_aws.get_function(FunctionName=function_name) + assert ( + fn_local["Configuration"]["FunctionArn"] + == fn_aws["Configuration"]["FunctionArn"] + ) + + # Invoke should be proxied when execute: True is set alongside read_only response = lambda_client.invoke( FunctionName=function_name, Payload=json.dumps({"key": "value"}), @@ -222,7 +266,7 @@ def _list_contains_function(): body = json.loads(payload["body"]) assert body["message"] == "hello" - # UpdateFunctionConfiguration is a write operation - should be blocked + # UpdateFunctionConfiguration is a write operation - should still be blocked with pytest.raises(ClientError) as ctx: lambda_client.update_function_configuration( FunctionName=function_name,