feat(data-classes): authorizer for http api and rest api

aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py

@@ -0,0 +1,300 @@
+from typing import Any, Dict, List, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import (
+    BaseRequestContext,
+    BaseRequestContextV2,
+    DictWrapper,
+    get_header_value,
+)
+
+
+class APIGatewayRouteArn:
+    """A parsed route arn"""
+
+    def __init__(
+        self,
+        region: str,
+        aws_account_id: str,
+        api_id: str,
+        stage: str,
+        http_method: str,
+        resource: str,
+    ):
+        self.partition = "aws"
+        self.region = region
+        self.aws_account_id = aws_account_id
+        self.api_id = api_id
+        self.stage = stage
+        self.http_method = http_method
+        self.resource = resource
+
+    @property
+    def arn(self) -> str:
+        """Build an arn from it's parts
+        eg: arn:aws:execute-api:us-east-1:123456789012:abcdef123/test/GET/request"""
+        return (
+            f"arn:{self.partition}:execute-api:{self.region}:{self.aws_account_id}:{self.api_id}/{self.stage}/"
+            f"{self.http_method}/{self.resource}"
+        )
+
+
+def parse_api_gateway_arn(arn: str) -> APIGatewayRouteArn:
+    arn_parts = arn.split(":")
+    api_gateway_arn_parts = arn_parts[5].split("/")
+    return APIGatewayRouteArn(
+        region=arn_parts[3],
+        aws_account_id=arn_parts[4],
+        api_id=api_gateway_arn_parts[0],
+        stage=api_gateway_arn_parts[1],
+        http_method=api_gateway_arn_parts[2],
+        resource=api_gateway_arn_parts[3] if len(api_gateway_arn_parts) == 4 else "",
+    )
+
+
+class APIGatewayAuthorizerV2Event(DictWrapper):
+    """API Gateway Authorizer Event Format 2.0
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html
+    """
+
+    @property
+    def version(self) -> str:
+        """Event payload version should always be 2.0"""
+        return self["version"]
+
+    @property
+    def get_type(self) -> str:
+        """Event type should always be request"""
+        return self["type"]
+
+    @property
+    def route_arn(self) -> str:
+        """ARN of the route being called
+
+        eg: arn:aws:execute-api:us-east-1:123456789012:abcdef123/test/GET/request"""
+        return self["routeArn"]
+
+    @property
+    def parsed_arn(self) -> APIGatewayRouteArn:
+        """Convenient property to return a parsed api gateway route arn"""
+        return parse_api_gateway_arn(self.route_arn)
+
+    @property
+    def identity_source(self) -> Optional[List[str]]:
+        """The identity source for which authorization is requested.
+
+        For a REQUEST authorizer, this is optional. The value is a set of one or more mapping expressions of the
+        specified request parameters. The identity source can be headers, query string parameters, stage variables,
+        and context parameters.
+        """
+        return self.get("identitySource")
+
+    @property
+    def route_key(self) -> str:
+        """The route key for the route. For HTTP APIs, the route key can be either $default,
+        or a combination of an HTTP method and resource path, for example, GET /pets."""
+        return self["routeKey"]
+
+    @property
+    def raw_path(self) -> str:
+        return self["rawPath"]
+
+    @property
+    def raw_query_string(self) -> str:
+        return self["rawQueryString"]
+
+    @property
+    def cookies(self) -> List[str]:
+        return self["cookies"]
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        return self["headers"]
+
+    @property
+    def query_string_parameters(self) -> Dict[str, str]:
+        return self["queryStringParameters"]
+
+    @property
+    def request_context(self) -> BaseRequestContextV2:
+        return BaseRequestContextV2(self._data)
+
+    @property
+    def path_parameters(self) -> Optional[Dict[str, str]]:
+        return self.get("pathParameters")
+
+    @property
+    def stage_variables(self) -> Optional[Dict[str, str]]:
+        return self.get("stageVariables")
+
+    def get_header_value(
+        self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False
+    ) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        case_sensitive: bool
+            Whether to use a case sensitive look up
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return get_header_value(self.headers, name, default_value, case_sensitive)
+
+
+class APIGatewayAuthorizerTokenEvent(DictWrapper):
+    """API Gateway Authorizer Token Event Format 1.0
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-use-lambda-authorizer.html
+    """
+
+    @property
+    def get_type(self) -> str:
+        return self["type"]
+
+    @property
+    def authorization_token(self) -> str:
+        return self["authorizationToken"]
+
+    @property
+    def method_arn(self) -> str:
+        """ARN of the incoming method request and is populated by API Gateway in accordance with the Lambda authorizer
+        configuration"""
+        return self["methodArn"]
+
+    @property
+    def parsed_arn(self) -> APIGatewayRouteArn:
+        """Convenient property to return a parsed api gateway method arn"""
+        return parse_api_gateway_arn(self.method_arn)
+
+
+class APIGatewayAuthorizerRequestEvent(DictWrapper):
+    """API Gateway Authorizer Request Event Format 1.0
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-use-lambda-authorizer.html
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html
+    """
+
+    @property
+    def version(self) -> str:
+        return self["version"]
+
+    @property
+    def get_type(self) -> str:
+        return self["type"]
+
+    @property
+    def method_arn(self) -> str:
+        return self["methodArn"]
+
+    @property
+    def parsed_arn(self) -> APIGatewayRouteArn:
+        return parse_api_gateway_arn(self.method_arn)
+
+    @property
+    def identity_source(self) -> str:
+        return self["identitySource"]
+
+    @property
+    def authorization_token(self) -> str:
+        return self["authorizationToken"]
+
+    @property
+    def resource(self) -> str:
+        return self["resource"]
+
+    @property
+    def path(self) -> str:
+        return self["path"]
+
+    @property
+    def http_method(self) -> str:
+        return self["httpMethod"]
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        return self["headers"]
+
+    @property
+    def query_string_parameters(self) -> Dict[str, str]:
+        return self["queryStringParameters"]
+
+    @property
+    def path_parameters(self) -> Dict[str, str]:
+        return self["pathParameters"]
+
+    @property
+    def stage_variables(self) -> Dict[str, str]:
+        return self["stageVariables"]
+
+    @property
+    def request_context(self) -> BaseRequestContext:
+        return BaseRequestContext(self._data)
+
+    def get_header_value(
+        self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False
+    ) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        case_sensitive: bool
+            Whether to use a case sensitive look up
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return get_header_value(self.headers, name, default_value, case_sensitive)
+
+
+class APIGatewayAuthorizerSimpleResponse:
+    """Api Gateway HTTP API V2 payload authorizer simple response helper
+
+    Parameters
+    ----------
+    authorize: bool
+        authorize is a boolean value indicating if the value in authorizationToken
+        is authorized to make calls to the GraphQL API. If this value is
+        true, execution of the GraphQL API continues. If this value is false,
+        an UnauthorizedException is raised
+    context: Dict[str, Any], optional
+        A JSON object visible as `event.requestContext.authorizer` lambda event
+
+        The context object only supports key-value pairs. Nested keys are not supported.
+
+        Warning: The total size of this JSON object must not exceed 5MB.
+    """
+
+    def __init__(
+        self,
+        authorize: bool = False,
+        context: Optional[Dict[str, Any]] = None,
+    ):
+        self.authorize = authorize
+        self.context = context
+
+    def asdict(self) -> dict:
+        """Return the response as a dict"""
+        response: Dict = {"isAuthorized": self.authorize}
+
+        if self.context:
+            response["context"] = self.context
+
+        return response