feat(guardian): Downstream permissions targets to groups

authentik/suse/guardian/shortcuts.py

@@ -0,0 +1,147 @@
+"""Convenient shortcuts to manage or check object permissions."""
+
+from functools import lru_cache
+from typing import Any
+
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Count, Q, QuerySet
+from django.db.models.functions import Cast
+from guardian.ctypes import get_content_type
+from guardian.exceptions import (
+    GuardianError,
+    MixedContentTypeError,
+)
+from guardian.utils import (
+    get_anonymous_user,
+    get_role_obj_perms_model,
+)
+
+
+@lru_cache(None)
+def _get_ct_cached(app_label: str, codename: str) -> ContentType:
+    """Caches `ContentType` instances like its `QuerySet` does."""
+    return ContentType.objects.get(app_label=app_label, permission__codename=codename)
+
+
+def get_objects_for_user(  # noqa: PLR0912 PLR0915
+    user: Any,
+    perms: str | list[str],
+    queryset: QuerySet | None = None,
+) -> QuerySet:
+    """Get objects that a user has *all* + descendant models the supplied permissions for."""
+    if isinstance(perms, str):
+        perms = [perms]
+    ctype = None
+    app_label = None
+    codenames = set()
+    pk_field = "object_pk"
+
+    # Compute codenames, app_label, ctype
+    for perm in perms:
+        if "." not in perm:
+            raise GuardianError(f"Cannot determine app label and content type from {perm}")
+        new_app_label, new_codename = perm.split(".", 1)
+        if not new_app_label or not new_codename:
+            raise GuardianError(f"Cannot determine app label and content type from {perm}")
+
+        if app_label is not None and app_label != new_app_label:
+            raise MixedContentTypeError(
+                f"Given perms must have same app label ({app_label} != {new_app_label})"
+            )
+
+        new_ctype = _get_ct_cached(new_app_label, new_codename)
+        if ctype is not None and ctype != new_ctype:
+            raise MixedContentTypeError(
+                f"ContentType was once computed to be {ctype} and another one {new_ctype}"
+            )
+
+        ctype = new_ctype
+        app_label = new_app_label
+        codenames.add(new_codename)
+
+    if queryset is None:
+        queryset = ctype.model_class()._default_manager.all()
+    elif ctype != get_content_type(queryset.model):
+        raise MixedContentTypeError("Content type for given perms and queryset differs")
+
+    # Superuser has access to all objects
+    if user.is_superuser:
+        return queryset
+
+    # The anonymous user can have permissions
+    if user.is_anonymous:
+        user = get_anonymous_user()
+
+    # If the user has a model-level permission, we don't need to filter on it
+    model_perms = {code for code in codenames if user.has_perm(ctype.app_label + "." + code)}
+    for code in model_perms:
+        codenames.discard(code)
+    # We may be done
+    if len(codenames) == 0:
+        return queryset
+
+    # Now we should extract the list of pk values for which we would filter the queryset
+    role_model = get_role_obj_perms_model(queryset.model)
+    perms_queryset = (
+        role_model.objects.filter(role__in=user.all_roles())
+        .filter(permission__content_type=ctype)
+        .filter(permission__codename__in=codenames)
+    )
+
+    if len(codenames) > 1:
+        perms_queryset = (
+            perms_queryset.values(pk_field)
+            .annotate(object_pk_count=Count(pk_field))
+            .filter(object_pk_count__gte=len(codenames))
+        )
+
+    # pk is either UUID or an integer type, while object_pk is a varchar
+    pk = queryset.model._meta.pk
+
+    # From here on, we'll play a type normalization game. Postgresql has strict
+    # type matching for types varchars can't be implicitly coerced into uuids
+    # and vice versa, same with integers.
+
+    # thus comparing uuid/int with varchar fails on query execution.
+    normalized_pk_field = f"t__normalized_{pk_field}"
+    normalized_pk_field_lookup = f"{normalized_pk_field}__in"
+
+    perms_queryset = perms_queryset.annotate(
+        **{normalized_pk_field: Cast(pk_field, pk)}
+    ).values_list(normalized_pk_field, flat=True)
+
+    if getattr(queryset.model, "parents", None) is not None:
+        # here, we play the same game pretty much, this time we extend the
+        # subject part of the permissions to include group children.
+        #
+        # Meaning: granting view permissions on a group parent, gives you view
+        # permissions on the children.
+
+        normalized_parents_field = f"t__normalized_{queryset.model.parents.field.name}"
+        normalized_parents_field_lookup = f"{normalized_parents_field}__in"
+
+        # Pull now groups matching the current pks returned by the original
+        # query _and_ matching the parents.
+        values_list = perms_queryset.values_list(normalized_pk_field, flat=True)
+        perms_queryset = (
+            queryset.model.objects.annotate(
+                **{
+                    normalized_pk_field: Cast(pk.name, pk),
+                    normalized_parents_field: Cast(queryset.model.parents.field.name, pk),
+                }
+            ).filter(
+                Q(**{normalized_pk_field_lookup: values_list})
+                | Q(**{normalized_parents_field_lookup: values_list})
+            )
+        ).values_list(normalized_pk_field, flat=True)
+
+    # at this point now the `normalized_pk_field` field in the `perms_queryset` is
+    # guaranteed to be the target type (uuid, integer, varchar).
+    queryset = queryset.annotate(**{normalized_pk_field: Cast(pk.name, pk)})
+    # Now at this point `normalized_pk_field` in the `queryset` is guaranteed to
+    # be the target type (uuid, integer, varchar).
+
+    # Now both sides of the comparison are homogeneus, the query planner can
+    # take _any_ decision with regards to evaluating these conditions.
+    # and still no type mismatch will happen.
+    return queryset.filter(**{normalized_pk_field_lookup: perms_queryset})

authentik/suse/settings.py

@@ -90,3 +90,5 @@
     endpoint_name: CONFIG.get_bool(f"suse.override_endpoint.{endpoint_name}", False)
     for endpoint_name in known_endpoints
 }
+
+USE_CUSTOM_GUARDIAN = CONFIG.get_bool("suse.use_custom_guardian", False)

authentik/suse/tests/test_rbac_permission_group_descendants.py

@@ -0,0 +1,88 @@
+"""Test RBAC permissions group descendants"""
+
+from django.test.utils import override_settings
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.models import Group
+from authentik.core.tests.utils import create_test_user
+from authentik.lib.generators import generate_id
+
+
+class TestRBACGroupDescendants(APITestCase):
+    """
+    Test that granting permissions upon a group, also grant the same
+    permissionsto the children groups
+    """
+
+    def setUp(self) -> None:
+        self.user = create_test_user()
+
+        self.group_admin = Group.objects.create(
+            name=f"admin_{generate_id(10)}",
+        )
+        self.group_admin.users.set([self.user])
+
+        self.target_parent_group = Group.objects.create(
+            name=f"parent_{generate_id(10)}",
+        )
+        self.target_child_group = Group.objects.create(
+            name=f"child_{generate_id(10)}",
+        )
+        self.target_child_group.parents.set([self.target_parent_group])
+
+        self.group_admin.assign_perms_to_managed_role(
+            "authentik_core.view_group", obj=self.target_parent_group
+        )
+
+    # Baseline
+
+    @override_settings(USE_CUSTOM_GUARDIAN=False)
+    def test_original_permission_on_parent(self):
+        """Baseline assert: permission granted to the parent, action performed upon parent"""
+        self.client.force_login(self.user)
+        response = self.client.get(
+            reverse(
+                "authentik_api:group-detail",
+                kwargs={"pk": self.target_parent_group.pk},
+            )
+        )
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(USE_CUSTOM_GUARDIAN=False)
+    def test_original_permission_on_child(self):
+        """New behavior: permission granted to the parent, action performed upon child"""
+        self.client.force_login(self.user)
+        response = self.client.get(
+            reverse(
+                "authentik_api:group-detail",
+                kwargs={"pk": self.target_child_group.pk},
+            )
+        )
+        self.assertEqual(response.status_code, 404)
+
+    # SUSE Behavior
+
+    @override_settings(USE_CUSTOM_GUARDIAN=True)
+    def test_permission_on_parent(self):
+        """Baseline assert: permission granted to the parent, action performed upon parent"""
+        self.client.force_login(self.user)
+        response = self.client.get(
+            reverse(
+                "authentik_api:group-detail",
+                kwargs={"pk": self.target_parent_group.pk},
+            )
+        )
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(USE_CUSTOM_GUARDIAN=True)
+    def test_permission_on_child(self):
+        """New behavior: permission granted to the parent, action performed upon child"""
+        self.client.force_login(self.user)
+        response = self.client.get(
+            reverse(
+                "authentik_api:group-detail",
+                kwargs={"pk": self.target_child_group.pk},
+            )
+        )
+        self.assertEqual(response.status_code, 200)

packages/ak-guardian/guardian/shortcuts.py

@@ -3,6 +3,7 @@
 from functools import lru_cache
 from typing import Any, TypeVar
 
+from django.conf import settings
 from django.contrib.auth.models import Permission
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import (
@@ -14,6 +15,7 @@
 )
 from django.db.models.expressions import RawSQL
 
+from authentik.suse.guardian import shortcuts
 from guardian.core import ObjectPermissionChecker
 from guardian.ctypes import get_content_type
 from guardian.exceptions import (
@@ -181,6 +183,16 @@ def get_objects_for_user(  # noqa: PLR0912 PLR0915
     user: Any,
     perms: str | list[str],
     queryset: QuerySet | None = None,
+) -> QuerySet:
+    if getattr(settings, "USE_CUSTOM_GUARDIAN", False):
+        return shortcuts.get_objects_for_user(user, perms, queryset=queryset)
+    return _get_objects_for_user(user, perms, queryset=queryset)
+
+
+def _get_objects_for_user(  # noqa: PLR0912 PLR0915
+    user: Any,
+    perms: str | list[str],
+    queryset: QuerySet | None = None,
 ) -> QuerySet:
     """Get objects that a user has *all* the supplied permissions for.