Add CloudWatch proxy tests

aws-proxy/AGENTS.md

@@ -38,3 +38,9 @@ When adding new integration tests, consider the following:
 * Make sure to either use fixtures (preferred), or reliable cleanups for removing the resources; several fixtures for creating AWS resources are available in the `localstack.testing.pytest.fixtures` module
 * If a test uses multiple resources with interdependencies (e.g., an SQS queue connected to an SNS topic), then the test needs to ensure that both resource types are proxied (i.e., created in real AWS), to avoid a situation where a resource in AWS is attempting to reference a local resource in LocalStack (using account ID `000000000000` in their ARN).
 * When waiting for the creation status of a resource, use the `localstack.utils.sync.retry(..)` utility function, rather than a manual `for` loop.
+
+## Fixing or Enhancing Logic in the Proxy
+
+Notes:
+* The AWS proxy is running as a LocalStack Extension, and the tests are currently set up in a way that they assume the container to be running with the Extension in dev mode. Hence, in order to make actual changes to the proxy logic, we'll need to restart the LocalStack main container. You can either ask me (the user) to restart the container whenever you're making changes in the core logic, or alternatively remove the `localstack-main` container, and then run `EXTENSION_DEV_MODE=1 DEBUG=1 localstack start -d` again to restart the container, which may reveal some error logs, stack traces, etc.
+* If the proxy raises errors or something seems off, you can grab and parse the output of the LocalStack container via `localstack logs`.

aws-proxy/aws_proxy/client/auth_proxy.py

@@ -1,3 +1,4 @@
+# Note/disclosure: This file has been partially modified by an AI agent.
 import json
 import logging
 import os
@@ -13,7 +14,6 @@
 from botocore.awsrequest import AWSPreparedRequest
 from botocore.model import OperationModel
 from localstack import config as localstack_config
-from localstack.aws.spec import load_service
 from localstack.config import external_service_url
 from localstack.constants import (
     AWS_REGION_US_EAST_1,
@@ -51,6 +51,11 @@
 if localstack_config.DEBUG:
     LOG.setLevel(logging.DEBUG)
 
+# Mapping from AWS service signing names to boto3 client names
+SERVICE_NAME_MAPPING = {
+    "monitoring": "cloudwatch",
+}
+
 # TODO make configurable
 CLI_PIP_PACKAGE = "localstack-extension-aws-proxy"
 # note: enable the line below temporarily for testing:
@@ -86,6 +91,8 @@ def proxy_request(self, request: Request, data: bytes) -> Response:
         if not parsed:
             return requests_response("", status_code=400)
         region_name, service_name = parsed
+        # Map AWS signing names to boto3 client names
+        service_name = SERVICE_NAME_MAPPING.get(service_name, service_name)
         query_string = to_str(request.query_string or "")
 
         LOG.debug(
@@ -97,10 +104,12 @@ def proxy_request(self, request: Request, data: bytes) -> Response:
             query_string,
         )
 
+        # Convert Quart headers to a dict for the LocalStack Request
+        headers_dict = dict(request.headers)
         request = Request(
             body=data,
             method=request.method,
-            headers=request.headers,
+            headers=headers_dict,
             path=request.path,
             query_string=query_string,
         )
@@ -172,12 +181,28 @@ def register_in_instance(self):
             )
             raise
 
+    def deregister_from_instance(self):
+        """Deregister this proxy from the LocalStack instance."""
+        port = getattr(self, "port", None)
+        if not port:
+            return
+        url = f"{external_service_url()}{HANDLER_PATH_PROXIES}/{port}"
+        LOG.debug("Deregistering proxy from main container via: %s", url)
+        try:
+            response = requests.delete(url)
+            return response
+        except Exception as e:
+            LOG.debug("Unable to deregister auth proxy: %s", e)
+
     def _parse_aws_request(
         self, request: Request, service_name: str, region_name: str, client
     ) -> Tuple[OperationModel, AWSPreparedRequest, Dict]:
         from localstack.aws.protocol.parser import create_parser
 
-        parser = create_parser(load_service(service_name))
+        # Use botocore's service model to ensure protocol compatibility
+        # (LocalStack's load_service may return newer protocol versions that don't match the client)
+        service_model = self._get_botocore_service_model(service_name)
+        parser = create_parser(service_model)
         operation_model, parsed_request = parser.parse(request)
         request_context = {
             "client_region": region_name,
@@ -315,6 +340,22 @@ def _query_account_id_from_aws(self) -> str:
         result = sts_client.get_caller_identity()
         return result["Account"]
 
+    @staticmethod
+    @cache
+    def _get_botocore_service_model(service_name: str):
+        """
+        Get the botocore service model for a service. This is used instead of LocalStack's
+        load_service() to ensure protocol compatibility, as LocalStack may use newer protocol
+        versions (e.g., smithy-rpc-v2-cbor) while clients use older protocols (e.g., query).
+        """
+        import botocore.session
+        from botocore.model import ServiceModel
+
+        session = botocore.session.get_session()
+        loader = session.get_component("data_loader")
+        api_data = loader.load_service_model(service_name, "service-2")
+        return ServiceModel(api_data)
+
 
 def start_aws_auth_proxy(config: ProxyConfig, port: int = None) -> AuthProxyAWS:
     setup_logging()

aws-proxy/aws_proxy/server/aws_request_forwarder.py

@@ -1,9 +1,12 @@
+# Note/disclosure: This file has been partially modified by an AI agent.
 import json
 import logging
 import re
 from typing import Dict, Optional
+from urllib.parse import urlencode
 
 import requests
+from botocore.serialize import create_serializer
 from localstack.aws.api import RequestContext
 from localstack.aws.chain import Handler, HandlerChain
 from localstack.constants import APPLICATION_JSON, LOCALHOST, LOCALHOST_HOSTNAME
@@ -134,7 +137,32 @@ def _request_matches_resource(
                     secret_id, account_id=context.account_id, region_name=context.region
                 )
                 return bool(re.match(resource_name_pattern, secret_arn))
-            # TODO: add more resource patterns
+            if service_name == "cloudwatch":
+                # CloudWatch alarm ARN format: arn:aws:cloudwatch:{region}:{account}:alarm:{alarm_name}
+                alarm_name = context.service_request.get("AlarmName") or ""
+                alarm_names = context.service_request.get("AlarmNames") or []
+                if alarm_name:
+                    alarm_names = [alarm_name]
+                if alarm_names:
+                    for name in alarm_names:
+                        alarm_arn = f"arn:aws:cloudwatch:{context.region}:{context.account_id}:alarm:{name}"
+                        if re.match(resource_name_pattern, alarm_arn):
+                            return True
+                    return False
+                # For metric operations without alarm names, 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 ""
+                log_group_prefix = (
+                    context.service_request.get("logGroupNamePrefix") or ""
+                )
+                name = log_group_name or log_group_prefix
+                if name:
+                    log_group_arn = f"arn:aws:logs:{context.region}:{context.account_id}:log-group:{name}:*"
+                    return bool(re.match(resource_name_pattern, log_group_arn))
+                # No log group name specified - check if pattern is generic
+                return bool(re.match(resource_name_pattern, ".*"))
         except re.error as e:
             raise Exception(
                 "Error evaluating regular expression - please verify proxy configuration"
@@ -168,6 +196,12 @@ def forward_request(
                 data = request.form
             elif request.data:
                 data = request.data
+
+            # Fallback: if data is empty and we have parsed service_request,
+            # reconstruct the request body (handles cases where form data was consumed)
+            if not data and context.service_request:
+                data = self._reconstruct_request_body(context, ctype)
+
             LOG.debug(
                 "Forward request: %s %s - %s - %s",
                 request.method,
@@ -256,6 +290,35 @@ def _get_resource_names(cls, service_config: ProxyServiceConfig) -> list[str]:
 
     @classmethod
     def _get_canonical_service_name(cls, service_name: str) -> str:
-        if service_name == "sqs-query":
-            return "sqs"
-        return service_name
+        # Map internal/signing service names to boto3 client names
+        mapping = {
+            "sqs-query": "sqs",
+            "monitoring": "cloudwatch",
+        }
+        return mapping.get(service_name, service_name)
+
+    def _reconstruct_request_body(
+        self, context: RequestContext, content_type: str
+    ) -> bytes:
+        """
+        Reconstruct the request body from the parsed service_request.
+        This is used when the original request body was consumed during parsing.
+        """
+        try:
+            protocol = context.service.protocol
+            if protocol == "query" or "x-www-form-urlencoded" in (content_type or ""):
+                # For Query protocol, serialize using botocore serializer
+                serializer = create_serializer(protocol)
+                operation_model = context.operation
+                serialized = serializer.serialize_to_request(
+                    context.service_request, operation_model
+                )
+                body = serialized.get("body", {})
+                if isinstance(body, dict):
+                    return urlencode(body, doseq=True)
+                return body
+            elif protocol == "json" or protocol == "rest-json":
+                return json.dumps(context.service_request)
+        except Exception as e:
+            LOG.debug("Failed to reconstruct request body: %s", e)
+        return b""

aws-proxy/aws_proxy/server/request_handler.py

@@ -1,3 +1,4 @@
+# Note/disclosure: This file has been partially modified by an AI agent.
 import json
 import logging
 import os.path
@@ -43,6 +44,11 @@ def add_proxy(self, request: Request, **kwargs):
         result = handle_proxies_request(req)
         return result or {}
 
+    @route(f"{HANDLER_PATH_PROXIES}/<int:port>", methods=["DELETE"])
+    def delete_proxy(self, request: Request, port: int, **kwargs):
+        removed = AwsProxyHandler.PROXY_INSTANCES.pop(port, None)
+        return {"removed": removed is not None}
+
     @route(f"{HANDLER_PATH_PROXIES}/status", methods=["GET"])
     def get_status(self, request: Request, **kwargs):
         containers = get_proxy_containers()

aws-proxy/tests/conftest.py

@@ -1,3 +1,4 @@
+# Note/disclosure: This file has been partially modified by an AI agent.
 import os
 
 import pytest
@@ -51,4 +52,6 @@ def _start(config: dict = None):
     yield _start
 
     for proxy in proxies:
+        # Deregister from LocalStack instance before shutting down
+        proxy.deregister_from_instance()
         proxy.shutdown()