feat(parser): add support for sourceIp with ipv6 and port

aws_lambda_powertools/utilities/parser/functions.py

@@ -101,7 +101,15 @@ def _validate_source_ip(value):
         IPvAnyNetwork(value)
     except ValueError:
         try:
-            ip_part = value.split(":")[0]
+            # Handle IPv6 with port: [IPv6]:port
+            if value.startswith("[") and "]:" in value:
+                ip_part = value.split("]:")[0][1:]  # Remove "[" and get IP part
+            elif ":" in value and value.count(":") <= 1:
+                ip_part = value.split(":")[0]
+            else:
+                # If it"s not in IP:port format, validate as-is
+                ip_part = value
+
             IPvAnyNetwork(ip_part)
         except (ValueError, IndexError) as e:
             raise ValueError(f"Invalid IP address in sourceIp: {ip_part}") from e

tests/unit/parser/_pydantic/test_apigw.py

@@ -120,6 +120,20 @@ def test_apigw_event_and_source_ip_with_random_string():
         APIGatewayProxyEventModel(**raw_event)
 
 
+def test_apigw_event_and_source_ip_ipv6():
+    raw_event = load_event("apiGatewayProxyEvent.json")
+    raw_event["requestContext"]["identity"]["sourceIp"] = "fe80::1ff:fe23:4567:890a"
+
+    APIGatewayProxyEventModel(**raw_event)
+
+
+def test_apigw_event_and_source_ip_ipv6_with_port():
+    raw_event = load_event("apiGatewayProxyEvent.json")
+    raw_event["requestContext"]["identity"]["sourceIp"] = "[fe80::1ff:fe23:4567:890a]:12345"
+
+    APIGatewayProxyEventModel(**raw_event)
+
+
 def test_apigw_event_with_invalid_websocket_request():
     # GIVEN an event with an eventType != MESSAGE and has  a messageId
     event = {

tests/unit/parser/_pydantic/test_apigwv2.py

@@ -89,6 +89,20 @@ def test_apigw_v2_event_and_source_ip_with_random_string():
         APIGatewayProxyEventV2Model(**raw_event)
 
 
+def test_apigw_v2_event_and_source_ip_ipv6():
+    raw_event = load_event("apiGatewayProxyV2Event.json")
+    raw_event["requestContext"]["http"]["sourceIp"] = "fe80::1ff:fe23:4567:890a"
+
+    APIGatewayProxyEventV2Model(**raw_event)
+
+
+def test_apigw_v2_event_and_source_ip_ipv6_with_port():
+    raw_event = load_event("apiGatewayProxyV2Event.json")
+    raw_event["requestContext"]["http"]["sourceIp"] = "[fe80::1ff:fe23:4567:890a]:12345"
+
+    APIGatewayProxyEventV2Model(**raw_event)
+
+
 def test_api_gateway_proxy_v2_event_lambda_authorizer():
     raw_event = load_event("apiGatewayProxyV2LambdaAuthorizerEvent.json")
     parsed_event: APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model(**raw_event)