Skip to content

docs(data-classes): make authorizer concise; use enum #630

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 21, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import enum
import re
from typing import Any, Dict, List, Optional

Expand Down Expand Up @@ -312,7 +313,7 @@ def asdict(self) -> dict:
return response


class HttpVerb:
class HttpVerb(enum.Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -433,29 +435,42 @@ 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
methods for the policy.

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
methods for the policy.

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
Expand Down
68 changes: 38 additions & 30 deletions docs/utilities/data_classes.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down
36 changes: 12 additions & 24 deletions tests/functional/data_classes/test_api_gateway_authorizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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",
Expand All @@ -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": {
Expand All @@ -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",
Expand All @@ -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": {
Expand Down