import abc import contextlib import contextvars import copy import dataclasses import enum import functools import inspect from collections.abc import Callable, Hashable, Iterable from typing import ( TYPE_CHECKING, Any, ClassVar, Optional, TypeVar, Union, cast, overload, ) import strawberry from asgiref.sync import sync_to_async from django.core.exceptions import PermissionDenied from django.db.models import Model, QuerySet from graphql.pyutils import AwaitableOrValue from strawberry import relay, schema_directive from strawberry.extensions.field_extension import ( AsyncExtensionResolver, FieldExtension, SyncExtensionResolver, ) from strawberry.schema_directive import Location from strawberry.types.base import StrawberryList, StrawberryOptional from strawberry.types.field import StrawberryField from strawberry.types.info import Info from strawberry.types.union import StrawberryUnion from typing_extensions import Literal, assert_never from strawberry_django.auth.utils import aget_current_user, get_current_user from strawberry_django.fields.types import OperationInfo, OperationMessage from strawberry_django.pagination import OffsetPaginated from strawberry_django.resolvers import django_resolver from .utils.query import filter_for_user from .utils.typing import UserType if TYPE_CHECKING: from strawberry.django.context import StrawberryDjangoContext from strawberry_django.fields.field import StrawberryDjangoField _M = TypeVar("_M", bound=Model) @functools.lru_cache def _get_user_or_anonymous_getter() -> Optional[Callable[[UserType], UserType]]: try: from .integrations.guardian import get_user_or_anonymous except (ImportError, RuntimeError): # pragma: no cover return None return get_user_or_anonymous @dataclasses.dataclass class PermContext: is_safe_list: list[bool] = dataclasses.field(default_factory=list) checkers: list["HasPerm"] = dataclasses.field(default_factory=list) def __copy__(self): return self.__class__( is_safe_list=self.is_safe_list[:], checkers=self.checkers[:], ) @property def is_safe(self): return bool(self.is_safe_list and all(self.is_safe_list)) perm_context: contextvars.ContextVar[PermContext] = contextvars.ContextVar( "perm-safe", default=PermContext(), # noqa: B039 ) @contextlib.contextmanager def with_perm_checker(checker: "HasPerm"): context = copy.copy(perm_context.get()) context.checkers.append(checker) token = perm_context.set(context) try: yield finally: perm_context.reset(token) def set_perm_safe(value: bool): perm_context.get().is_safe_list.append(value) def filter_with_perms(qs: QuerySet[_M], info: Info) -> QuerySet[_M]: context = perm_context.get() if not context.checkers or context.is_safe: return qs if isinstance(qs, list): # return sliced queryset as-is return qs # Do not do anything is results are cached if qs._result_cache is not None: # type: ignore set_perm_safe(False) return qs user = cast("StrawberryDjangoContext", info.context).request.user # If the user is anonymous, we can't filter object permissions for it if user.is_anonymous: set_perm_safe(False) return qs.none() for check in context.checkers: if check.target != PermTarget.RETVAL: continue qs = filter_for_user( qs, user, [p.perm for p in check.perms], any_perm=check.any_perm, with_superuser=check.with_superuser, ) set_perm_safe(True) return qs @overload def get_with_perms( pk: strawberry.ID, info: Info, *, required: Literal[True], model: type[_M], key_attr: str = ..., ) -> _M: ... @overload def get_with_perms( pk: strawberry.ID, info: Info, *, required: bool = ..., model: type[_M], key_attr: str = ..., ) -> Optional[_M]: ... @overload def get_with_perms( pk: relay.GlobalID, info: Info, *, required: Literal[True], model: type[_M], key_attr: str = ..., ) -> _M: ... @overload def get_with_perms( pk: relay.GlobalID, info: Info, *, required: bool = ..., model: type[_M], key_attr: str = ..., ) -> Optional[_M]: ... @overload def get_with_perms( pk: relay.GlobalID, info: Info, *, required: Literal[True], key_attr: str = ..., ) -> Any: ... @overload def get_with_perms( pk: relay.GlobalID, info: Info, *, required: bool = ..., key_attr: str = ..., ) -> Optional[Any]: ... def get_with_perms( pk, info, *, required=False, model=None, key_attr="pk", ): if isinstance(pk, relay.GlobalID): instance = pk.resolve_node_sync(info, required=required, ensure_type=model) else: assert model instance = model._default_manager.get(**{key_attr: pk}) if instance is None: return None context = perm_context.get() if not context.checkers or context.is_safe: return instance user = cast("StrawberryDjangoContext", info.context).request.user if user and (get_user_or_anonymous := _get_user_or_anonymous_getter()) is not None: user = get_user_or_anonymous(user) for check in context.checkers: f = any if check.any_perm else all checker = check.obj_perm_checker(info, user) if not f(checker(p, instance) for p in check.perms): raise PermissionDenied(check.message) return instance _return_condition = """\ When the condition fails, the following can be returned (following this priority): 1) `OperationInfo`/`OperationMessage` if those types are allowed at the return type 2) `null` in case the field is not mandatory (e.g. `String` or `[String]`) 3) An empty list in case the field is a list (e.g. `[String]!`) 4) An empty `Connection` in case the return type is a relay connection 2) Otherwise, an error will be raised """ def _desc(desc): return f"{desc}\n\n{_return_condition.strip()}" class DjangoNoPermission(Exception): # noqa: N818 """Raise to identify that the user doesn't have perms for a given retval.""" class DjangoPermissionExtension(FieldExtension, abc.ABC): """Base django permission extension.""" DEFAULT_ERROR_MESSAGE: ClassVar[str] = "User does not have permission." SCHEMA_DIRECTIVE_LOCATIONS: ClassVar[list[Location]] = [Location.FIELD_DEFINITION] SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = None def __init__( self, *, message: Optional[str] = None, use_directives: bool = True, fail_silently: bool = True, ): super().__init__() self.message = message if message is not None else self.DEFAULT_ERROR_MESSAGE self.fail_silently = fail_silently self.use_directives = use_directives def apply(self, field: StrawberryField) -> None: # pragma: no cover if self.use_directives: directive = self.schema_directive # Avoid interfaces duplicating the directives if directive not in field.directives: field.directives.append(self.schema_directive) @functools.cached_property def schema_directive(self) -> object: key = "__strawberry_directive_type__" directive_class = getattr(self.__class__, key, None) if directive_class is None: @schema_directive( name=self.__class__.__name__, locations=self.SCHEMA_DIRECTIVE_LOCATIONS, description=self.SCHEMA_DIRECTIVE_DESCRIPTION, repeatable=True, ) class AutoDirective: ... directive_class = AutoDirective return directive_class() @django_resolver(qs_hook=None) def resolve( self, next_: SyncExtensionResolver, source: Any, info: Info, **kwargs: dict[str, Any], ) -> Any: user = get_current_user(info) if ( user and (get_user_or_anonymous := _get_user_or_anonymous_getter()) is not None ): user = get_user_or_anonymous(user) # make sure the user is loaded user.is_authenticated # noqa: B018 try: retval = self.resolve_for_user( functools.partial(next_, source, info, **kwargs), user, info=info, source=source, ) except DjangoNoPermission as e: retval = self.handle_no_permission(e, info=info) return retval async def resolve_async( self, next_: AsyncExtensionResolver, source: Any, info: Info, **kwargs: dict[str, Any], ) -> Any: user = await aget_current_user(info) try: from .integrations.guardian import get_user_or_anonymous except (ImportError, RuntimeError): # pragma: no cover pass else: user = user and await sync_to_async(get_user_or_anonymous)(user) # make sure the user is loaded await sync_to_async(getattr)(user, "is_anonymous") try: retval = self.resolve_for_user( functools.partial(next_, source, info, **kwargs), user, info=info, source=source, ) while inspect.isawaitable(retval): retval = await retval except DjangoNoPermission as e: retval = self.handle_no_permission(e, info=info) return retval def handle_no_permission(self, exception: BaseException, *, info: Info): if not self.fail_silently: raise PermissionDenied(self.message) from exception ret_type = info.return_type if isinstance(ret_type, StrawberryOptional): ret_type = ret_type.of_type is_optional = True else: is_optional = False if isinstance(ret_type, StrawberryUnion): ret_types = [] for type_ in ret_type.types: ret_types.append(ret_type) if not isinstance(type_, type): continue if issubclass(type_, OperationInfo): return type_( messages=[ OperationMessage( kind=OperationMessage.Kind.PERMISSION, message=self.message, field=info.field_name, ), ], ) if issubclass(type_, OperationMessage): return type_( kind=OperationMessage.Kind.PERMISSION, message=self.message, field=info.field_name, ) else: ret_types = [ret_type] if is_optional: return None if isinstance(ret_type, StrawberryList): return [] if isinstance(ret_type, type) and issubclass(ret_type, OffsetPaginated): django_model = cast("StrawberryDjangoField", info._field).django_model assert django_model return django_model._default_manager.none() # If it is a Connection, try to return an empty connection, but only if # it is the only possibility available... for ret_possibility in ret_types: if isinstance(ret_possibility, type) and issubclass( ret_possibility, relay.Connection, ): return [] # In last case, raise an error raise PermissionDenied(self.message) from exception @abc.abstractmethod def resolve_for_user( # pragma: no cover self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ) -> AwaitableOrValue[Any]: ... class IsAuthenticated(DjangoPermissionExtension): """Mark a field as only resolvable by authenticated users.""" DEFAULT_ERROR_MESSAGE: ClassVar[str] = "User is not authenticated." SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Can only be resolved by authenticated users.", ) @django_resolver(qs_hook=None) def resolve_for_user( self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ): if user is None or not user.is_authenticated or not user.is_active: raise DjangoNoPermission return resolver() class IsStaff(DjangoPermissionExtension): """Mark a field as only resolvable by staff users.""" DEFAULT_ERROR_MESSAGE: ClassVar[str] = "User is not a staff member." SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Can only be resolved by staff users.", ) @django_resolver(qs_hook=None) def resolve_for_user( self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ): if ( user is None or not user.is_authenticated or not getattr(user, "is_staff", False) ): raise DjangoNoPermission return resolver() class IsSuperuser(DjangoPermissionExtension): """Mark a field as only resolvable by superuser users.""" DEFAULT_ERROR_MESSAGE: ClassVar[str] = "User is not a superuser." SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Can only be resolved by superuser users.", ) @django_resolver(qs_hook=None) def resolve_for_user( self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ): if ( user is None or not user.is_authenticated or not getattr(user, "is_superuser", False) ): raise DjangoNoPermission return resolver() @strawberry.input(description="Permission definition for schema directives.") class PermDefinition: """Permission definition. Attributes ---------- app: The app to which we are requiring permission. permission: The permission itself """ app: Optional[str] = strawberry.field( description=( "The app to which we are requiring permission. If this is " "empty that means that we are checking the permission directly." ), ) permission: Optional[str] = strawberry.field( description=( "The permission itself. If this is empty that means that we " "are checking for any permission for the given app." ), ) @classmethod def from_perm(cls, perm: str): parts = perm.split(".") if len(parts) != 2: # noqa: PLR2004 raise TypeError( "Permissions need to be defined as `app_label.perm`, `app_label.`" " or `.perm`", ) return cls( app=parts[0].strip() or None, permission=parts[1].strip() or None, ) @property def perm(self): return f"{self.app or ''}.{self.permission or ''}".strip(".") def __eq__(self, other: object): if not isinstance(other, PermDefinition): return NotImplemented return self.perm == other.perm def __hash__(self): return hash((self.__class__, self.perm)) class PermTarget(enum.IntEnum): """Permission location.""" GLOBAL = enum.auto() SOURCE = enum.auto() RETVAL = enum.auto() def _default_perm_checker(info: Info, user: UserType): def perm_checker(perm: PermDefinition) -> bool: return ( user.has_perm(perm.perm) # type: ignore if perm.permission else user.has_module_perms(cast("str", perm.app)) # type: ignore ) return perm_checker def _default_obj_perm_checker(info: Info, user: UserType): def perm_checker(perm: PermDefinition, obj: Any) -> bool: # Check global perms first, then object specific return user.has_perm(perm.perm) or user.has_perm( # type: ignore perm.perm, obj=obj, ) return perm_checker class HasPerm(DjangoPermissionExtension): """Defines permissions required to access the given object/field. Given a `app` name, the user can access the decorated object/field if he has any of the permissions defined in this directive. Examples -------- To indicate that a mutation can only be done by someone who has "product.add_product" perm in the django system: >>> @strawberry.type ... class Query: ... @strawberry.mutation(directives=[HasPerm("product.add_product")]) ... def create_product(self, name: str) -> ProductType: ... ... Attributes ---------- perms: Perms required to access this app. any_perm: If any perm or all perms are required to resolve the object/field. target: The target to check for permissions. Use `HasSourcePerm` or `HasRetvalPerm` as a shortcut for this. with_anonymous: If we should optimize the permissions check and consider an anonymous user as not having any permissions. This is true by default, which means that anonymous users will not trigger has_perm checks. with_superuser: If we should optimize the permissions check and consider a superuser as having permissions foe everything. This is false by default to avoid returning unexpected results. Setting this to true will avoid triggering has_perm checks. """ DEFAULT_TARGET: ClassVar[PermTarget] = PermTarget.GLOBAL DEFAULT_ERROR_MESSAGE: ClassVar[str] = ( "You don't have permission to access this app." ) SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Will check if the user has any/all permissions to resolve this.", ) def __init__( self, perms: Union[list[str], str], *, message: Optional[str] = None, use_directives: bool = True, fail_silently: bool = True, target: Optional[PermTarget] = None, any_perm: bool = True, perm_checker: Optional[ Callable[[Info, UserType], Callable[[PermDefinition], bool]] ] = None, obj_perm_checker: Optional[ Callable[[Info, UserType], Callable[[PermDefinition, Any], bool]] ] = None, with_anonymous: bool = True, with_superuser: bool = False, ): super().__init__( message=message, use_directives=use_directives, fail_silently=fail_silently, ) if isinstance(perms, str): perms = [perms] if not perms: raise TypeError(f"At least one perm is required for {self!r}") self.perms: tuple[PermDefinition, ...] = tuple( PermDefinition.from_perm(p) if isinstance(p, str) else p for p in perms ) assert all(isinstance(p, PermDefinition) for p in self.perms) self.target = target if target is not None else self.DEFAULT_TARGET self.permissions = perms self.any_perm = any_perm self.perm_checker = ( perm_checker if perm_checker is not None else _default_perm_checker ) self.obj_perm_checker = ( obj_perm_checker if obj_perm_checker is not None else _default_obj_perm_checker ) self.with_anonymous = with_anonymous self.with_superuser = with_superuser @functools.cached_property def schema_directive(self) -> object: key = "__strawberry_directive_class__" directive_class = getattr(self.__class__, key, None) if directive_class is None: @schema_directive( name=self.__class__.__name__, locations=self.SCHEMA_DIRECTIVE_LOCATIONS, description=self.SCHEMA_DIRECTIVE_DESCRIPTION, repeatable=True, ) class AutoDirective: permissions: list[PermDefinition] = strawberry.field( description="Required perms to access this resource.", default_factory=list, ) any: bool = strawberry.field( description="If any or all perms listed are required.", default=True, ) directive_class = AutoDirective return directive_class( permissions=list(self.perms), any=self.any_perm, ) @django_resolver(qs_hook=None) def resolve_for_user( self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ): if user is None or (self.with_anonymous and user.is_anonymous): raise DjangoNoPermission if ( self.with_superuser and user.is_active and getattr(user, "is_superuser", False) ): return resolver() return self.resolve_for_user_with_perms( resolver, user, info=info, source=source, ) def resolve_for_user_with_perms( self, resolver: Callable, user: Optional[UserType], *, info: Info, source: Any, ): if user is None: raise DjangoNoPermission if self.target == PermTarget.GLOBAL: if not self._has_perm(source, user, info=info): raise DjangoNoPermission retval = resolver() elif self.target == PermTarget.SOURCE: # Just call _resolve_obj, it will raise DjangoNoPermission # if the user doesn't have permission for it self._resolve_obj(source, user, source, info=info) retval = resolver() elif self.target == PermTarget.RETVAL: with with_perm_checker(self): obj = resolver() retval = self._resolve_obj(source, user, obj, info=info) else: assert_never(self.target) return retval def _get_cache( self, info: Info, user: UserType, ) -> dict[tuple[Hashable, ...], bool]: cache_key = "_strawberry_django_permissions_cache" cache = getattr(user, cache_key, None) if cache is None: cache = {} setattr(user, cache_key, cache) return cache def _has_perm( self, source: Any, user: UserType, *, info: Info, ) -> bool: cache = self._get_cache(info, user) # Maybe the result ended up in the cache in the meantime cache_key = (self.perms, self.any_perm) if cache_key in cache: return cache[cache_key] f = any if self.any_perm else all checker = self.perm_checker(info, user) has_perm = f(checker(p) for p in self.perms) cache[cache_key] = has_perm return has_perm def _resolve_obj( self, source: Any, user: UserType, obj: Any, *, info: Info, ) -> Any: context = perm_context.get() if context.is_safe: return obj if isinstance(obj, Iterable): return list(self._resolve_iterable_obj(source, user, obj, info=info)) cache = self._get_cache(info, user) cache_key = (self.perms, self.any_perm, obj) has_perm = cache.get(cache_key) if has_perm is None: if isinstance(obj, OperationInfo): has_perm = True else: f = any if self.any_perm else all checker = self.obj_perm_checker(info, user) has_perm = f(checker(p, obj) for p in self.perms) cache[cache_key] = has_perm if not has_perm: raise DjangoNoPermission return obj def _resolve_iterable_obj( self, source: Any, user: UserType, objs: Iterable[Any], *, info: Info, ) -> Any: cache = self._get_cache(info, user) f = any if self.any_perm else all checker = self.obj_perm_checker(info, user) for obj in objs: cache_key = (self.perms, self.any_perm, obj) has_perm = cache.get(cache_key) if has_perm is None: if isinstance(obj, OperationInfo): has_perm = True else: has_perm = f(checker(p, obj) for p in self.perms) cache[cache_key] = has_perm if has_perm: yield obj class HasSourcePerm(HasPerm): """Defines permissions required to access the given field at object level. This will check the permissions for the source object to access the given field. Unlike `HasRetvalPerm`, this uses the source value (the object where the field is defined) to resolve the field, which means that this cannot be used for source queries and types. Examples -------- To indicate that a field inside a `ProductType` can only be accessed if the user has "product.view_field" in it in the django system: >>> @gql.django.type(Product) ... class ProductType: ... some_field: str = strawberry.field( ... directives=[HasSourcePerm(".add_product")], ... ) """ DEFAULT_TARGET: ClassVar[PermTarget] = PermTarget.SOURCE SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Will check if the user has any/all permissions for the parent " "of this field to resolve this.", ) class HasRetvalPerm(HasPerm): """Defines permissions required to access the given object/field at object level. Given a `app` name, the user can access the decorated object/field if he has any of the permissions defined in this directive. Note that this depends on resolving the object to check the permissions specifically for that object, unlike `HasPerm` which checks it before resolving. Examples -------- To indicate that a field that returns a `ProductType` can only be accessed by someone who has "product.view_product" has "product.view_product" perm in the django system: >>> @strawberry.type ... class SomeType: ... product: ProductType = strawberry.field( ... directives=[HasRetvalPerm(".add_product")], ... ) """ DEFAULT_TARGET: ClassVar[PermTarget] = PermTarget.RETVAL SCHEMA_DIRECTIVE_DESCRIPTION: ClassVar[Optional[str]] = _desc( "Will check if the user has any/all permissions for the resolved " "value of this field before returning it.", )