20839373803

Committed 08 Jan 2026 09:37PM UTC coverage: 86.934% (+0.02%) from 86.913%

Build # 20839373803

Build Type

push

github

Committed by

web-flow

Commit Message

fix/openapi spec (#13599)

Coverage Stats

2 of 2 new or added lines in 1 file covered. (100.0%)

25 existing lines in 4 files now uncovered.

70176 of 80723 relevant lines covered (86.93%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.14

/localstack-core/localstack/aws/handlers/validation.py

"""
Handlers for validating request and response schema against OpenAPI specs.
"""

import logging

from openapi_core import OpenAPI
from openapi_core.contrib.werkzeug import WerkzeugOpenAPIRequest, WerkzeugOpenAPIResponse
from openapi_core.exceptions import OpenAPIError
from openapi_core.validation.request.exceptions import (
    RequestValidationError,
)
from openapi_core.validation.response.exceptions import ResponseValidationError
from plux import PluginManager

from localstack import config
from localstack.aws.api import RequestContext
from localstack.aws.chain import Handler, HandlerChain
from localstack.constants import INTERNAL_RESOURCE_PATH
from localstack.http import Response

LOG = logging.getLogger(__name__)


class OpenAPIValidator(Handler):
    open_apis: list["OpenAPI"]

    def __init__(self) -> None:
        self._load_specs()

    def _load_specs(self) -> None:
        """Load the openapi spec plugins iff at least one between request and response validation is set."""
        if not (config.OPENAPI_VALIDATE_REQUEST or config.OPENAPI_VALIDATE_RESPONSE):
            return
        specs = PluginManager("localstack.openapi.spec").load_all()
        self.open_apis = []
        for spec in specs:
            self.open_apis.append(OpenAPI.from_path(spec.spec_path))


class OpenAPIRequestValidator(OpenAPIValidator):
    """
    Validates the requests to the LocalStack public endpoints (the ones with a _localstack or _aws prefix) against
    a OpenAPI specification.
    """

    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
        if not config.OPENAPI_VALIDATE_REQUEST:
            return

        hasattr(self, "open_apis") or self._load_specs()
        path = context.request.path

        if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):
            for openapi in self.open_apis:
                try:
                    openapi.validate_request(WerkzeugOpenAPIRequest(context.request))
                    # We stop the handler at the first succeeded validation, as the other spec might not even specify
                    #   this path.
                    break
                except RequestValidationError as e:
                    # Note: in this handler we only check validation errors, e.g., wrong body, missing required.
                    response.status_code = 400
                    response.set_json({"error": "Bad Request", "message": str(e)})
                    chain.stop()
                except OpenAPIError:
                    # Other errors can be raised when validating a request against the OpenAPI specification.
                    #   The most common are: ServerNotFound, OperationNotFound, or PathNotFound.
                    #   We explicitly do not check any other error but RequestValidationError ones.
                    #   We shallow the exception to avoid excessive logging (e.g., a lot of ServerNotFound), as the only
                    #   purpose of this handler is to check for request validation errors.
                    pass


class OpenAPIResponseValidator(OpenAPIValidator):
    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
        # The use of this flag is intended for test only. Eventual errors are due to LocalStack implementation and not
        #   to improper user usage of the endpoints.
        if not config.OPENAPI_VALIDATE_RESPONSE:
            return

        hasattr(self, "open_apis") or self._load_specs()
        path = context.request.path

        if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):
            for openapi in self.open_apis:
                try:
                    openapi.validate_response(
                        WerkzeugOpenAPIRequest(context.request),
                        WerkzeugOpenAPIResponse(response),
                    )
                    break
                except ResponseValidationError as exc:
                    LOG.error("Response validation failed for %s: %s", path, exc)
                    response.status_code = 500
                    response.set_json({"error": exc.__class__.__name__, "message": str(exc)})
                    chain.terminate()
                except OpenAPIError:
                    # Same logic from the request validator applies here.
                    pass

1	"""
2	Handlers for validating request and response schema against OpenAPI specs.
3	"""
4
5	import logging	1✔
6
7	from openapi_core import OpenAPI	1✔
8	from openapi_core.contrib.werkzeug import WerkzeugOpenAPIRequest, WerkzeugOpenAPIResponse	1✔
9	from openapi_core.exceptions import OpenAPIError	1✔
10	from openapi_core.validation.request.exceptions import (	1✔
11	RequestValidationError,
12	)
13	from openapi_core.validation.response.exceptions import ResponseValidationError	1✔
14	from plux import PluginManager	1✔
15
16	from localstack import config	1✔
17	from localstack.aws.api import RequestContext	1✔
18	from localstack.aws.chain import Handler, HandlerChain	1✔
19	from localstack.constants import INTERNAL_RESOURCE_PATH	1✔
20	from localstack.http import Response	1✔
21
22	LOG = logging.getLogger(__name__)	1✔
23
24
25	class OpenAPIValidator(Handler):	1✔
26	open_apis: list["OpenAPI"]	1✔
27
28	def __init__(self) -> None:	1✔
29	self._load_specs()	1✔
30
31	def _load_specs(self) -> None:	1✔
32	"""Load the openapi spec plugins iff at least one between request and response validation is set."""
33	if not (config.OPENAPI_VALIDATE_REQUEST or config.OPENAPI_VALIDATE_RESPONSE):	1✔
34	return	1✔
35	specs = PluginManager("localstack.openapi.spec").load_all()	1✔
36	self.open_apis = []	1✔
37	for spec in specs:	1✔
38	self.open_apis.append(OpenAPI.from_path(spec.spec_path))	1✔
39
40
41	class OpenAPIRequestValidator(OpenAPIValidator):	1✔
42	"""
43	Validates the requests to the LocalStack public endpoints (the ones with a _localstack or _aws prefix) against
44	a OpenAPI specification.
45	"""
46
47	def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):	1✔
48	if not config.OPENAPI_VALIDATE_REQUEST:	1✔
49	return	1✔
50
51	hasattr(self, "open_apis") or self._load_specs()	1✔
52	path = context.request.path	1✔
53
54	if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):	1✔
55	for openapi in self.open_apis:	1✔
56	try:	1✔
57	openapi.validate_request(WerkzeugOpenAPIRequest(context.request))	1✔
58	# We stop the handler at the first succeeded validation, as the other spec might not even specify
59	# this path.
60	break	1✔
61	except RequestValidationError as e:	1✔
62	# Note: in this handler we only check validation errors, e.g., wrong body, missing required.
63	response.status_code = 400	1✔
64	response.set_json({"error": "Bad Request", "message": str(e)})	1✔
65	chain.stop()	1✔
66	except OpenAPIError:	1✔
67	# Other errors can be raised when validating a request against the OpenAPI specification.
68	# The most common are: ServerNotFound, OperationNotFound, or PathNotFound.
69	# We explicitly do not check any other error but RequestValidationError ones.
70	# We shallow the exception to avoid excessive logging (e.g., a lot of ServerNotFound), as the only
71	# purpose of this handler is to check for request validation errors.
72	pass	1✔
73
74
75	class OpenAPIResponseValidator(OpenAPIValidator):	1✔
76	def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):	1✔
77	# The use of this flag is intended for test only. Eventual errors are due to LocalStack implementation and not
78	# to improper user usage of the endpoints.
79	if not config.OPENAPI_VALIDATE_RESPONSE:	1✔
80	return	1✔
81
82	hasattr(self, "open_apis") or self._load_specs()	1✔
83	path = context.request.path	1✔
84
85	if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):	1✔
86	for openapi in self.open_apis:	1✔
87	try:	1✔
88	openapi.validate_response(	1✔
89	WerkzeugOpenAPIRequest(context.request),
90	WerkzeugOpenAPIResponse(response),
91	)
92	break	1✔
UNCOV 93	except ResponseValidationError as exc:	×
94	LOG.error("Response validation failed for %s: %s", path, exc)	×
95	response.status_code = 500	×
96	response.set_json({"error": exc.__class__.__name__, "message": str(exc)})	×
97	chain.terminate()	×
UNCOV 98	except OpenAPIError:	×
99	# Same logic from the request validator applies here.
UNCOV 100	pass	×

localstack / localstack / 20839373803

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous