add proxy tests for the CloudWatch service (metrics + logs)

- Add test file with tests for CloudWatch Metrics and CloudWatch Logs
- Fix service name mapping (monitoring -> cloudwatch) in auth_proxy and forwarder
- Use botocore service model for protocol compatibility (LocalStack uses smithy-rpc-v2-cbor, boto3 uses query protocol)
- Implement resource name matching for CloudWatch and CloudWatch Logs

Tests added:
- test_cloudwatch_metric_operations: PutMetricData and ListMetrics
- test_cloudwatch_alarm_operations: PutMetricAlarm and DescribeAlarms
- test_cloudwatch_readonly_operations (xfail): read-only mode
- test_cloudwatch_resource_name_matching (xfail): resource pattern matching
- test_logs_group_operations: CreateLogGroup and DescribeLogGroups
- test_logs_stream_and_events: log streams and events
- test_logs_readonly_operations: read-only mode for logs
- test_logs_resource_name_matching: resource pattern matching for logs
- test_logs_filter_log_events: FilterLogEvents operation

Known limitations (2 xfail tests):
- CloudWatch read_only and resource_name_matching: form data stream consumed by LocalStack before proxy can access it (Query protocol issue)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: whummer

aws-proxy/aws_proxy/client/auth_proxy.py

@@ -13,7 +13,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 +50,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 +90,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 +103,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,
         )
@@ -177,7 +185,10 @@ def _parse_aws_request(
     ) -> 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 +326,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

@@ -134,7 +134,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"
@@ -261,6 +286,9 @@ 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)