feat(data-classes): AppSync Lambda authorizer event

aws_lambda_powertools/logging/correlation_paths.py

@@ -2,6 +2,7 @@
 
 API_GATEWAY_REST = "requestContext.requestId"
 API_GATEWAY_HTTP = API_GATEWAY_REST
+APPSYNC_AUTHORIZER = "requestContext.requestId"
 APPSYNC_RESOLVER = 'request.headers."x-amzn-trace-id"'
 APPLICATION_LOAD_BALANCER = 'headers."x-amzn-trace-id"'
 EVENT_BRIDGE = "id"

aws_lambda_powertools/utilities/data_classes/appsync_authorizer_event.py

@@ -0,0 +1,116 @@
+from typing import Any, Dict, List, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class AppSyncAuthorizerEventRequestContext(DictWrapper):
+    """Request context"""
+
+    @property
+    def api_id(self) -> str:
+        """AppSync API ID"""
+        return self["requestContext"]["apiId"]
+
+    @property
+    def account_id(self) -> str:
+        """AWS Account ID"""
+        return self["requestContext"]["accountId"]
+
+    @property
+    def request_id(self) -> str:
+        """Requestt ID"""
+        return self["requestContext"]["requestId"]
+
+    @property
+    def query_string(self) -> str:
+        """GraphQL query string"""
+        return self["requestContext"]["queryString"]
+
+    @property
+    def operation_name(self) -> Optional[str]:
+        """GraphQL operation name, optional"""
+        return self["requestContext"].get("operationName")
+
+    @property
+    def variables(self) -> Dict:
+        """GraphQL variables"""
+        return self["requestContext"]["variables"]
+
+
+class AppSyncAuthorizerEvent(DictWrapper):
+    """AppSync lambda authorizer event
+
+    Documentation:
+    -------------
+    - https://aws.amazon.com/blogs/mobile/appsync-lambda-auth/
+    - https://docs.aws.amazon.com/appsync/latest/devguide/security-authz.html#aws-lambda-authorization
+    - https://docs.amplify.aws/lib/graphqlapi/authz/q/platform/js#aws-lambda
+    """
+
+    @property
+    def authorization_token(self) -> str:
+        """Authorization token"""
+        return self["authorizationToken"]
+
+    @property
+    def request_context(self) -> AppSyncAuthorizerEventRequestContext:
+        """Request context"""
+        return AppSyncAuthorizerEventRequestContext(self._data)
+
+
+class AppSyncAuthorizerResponse:
+    """AppSync Lambda authorizer response helper
+
+    Parameters
+    ----------
+    authorize: bool
+        authorize is a boolean value indicating if the value in authorizationToken
+        is authorized to make calls to the GraphQL API. If this value is
+        true, execution of the GraphQL API continues. If this value is false,
+        an UnauthorizedException is raised
+    max_age: Optional[int]
+        Set the ttlOverride. The number of seconds that the response should be
+        cached for. If no value is returned, the value from the API (if configured)
+        or the default of 300 seconds (five minutes) is used. If this is 0, the response
+        is not cached.
+    resolver_context: Optional[Dict[str, Any]]
+        A JSON object visible as `$ctx.identity.resolverContext` in resolver templates
+
+        The resolverContext object only supports key-value pairs. Nested keys are not supported.
+
+        Warning: The total size of this JSON object must not exceed 5MB.
+    deny_fields: Optional[List[str]]
+        A list of fields that will be set to `null` regardless of the resolver's return.
+
+        A field is either `TypeName.FieldName`, or an ARN such as
+        `arn:aws:appsync:us-east-1:111122223333:apis/GraphQLApiId/types/TypeName/fields/FieldName`
+
+        Use the full ARN for correctness when sharing a Lambda function authorizer between APIs.
+    """
+
+    def __init__(
+        self,
+        authorize: bool = False,
+        max_age: Optional[int] = None,
+        resolver_context: Optional[Dict[str, Any]] = None,
+        deny_fields: Optional[List[str]] = None,
+    ):
+        self.authorize = authorize
+        self.max_age = max_age
+        self.deny_fields = deny_fields
+        self.resolver_context = resolver_context
+
+    def asdict(self) -> dict:
+        """Return the response as a dict"""
+        response: Dict = {"isAuthorized": self.authorize}
+
+        if self.max_age is not None:
+            response["ttlOverride"] = self.max_age
+
+        if self.deny_fields:
+            response["deniedFields"] = self.deny_fields
+
+        if self.resolver_context:
+            response["resolverContext"] = self.resolver_context
+
+        return response

docs/utilities/data_classes.md

@@ -63,6 +63,7 @@ Event Source | Data_class
 [API Gateway Proxy V2](#api-gateway-proxy-v2) | `APIGatewayProxyEventV2`
 [Application Load Balancer](#application-load-balancer) | `ALBEvent`
 [AppSync Resolver](#appsync-resolver) | `AppSyncResolverEvent`
+[AppSync Authorizer](#appsync-authorizer) | `AppSyncAuthorizerEvent`
 [CloudWatch Logs](#cloudwatch-logs) | `CloudWatchLogsEvent`
 [CodePipeline Job Event](#codepipeline-job) | `CodePipelineJobEvent`
 [Cognito User Pool](#cognito-user-pool) | Multiple available under `cognito_user_pool_event`
@@ -128,6 +129,53 @@ Is it used for Application load balancer event.
             do_something_with(event.json_body, event.query_string_parameters)
     ```
 
+## AppSync Authorizer
+
+> New in 1.20.0
+
+Used when building an [AWS_LAMBDA Authorization](https://docs.aws.amazon.com/appsync/latest/devguide/security-authz.html#aws-lambda-authorization){target="_blank"} with AppSync.
+See blog post [Introducing Lambda authorization for AWS AppSync GraphQL APIs](https://aws.amazon.com/blogs/mobile/appsync-lambda-auth/){target="_blank"}
+or read the Amplify documentation on using [AWS Lambda for authorization](https://docs.amplify.aws/lib/graphqlapi/authz/q/platform/js#aws-lambda){target="_blank"} with AppSync.
+
+In this example extract the `requestId` as the `correlation_id` for logging, used `@event_source` decorator and builds the AppSync authorizer using the `AppSyncAuthorizerResponse` helper.
+
+=== "app.py"
+
+    ```python
+    from typing import Dict
+
+    from aws_lambda_powertools.logging import correlation_paths
+    from aws_lambda_powertools.logging.logger import Logger
+    from aws_lambda_powertools.utilities.data_classes.appsync_authorizer_event import (
+        AppSyncAuthorizerEvent,
+        AppSyncAuthorizerResponse,
+    )
+    from aws_lambda_powertools.utilities.data_classes.event_source import event_source
+
+    logger = Logger()
+
+
+    def get_user_by_token(token: str):
+        """Look a user by token"""
+
+
+    @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_AUTHORIZER)
+    @event_source(data_class=AppSyncAuthorizerEvent)
+    def lambda_handler(event: AppSyncAuthorizerEvent, context) -> Dict:
+        user = get_user_by_token(event.authorization_token)
+
+        if not user:
+            # No user found, return not authorized
+            return AppSyncAuthorizerResponse().to_dict()
+
+        return AppSyncAuthorizerResponse(
+            authorize=True,
+            resolver_context={"id": user.id},
+            # Only allow admins to delete events
+            deny_fields=None if user.is_admin else ["Mutation.deleteEvent"],
+        ).asdict()
+    ```
+
 ### AppSync Resolver
 
 > New in 1.12.0

tests/events/appSyncAuthorizerEvent.json

@@ -0,0 +1,13 @@
+{
+    "authorizationToken": "BE9DC5E3-D410-4733-AF76-70178092E681",
+    "requestContext": {
+        "apiId": "giy7kumfmvcqvbedntjwjvagii",
+        "accountId": "254688921111",
+        "requestId": "b80ed838-14c6-4500-b4c3-b694c7bef086",
+        "queryString": "mutation MyNewTask($desc: String!) {\n  createTask(description: $desc, owner: \"ccc\", taskStatus: \"cc\", title: \"ccc\") {\n    id\n  }\n}\n",
+        "operationName": "MyNewTask",
+        "variables": {
+            "desc": "Foo"
+        }
+    }
+}

tests/events/appSyncAuthorizerResponse.json

@@ -0,0 +1,9 @@
+{
+    "isAuthorized": true,
+    "resolverContext": {
+        "name": "Foo Man",
+        "balance": 100
+    },
+    "deniedFields": ["Mutation.createEvent"],
+    "ttlOverride": 15
+}

tests/functional/test_data_classes.py

@@ -30,6 +30,10 @@
     aws_timestamp,
     make_id,
 )
+from aws_lambda_powertools.utilities.data_classes.appsync_authorizer_event import (
+    AppSyncAuthorizerEvent,
+    AppSyncAuthorizerResponse,
+)
 from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import (
     AppSyncIdentityCognito,
     AppSyncIdentityIAM,
@@ -1419,3 +1423,34 @@ def lambda_handler(event: APIGatewayProxyEventV2, _):
 
     # WHEN calling the lambda handler
     lambda_handler({"headers": {"X-Foo": "Foo"}}, None)
+
+
+def test_appsync_authorizer_event():
+    event = AppSyncAuthorizerEvent(load_event("appSyncAuthorizerEvent.json"))
+
+    assert event.authorization_token == "BE9DC5E3-D410-4733-AF76-70178092E681"
+    assert event.authorization_token == event["authorizationToken"]
+    assert event.request_context.api_id == event["requestContext"]["apiId"]
+    assert event.request_context.account_id == event["requestContext"]["accountId"]
+    assert event.request_context.request_id == event["requestContext"]["requestId"]
+    assert event.request_context.query_string == event["requestContext"]["queryString"]
+    assert event.request_context.operation_name == event["requestContext"]["operationName"]
+    assert event.request_context.variables == event["requestContext"]["variables"]
+
+
+def test_appsync_authorizer_response():
+    """Check various helper functions for AppSync authorizer response"""
+    expected = load_event("appSyncAuthorizerResponse.json")
+    response = AppSyncAuthorizerResponse(
+        authorize=True,
+        max_age=15,
+        resolver_context={"balance": 100, "name": "Foo Man"},
+        deny_fields=["Mutation.createEvent"],
+    )
+    assert expected == response.asdict()
+
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse().asdict()
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse(deny_fields=[]).asdict()
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse(resolver_context={}).asdict()
+    assert {"isAuthorized": True} == AppSyncAuthorizerResponse(authorize=True).asdict()
+    assert {"isAuthorized": False, "ttlOverride": 0} == AppSyncAuthorizerResponse(max_age=0).asdict()