Skip to content

Commit d04d30f

Browse files
fix(event-handler): prevent OpenAPI schema bleed when reusing response dictionaries (#7952)
* fix(event-handler): prevent OpenAPI schema bleed when reusing response dictionaries Fixes #7711 When multiple routes shared the same response dictionary object, the OpenAPI schema generator was mutating the shared dictionary by directly modifying it. This caused schema bleeding where one route's return type would incorrectly appear in another route's OpenAPI schema. The fix uses copy.deepcopy() to create independent copies of response dictionaries before mutation, ensuring each route gets its own correct OpenAPI schema based on its return type annotation. * test(event-handler): add tests for OpenAPI schema bleed fix Relates to #7711 Add comprehensive tests to verify that when multiple routes share the same response dictionary, each route gets its own correct OpenAPI schema without bleeding return types between routes. Tests cover: - Different return types (list vs single object) with shared responses - Verification that shared dictionaries are not mutated - Regression testing for standard behavior * moving additional tests --------- Co-authored-by: Leandro Damascena <lcdama@amazon.pt>
1 parent 8761fe7 commit d04d30f

File tree

2 files changed

+174
-1
lines changed

2 files changed

+174
-1
lines changed

aws_lambda_powertools/event_handler/api_gateway.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
from __future__ import annotations
22

33
import base64
4+
import copy
45
import json
56
import logging
67
import re
@@ -666,7 +667,8 @@ def _get_openapi_path( # noqa PLR0912
666667
# Add the response to the OpenAPI operation
667668
if self.responses:
668669
for status_code in list(self.responses):
669-
response = self.responses[status_code]
670+
# Create a deep copy to prevent mutation of the shared dictionary
671+
response = copy.deepcopy(self.responses[status_code])
670672

671673
# Case 1: there is not 'content' key
672674
if "content" not in response:
Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,171 @@
1+
from typing import Dict, List
2+
3+
from pydantic import BaseModel
4+
5+
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
6+
from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse
7+
8+
9+
class ExamSummary(BaseModel):
10+
"""Summary information about an exam"""
11+
12+
id: str
13+
name: str
14+
duration_minutes: int
15+
16+
17+
class ExamConfig(BaseModel):
18+
"""Detailed configuration for an exam"""
19+
20+
id: str
21+
name: str
22+
duration_minutes: int
23+
max_attempts: int
24+
passing_score: int
25+
26+
27+
class Responses:
28+
"""Pre-configured OpenAPI response schemas."""
29+
30+
# Base responses
31+
OK = {200: OpenAPIResponse(description="Successful operation")}
32+
NOT_FOUND = {404: OpenAPIResponse(description="Resource not found")}
33+
VALIDATION_ERROR = {422: OpenAPIResponse(description="Validation error")}
34+
SERVER_ERROR = {500: OpenAPIResponse(description="Internal server error")}
35+
36+
# Common combinations
37+
STANDARD_ERRORS = {**NOT_FOUND, **VALIDATION_ERROR, **SERVER_ERROR}
38+
39+
@classmethod
40+
def combine(cls, *response_dicts: Dict[int, OpenAPIResponse]) -> Dict[int, OpenAPIResponse]:
41+
"""Combine multiple response dictionaries."""
42+
result = {}
43+
for response_dict in response_dicts:
44+
result.update(response_dict)
45+
return result
46+
47+
48+
def test_openapi_shared_response_no_bleed():
49+
"""
50+
Test that when reusing the same response dictionary across multiple routes,
51+
each route gets the correct return type in its OpenAPI schema.
52+
53+
This reproduces bug #7711 where the schema from one route bleeds into another
54+
when they share the same response dictionary object.
55+
"""
56+
app = APIGatewayRestResolver(enable_validation=True)
57+
58+
@app.get(
59+
"/exams",
60+
tags=["Exams"],
61+
responses=Responses.combine(Responses.OK, Responses.STANDARD_ERRORS),
62+
)
63+
def list_exams() -> Response[List[ExamSummary]]:
64+
"""Lists all available exams."""
65+
return Response(
66+
status_code=200,
67+
body=[
68+
ExamSummary(id="1", name="Math", duration_minutes=60),
69+
ExamSummary(id="2", name="Science", duration_minutes=90),
70+
],
71+
)
72+
73+
@app.get(
74+
"/exams/<exam_id>",
75+
tags=["Exams"],
76+
responses=Responses.combine(Responses.OK, Responses.STANDARD_ERRORS), # Reusing the shared Responses.OK
77+
)
78+
def get_exam_config(exam_id: str) -> Response[ExamConfig]:
79+
"""Get the configuration for a specific exam"""
80+
return Response(
81+
status_code=200,
82+
body=ExamConfig(
83+
id=exam_id,
84+
name="Math",
85+
duration_minutes=60,
86+
max_attempts=3,
87+
passing_score=70,
88+
),
89+
)
90+
91+
# Generate the OpenAPI schema
92+
schema = app.get_openapi_schema()
93+
94+
# Verify /exams endpoint has the correct list[ExamSummary] schema
95+
exams_response = schema.paths["/exams"].get.responses[200]
96+
exams_schema = exams_response.content["application/json"].schema_
97+
98+
# The schema should be an array type
99+
assert exams_schema.type == "array", f"/exams should return an array, got {exams_schema.type}"
100+
assert exams_schema.items is not None, "/exams should have items definition"
101+
102+
# The items should reference ExamSummary
103+
if hasattr(exams_schema.items, "ref"):
104+
assert "ExamSummary" in exams_schema.items.ref, (
105+
f"/exams should return list[ExamSummary], got {exams_schema.items.ref}"
106+
)
107+
elif hasattr(exams_schema.items, "title"):
108+
assert exams_schema.items.title == "ExamSummary", (
109+
f"/exams should return list[ExamSummary], got {exams_schema.items.title}"
110+
)
111+
112+
# Verify /exams/{exam_id} endpoint has the correct ExamConfig schema
113+
exam_detail_response = schema.paths["/exams/{exam_id}"].get.responses[200]
114+
exam_detail_schema = exam_detail_response.content["application/json"].schema_
115+
116+
# The schema should NOT be an array - it should be an object
117+
assert exam_detail_schema.type != "array", "/exams/{exam_id} should not return an array (bug #7711 - schema bleed)"
118+
119+
# The schema should reference ExamConfig
120+
if hasattr(exam_detail_schema, "ref"):
121+
assert "ExamConfig" in exam_detail_schema.ref, (
122+
f"/exams/{{exam_id}} should return ExamConfig, got {exam_detail_schema.ref}"
123+
)
124+
elif hasattr(exam_detail_schema, "title"):
125+
assert exam_detail_schema.title == "ExamConfig", (
126+
f"/exams/{{exam_id}} should return ExamConfig, got {exam_detail_schema.title}"
127+
)
128+
129+
130+
def test_openapi_shared_response_dict_not_mutated():
131+
"""
132+
Test that the original shared response dictionary is not mutated
133+
when generating OpenAPI schemas.
134+
"""
135+
app = APIGatewayRestResolver(enable_validation=True)
136+
137+
# Create a shared response dictionary
138+
shared_responses = Responses.combine(Responses.OK, Responses.STANDARD_ERRORS)
139+
140+
# Store the original state - the 200 response should not have 'content' key
141+
original_200_response = shared_responses[200].copy()
142+
assert "content" not in original_200_response, "200 response should not have content initially"
143+
144+
@app.get("/route1", responses=shared_responses)
145+
def route1() -> Response[ExamSummary]:
146+
return Response(
147+
status_code=200,
148+
body=ExamSummary(id="1", name="Test", duration_minutes=60),
149+
)
150+
151+
@app.get("/route2", responses=shared_responses)
152+
def route2() -> Response[ExamConfig]:
153+
return Response(
154+
status_code=200,
155+
body=ExamConfig(
156+
id="1",
157+
name="Test",
158+
duration_minutes=60,
159+
max_attempts=3,
160+
passing_score=70,
161+
),
162+
)
163+
164+
# Generate the OpenAPI schema
165+
app.get_openapi_schema()
166+
167+
# Verify the shared dictionary was not mutated
168+
# The original 200 response should still not have 'content' key
169+
assert "content" not in shared_responses[200], (
170+
"Shared response dictionary should not be mutated during OpenAPI schema generation (bug #7711)"
171+
)

0 commit comments

Comments
 (0)