How to restrict access to queries

[iAlexMili]

Hello,

I would like to restrict access to some of the queries and mutations available in my schema. As mentioned in #506 it is not possible to add custom directives on queries. So I added my custom directive on the object I want to restrict access to:

from functools import partial

from ariadne import QueryType, make_executable_schema
from ariadne.asgi import GraphQL
from ariadne.schema_visitor import SchemaDirectiveVisitor
from graphql import default_field_resolver
from graphql.type.definition import GraphQLObjectType
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
import uvicorn


type_def = """
    directive @checkAccess(needs: [String]) on OBJECT

    type MyObject @checkAccess(needs: ["admin"]) {
        id: Int!
        name: String!
    }

    type Query {
        getObject: MyObject
    }
"""

query = QueryType()


@query.field("getObject")
def resolve_getObject(obj, info):
    return {"id": 1, "name": "object1"}


class MyObjectDirective(SchemaDirectiveVisitor):
    def visit_object(self, obj: GraphQLObjectType):
        self.check_myobject(obj)

    def check_myobject(self, obj: GraphQLObjectType):

        def _resolver(_, info, *, field=None, obj_=None):
            return original_resolver(_, info)

        for _, field in obj.fields.items():
            original_resolver = field.resolve or default_field_resolver
            field.resolve = partial(_resolver, field=field, obj_=obj)


schema = make_executable_schema(type_def, query,
                                directives={"checkAccess": MyObjectDirective})
graphqlapp = GraphQL(schema, debug=True)

app = Starlette(debug=True, middleware=[Middleware(CORSMiddleware,
                allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])])

app.mount("/graphql", graphqlapp)

if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=5000, log_level="info")

The problem is that the query's resolver resolve_getObject() is called before the new resolver _resolver() edited by my directive. I understand that it is the expected behavior. Is there any way to call a directive before the resolution of the query or any alternative with such result ? I checked Middlewares and Extensions but they do not apply to subscriptions, which is mandatory in my case.

[rafalp]

Your MyObjectDirective.visit_object is not run against the resolvers on Query type, but instead on resolvers on MyObject fields.

What you are after doesn't seem like something that can be achieved out of the box with SchemaDirectiveVisitor. You will have to write custom logic that would walk the schema object and wrap resolvers returning MyObject with custom resolver checking permissions.

[iAlexMili]

Thank you for your suggestion. GraphQL doc reminded me to Delegate authorization logic to the business logic layer. So I implemented a classic approach as follows:

@query.field("getObject")
def resolve_getObject(obj, info):
    result = run_with_perm(info, get_object)
    return result

This new approach led me to a new problematic: doing so, would either require to do a DB call to check permission on every query object or to load all permissions in memory and check in the resolver. Again in order to be as efficient as possible, I would like to load exclusively permissions that are necessary to execute the graphql request.
To do so, I need the request to be parsed and validated before doing any processing. As I understand, the validation is done in graphql.py. So I re-implemented the method graphql_http_server() of GraphQL class by copy/pasting the part of handling query validation:

            validate_data(data)
            query, variables, operation_name = (
                data["query"],
                data.get("variables"),
                data.get("operationName"),
            )

            document = parse_query(query)

            if callable(validation_rules):
                validation_rules = cast(
                    Optional[Sequence[RuleType]],
                    validation_rules(context_value, document, data),
                )

            validation_errors = validate_query(
                schema, document, validation_rules, enable_introspection=introspection
            )

Ideally integrating a hook before the execute() function in graphql.py would be ideal for me. But I don't know if it could be good for the project. What do you think ?
Interestingly, this specific part of the code, validating the query, is the same across graphql(), graphql_sync() and suscribe(). So I was wondering if you could consider accepting a PR wrapping this part of the code in a function? This way, the process would not change, the code would be more DRY and in my case, it could be possible to call the validation function from my custom implementation of graphql_http_server().

The topic is shifting away from my original question, if this is problematic, I can create a new one.

[rafalp]

Have you considered loading all permissions in either context_value function or running auth in the ASGI middleware wrapping Ariadne's GraphQL app and using request.scope to pass those perms to the resolvers?

Unless you really have thousands of permissions associated with the user, in which case you could also use dataloader to load those efficiently and lazily on demand. ;)

[iAlexMili]

Actually that's exactly what I am doing right now :) I load all permissions in an ASGI middleware and use request.scope to pass the information.

I guess I was just curious how far it was possible to optimize my code, but you're right the overhead of loading all permissions should be low. Anyway, thank you for your answers !