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,107 @@
+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
+    """
+
+    @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
+    ttl: 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
+        Warning: The total size of this JSON object must not exceed 5MB.
+    denied_fields: Optional[List[str]]
+        A list of which are forcibly changed to null, even if a value was returned from a resolver.
+        Each item is either a fully qualified field ARN in the form of
+        `arn:aws:appsync:us-east-1:111122223333:apis/GraphQLApiId/types/TypeName/fields/FieldName`
+        or a short form of TypeName.FieldName. The full ARN form should be used when two APIs
+        share a lambda function authorizer and there might be ambiguity between common types
+        and fields between the two APIs.
+    """
+
+    def __init__(
+        self,
+        authorize: bool = False,
+        ttl: Optional[int] = None,
+        resolver_context: Optional[Dict[str, Any]] = None,
+        denied_fields: Optional[List[str]] = None,
+    ):
+        self._data: Dict = {"isAuthorized": authorize}
+
+        if ttl is not None:
+            self._data["ttlOverride"] = ttl
+
+        if denied_fields:
+            self._data["deniedFields"] = denied_fields
+
+        if resolver_context:
+            self._data["resolverContext"] = resolver_context
+
+    def to_dict(self) -> dict:
+        """Return the response as a dict"""
+        return self._data

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,
+        ttl=15,
+        resolver_context={"balance": 100, "name": "Foo Man"},
+        denied_fields=["Mutation.createEvent"],
+    )
+    assert expected == response.to_dict()
+
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse().to_dict()
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse(denied_fields=[]).to_dict()
+    assert {"isAuthorized": False} == AppSyncAuthorizerResponse(resolver_context={}).to_dict()
+    assert {"isAuthorized": True} == AppSyncAuthorizerResponse(authorize=True).to_dict()
+    assert {"isAuthorized": True, "ttlOverride": 0} == AppSyncAuthorizerResponse(authorize=True, ttl=0).to_dict()