feat(event_handler): support Header parameter validation in OpenAPI schema

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -81,9 +81,22 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) ->
             query_string,
         )
 
+        # Normalize header values before validate this
+        headers = _normalize_multi_header_values_with_param(
+            app.current_event.resolved_headers_field,
+            route.dependant.header_params,
+        )
+
+        # Process header values
+        header_values, header_errors = _request_params_to_args(
+            route.dependant.header_params,
+            headers,
+        )
+
         values.update(path_values)
         values.update(query_values)
-        errors += path_errors + query_errors
+        values.update(header_values)
+        errors += path_errors + query_errors + header_errors
 
         # Process the request body, if it exists
         if route.dependant.body_params:
@@ -243,12 +256,14 @@ def _request_params_to_args(
     errors = []
 
     for field in required_params:
-        value = received_params.get(field.alias)
-
         field_info = field.field_info
+
+        # To ensure early failure, we check if it's not an instance of Param.
         if not isinstance(field_info, Param):
             raise AssertionError(f"Expected Param field_info, got {field_info}")
 
+        value = received_params.get(field.alias)
+
         loc = (field_info.in_.value, field.alias)
 
         # If we don't have a value, see if it's required or has a default
@@ -377,3 +392,30 @@ def _normalize_multi_query_string_with_param(query_string: Optional[Dict[str, st
             except KeyError:
                 pass
     return query_string
+
+
+def _normalize_multi_header_values_with_param(headers: Optional[Dict[str, str]], params: Sequence[ModelField]):
+    """
+    Extract and normalize resolved_headers_field
+
+    Parameters
+    ----------
+    headers: Dict
+        A dictionary containing the initial header parameters.
+    params: Sequence[ModelField]
+        A sequence of ModelField objects representing parameters.
+
+    Returns
+    -------
+    A dictionary containing the processed headers.
+    """
+    if headers:
+        for param in filter(is_scalar_field, params):
+            try:
+                if len(headers[param.alias]) == 1:
+                    # if the target parameter is a scalar and the list contains only 1 element
+                    # we keep the first value of the headers regardless if there are more in the payload
+                    headers[param.alias] = headers[param.alias][0]
+            except KeyError:
+                pass
+    return headers

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -14,12 +14,12 @@
 from aws_lambda_powertools.event_handler.openapi.params import (
     Body,
     Dependant,
+    Header,
     Param,
     ParamTypes,
     Query,
     _File,
     _Form,
-    _Header,
     analyze_param,
     create_response_field,
     get_flat_dependant,
@@ -59,16 +59,21 @@ def add_param_to_fields(
 
     """
     field_info = cast(Param, field.field_info)
-    if field_info.in_ == ParamTypes.path:
-        dependant.path_params.append(field)
-    elif field_info.in_ == ParamTypes.query:
-        dependant.query_params.append(field)
-    elif field_info.in_ == ParamTypes.header:
-        dependant.header_params.append(field)
+
+    # Dictionary to map ParamTypes to their corresponding lists in dependant
+    param_type_map = {
+        ParamTypes.path: dependant.path_params,
+        ParamTypes.query: dependant.query_params,
+        ParamTypes.header: dependant.header_params,
+        ParamTypes.cookie: dependant.cookie_params,
+    }
+
+    # Check if field_info.in_ is a valid key in param_type_map and append the field to the corresponding list
+    # or raise an exception if it's not a valid key.
+    if field_info.in_ in param_type_map:
+        param_type_map[field_info.in_].append(field)
     else:
-        if field_info.in_ != ParamTypes.cookie:
-            raise AssertionError(f"Unsupported param type: {field_info.in_}")
-        dependant.cookie_params.append(field)
+        raise AssertionError(f"Unsupported param type: {field_info.in_}")
 
 
 def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
@@ -265,7 +270,7 @@ def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
         return False
     elif is_scalar_field(field=param_field):
         return False
-    elif isinstance(param_field.field_info, (Query, _Header)) and is_scalar_sequence_field(param_field):
+    elif isinstance(param_field.field_info, (Query, Header)) and is_scalar_sequence_field(param_field):
         return False
     else:
         if not isinstance(param_field.field_info, Body):

aws_lambda_powertools/event_handler/openapi/params.py

@@ -486,7 +486,7 @@ def __init__(
         )
 
 
-class _Header(Param):
+class Header(Param):
     """
     A class used internally to represent a header parameter in a path operation.
     """
@@ -527,12 +527,75 @@ def __init__(
         json_schema_extra: Union[Dict[str, Any], None] = None,
         **extra: Any,
     ):
+        """
+        Constructs a new Query param.
+
+        Parameters
+        ----------
+        default: Any
+            The default value of the parameter
+        default_factory: Callable[[], Any], optional
+            Callable that will be called when a default value is needed for this field
+        annotation: Any, optional
+            The type annotation of the parameter
+        alias: str, optional
+            The public name of the field
+        alias_priority: int, optional
+            Priority of the alias. This affects whether an alias generator is used
+        validation_alias: str | AliasPath | AliasChoices | None, optional
+            Alias to be used for validation only
+        serialization_alias: str | AliasPath | AliasChoices | None, optional
+            Alias to be used for serialization only
+        convert_underscores: bool
+            If true convert "_" to "-"
+            See RFC: https://www.rfc-editor.org/rfc/rfc9110.html#name-field-name-registry
+        title: str, optional
+            The title of the parameter
+        description: str, optional
+            The description of the parameter
+        gt: float, optional
+            Only applies to numbers, required the field to be "greater than"
+        ge: float, optional
+            Only applies to numbers, required the field to be "greater than or equal"
+        lt: float, optional
+            Only applies to numbers, required the field to be "less than"
+        le: float, optional
+            Only applies to numbers, required the field to be "less than or equal"
+        min_length: int, optional
+            Only applies to strings, required the field to have a minimum length
+        max_length: int, optional
+            Only applies to strings, required the field to have a maximum length
+        pattern: str, optional
+            Only applies to strings, requires the field match against a regular expression pattern string
+        discriminator: str, optional
+            Parameter field name for discriminating the type in a tagged union
+        strict: bool, optional
+            Enables Pydantic's strict mode for the field
+        multiple_of: float, optional
+            Only applies to numbers, requires the field to be a multiple of the given value
+        allow_inf_nan: bool, optional
+            Only applies to numbers, requires the field to allow infinity and NaN values
+        max_digits: int, optional
+            Only applies to Decimals, requires the field to have a maxmium number of digits within the decimal.
+        decimal_places: int, optional
+            Only applies to Decimals, requires the field to have at most a number of decimal places
+        examples: List[Any], optional
+            A list of examples for the parameter
+        deprecated: bool, optional
+            If `True`, the parameter will be marked as deprecated
+        include_in_schema: bool, optional
+            If `False`, the parameter will be excluded from the generated OpenAPI schema
+        json_schema_extra: Dict[str, Any], optional
+            Extra values to include in the generated OpenAPI schema
+        """
         self.convert_underscores = convert_underscores
+        self._alias = alias
+
         super().__init__(
             default=default,
             default_factory=default_factory,
             annotation=annotation,
-            alias=alias,
+            alias=self._alias,
             alias_priority=alias_priority,
             validation_alias=validation_alias,
             serialization_alias=serialization_alias,
@@ -558,6 +621,18 @@ def __init__(
             **extra,
         )
 
+    @property
+    def alias(self):
+        return self._alias
+
+    @alias.setter
+    def alias(self, value: Optional[str] = None):
+        if value is not None:
+            # Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the parameter name
+            # This ensures that customers can access headers with any casing, as per the RFC guidelines.
+            # Reference: https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2
+            self._alias = value.lower()
+
 
 class Body(FieldInfo):
     """

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -42,6 +42,17 @@ def resolved_query_string_parameters(self) -> Optional[Dict[str, Any]]:
 
         return self.query_string_parameters
 
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        headers: Dict[str, Any] = {}
+
+        if self.multi_value_headers:
+            headers = self.multi_value_headers
+        else:
+            headers = self.headers
+
+        return {key.lower(): value for key, value in headers.items()}
+
     @property
     def multi_value_headers(self) -> Optional[Dict[str, List[str]]]:
         return self.get("multiValueHeaders")

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -125,6 +125,17 @@ def resolved_query_string_parameters(self) -> Optional[Dict[str, Any]]:
 
         return self.query_string_parameters
 
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        headers: Dict[str, Any] = {}
+
+        if self.multi_value_headers:
+            headers = self.multi_value_headers
+        else:
+            headers = self.headers
+
+        return {key.lower(): value for key, value in headers.items()}
+
     @property
     def request_context(self) -> APIGatewayEventRequestContext:
         return APIGatewayEventRequestContext(self._data)
@@ -316,3 +327,11 @@ def resolved_query_string_parameters(self) -> Optional[Dict[str, Any]]:
             return query_string
 
         return {}
+
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        if self.headers is not None:
+            headers = {key.lower(): value.split(",") if "," in value else value for key, value in self.headers.items()}
+            return headers
+
+        return {}

aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
 
@@ -112,3 +112,7 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
     @property
     def resolved_query_string_parameters(self) -> Optional[Dict[str, str]]:
         return self.query_string_parameters
+
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        return {}

aws_lambda_powertools/utilities/data_classes/common.py

@@ -114,6 +114,21 @@ def resolved_query_string_parameters(self) -> Optional[Dict[str, str]]:
         """
         return self.query_string_parameters
 
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        """
+        This property determines the appropriate header to be used
+        as a trusted source for validating OpenAPI.
+
+        This is necessary because different resolvers use different formats to encode
+        headers parameters.
+
+        Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the header name
+        This ensures that customers can access headers with any casing, as per the RFC guidelines.
+        Reference: https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2
+        """
+        return self.headers
+
     @property
     def is_base64_encoded(self) -> Optional[bool]:
         return self.get("isBase64Encoded")

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -145,6 +145,14 @@ def query_string_parameters(self) -> Dict[str, str]:
     def resolved_query_string_parameters(self) -> Optional[Dict[str, str]]:
         return self.query_string_parameters
 
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+        if self.headers is not None:
+            headers = {key.lower(): value.split(",") if "," in value else value for key, value in self.headers.items()}
+            return headers
+
+        return {}
+
 
 class vpcLatticeEventV2Identity(DictWrapper):
     @property
@@ -259,3 +267,10 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
     @property
     def resolved_query_string_parameters(self) -> Optional[Dict[str, str]]:
         return self.query_string_parameters
+
+    @property
+    def resolved_headers_field(self) -> Optional[Dict[str, str]]:
+        if self.headers is not None:
+            return {key.lower(): value for key, value in self.headers.items()}
+
+        return {}