diff --git a/aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py b/aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py index 7c81678f305..29694eacd97 100644 --- a/aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py +++ b/aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py @@ -1,3 +1,4 @@ +import enum import re from typing import Any, Dict, List, Optional @@ -312,7 +313,7 @@ def asdict(self) -> dict: return response -class HttpVerb: +class HttpVerb(enum.Enum): GET = "GET" POST = "POST" PUT = "PUT" @@ -386,8 +387,9 @@ def _add_route(self, effect: str, verb: str, resource: str, conditions: List[Dic """Adds a route to the internal lists of allowed or denied routes. Each object in the internal list contains a resource ARN and a condition statement. The condition statement can be null.""" - if verb != "*" and not hasattr(HttpVerb, verb): - raise ValueError(f"Invalid HTTP verb {verb}. Allowed verbs in HttpVerb class") + if verb != "*" and verb not in HttpVerb.__members__: + allowed_values = [verb.value for verb in HttpVerb] + raise ValueError(f"Invalid HTTP verb: '{verb}'. Use either '{allowed_values}'") resource_pattern = re.compile(self.path_regex) if not resource_pattern.match(resource): @@ -433,13 +435,24 @@ def _get_statement_for_effect(self, effect: str, methods: List) -> List: return statements - def allow_all_routes(self): - """Adds a '*' allow to the policy to authorize access to all methods of an API""" - self._add_route("Allow", HttpVerb.ALL, "*", []) + def allow_all_routes(self, http_method: str = HttpVerb.ALL.value): + """Adds a '*' allow to the policy to authorize access to all methods of an API - def deny_all_routes(self): - """Adds a '*' allow to the policy to deny access to all methods of an API""" - self._add_route("Deny", HttpVerb.ALL, "*", []) + Parameters + ---------- + http_method: str + """ + self._add_route(effect="Allow", verb=http_method, resource="*", conditions=[]) + + def deny_all_routes(self, http_method: str = HttpVerb.ALL.value): + """Adds a '*' allow to the policy to deny access to all methods of an API + + Parameters + ---------- + http_method: str + """ + + self._add_route(effect="Deny", verb=http_method, resource="*", conditions=[]) def allow_route(self, http_method: str, resource: str, conditions: Optional[List[Dict]] = None): """Adds an API Gateway method (Http verb + Resource path) to the list of allowed @@ -447,7 +460,8 @@ def allow_route(self, http_method: str, resource: str, conditions: Optional[List Optionally includes a condition for the policy statement. More on AWS policy conditions here: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition""" - self._add_route("Allow", http_method, resource, conditions or []) + conditions = conditions or [] + self._add_route(effect="Allow", verb=http_method, resource=resource, conditions=conditions) def deny_route(self, http_method: str, resource: str, conditions: Optional[List[Dict]] = None): """Adds an API Gateway method (Http verb + Resource path) to the list of denied @@ -455,7 +469,8 @@ def deny_route(self, http_method: str, resource: str, conditions: Optional[List[ Optionally includes a condition for the policy statement. More on AWS policy conditions here: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition""" - self._add_route("Deny", http_method, resource, conditions or []) + conditions = conditions or [] + self._add_route(effect="Deny", verb=http_method, resource=resource, conditions=conditions) def asdict(self) -> Dict[str, Any]: """Generates the policy document based on the internal lists of allowed and denied diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md index c3a8b5a415f..6a3492bda35 100644 --- a/docs/utilities/data_classes.md +++ b/docs/utilities/data_classes.md @@ -87,19 +87,17 @@ Event Source | Data_class > New in 1.20.0 -It is used for API Gateway Rest API lambda authorizer payload. See docs on -[Use API Gateway Lambda authorizers](https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-use-lambda-authorizer.html){target="_blank"} -for more details. Use `APIGatewayAuthorizerRequestEvent` for type "REQUEST" and `APIGatewayAuthorizerTokenEvent` for -type "TOKEN". +It is used for [API Gateway Rest API Lambda Authorizer payload](https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-use-lambda-authorizer.html){target="_blank"}. -Below is 2 examples of a Rest API lambda authorizer. One looking up user details by `Authorization` header and using -`APIGatewayAuthorizerResponse` to return the declined response when user is not found or authorized and include -the user details in the request context and full access for admin users. And another using -`APIGatewayAuthorizerTokenEvent` to get the `authorization_token`. +Use **`APIGatewayAuthorizerRequestEvent`** for type `REQUEST` and **`APIGatewayAuthorizerTokenEvent`** for type `TOKEN`. === "app_type_request.py" - ```python + This example uses the `APIGatewayAuthorizerResponse` to decline a given request if the user is not found. + + When the user is found, it includes the user details in the request context that will be available to the back-end, and returns a full access policy for admin users. + + ```python hl_lines="2-5 26-31 36-37 40 44 46" from aws_lambda_powertools.utilities.data_classes import event_source from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import ( APIGatewayAuthorizerRequestEvent, @@ -125,27 +123,33 @@ the user details in the request context and full access for admin users. And ano # parse the `methodArn` as an `APIGatewayRouteArn` arn = event.parsed_arn # Create the response builder from parts of the `methodArn` - builder = APIGatewayAuthorizerResponse("user", arn.region, arn.aws_account_id, arn.api_id, arn.stage) + policy = APIGatewayAuthorizerResponse( + principal_id="user", + region=arn.region, + aws_account_id=arn.aws_account_id, + api_id=arn.api_id, + stage=arn.stage + ) if user is None: # No user was found, so we return not authorized - builder.deny_all_routes() - return builder.asdict() + policy.deny_all_routes() + return policy.asdict() # Found the user and setting the details in the context - builder.context = user + policy.context = user # Conditional IAM Policy if user.get("isAdmin", False): - builder.allow_all_routes() + policy.allow_all_routes() else: - builder.allow_route(HttpVerb.GET, "/user-profile") + policy.allow_route(HttpVerb.GET, "/user-profile") - return builder.asdict() + return policy.asdict() ``` === "app_type_token.py" - ```python + ```python hl_lines="2-5 12-18 21 23-24" from aws_lambda_powertools.utilities.data_classes import event_source from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import ( APIGatewayAuthorizerTokenEvent, @@ -156,30 +160,34 @@ the user details in the request context and full access for admin users. And ano @event_source(data_class=APIGatewayAuthorizerTokenEvent) def handler(event: APIGatewayAuthorizerTokenEvent, context): arn = event.parsed_arn - builder = APIGatewayAuthorizerResponse("user", arn.region, arn.aws_account_id, arn.api_id, arn.stage) + + policy = APIGatewayAuthorizerResponse( + principal_id="user", + region=arn.region, + aws_account_id=arn.aws_account_id, + api_id=arn.api_id, + stage=arn.stage + ) + if event.authorization_token == "42": - builder.allow_all_methods() + policy.allow_all_routes() else: - builder.deny_all_methods() - return builder.asdict() + policy.deny_all_routes() + return policy.asdict() ``` ### API Gateway Authorizer V2 > New in 1.20.0 -It is used for API Gateway HTTP API lambda authorizer payload version 2. See blog post -[Introducing IAM and Lambda authorizers for Amazon API Gateway HTTP APIs](https://aws.amazon.com/blogs/compute/introducing-iam-and-lambda-authorizers-for-amazon-api-gateway-http-apis/){target="_blank"} -or [Working with AWS Lambda authorizers for HTTP APIs](https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html){target="_blank"} -for more details - -Below is a simple example of an HTTP API lambda authorizer looking up user details by `x-token` header and using -`APIGatewayAuthorizerResponseV2` to return the declined response when user is not found or authorized and include -the user details in the request context. +It is used for [API Gateway HTTP API Lambda Authorizer payload version 2](https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html){target="_blank"}. +See also [this blog post](https://aws.amazon.com/blogs/compute/introducing-iam-and-lambda-authorizers-for-amazon-api-gateway-http-apis/){target="_blank"} for more details. === "app.py" - ```python + This example looks up user details via `x-token` header. It uses `APIGatewayAuthorizerResponseV2` to return a deny policy when user is not found or authorized. + + ```python hl_lines="2-5 21 24" from aws_lambda_powertools.utilities.data_classes import event_source from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import ( APIGatewayAuthorizerEventV2, diff --git a/tests/functional/data_classes/test_api_gateway_authorizer.py b/tests/functional/data_classes/test_api_gateway_authorizer.py index 5310b28e634..7dac6cb7791 100644 --- a/tests/functional/data_classes/test_api_gateway_authorizer.py +++ b/tests/functional/data_classes/test_api_gateway_authorizer.py @@ -22,23 +22,17 @@ def test_authorizer_response_no_statement(builder: APIGatewayAuthorizerResponse) def test_authorizer_response_invalid_verb(builder: APIGatewayAuthorizerResponse): - with pytest.raises(ValueError) as ex: + with pytest.raises(ValueError, match="Invalid HTTP verb: 'INVALID'"): # GIVEN a invalid http_method # WHEN calling deny_method - builder.deny_route("INVALID", "foo") - - # THEN raise a name error for invalid http verb - assert str(ex.value) == "Invalid HTTP verb INVALID. Allowed verbs in HttpVerb class" + builder.deny_route(http_method="INVALID", resource="foo") def test_authorizer_response_invalid_resource(builder: APIGatewayAuthorizerResponse): - with pytest.raises(ValueError) as ex: + with pytest.raises(ValueError, match="Invalid resource path: \$."): # noqa: W605 # GIVEN a invalid resource path "$" # WHEN calling deny_method - builder.deny_route(HttpVerb.GET, "$") - - # THEN raise a name error for invalid resource - assert "Invalid resource path: $" in str(ex.value) + builder.deny_route(http_method=HttpVerb.GET.value, resource="$") def test_authorizer_response_allow_all_routes_with_context(): @@ -78,7 +72,7 @@ def test_authorizer_response_deny_all_routes(builder: APIGatewayAuthorizerRespon def test_authorizer_response_allow_route(builder: APIGatewayAuthorizerResponse): - builder.allow_route(HttpVerb.GET, "/foo") + builder.allow_route(http_method=HttpVerb.GET.value, resource="/foo") assert builder.asdict() == { "policyDocument": { "Version": "2012-10-17", @@ -95,7 +89,7 @@ def test_authorizer_response_allow_route(builder: APIGatewayAuthorizerResponse): def test_authorizer_response_deny_route(builder: APIGatewayAuthorizerResponse): - builder.deny_route(HttpVerb.PUT, "foo") + builder.deny_route(http_method=HttpVerb.PUT.value, resource="foo") assert builder.asdict() == { "principalId": "foo", "policyDocument": { @@ -112,12 +106,11 @@ def test_authorizer_response_deny_route(builder: APIGatewayAuthorizerResponse): def test_authorizer_response_allow_route_with_conditions(builder: APIGatewayAuthorizerResponse): + condition = {"StringEquals": {"method.request.header.Content-Type": "text/html"}} builder.allow_route( - HttpVerb.POST, - "/foo", - [ - {"StringEquals": {"method.request.header.Content-Type": "text/html"}}, - ], + http_method=HttpVerb.POST.value, + resource="/foo", + conditions=[condition], ) assert builder.asdict() == { "principalId": "foo", @@ -136,13 +129,8 @@ def test_authorizer_response_allow_route_with_conditions(builder: APIGatewayAuth def test_authorizer_response_deny_route_with_conditions(builder: APIGatewayAuthorizerResponse): - builder.deny_route( - HttpVerb.POST, - "/foo", - [ - {"StringEquals": {"method.request.header.Content-Type": "application/json"}}, - ], - ) + condition = {"StringEquals": {"method.request.header.Content-Type": "application/json"}} + builder.deny_route(http_method=HttpVerb.POST.value, resource="/foo", conditions=[condition]) assert builder.asdict() == { "principalId": "foo", "policyDocument": {