fix: fk resolver permissions leak (#1411)

* fix: fk resolver permissions leak

* fix: only one query for 1o1 relation

* tests: added queries count check

* fix: docstring

* fix: typo

* docs: added warning to authorization

* feat: added bypass_get_queryset decorator

Author: superlevure

.gitignore

@@ -11,6 +11,7 @@ __pycache__/
 # Distribution / packaging
 .Python
 env/
+.env/
 venv/
 .venv/
 build/

docs/authorization.rst

@@ -144,6 +144,21 @@ If you are using ``DjangoObjectType`` you can define a custom `get_queryset`.
                 return queryset.filter(published=True)
             return queryset
 
+.. warning::
+
+    Defining a custom ``get_queryset`` gives the guaranteed it will be called
+    when resolving the ``DjangoObjectType``, even through related objects.
+    Note that because of this, benefits from using ``select_related``
+    in objects that define a relation to this ``DjangoObjectType`` will be canceled out.
+    In the case of ``prefetch_related``, the benefits of the optimization will be lost only
+    if the custom ``get_queryset`` modifies the queryset. For more information about this, refers
+    to Django documentation about ``prefetch_related``: https://docs.djangoproject.com/en/4.2/ref/models/querysets/#prefetch-related.
+
+
+    If you want to explicitly disable the execution of the custom ``get_queryset`` when resolving,
+    you can decorate the resolver with `@graphene_django.bypass_get_queryset`. Note that this
+    can lead to authorization leaks if you are performing authorization checks in the custom
+    ``get_queryset``.
 
 Filtering ID-based Node Access
 ------------------------------
@@ -197,8 +212,8 @@ For Django 2.2 and above:
 .. code:: python
 
     urlpatterns = [
-      # some other urls
-      path('graphql/', PrivateGraphQLView.as_view(graphiql=True, schema=schema)),
+        # some other urls
+        path('graphql/', PrivateGraphQLView.as_view(graphiql=True, schema=schema)),
     ]
 
 .. _LoginRequiredMixin: https://docs.djangoproject.com/en/dev/topics/auth/default/#the-loginrequired-mixin

docs/introspection.rst

@@ -57,9 +57,9 @@ specify the parameters in your settings.py:
 .. code:: python
 
     GRAPHENE = {
-    	'SCHEMA': 'tutorial.quickstart.schema',
-    	'SCHEMA_OUTPUT': 'data/schema.json',  # defaults to schema.json,
-    	'SCHEMA_INDENT': 2,  # Defaults to None (displays all data on a single line)
+        'SCHEMA': 'tutorial.quickstart.schema',
+        'SCHEMA_OUTPUT': 'data/schema.json',  # defaults to schema.json,
+        'SCHEMA_INDENT': 2,  # Defaults to None (displays all data on a single line)
     }
 
 

graphene_django/__init__.py

@@ -1,5 +1,6 @@
 from .fields import DjangoConnectionField, DjangoListField
 from .types import DjangoObjectType
+from .utils import bypass_get_queryset
 
 __version__ = "3.1.3"
 
@@ -8,4 +9,5 @@
     "DjangoObjectType",
     "DjangoListField",
     "DjangoConnectionField",
+    "bypass_get_queryset",
 ]

graphene_django/converter.py

@@ -1,5 +1,6 @@
+import inspect
 from collections import OrderedDict
-from functools import singledispatch, wraps
+from functools import partial, singledispatch, wraps
 
 from django.db import models
 from django.utils.encoding import force_str
@@ -25,6 +26,7 @@
 )
 from graphene.types.json import JSONString
 from graphene.types.scalars import BigInt
+from graphene.types.resolver import get_default_resolver
 from graphene.utils.str_converters import to_camel_case
 from graphql import GraphQLError
 
@@ -258,14 +260,62 @@ def convert_time_to_string(field, registry=None):
 
 @convert_django_field.register(models.OneToOneRel)
 def convert_onetoone_field_to_djangomodel(field, registry=None):
+    from graphene.utils.str_converters import to_snake_case
+    from .types import DjangoObjectType
+
     model = field.related_model
 
     def dynamic_type():
         _type = registry.get_type_for_model(model)
         if not _type:
             return
 
-        return Field(_type, required=not field.null)
+        class CustomField(Field):
+            def wrap_resolve(self, parent_resolver):
+                """
+                Implements a custom resolver which goes through the `get_node` method to ensure that
+                it goes through the `get_queryset` method of the DjangoObjectType.
+                """
+                resolver = super().wrap_resolve(parent_resolver)
+
+                # If `get_queryset` was not overridden in the DjangoObjectType
+                # or if we explicitly bypass the `get_queryset` method,
+                # we can just return the default resolver.
+                if (
+                    _type.get_queryset.__func__
+                    is DjangoObjectType.get_queryset.__func__
+                    or getattr(resolver, "_bypass_get_queryset", False)
+                ):
+                    return resolver
+
+                def custom_resolver(root, info, **args):
+                    # Note: this function is used to resolve 1:1 relation fields
+
+                    is_resolver_awaitable = inspect.iscoroutinefunction(resolver)
+
+                    if is_resolver_awaitable:
+                        fk_obj = resolver(root, info, **args)
+                        # In case the resolver is a custom awaitable resolver that overwrites
+                        # the default Django resolver
+                        return fk_obj
+
+                    field_name = to_snake_case(info.field_name)
+                    reversed_field_name = root.__class__._meta.get_field(
+                        field_name
+                    ).remote_field.name
+                    return _type.get_queryset(
+                        _type._meta.model.objects.filter(
+                            **{reversed_field_name: root.pk}
+                        ),
+                        info,
+                    ).get()
+
+                return custom_resolver
+
+        return CustomField(
+            _type,
+            required=not field.null,
+        )
 
     return Dynamic(dynamic_type)
 
@@ -313,14 +363,89 @@ def dynamic_type():
 @convert_django_field.register(models.OneToOneField)
 @convert_django_field.register(models.ForeignKey)
 def convert_field_to_djangomodel(field, registry=None):
+    from graphene.utils.str_converters import to_snake_case
+    from .types import DjangoObjectType
+
     model = field.related_model
 
     def dynamic_type():
         _type = registry.get_type_for_model(model)
         if not _type:
             return
 
-        return Field(
+        class CustomField(Field):
+            def wrap_resolve(self, parent_resolver):
+                """
+                Implements a custom resolver which goes through the `get_node` method to ensure that
+                it goes through the `get_queryset` method of the DjangoObjectType.
+                """
+                resolver = super().wrap_resolve(parent_resolver)
+
+                # If `get_queryset` was not overridden in the DjangoObjectType
+                # or if we explicitly bypass the `get_queryset` method,
+                # we can just return the default resolver.
+                if (
+                    _type.get_queryset.__func__
+                    is DjangoObjectType.get_queryset.__func__
+                    or getattr(resolver, "_bypass_get_queryset", False)
+                ):
+                    return resolver
+
+                def custom_resolver(root, info, **args):
+                    # Note: this function is used to resolve FK or 1:1 fields
+                    #   it does not differentiate between custom-resolved fields
+                    #   and default resolved fields.
+
+                    # because this is a django foreign key or one-to-one field, the primary-key for
+                    # this node can be accessed from the root node.
+                    # ex: article.reporter_id
+
+                    # get the name of the id field from the root's model
+                    field_name = to_snake_case(info.field_name)
+                    db_field_key = root.__class__._meta.get_field(field_name).attname
+                    if hasattr(root, db_field_key):
+                        # get the object's primary-key from root
+                        object_pk = getattr(root, db_field_key)
+                    else:
+                        return None
+
+                    is_resolver_awaitable = inspect.iscoroutinefunction(resolver)
+
+                    if is_resolver_awaitable:
+                        fk_obj = resolver(root, info, **args)
+                        # In case the resolver is a custom awaitable resolver that overwrites
+                        # the default Django resolver
+                        return fk_obj
+
+                    instance_from_get_node = _type.get_node(info, object_pk)
+
+                    if instance_from_get_node is None:
+                        # no instance to return
+                        return
+                    elif (
+                        isinstance(resolver, partial)
+                        and resolver.func is get_default_resolver()
+                    ):
+                        return instance_from_get_node
+                    elif resolver is not get_default_resolver():
+                        # Default resolver is overridden
+                        # For optimization, add the instance to the resolver
+                        setattr(root, field_name, instance_from_get_node)
+                        # Explanation:
+                        #   previously, _type.get_node` is called which results in at least one hit to the database.
+                        #   But, if we did not pass the instance to the root, calling the resolver will result in
+                        #   another call to get the instance which results in at least two database queries in total
+                        #   to resolve this node only.
+                        #   That's why the value of the object is set in the root so when the object is accessed
+                        #   in the resolver (root.field_name) it does not access the database unless queried explicitly.
+                        fk_obj = resolver(root, info, **args)
+                        return fk_obj
+                    else:
+                        return instance_from_get_node
+
+                return custom_resolver
+
+        return CustomField(
             _type,
             description=get_django_field_description(field),
             required=not field.null,