Starlette: [Question] Add middleware to GraphQLApp

[cglacet]

I'm wondering if it's possible to add middlewares to a GraphQLApp directly, for example is it possible to add authentication on top of it?

I think I could make it work using the following strategy, but is there a better way?

def authentication_check(func):

    async def check(*args, **kwargs):
        print("We could check for authentication here")
        response = await func(*args, **kwargs)
        print("...")
        return response

    return check


graphql_app = GraphQLApp(schema=schema, executor_class=AsyncioExecutor)
graphql_app.handle_graphql = authentication_check(graphql_app.handle_graphql)

Maybe it would be better to intercept calls to graphql_app.execute so we can decide which request are allowed depending on the kind of user? Or, as suggested in graphene documentation we could just allow to add middleware directly to schema.execute right here.

async def execute(  # type: ignore
        self, query, variables=None, context=None, operation_name=None, middleware=None
    ):

    return await self.schema.execute(query, middleware=middleware, ...)

Or maybe even better, allow the user to pass any parameter to schema.execute:

async def execute(  # type: ignore
        self, query, **optional_args
    ):

    return await self.schema.execute(query, **optional_args)

I found this location to add a middleware very strange (adding it on the schema directly would make more sense to me), but if that's what graphene does I guess that's the better way? I'm very new to this so maybe I'm just lost ^^.

[tomchristie]

I don't have a lot of time to dig into what the best patterns would be here, but very happy to hear what other users are doing here atm?

Related https://github.com/encode/starlette/issues/619

[cglacet]

For now I decided to only use basic authentication rules (independent from the GraphQL requests) in graphql_app.handle_graphql (for example: only authenticated users can make requests). And then check for authorization in graphql_app.execute (for example: only admins can request for customerList). In the end it looks something like this:

# Starlette responsible for authentication 
graphql_app.handle_graphql = requires(['authenticated'])(graphql_app.handle_graphql)
# My code is responsible for authorization
graphql_app.execute = check_authorization(graphql_app.execute)

With check_authorization looking like this:

def check_authorization(execute_function):

    async def decorator(graphql_request, *args, variables=None, context=None, operation_name=None, **kwargs):
        authorization = context['request'].user.authorization_group
        if authorization == 'admin' or 'customersList' not in graphql_request:
            return await execute_function(graphql_request, *args, **kwargs)
        else:
            raise AuthenticationError("You can't execute this request unless you are admin.")

    return decorator

The verification done here is very dumb as I only check if the input graphql contains a given string, but you get the idea.

Note that in my authentication process I add this authorization_group to my users:

def authenticate_customer(customer, authorization_group=None):
    customer.is_authenticated = True
    customer.authorization_group = authorization_group
    return customer

class BasicAuthBackend(AuthenticationBackend):

    async def authenticate(self, request):
        ... 
        if is_admin(user):
               authorization_group = 'admin'
               user = authenticate_customer(customer, authorization_group)
               credentials = AuthCredentials(["authenticated", authorization_group])

        return credentials, user

All of this is not very sexy, but at least I fell like the code is located where it belongs.

[cglacet]

I'm still investigating this, but another (probably better) option to check for authorization is what I suggested in the original question: Modify the graphQL schema on the flight to add a middlewares directly to schema.execute:

def add_execute_middleware(graphql_schema, middleware):

    def graphql_middleware_decorator(graphql_execute):

        def execute_with_middleware(*args, **kwargs):
            return graphql_execute(*args, **kwargs, middleware=middleware)

        return execute_with_middleware

    graphql_schema.execute = graphql_middleware_decorator(graphql_schema.execute)
    return graphql_schema


async def graphql_auth_middleware(next, root, info, **kwargs):
    authorization = info.context['request'].user.authorization_group
    if authorization != 'admin' and 'customersList' in info.path:
        raise ValueError("customersList is not available for your group.")
    if info.field_name == 'password':
        return "even if it's encrypted, we wont send this to you."
    return await next(root, info, **kwargs)


schema = add_execute_middleware(schema, [graphql_authorization_middleware])
graphql_app = GraphQLApp(schema=schema)

This way we can access to parsed graphQL queries, which allows us to get more accurate information faster. For example we just need to search a list to check things like 'customersList' in info.path.

Here is another example limiting the maximum query depth (like some people tend to do):

async def graphql_max_depth_middleware(next, root, info, **kwargs):
    if len(info.path) > GRAPHQL_MAX_QUERY_DEPTH:
        raise ValueError("GraphQL request is too complex.")
    return await next(root, info, **kwargs)

schema = add_execute_middleware(schema, [graphql_authorization_middleware, graphql_max_depth_middleware])
graphql_app = GraphQLApp(schema=schema)

This also have the benefit or returning meaningful results to the client:

{
    "data": {
        "customer": {
            "lastname": "glacet",
            "password": "even if it's encrypted, we wont send this to you."
        },
        "customers": {
            "edges": [
                {
                    "node": null
                },
            ]
        },
        "customersList": null
    },
    "errors": [
        {
            "message": "customersList is not available for your group.",
            "locations": [  ],
            "path": ["customersList"]
        },
        {
            "message": "GraphQL request is too complex.",
            "locations": [  ],
            "path": ["customers", "edges", 0, "node"]
        },
    ]
}

That would be pretty easy to update starlette to allow for such middleware to be passed directly to GraphQLApp.

---

Edit I noticed I should probably use Starlette's scopes for user groups authorizations, but that's rather not clear to me. It would only require to update the code like this:

async def graphql_auth_middleware(next, root, info, **kwargs):
    user_scopes = info.context['request'].auth.scopes
    if 'admin' not in user_scopes and 'customersList' in info.path:
        raise ValueError("customersList is not available for your group.")
   ...

And have credentials adding user groups like so: AuthCredentials(["authenticated", authorization_group]).

[tazimmerman]

Couldn't a middleware argument be added to GraphQLApp constructor and passed on later to the schema.execute call? I'd like to do this in order to apply custom directives (which you can already pass in).

[cglacet]

That's the solution I ended up with. I could provide some code for it in the future if needed.

[tazimmerman]

Here's what I did, in case anyone finds it useful (or has a better suggestion). I derived my own app from GraphQLApp, and re-implemented execute to inject the middleware.

class CustomGraphQLApp(GraphQLApp):
    """App that injects our custom directive middleware when calling `schema.execute`."""

    middleware = [CustomDirectiveMiddleware()]

    async def execute(self, query, variables=None, context=None, operation_name=None):
        if self.is_async:
            return await self.schema.execute(
                query,
                variables=variables,
                operation_name=operation_name,
                executor=self.executor,
                return_promise=True,
                context=context,
                middleware=self.middleware,
            )
        else:
            raise NotImplementedError("Synchronous execution is not supported.")

[jawahar273]

Hi, I am trying to create app based graphql. I have adding AuthenticationMiddleware as a middleware as you can see i am trying to raising exception but the exception is not handled by graphql. Is there a way to handle exception or should i raise new issue i am just confused.

class CustomAuthBackend(AuthenticationBackend):
    async def authenticate(self, request: HTTPConnection):
        raise ValueError("Testing Expection")
       username="admin"
        return AuthCredentials(["authenticated"]), SimpleUser(username)

app.add_middleware(
    AuthenticationMiddleware, backend=CustomAuthBackend()
)

current behaviour

ValueError: Testing expection

expected

graphql.error.located_error.GraphQLLocatedError: Testing expection

[miguescri]

In case someone is willing to patch the libray manually to pass middleware as a parameter of GraphQLApp, there is this merge request https://github.com/encode/starlette/pull/1044

[JayH5]

Thank you for filing this issue. We have decided to deprecate GraphQL support within Starlette itself so I am going to close this issue. See #619.