File tree Expand file tree Collapse file tree 3 files changed +37
-1
lines changed
aws_lambda_powertools/utilities/parser
tests/unit/parser/_pydantic Expand file tree Collapse file tree 3 files changed +37
-1
lines changed Original file line number Diff line number Diff line change @@ -101,7 +101,15 @@ def _validate_source_ip(value):
101
101
IPvAnyNetwork (value )
102
102
except ValueError :
103
103
try :
104
- ip_part = value .split (":" )[0 ]
104
+ # Handle IPv6 with port: [IPv6]:port
105
+ if value .startswith ("[" ) and "]:" in value :
106
+ ip_part = value .split ("]:" )[0 ][1 :] # Remove "[" and get IP part
107
+ elif ":" in value and value .count (":" ) <= 1 :
108
+ ip_part = value .split (":" )[0 ]
109
+ else :
110
+ # If it"s not in IP:port format, validate as-is
111
+ ip_part = value
112
+
105
113
IPvAnyNetwork (ip_part )
106
114
except (ValueError , IndexError ) as e :
107
115
raise ValueError (f"Invalid IP address in sourceIp: { ip_part } " ) from e
Original file line number Diff line number Diff line change @@ -120,6 +120,20 @@ def test_apigw_event_and_source_ip_with_random_string():
120
120
APIGatewayProxyEventModel (** raw_event )
121
121
122
122
123
+ def test_apigw_event_and_source_ip_ipv6 ():
124
+ raw_event = load_event ("apiGatewayProxyEvent.json" )
125
+ raw_event ["requestContext" ]["identity" ]["sourceIp" ] = "fe80::1ff:fe23:4567:890a"
126
+
127
+ APIGatewayProxyEventModel (** raw_event )
128
+
129
+
130
+ def test_apigw_event_and_source_ip_ipv6_with_port ():
131
+ raw_event = load_event ("apiGatewayProxyEvent.json" )
132
+ raw_event ["requestContext" ]["identity" ]["sourceIp" ] = "[fe80::1ff:fe23:4567:890a]:12345"
133
+
134
+ APIGatewayProxyEventModel (** raw_event )
135
+
136
+
123
137
def test_apigw_event_with_invalid_websocket_request ():
124
138
# GIVEN an event with an eventType != MESSAGE and has a messageId
125
139
event = {
Original file line number Diff line number Diff line change @@ -89,6 +89,20 @@ def test_apigw_v2_event_and_source_ip_with_random_string():
89
89
APIGatewayProxyEventV2Model (** raw_event )
90
90
91
91
92
+ def test_apigw_v2_event_and_source_ip_ipv6 ():
93
+ raw_event = load_event ("apiGatewayProxyV2Event.json" )
94
+ raw_event ["requestContext" ]["http" ]["sourceIp" ] = "fe80::1ff:fe23:4567:890a"
95
+
96
+ APIGatewayProxyEventV2Model (** raw_event )
97
+
98
+
99
+ def test_apigw_v2_event_and_source_ip_ipv6_with_port ():
100
+ raw_event = load_event ("apiGatewayProxyV2Event.json" )
101
+ raw_event ["requestContext" ]["http" ]["sourceIp" ] = "[fe80::1ff:fe23:4567:890a]:12345"
102
+
103
+ APIGatewayProxyEventV2Model (** raw_event )
104
+
105
+
92
106
def test_api_gateway_proxy_v2_event_lambda_authorizer ():
93
107
raw_event = load_event ("apiGatewayProxyV2LambdaAuthorizerEvent.json" )
94
108
parsed_event : APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model (** raw_event )
You can’t perform that action at this time.
0 commit comments