Graphene: Offer a way to calculate and limit the total cost of a query

Created on 18 Jun 2018  路  7Comments  路  Source: graphql-python/graphene

When offering a public API there's the problem of malicious clients preparing intentionally expensive queries. For example one could abuse mutually related objects to arbitrarily make a query more expensive: book -> author -> books -> author -> books -> ...and so on.

I'd like to request a method to evaluate the estimated cost of query before actually executing any resolvers and a way to prevent execution of queries with the estimate cost above a certain threshold (ideally in a programmatic fashion so we could for example vary the limits depending on the currently logged in user's role).

Most helpful comment

Ah sorry I missed that part. As far as I know there isn't an "official" way of doing this but I've been playing around with then new backends feature that @syrusakbary introduced in graphql-core (https://github.com/graphql-python/graphql-core/pull/185) and I think this would be a great use case for it.

You should be able to do something like this with graphql-core 2.1rc:

from graphql.backend.core import GraphQLCoreBackend

def measure_depth(selection_set, level=1):
    max_depth = level
    for field in selection_set.selections:
        if field.selection_set:
            new_depth = measure_depth(field.selection_set, level=level + 1)
            if new_depth > max_depth:
                max_depth = new_depth
    return max_depth

class DepthAnalysisBackend(GraphQLCoreBackend):
    def document_from_string(self, schema, document_string):
        document = super().document_from_string(schema, document_string)
        ast = document.document_ast
        for definition in ast.definitions:
            # We are only interested in queries
            if definition.operation != 'query':
                continue

            depth = measure_depth(definition.selection_set)
            if depth > 3: # set your depth max here
                raise Exception('Query is too complex')

        return document

Then when you're executing the following query it will bail out before trying to execute the query:

query_string = '''
    query {
        myFavouriteBook {
            author {
                bestBook {
                    author {
                        name
                    }
                }
            }
        }
    }
'''

backend = DepthAnalysisBackend()
schema = graphene.Schema(query=Query)
result = schema.execute(query_string, backend=backend)

assert result.errors

I think there is a lot more that could be done here (like being able to assign complexity values to each field) but hopefully this helps you for now.

All 7 comments

@patrys I think you can achieve this using middleware. I'd imagine you set up some counter in the execution context and then increment it in the middleware. I've not actually tried this though but if you're willing to try it out do let us know how it goes.

@dan98765 I found this comment that implies that you already have something working? Could you share how you do it?

@jkimbo I want to calculate this and reject a query before it starts to be evaluated.

Ah sorry I missed that part. As far as I know there isn't an "official" way of doing this but I've been playing around with then new backends feature that @syrusakbary introduced in graphql-core (https://github.com/graphql-python/graphql-core/pull/185) and I think this would be a great use case for it.

You should be able to do something like this with graphql-core 2.1rc:

from graphql.backend.core import GraphQLCoreBackend

def measure_depth(selection_set, level=1):
    max_depth = level
    for field in selection_set.selections:
        if field.selection_set:
            new_depth = measure_depth(field.selection_set, level=level + 1)
            if new_depth > max_depth:
                max_depth = new_depth
    return max_depth

class DepthAnalysisBackend(GraphQLCoreBackend):
    def document_from_string(self, schema, document_string):
        document = super().document_from_string(schema, document_string)
        ast = document.document_ast
        for definition in ast.definitions:
            # We are only interested in queries
            if definition.operation != 'query':
                continue

            depth = measure_depth(definition.selection_set)
            if depth > 3: # set your depth max here
                raise Exception('Query is too complex')

        return document

Then when you're executing the following query it will bail out before trying to execute the query:

query_string = '''
    query {
        myFavouriteBook {
            author {
                bestBook {
                    author {
                        name
                    }
                }
            }
        }
    }
'''

backend = DepthAnalysisBackend()
schema = graphene.Schema(query=Query)
result = schema.execute(query_string, backend=backend)

assert result.errors

I think there is a lot more that could be done here (like being able to assign complexity values to each field) but hopefully this helps you for now.

Closing this issue in favour of tracking multiple approaches to protecting against malicious queries in #907

Ah sorry I missed that part. As far as I know there isn't an "official" way of doing this but I've been playing around with then new backends feature that @syrusakbary introduced in graphql-core (https://github.com/graphql-python/graphql-core/pull/185) and I think this would be a great use case for it.

You should be able to do something like this with graphql-core 2.1rc:

from graphql.backend.core import GraphQLCoreBackend

def measure_depth(selection_set, level=1):
    max_depth = level
    for field in selection_set.selections:
        if field.selection_set:
            new_depth = measure_depth(field.selection_set, level=level + 1)
            if new_depth > max_depth:
                max_depth = new_depth
    return max_depth

class DepthAnalysisBackend(GraphQLCoreBackend):
    def document_from_string(self, schema, document_string):
        document = super().document_from_string(schema, document_string)
        ast = document.document_ast
        for definition in ast.definitions:
            # We are only interested in queries
            if definition.operation != 'query':
                continue

            depth = measure_depth(definition.selection_set)
            if depth > 3: # set your depth max here
                raise Exception('Query is too complex')

        return document

Then when you're executing the following query it will bail out before trying to execute the query:

query_string = '''
    query {
      myFavouriteBook {
          author {
              bestBook {
                  author {
                      name
                  }
              }
          }
      }
    }
'''

backend = DepthAnalysisBackend()
schema = graphene.Schema(query=Query)
result = schema.execute(query_string, backend=backend)

assert result.errors

I think there is a lot more that could be done here (like being able to assign complexity values to each field) but hopefully this helps you for now.

Ah sorry I missed that part. As far as I know there isn't an "official" way of doing this but I've been playing around with then new backends feature that @syrusakbary introduced in graphql-core (https://github.com/graphql-python/graphql-core/pull/185) and I think this would be a great use case for it.

You should be able to do something like this with graphql-core 2.1rc:

from graphql.backend.core import GraphQLCoreBackend

def measure_depth(selection_set, level=1):
    max_depth = level
    for field in selection_set.selections:
        if field.selection_set:
            new_depth = measure_depth(field.selection_set, level=level + 1)
            if new_depth > max_depth:
                max_depth = new_depth
    return max_depth

class DepthAnalysisBackend(GraphQLCoreBackend):
    def document_from_string(self, schema, document_string):
        document = super().document_from_string(schema, document_string)
        ast = document.document_ast
        for definition in ast.definitions:
            # We are only interested in queries
            if definition.operation != 'query':
                continue

            depth = measure_depth(definition.selection_set)
            if depth > 3: # set your depth max here
                raise Exception('Query is too complex')

        return document

Then when you're executing the following query it will bail out before trying to execute the query:

query_string = '''
    query {
      myFavouriteBook {
          author {
              bestBook {
                  author {
                      name
                  }
              }
          }
      }
    }
'''

backend = DepthAnalysisBackend()
schema = graphene.Schema(query=Query)
result = schema.execute(query_string, backend=backend)

assert result.errors

I think there is a lot more that could be done here (like being able to assign complexity values to each field) but hopefully this helps you for now.

Hey,
Have you developed an official way until now?
If not, how can I use this backend in my graphene project?

I am trying to add this code into my settings file, but it doesn't work!

GRAPHENE = {
    'SCHEMA': 'django_app.schema.schema',
    'MIDDLEWARE': [
        'graphql_jwt.middleware.JSONWebTokenMiddleware',
    ],
    "BACKENDS": ["django_app.custom_backends.depth_analysis_backend.DepthAnalysisBackend"],

}
Was this page helpful?
0 / 5 - 0 ratings