feat(event_handler): add support to VPC Lattice payload v2 (aws-powertools#3153)

Co-authored-by: Leandro Damascena <[email protected]>

Author: stephenbawks

aws_lambda_powertools/event_handler/__init__.py

@@ -14,7 +14,7 @@
 from aws_lambda_powertools.event_handler.lambda_function_url import (
     LambdaFunctionUrlResolver,
 )
-from aws_lambda_powertools.event_handler.vpc_lattice import VPCLatticeResolver
+from aws_lambda_powertools.event_handler.vpc_lattice import VPCLatticeResolver, VPCLatticeV2Resolver
 
 __all__ = [
     "AppSyncResolver",
@@ -26,4 +26,5 @@
     "LambdaFunctionUrlResolver",
     "Response",
     "VPCLatticeResolver",
+    "VPCLatticeV2Resolver",
 ]

aws_lambda_powertools/event_handler/api_gateway.py

@@ -22,6 +22,7 @@
     APIGatewayProxyEventV2,
     LambdaFunctionUrlEvent,
     VPCLatticeEvent,
+    VPCLatticeEventV2,
 )
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -43,6 +44,7 @@ class ProxyEventType(Enum):
     APIGatewayProxyEventV2 = "APIGatewayProxyEventV2"
     ALBEvent = "ALBEvent"
     VPCLatticeEvent = "VPCLatticeEvent"
+    VPCLatticeEventV2 = "VPCLatticeEventV2"
     LambdaFunctionUrlEvent = "LambdaFunctionUrlEvent"
 
 
@@ -999,6 +1001,9 @@ def _to_proxy_event(self, event: Dict) -> BaseProxyEvent:
         if self._proxy_type == ProxyEventType.VPCLatticeEvent:
             logger.debug("Converting event to VPC Lattice contract")
             return VPCLatticeEvent(event)
+        if self._proxy_type == ProxyEventType.VPCLatticeEventV2:
+            logger.debug("Converting event to VPC LatticeV2 contract")
+            return VPCLatticeEventV2(event)
         logger.debug("Converting event to ALB contract")
         return ALBEvent(event)
 

aws_lambda_powertools/event_handler/vpc_lattice.py

@@ -5,7 +5,7 @@
     ApiGatewayResolver,
     ProxyEventType,
 )
-from aws_lambda_powertools.utilities.data_classes import VPCLatticeEvent
+from aws_lambda_powertools.utilities.data_classes import VPCLatticeEvent, VPCLatticeEventV2
 
 
 class VPCLatticeResolver(ApiGatewayResolver):
@@ -51,3 +51,48 @@ def __init__(
     ):
         """Amazon VPC Lattice resolver"""
         super().__init__(ProxyEventType.VPCLatticeEvent, cors, debug, serializer, strip_prefixes)
+
+
+class VPCLatticeV2Resolver(ApiGatewayResolver):
+    """VPC Lattice resolver
+
+    Documentation:
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-vpc-lattice.html
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-vpc-lattice.html#vpc-lattice-receiving-events
+
+    Examples
+    --------
+    Simple example integrating with Tracer
+
+    ```python
+    from aws_lambda_powertools import Tracer
+    from aws_lambda_powertools.event_handler import VPCLatticeV2Resolver
+
+    tracer = Tracer()
+    app = VPCLatticeV2Resolver()
+
+    @app.get("/get-call")
+    def simple_get():
+        return {"message": "Foo"}
+
+    @app.post("/post-call")
+    def simple_post():
+        post_data: dict = app.current_event.json_body
+        return {"message": post_data}
+
+    @tracer.capture_lambda_handler
+    def lambda_handler(event, context):
+        return app.resolve(event, context)
+    """
+
+    current_event: VPCLatticeEventV2
+
+    def __init__(
+        self,
+        cors: Optional[CORSConfig] = None,
+        debug: Optional[bool] = None,
+        serializer: Optional[Callable[[Dict], str]] = None,
+        strip_prefixes: Optional[List[Union[str, Pattern]]] = None,
+    ):
+        """Amazon VPC Lattice resolver"""
+        super().__init__(ProxyEventType.VPCLatticeEventV2, cors, debug, serializer, strip_prefixes)

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -27,7 +27,7 @@
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
-from .vpc_lattice import VPCLatticeEvent
+from .vpc_lattice import VPCLatticeEvent, VPCLatticeEventV2
 
 __all__ = [
     "APIGatewayProxyEvent",
@@ -56,4 +56,5 @@
     "event_source",
     "AWSConfigRuleEvent",
     "VPCLatticeEvent",
+    "VPCLatticeEventV2",
 ]

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -4,15 +4,15 @@
     BaseHeadersSerializer,
     HttpApiHeadersSerializer,
 )
-from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
+from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
 from aws_lambda_powertools.utilities.data_classes.shared_functions import (
     base64_decode,
     get_header_value,
     get_query_string_value,
 )
 
 
-class VPCLatticeEvent(BaseProxyEvent):
+class VPCLatticeEventBase(BaseProxyEvent):
     @property
     def body(self) -> str:
         """The VPC Lattice body."""
@@ -30,11 +30,6 @@ def headers(self) -> Dict[str, str]:
         """The VPC Lattice event headers."""
         return self["headers"]
 
-    @property
-    def is_base64_encoded(self) -> bool:
-        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
-        return self["is_base64_encoded"]
-
     @property
     def decoded_body(self) -> str:
         """Dynamically base64 decode body as a str"""
@@ -48,24 +43,6 @@ def method(self) -> str:
         """The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
         return self["method"]
 
-    @property
-    def query_string_parameters(self) -> Dict[str, str]:
-        """The request query string parameters."""
-        return self["query_string_parameters"]
-
-    @property
-    def raw_path(self) -> str:
-        """The raw VPC Lattice request path."""
-        return self["raw_path"]
-
-    # VPCLattice event has no path field
-    # Added here for consistency with the BaseProxyEvent class
-    @property
-    def path(self) -> str:
-        return self["raw_path"]
-
-    # VPCLattice event has no http_method field
-    # Added here for consistency with the BaseProxyEvent class
     @property
     def http_method(self) -> str:
         """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
@@ -140,3 +117,137 @@ def get_header_value(
     def header_serializer(self) -> BaseHeadersSerializer:
         # When using the VPC Lattice integration, we have multiple HTTP Headers.
         return HttpApiHeadersSerializer()
+
+
+class VPCLatticeEvent(VPCLatticeEventBase):
+    @property
+    def raw_path(self) -> str:
+        """The raw VPC Lattice request path."""
+        return self["raw_path"]
+
+    @property
+    def is_base64_encoded(self) -> bool:
+        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
+        return self["is_base64_encoded"]
+
+    # VPCLattice event has no path field
+    # Added here for consistency with the BaseProxyEvent class
+    @property
+    def path(self) -> str:
+        return self["raw_path"]
+
+    @property
+    def query_string_parameters(self) -> Dict[str, str]:
+        """The request query string parameters."""
+        return self["query_string_parameters"]
+
+
+class vpcLatticeEventV2Identity(DictWrapper):
+    @property
+    def source_vpc_arn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
+        return self.get("sourceVpcArn")
+
+    @property
+    def get_type(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext Identity type"""
+        return self.get("type")
+
+    @property
+    def principal(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext principal"""
+        return self.get("principal")
+
+    @property
+    def principal_org_id(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext principalOrgID"""
+        return self.get("principalOrgID")
+
+    @property
+    def session_name(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext sessionName"""
+        return self.get("sessionName")
+
+    @property
+    def x509_subject_cn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SubjectCn"""
+        return self.get("X509SubjectCn")
+
+    @property
+    def x509_issuer_ou(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509IssuerOu"""
+        return self.get("X509IssuerOu")
+
+    @property
+    def x509_san_dns(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanDns"""
+        return self.get("x509SanDns")
+
+    @property
+    def x509_san_uri(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanUri"""
+        return self.get("X509SanUri")
+
+    @property
+    def x509_san_name_cn(self) -> Optional[str]:
+        """The VPC Lattice v2 Event requestContext X509SanNameCn"""
+        return self.get("X509SanNameCn")
+
+
+class vpcLatticeEventV2RequestContext(DictWrapper):
+    @property
+    def service_network_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
+        return self["serviceNetworkArn"]
+
+    @property
+    def service_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceArn"""
+        return self["serviceArn"]
+
+    @property
+    def target_group_arn(self) -> str:
+        """The VPC Lattice v2 Event requestContext targetGroupArn"""
+        return self["targetGroupArn"]
+
+    @property
+    def identity(self) -> vpcLatticeEventV2Identity:
+        """The VPC Lattice v2 Event requestContext identity"""
+        return vpcLatticeEventV2Identity(self["identity"])
+
+    @property
+    def region(self) -> str:
+        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
+        return self["region"]
+
+    @property
+    def time_epoch(self) -> float:
+        """The VPC Lattice v2 Event requestContext timeEpoch"""
+        return self["timeEpoch"]
+
+
+class VPCLatticeEventV2(VPCLatticeEventBase):
+    @property
+    def version(self) -> str:
+        """The VPC Lattice v2 Event version"""
+        return self["version"]
+
+    @property
+    def is_base64_encoded(self) -> Optional[bool]:
+        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
+        return self.get("isBase64Encoded")
+
+    @property
+    def path(self) -> str:
+        """The VPC Lattice v2 Event path"""
+        return self["path"]
+
+    @property
+    def request_context(self) -> vpcLatticeEventV2RequestContext:
+        """he VPC Lattice v2 Event request context."""
+        return vpcLatticeEventV2RequestContext(self["requestContext"])
+
+    @property
+    def query_string_parameters(self) -> Optional[Dict[str, str]]:
+        """The request query string parameters."""
+        return self.get("queryStringParameters")

aws_lambda_powertools/utilities/parser/envelopes/__init__.py

@@ -11,6 +11,7 @@
 from .sns import SnsEnvelope, SnsSqsEnvelope
 from .sqs import SqsEnvelope
 from .vpc_lattice import VpcLatticeEnvelope
+from .vpc_latticev2 import VpcLatticeV2Envelope
 
 __all__ = [
     "ApiGatewayEnvelope",
@@ -27,4 +28,5 @@
     "KafkaEnvelope",
     "BaseEnvelope",
     "VpcLatticeEnvelope",
+    "VpcLatticeV2Envelope",
 ]

aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py

@@ -0,0 +1,32 @@
+import logging
+from typing import Any, Dict, Optional, Type, Union
+
+from ..models import VpcLatticeV2Model
+from ..types import Model
+from .base import BaseEnvelope
+
+logger = logging.getLogger(__name__)
+
+
+class VpcLatticeV2Envelope(BaseEnvelope):
+    """Amazon VPC Lattice envelope to extract data within body key"""
+
+    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+        """Parses data found with model provided
+
+        Parameters
+        ----------
+        data : Dict
+            Lambda event to be parsed
+        model : Type[Model]
+            Data model provided to parse after extracting data using envelope
+
+        Returns
+        -------
+        Optional[Model]
+            Parsed detail payload with model provided
+        """
+        logger.debug(f"Parsing incoming data with VPC Lattice V2 model {VpcLatticeV2Model}")
+        parsed_envelope: VpcLatticeV2Model = VpcLatticeV2Model.parse_obj(data)
+        logger.debug(f"Parsing event payload in `detail` with {model}")
+        return self._parse(data=parsed_envelope.body, model=model)

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -89,6 +89,7 @@
 from .sns import SnsModel, SnsNotificationModel, SnsRecordModel
 from .sqs import SqsAttributesModel, SqsModel, SqsMsgAttributeModel, SqsRecordModel
 from .vpc_lattice import VpcLatticeModel
+from .vpc_latticev2 import VpcLatticeV2Model
 
 __all__ = [
     "APIGatewayProxyEventV2Model",
@@ -163,4 +164,5 @@
     "CloudFormationCustomResourceCreateModel",
     "CloudFormationCustomResourceBaseModel",
     "VpcLatticeModel",
+    "VpcLatticeV2Model",
 ]