fix(data-classes,parser): read VPC Lattice v2 x509 identity fields with the casing AWS sends

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -110,28 +110,28 @@ def session_name(self) -> str | None:
 
     @property
     def x509_subject_cn(self) -> str | None:
-        """The VPC Lattice v2 Event requestContext X509SubjectCn"""
-        return self.get("X509SubjectCn")
+        """The VPC Lattice v2 Event requestContext x509SubjectCn"""
+        return self.get("x509SubjectCn")
 
     @property
     def x509_issuer_ou(self) -> str | None:
-        """The VPC Lattice v2 Event requestContext X509IssuerOu"""
-        return self.get("X509IssuerOu")
+        """The VPC Lattice v2 Event requestContext x509IssuerOu"""
+        return self.get("x509IssuerOu")
 
     @property
     def x509_san_dns(self) -> str | None:
-        """The VPC Lattice v2 Event requestContext X509SanDns"""
+        """The VPC Lattice v2 Event requestContext x509SanDns"""
         return self.get("x509SanDns")
 
     @property
     def x509_san_uri(self) -> str | None:
-        """The VPC Lattice v2 Event requestContext X509SanUri"""
-        return self.get("X509SanUri")
+        """The VPC Lattice v2 Event requestContext x509SanUri"""
+        return self.get("x509SanUri")
 
     @property
     def x509_san_name_cn(self) -> str | None:
-        """The VPC Lattice v2 Event requestContext X509SanNameCn"""
-        return self.get("X509SanNameCn")
+        """The VPC Lattice v2 Event requestContext x509SanNameCn"""
+        return self.get("x509SanNameCn")
 
 
 class vpcLatticeEventV2RequestContext(DictWrapper):

aws_lambda_powertools/utilities/parser/models/vpc_latticev2.py

@@ -36,13 +36,13 @@ class VpcLatticeV2RequestContextIdentity(BaseModel):
     )
     x509_subject_cn: str | None = Field(
         None,
-        alias="X509SubjectCn",
+        alias="x509SubjectCn",
         description="The X.509 certificate subject common name.",
         examples=["example.com"],
     )
     x509_issuer_ou: str | None = Field(
         None,
-        alias="X509IssuerOu",
+        alias="x509IssuerOu",
         description="The X.509 certificate issuer organizational unit.",
         examples=["IT Department"],
     )
@@ -54,13 +54,13 @@ class VpcLatticeV2RequestContextIdentity(BaseModel):
     )
     x509_san_uri: str | None = Field(
         None,
-        alias="X509SanUri",
+        alias="x509SanUri",
         description="The X.509 certificate Subject Alternative Name URI entry.",
         examples=["https://example.com"],
     )
     x509_san_name_cn: str | None = Field(
         None,
-        alias="X509SanNameCn",
+        alias="x509SanNameCn",
         description="The X.509 certificate Subject Alternative Name common name.",
         examples=["example.com"],
     )

tests/unit/data_classes/required_dependencies/test_vpc_lattice_eventv2.py

@@ -73,3 +73,27 @@ def test_vpc_lattice_v2_event_path_excludes_query_string(event_path, expected_pa
 
     assert parsed_event.path == expected_path
     assert parsed_event.raw_event["path"] == event_path
+
+
+def test_vpc_lattice_v2_event_x509_identity():
+    # GIVEN a VPC Lattice v2 event authenticated with IAM Roles Anywhere
+    raw_event = load_event("vpcLatticeV2Event.json")
+    raw_event["requestContext"]["identity"].update(
+        {
+            "x509IssuerOu": "IT Department",
+            "x509SanDns": "example.com",
+            "x509SanNameCn": "example-name",
+            "x509SanUri": "https://example.com",
+            "x509SubjectCn": "example-subject",
+        },
+    )
+
+    # WHEN parsing the event
+    identity = VPCLatticeEventV2(raw_event).request_context.identity
+
+    # THEN all x509 fields are read from the event
+    assert identity.x509_issuer_ou == "IT Department"
+    assert identity.x509_san_dns == "example.com"
+    assert identity.x509_san_name_cn == "example-name"
+    assert identity.x509_san_uri == "https://example.com"
+    assert identity.x509_subject_cn == "example-subject"

tests/unit/parser/_pydantic/test_vpc_latticev2.py

@@ -49,6 +49,30 @@ def test_vpc_lattice_v2_event():
     assert model.request_context.identity.principal_org_id is None
 
 
+def test_vpc_lattice_v2_event_x509_identity():
+    # GIVEN a VPC Lattice v2 event authenticated with IAM Roles Anywhere
+    raw_event = load_event("vpcLatticeV2Event.json")
+    raw_event["requestContext"]["identity"].update(
+        {
+            "x509IssuerOu": "IT Department",
+            "x509SanDns": "example.com",
+            "x509SanNameCn": "example-name",
+            "x509SanUri": "https://example.com",
+            "x509SubjectCn": "example-subject",
+        },
+    )
+
+    # WHEN parsing the event
+    identity = VpcLatticeV2Model(**raw_event).request_context.identity
+
+    # THEN all x509 fields are read from the event
+    assert identity.x509_issuer_ou == "IT Department"
+    assert identity.x509_san_dns == "example.com"
+    assert identity.x509_san_name_cn == "example-name"
+    assert identity.x509_san_uri == "https://example.com"
+    assert identity.x509_subject_cn == "example-subject"
+
+
 def test_vpc_lattice_v2_event_custom_model():
     class MyCustomResource(VpcLatticeV2Model):
         body: str