fix(integ-tests): Add IAM auth to API Gateway resources in integration tests (#3963)

Author: licjun

integration/combination/test_function_with_http_api.py

@@ -18,15 +18,15 @@ def test_function_with_http_api(self):
 
         stack_outputs = self.get_stack_outputs()
         base_url = stack_outputs["ApiUrl"]
-        self.verify_get_request_response(base_url + "some/path", 200)
-        self.verify_get_request_response(base_url + "something", 404)
-        self.verify_get_request_response(base_url + "another/endpoint", 404)
+        self.verify_get_request_response_sigv4(base_url + "some/path", 200)
+        self.verify_get_request_response_sigv4(base_url + "something", 404)
+        self.verify_get_request_response_sigv4(base_url + "another/endpoint", 404)
 
     def test_function_with_http_api_default_path(self):
         self.create_and_verify_stack("combination/function_with_http_api_default_path")
 
         stack_outputs = self.get_stack_outputs()
         base_url = stack_outputs["ApiUrl"]
         # The $default route catches requests that don't explicitly match other routes
-        self.verify_get_request_response(base_url, 200)
-        self.verify_get_request_response(base_url + "something", 200)
+        self.verify_get_request_response_sigv4(base_url, 200)
+        self.verify_get_request_response_sigv4(base_url + "something", 200)

integration/combination/test_function_with_implicit_http_api.py

@@ -12,6 +12,6 @@ def test_function_with_implicit_api(self):
 
         stack_outputs = self.get_stack_outputs()
         base_url = stack_outputs["ApiUrl"]
-        self.verify_get_request_response(base_url, 200)
-        self.verify_get_request_response(base_url + "something", 200)
-        self.verify_get_request_response(base_url + "another/endpoint", 200)
+        self.verify_get_request_response_sigv4(base_url, 200)
+        self.verify_get_request_response_sigv4(base_url + "something", 200)
+        self.verify_get_request_response_sigv4(base_url + "another/endpoint", 200)

integration/helpers/base_test.py

@@ -4,11 +4,14 @@
 import shutil
 from pathlib import Path
 from unittest.case import TestCase
+from urllib.parse import urlparse
 
 import boto3
 import botocore
 import pytest
 import requests
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
 from samtranslator.translator.arn_generator import ArnGenerator
 from samtranslator.yaml_helper import yaml_parse
 from tenacity import (
@@ -538,6 +541,25 @@ def verify_get_request_response(self, url, expected_status_code, headers=None):
             )
         return response
 
+    @retry(
+        stop=stop_after_attempt(6),
+        wait=wait_exponential(multiplier=1, min=16, max=64) + wait_random(0, 1),
+        retry=retry_if_exception_type(StatusCodeError),
+        after=after_log(LOG, logging.WARNING),
+        reraise=True,
+    )
+    def verify_get_request_response_sigv4(self, url, expected_status_code, headers=None):
+        """
+        Verify if a SigV4-signed get request to a certain url returns the expected status code.
+        Use this for APIs with IAM authorization.
+        """
+        response = self.do_get_request_with_sigv4(url, headers)
+        if response.status_code != expected_status_code:
+            raise StatusCodeError(
+                f"SigV4 request to {url} failed with status: {response.status_code}, expected status: {expected_status_code}"
+            )
+        return response
+
     @retry(
         stop=stop_after_attempt(6),
         wait=wait_exponential(multiplier=1, min=16, max=64) + wait_random(0, 1),
@@ -581,6 +603,22 @@ def verify_post_request(self, url: str, body_obj, expected_status_code: int, hea
             )
         return response
 
+    @retry(
+        stop=stop_after_attempt(6),
+        wait=wait_exponential(multiplier=1, min=16, max=64) + wait_random(0, 1),
+        retry=retry_if_exception_type(StatusCodeError),
+        after=after_log(LOG, logging.WARNING),
+        reraise=True,
+    )
+    def verify_post_request_sigv4(self, url: str, body_obj, expected_status_code: int, headers=None):
+        """Return response to SigV4-signed POST request and verify matches expected status code."""
+        response = self.do_post_request_with_sigv4(url, body_obj, headers)
+        if response.status_code != expected_status_code:
+            raise StatusCodeError(
+                f"SigV4 POST request to {url} failed with status: {response.status_code}, expected status: {expected_status_code}"
+            )
+        return response
+
     def get_default_test_template_parameters(self):
         """
         get the default template parameters
@@ -636,6 +674,29 @@ def do_get_request_with_logging(self, url, headers=None):
             )
         return response
 
+    def do_get_request_with_sigv4(self, url, headers=None):
+        """
+        Perform a SigV4-signed GET request to an APIGW endpoint with IAM auth.
+        """
+        parsed = urlparse(url)
+        request_headers = {"host": parsed.hostname}
+        if headers:
+            request_headers.update(headers)
+
+        aws_request = AWSRequest(method="GET", url=url, headers=request_headers)
+        session = botocore.session.Session()
+        credentials = session.get_credentials().get_frozen_credentials()
+        SigV4Auth(credentials, "execute-api", self.my_region).add_auth(aws_request)
+
+        response = requests.get(url, headers=dict(aws_request.headers))
+        amazon_headers = RequestUtils(response).get_amazon_headers()
+        if self.internal:
+            REQUEST_LOGGER.info(
+                "SigV4 request made to " + url,
+                extra={"test": self.testcase, "status": response.status_code, "headers": amazon_headers},
+            )
+        return response
+
     def do_options_request_with_logging(self, url, headers=None):
         """
         Perform a options request to an APIGW endpoint and log relevant info
@@ -669,3 +730,25 @@ def do_post_request_with_logging(self, url: str, body_obj, requestHeaders=None):
                 extra={"test": self.testcase, "status": response.status_code, "headers": amazon_headers},
             )
         return response
+
+    def do_post_request_with_sigv4(self, url: str, body_obj, headers=None):
+        """Perform a SigV4-signed POST request to an APIGW endpoint with IAM auth."""
+        parsed = urlparse(url)
+        body = json.dumps(body_obj)
+        request_headers = {"host": parsed.hostname, "content-type": "application/json"}
+        if headers:
+            request_headers.update(headers)
+
+        aws_request = AWSRequest(method="POST", url=url, headers=request_headers, data=body)
+        session = botocore.session.Session()
+        credentials = session.get_credentials().get_frozen_credentials()
+        SigV4Auth(credentials, "execute-api", self.my_region).add_auth(aws_request)
+
+        response = requests.post(url, data=body, headers=dict(aws_request.headers))
+        amazon_headers = RequestUtils(response).get_amazon_headers()
+        if self.internal:
+            REQUEST_LOGGER.info(
+                "SigV4 POST request made to " + url,
+                extra={"test": self.testcase, "status": response.status_code, "headers": amazon_headers},
+            )
+        return response

integration/resources/templates/combination/api_with_binary_media_types.yaml

@@ -17,6 +17,8 @@ Resources:
   MyApi:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       DefinitionUri: ${definitionuri}
       BinaryMediaTypes:

integration/resources/templates/combination/api_with_binary_media_types_with_definition_body.yaml

@@ -22,6 +22,8 @@ Resources:
   MyApi:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       DefinitionBody:
         # Simple HTTP Proxy API

integration/resources/templates/combination/api_with_binary_media_types_with_definition_body_openapi.yaml

@@ -26,6 +26,8 @@ Resources:
   MyApi:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       DefinitionBody:
         # Simple HTTP Proxy API

integration/resources/templates/combination/api_with_custom_domains_edge.yaml

@@ -32,6 +32,8 @@ Resources:
   MyApi:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       OpenApiVersion: 3.0.1
       StageName: Prod
       Domain:

integration/resources/templates/combination/api_with_disable_execute_api_endpoint.yaml

@@ -12,6 +12,8 @@ Resources:
   RestApiGateway:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       DisableExecuteApiEndpoint:
         Ref: DisableExecuteApiEndpointValue

integration/resources/templates/combination/api_with_disable_execute_api_endpoint_openapi_3.yaml

@@ -12,6 +12,8 @@ Resources:
   RestApiGateway:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       OpenApiVersion: 3.0
       DisableExecuteApiEndpoint:

integration/resources/templates/combination/api_with_endpoint_configuration.yaml

@@ -13,6 +13,8 @@ Resources:
   MyApi:
     Type: AWS::Serverless::Api
     Properties:
+      Auth:
+        DefaultAuthorizer: AWS_IAM
       StageName: Prod
       EndpointConfiguration: {Ref: Config}
       DefinitionUri: ${definitionuri}