localstack · bentsku · Jul 1, 2024 · Jun 27, 2024 · Jun 27, 2024 · Jun 27, 2024
@@ -8,6 +8,8 @@
 from localstack.aws.api.apigateway import Method, Resource
 from localstack.services.apigateway.models import RestApiDeployment
 
+from .variables import ContextVariables, LoggingContextVariables
+
 
 class InvocationRequest(TypedDict, total=False):
     http_method: Optional[HTTPMethod]
@@ -33,55 +35,21 @@ class InvocationRequest(TypedDict, total=False):
     """Body content of the request"""
 
 
-class AuthorizerContext(TypedDict, total=False):
-    # https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
-    claims: Optional[dict[str, str]]
-    """Claims returned from the Amazon Cognito user pool after the method caller is successfully authenticated"""
-    principal_id: Optional[str]
-    """The principal user identification associated with the token sent by the client and returned from an API Gateway Lambda authorizer"""
-    context: Optional[dict[str, str]]
-    """The stringified value of the specified key-value pair of the context map returned from an API Gateway Lambda authorizer function"""
-
-
-class IdentityContext(TypedDict, total=False):
-    # https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
-    accountId: Optional[str]
-    """The AWS account ID associated with the request."""
-    apiKey: Optional[str]
-    """For API methods that require an API key, this variable is the API key associated with the method request."""
-    apiKeyId: Optional[str]
-    """The API key ID associated with an API request that requires an API key."""
-    caller: Optional[str]
-    """The principal identifier of the caller that signed the request. Supported for resources that use IAM authorization."""
-    cognitoAuthenticationProvider: Optional[str]
-    """A comma-separated list of the Amazon Cognito authentication providers used by the caller making the request"""
-    cognitoAuthenticationType: Optional[str]
-    """The Amazon Cognito authentication type of the caller making the request"""
-    cognitoIdentityId: Optional[str]
-    """The Amazon Cognito identity ID of the caller making the request"""
-    cognitoIdentityPoolId: Optional[str]
-    """The Amazon Cognito identity pool ID of the caller making the request"""
-    principalOrgId: Optional[str]
-    """The AWS organization ID."""
-    sourceIp: Optional[str]
-    """The source IP address of the immediate TCP connection making the request to the API Gateway endpoint"""
-    clientCert: Optional[dict[str, str]]
-    """Certificate that a client presents. Present only in access logs if mutual TLS authentication fails."""
-    vpcId: Optional[str]
-    """The VPC ID of the VPC making the request to the API Gateway endpoint."""
-    vpceId: Optional[str]
-    """The VPC endpoint ID of the VPC endpoint making the request to the API Gateway endpoint."""
-    user: Optional[str]
-    """The principal identifier of the user that will be authorized against resource access for resources that use IAM authorization."""
-    userAgent: Optional[str]
-    """The User-Agent header of the API caller."""
-    userArn: Optional[str]
-    """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
-
-
-class ContextVariables(TypedDict, total=False):
-    authorizer: AuthorizerContext
-    identity: IdentityContext
+class IntegrationRequest(TypedDict, total=False):
+    http_method: Optional[HTTPMethod]
+    """HTTP Method of the incoming request"""
+    uri: Optional[str]
+    """URI of the integration"""
+    query_string_parameters: Optional[dict[str, str]]
+    """Query string parameters of the request"""
+    headers: Optional[dict[str, str]]
+    """Headers of the request"""
+    multi_value_query_string_parameters: Optional[dict[str, list[str]]]
+    """Multi value query string parameters of the request"""
+    multi_value_headers: Optional[dict[str, list[str]]]
+    """Multi value headers of the request"""
+    body: Optional[bytes]
+    """Body content of the request"""
 
 
 class RestApiInvocationContext(RequestContext):
@@ -97,6 +65,8 @@ class RestApiInvocationContext(RequestContext):
     """The REST API identifier of the invoked API"""
     stage: Optional[str]
     """The REST API stage linked to this invocation"""
+    deployment_id: Optional[str]
+    """The REST API deployment linked to this invocation"""
     region: Optional[str]
     """The region the REST API is living in."""
     account_id: Optional[str]
@@ -105,17 +75,27 @@ class RestApiInvocationContext(RequestContext):
     """The resource the invocation matched"""  # TODO: verify if needed through the invocation
     resource_method: Optional[Method]
     """The method of the resource the invocation matched"""
+    stage_variables: Optional[dict[str, str]]
+    """The Stage variables, also used in parameters mapping and mapping templates"""
     context_variables: Optional[ContextVariables]
-    """Variables can be used in data models, authorizers, mapping templates, and CloudWatch access logging."""
+    """The $context used in data models, authorizers, mapping templates, and CloudWatch access logging"""
+    logging_context_variables: Optional[LoggingContextVariables]
+    """Additional $context variables available only for access logging, not yet implemented"""
+    integration_request: Optional[IntegrationRequest]
+    """Contains the data needed to construct an HTTP request to an Integration"""
 
     def __init__(self, request: Request):
         super().__init__(request)
         self.deployment = None
         self.api_id = None
         self.stage = None
+        self.deployment_id = None
         self.account_id = None
         self.region = None
         self.invocation_request = None
         self.resource = None
         self.resource_method = None
+        self.stage_variables = None
         self.context_variables = None
+        self.logging_context_variables = None
+        self.integration_request = None
@@ -1,5 +1,6 @@
 import logging
 
+from localstack.aws.api.apigateway import Integration, IntegrationType
 from localstack.http import Response
 
 from ..api import RestApiGatewayHandler, RestApiGatewayHandlerChain
@@ -20,4 +21,20 @@ def __call__(
         context: RestApiInvocationContext,
         response: Response,
     ):
+        integration: Integration = context.resource_method["methodIntegration"]
+        integration_type = integration["type"]
+
+        if integration_type in (IntegrationType.AWS_PROXY, IntegrationType.HTTP_PROXY):
+            # `PROXY` types cannot use integration mapping templates
+            # TODO: check if PROXY can still parameters mapping and substitution in URI for example?
+            # See
+            return
+
+        if integration_type == IntegrationType.MOCK:
+            # TODO: only apply partial rendering of the VTL context
+            return
+
+        # TODO: apply rendering, and attach the Integration Request needed for the Integration to construct its HTTP
+        #  request to send
+
         return
@@ -1,14 +1,21 @@
+import datetime
 import logging
 from collections import defaultdict
+from typing import Optional
 from urllib.parse import urlparse
 
 from rolo.request import Request, restore_payload
 from werkzeug.datastructures import Headers, MultiDict
 
 from localstack.http import Response
+from localstack.services.apigateway.helpers import REQUEST_TIME_DATE_FORMAT
+from localstack.utils.strings import short_uid
+from localstack.utils.time import timestamp
 
 from ..api import RestApiGatewayHandler, RestApiGatewayHandlerChain
 from ..context import InvocationRequest, RestApiInvocationContext
+from ..moto_helpers import get_stage_variables
+from ..variables import ContextVariables
 
 LOG = logging.getLogger(__name__)
 
@@ -27,6 +34,11 @@ def __call__(
     def parse_and_enrich(self, context: RestApiInvocationContext):
         # first, create the InvocationRequest with the incoming request
         context.invocation_request = self.create_invocation_request(context.request)
+        # then we can create the ContextVariables, used throughout the invocation as payload and to render authorizer
+        # payload, mapping templates and such.
+        context.context_variables = self.create_context_variables(context)
+        # then populate the stage variables
+        context.stage_variables = self.fetch_stage_variables(context)
 
     def create_invocation_request(self, request: Request) -> InvocationRequest:
         params, multi_value_params = self._get_single_and_multi_values_from_multidict(request.args)
@@ -99,3 +111,44 @@ def _get_single_and_multi_values_from_headers(
             multi_values[key] = headers.getlist(key)
 
         return single_values, multi_values
+
+    @staticmethod
+    def create_context_variables(context: RestApiInvocationContext) -> ContextVariables:
+        invocation_request: InvocationRequest = context.invocation_request
+        domain_name = invocation_request["raw_headers"].get("Host", "")
+        domain_prefix = domain_name.split(".")[0]
+        now = datetime.datetime.now()
+
+        # TODO: verify which values needs to explicitly have None set
+        context_variables = ContextVariables(
+            accountId=context.account_id,
+            apiId=context.api_id,
+            deploymentId=context.deployment_id,
+            domainName=domain_name,
+            domainPrefix=domain_prefix,
+            extendedRequestId=short_uid(),  # TODO: use snapshot tests to verify format
+            httpMethod=invocation_request["http_method"],
+            path=invocation_request[
+                "path"
+            ],  # TODO: check if we need the raw path? with forward slashes
+            protocol="HTTP/1.1",
+            requestId=short_uid(),  # TODO: use snapshot tests to verify format
+            requestTime=timestamp(time=now, format=REQUEST_TIME_DATE_FORMAT),
+            requestTimeEpoch=int(now.timestamp() * 1000),
+            stage=context.stage,
+        )
+        return context_variables
+
+    @staticmethod
+    def fetch_stage_variables(context: RestApiInvocationContext) -> Optional[dict[str, str]]:
+        stage_variables = get_stage_variables(
+            account_id=context.account_id,
+            region=context.region,
+            api_id=context.api_id,
+            stage_name=context.stage,
+        )
+        if not stage_variables:
+            # we need to set the stage variables to None in the context if we don't have at least one
+            return None
+
+        return stage_variables
@@ -18,6 +18,7 @@
 
 from ..api import RestApiGatewayHandler, RestApiGatewayHandlerChain
 from ..context import RestApiInvocationContext
+from ..variables import ContextVariables
 
 LOG = logging.getLogger(__name__)
 
@@ -84,6 +85,7 @@ def route_and_enrich(self, context: RestApiInvocationContext):
         router = self.get_router_for_deployment(context.deployment)
 
         resource, path_parameters = router.match(context)
+        resource: Resource
 
         context.invocation_request["path_parameters"] = path_parameters
         context.resource = resource
@@ -94,6 +96,16 @@ def route_and_enrich(self, context: RestApiInvocationContext):
         )
         context.resource_method = method
 
+        self.update_context_variables_with_resource(context.context_variables, resource)
+
+    @staticmethod
+    def update_context_variables_with_resource(
+        context_variables: ContextVariables, resource: Resource
+    ):
+        # TODO: log updating the context_variables?
+        context_variables["resourcePath"] = resource["path"]
+        context_variables["resourceId"] = resource["id"]
+
     @staticmethod
     @cache
     def get_router_for_deployment(deployment: RestApiDeployment) -> RestAPIResourceRouter:

@@ -1,3 +1,4 @@
+from moto.apigateway.models import APIGatewayBackend, apigateway_backends
 from moto.apigateway.models import RestAPI as MotoRestAPI
 
 from localstack.aws.api.apigateway import Resource
@@ -28,3 +29,12 @@ def get_resources_from_moto_rest_api(moto_rest_api: MotoRestAPI) -> dict[str, Re
         resources[moto_resource.id] = resource
 
     return resources
+
+
+def get_stage_variables(
+    account_id: str, region: str, api_id: str, stage_name: str
+) -> dict[str, str]:
+    apigateway_backend: APIGatewayBackend = apigateway_backends[account_id][region]
+    moto_rest_api = apigateway_backend.apis[api_id]
+    stage = moto_rest_api.stages[stage_name]
+    return stage.variables
@@ -64,6 +64,7 @@ def populate_rest_api_invocation_context(
         context.deployment = frozen_deployment
         context.api_id = api_id
         context.stage = stage
+        context.deployment_id = deployment_id
 
 
 class ApiGatewayRouter: