"""Authoritative bulk capability resolution must equal the single-capability path. ``apps.access.authorization.authorized_capabilities`` resolves many permissions in a bounded number of queries. Every value it returns must be identical to what ``apps.operations.queries.capable`` answers one permission at a time, for every actor, scope and permission. Nothing here relaxes an existing assertion; it pins the bulk resolver to the single resolver so the two can never drift. """ import itertools from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser, Permission from django.db import connection from django.test import TestCase from django.test.utils import CaptureQueriesContext from apps.access.authorization import ( authorized_capabilities, authorized_capability_map, authorized_queryset) from apps.access.models import Role from apps.access.scopes import ADAPTERS, assignment_reaches from apps.access.services import create_role_assignment, set_role_permissions from apps.catalog.tests import make_catalog from apps.organization.assignment_services import create_assignment from apps.organization.assignments import UserOrganizationAssignment from apps.organization.models import Company, Department, Region, ServiceCenter from apps.organization.test_lifecycle import make_tree from apps.operations import queries as operations_queries BUSINESS_APPS = ("service", "commercial", "inventory", "devices", "frontdesk", "customers", "parts", "sla", "communications", "reporting") CAPABILITY_PERMISSIONS = sorted({ "{}.{}".format(app_label, codename) for app_label, codename in Permission.objects.values_list("content_type__app_label", "codename").distinct() if app_label in BUSINESS_APPS}) class BulkFixture(TestCase): @classmethod def setUpTestData(cls): cls.company, cls.region, cls.center, cls.department = make_tree("BULK") cls.other_company, cls.other_region, cls.other_center, _ = make_tree("BULKOTHER") make_catalog("BULK") cls.sibling_center = ServiceCenter.objects.create( company=cls.company, region=cls.region, code="BULKB", name="Sibling", center_type="OWN") cls.sibling_region = Region.objects.create(company=cls.company, code="BULKR2", name="Other region") cls.other_department = Department.objects.create(company=cls.company, code="BULK2", name="Other department") _names = itertools.count() def persona(self, permissions, *, scope=None, superuser=False): """A scoped persona. `permissions` may be empty, which grants nothing.""" name = "bulk-{}".format(next(self._names)) if superuser: return get_user_model().objects.create_superuser(username=name) user = get_user_model().objects.create_user(username=name) if permissions: role = Role.objects.create(code=name.upper(), name=name) set_role_permissions(role=role, permissions=list(permissions)) create_role_assignment(user=user, role=role, organization_assignment=create_assignment(user=user, **(scope or {"company": self.company}))) return user def permissions_for(self, permissions): keys = set(permissions) return Permission.objects.filter( content_type__app_label__in={key.split(".")[0] for key in keys}, codename__in={key.split(".")[1] for key in keys}) def sample(self, count=8): return CAPABILITY_PERMISSIONS[:count] def scoped(self, **overrides): return {"company": self.company, **overrides} class SemanticEquivalenceTests(BulkFixture): """A. Bulk equals single across actors, scopes, permission sets and scope modes.""" def actor_matrix(self): scopes = { "company": {}, "service_center": {"region": self.region, "service_center": self.center}, "sibling_center": {"region": self.region, "service_center": self.sibling_center}, "region_only": {"region": self.sibling_region}, "department": {"department": self.department}, "other_department": {"department": self.other_department}, "other_company": {"company": self.other_company, "region": self.other_region, "service_center": self.other_center}, } for name, scope in scopes.items(): built = {"company": self.company, **scope} yield "{}:sample".format(name), self.persona(self.permissions_for(self.sample()), scope=built) yield "{}:none".format(name), self.persona([], scope=built) # The full permission set is exercised on the two most common scopes only, # so the equivalence matrix stays large but the suite stays quick. yield "company:all".format(name), self.persona( self.permissions_for(CAPABILITY_PERMISSIONS), scope=built) yield "service_center:all".format(name), self.persona( self.permissions_for(CAPABILITY_PERMISSIONS), scope=built) yield "superuser", self.persona([], superuser=True) yield "superuser+assignment", self.persona( self.permissions_for(["service.view_servicecase"])) def test_bulk_matches_single_for_every_actor_scope_and_permission(self): checked = 0 for label, actor in self.actor_matrix(): for permissions in (CAPABILITY_PERMISSIONS, self.sample(), []): for company_only in (False, True): bulk = authorized_capabilities( user=actor, permissions=permissions, company_only=company_only) for permission in permissions: checked += 1 self.assertEqual( bulk[permission], operations_queries.capable(actor, permission, company_only=company_only), "{} {} company_only={}".format(label, permission, company_only)) self.assertGreater(checked, 3000, "the matrix must actually cover the permission set") def test_capability_map_matches_both_single_scope_modes(self): for label, actor in self.actor_matrix(): with self.subTest(actor=label): combined = authorized_capability_map(user=actor, permissions=self.sample()) for permission in self.sample(): for company_only in (False, True): self.assertEqual( combined[permission, company_only], operations_queries.capable(actor, permission, company_only=company_only), "{} {} company_only={}".format(label, permission, company_only)) def test_capability_map_is_exactly_the_two_modes_of_the_single_call(self): for _label, actor in self.actor_matrix(): with self.subTest(actor=actor.username): combined = authorized_capability_map(user=actor, permissions=self.sample()) general = authorized_capabilities(user=actor, permissions=self.sample()) company = authorized_capabilities( user=actor, permissions=self.sample(), company_only=True) for permission in self.sample(): self.assertEqual(combined[permission, False], general[permission]) self.assertEqual(combined[permission, True], company[permission]) def test_company_only_is_never_satisfied_by_a_service_center_scope(self): reader = self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(region=self.region, service_center=self.center)) self.assertFalse(authorized_capabilities( user=reader, permissions=["service.view_servicecase"], company_only=True)["service.view_servicecase"]) self.assertTrue(authorized_capabilities( user=reader, permissions=["service.view_servicecase"], company_only=False)["service.view_servicecase"]) def test_mixed_allowed_and_denied_set_is_answered_per_permission(self): allowed, denied = "service.view_servicecase", "commercial.view_serviceinvoice" actor = self.persona(self.permissions_for([allowed]), scope=self.scoped()) self.assertEqual(authorized_capabilities(user=actor, permissions=[allowed, denied]), {allowed: True, denied: False}) def test_unknown_and_malformed_permissions_are_denied(self): actor = self.persona(self.permissions_for(CAPABILITY_PERMISSIONS)) unusable = ["service.no_such_permission", "not-a-permission", "service.view_servicecase.extra", "", "service.", ".view_servicecase", None, 7] results = authorized_capabilities(user=actor, permissions=["service.view_servicecase", *unusable]) self.assertTrue(results["service.view_servicecase"]) for denied in unusable: self.assertFalse(results[denied], denied) self.assertFalse(operations_queries.capable(actor, denied)) def test_duplicate_requests_are_answered_once(self): actor = self.persona(self.permissions_for(["service.view_servicecase"])) self.assertEqual(authorized_capabilities( user=actor, permissions=["service.view_servicecase"] * 3), {"service.view_servicecase": True}) def test_empty_request_is_safe(self): actor = self.persona(self.permissions_for(["service.view_servicecase"])) self.assertEqual(authorized_capabilities(user=actor, permissions=[]), {}) self.assertEqual(authorized_capability_map(user=actor, permissions=[]), {}) def test_bulk_never_claims_more_than_authorized_queryset(self): for label, actor in self.actor_matrix(): for permission in self.sample(3): with self.subTest(actor=label, permission=permission): for model, company_only in ((Company, True), (ServiceCenter, False)): holds = authorized_capabilities( user=actor, permissions=[permission], company_only=company_only)[permission] direct = authorized_queryset(user=actor, permission=permission, queryset=model._default_manager.all()).exists() if holds: self.assertTrue(direct, "{} {} {}".format(label, permission, model.__name__)) elif company_only: self.assertFalse(direct) class ScopeIsolationTests(BulkFixture): """D and H. Cross-company, sibling-center and department isolation.""" def test_cross_company_scope_reaches_only_its_own_center(self): actor = self.persona(self.permissions_for(["service.view_servicecase"]), scope={"company": self.other_company, "region": self.other_region, "service_center": self.other_center}) self.assertTrue(authorized_capabilities( user=actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) self.assertEqual( set(authorized_queryset(user=actor, permission="service.view_servicecase", queryset=ServiceCenter.objects.all()).values_list("pk", flat=True)), {self.other_center.pk}) def test_sibling_center_does_not_grant_another_center_in_the_same_region(self): actor = self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(region=self.region, service_center=self.sibling_center)) self.assertEqual( set(authorized_queryset(user=actor, permission="service.view_servicecase", queryset=ServiceCenter.objects.all()).values_list("pk", flat=True)), {self.sibling_center.pk}) def test_department_scope_does_not_leak_to_other_departments_or_centers(self): actor = self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(department=self.department)) self.assertEqual( set(authorized_queryset(user=actor, permission="service.view_servicecase", queryset=Department.objects.all()).values_list("pk", flat=True)), {self.department.pk}) self.assertFalse(authorized_queryset(user=actor, permission="service.view_servicecase", queryset=ServiceCenter.objects.all()).exists(), "department scope must not reach a service center") def test_one_actor_never_inherits_another_actors_result(self): capable_actor = self.persona(self.permissions_for(["service.view_servicecase"])) denied_actor = self.persona(self.permissions_for(["commercial.view_serviceinvoice"])) self.assertTrue(authorized_capabilities( user=capable_actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) self.assertFalse(authorized_capabilities( user=denied_actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) self.assertFalse(authorized_capabilities( user=capable_actor, permissions=["commercial.view_serviceinvoice"])["commercial.view_serviceinvoice"]) self.assertTrue(authorized_capabilities( user=denied_actor, permissions=["commercial.view_serviceinvoice"])["commercial.view_serviceinvoice"]) def test_deactivating_a_center_removes_the_capability_immediately(self): actor = self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(region=self.region, service_center=self.center)) self.assertTrue(authorized_capabilities( user=actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) ServiceCenter.objects.filter(pk=self.center.pk).update(is_active=False) fresh = get_user_model().objects.get(pk=actor.pk) self.assertFalse(authorized_capabilities( user=fresh, permissions=["service.view_servicecase"])["service.view_servicecase"], "a capability must be recomputed from current state, never remembered") def test_deactivating_a_role_removes_the_capability_immediately(self): actor = self.persona(self.permissions_for(["service.view_servicecase"])) self.assertTrue(authorized_capabilities( user=actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) # Deactivate below the model guard; role lifecycle is covered by its own tests. Role.objects.filter(code=actor.username.upper()).update(is_active=False) fresh = get_user_model().objects.get(pk=actor.pk) self.assertFalse(authorized_capabilities( user=fresh, permissions=["service.view_servicecase"])["service.view_servicecase"]) def test_deactivating_a_role_assignment_removes_the_capability_immediately(self): actor = self.persona(self.permissions_for(["service.view_servicecase"])) self.assertTrue(authorized_capabilities( user=actor, permissions=["service.view_servicecase"])["service.view_servicecase"]) actor.business_role_assignments.update(is_active=False) fresh = get_user_model().objects.get(pk=actor.pk) self.assertFalse(authorized_capabilities( user=fresh, permissions=["service.view_servicecase"])["service.view_servicecase"]) def test_inactive_and_anonymous_actors_have_no_capability(self): actor = self.persona(self.permissions_for(["service.view_servicecase"])) get_user_model().objects.filter(pk=actor.pk).update(is_active=False) self.assertEqual(authorized_capabilities(user=get_user_model().objects.get(pk=actor.pk), permissions=["service.view_servicecase"]), {"service.view_servicecase": False}) self.assertEqual(authorized_capabilities(user=AnonymousUser(), permissions=["service.view_servicecase"]), {"service.view_servicecase": False}) class ContainmentMirrorTests(BulkFixture): """The bulk resolver's inverse containment clause must equal the adapter clause.""" def test_every_scope_model_with_an_adapter_has_an_inverse_clause(self): for model in (Company, Region, ServiceCenter, Department): with self.subTest(model=model.__name__): self.assertIn(model, ADAPTERS) self.assertIsNotNone(assignment_reaches(model)) def test_inverse_clause_never_widens_or_narrows_authorization(self): actors = [ self.persona(self.permissions_for(["service.view_servicecase"])), self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(region=self.region, service_center=self.center)), self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(department=self.department)), self.persona(self.permissions_for(["service.view_servicecase"]), scope={"company": self.other_company}), ] for actor in actors: for assignment in UserOrganizationAssignment.objects.filter(user=actor): for model in (Company, Region, ServiceCenter, Department): # The mirror answers "does this assignment reach any valid # target of this model", which must equal "the single-capability # path authorizes at least one target of this model". bulk_reaches = UserOrganizationAssignment.objects.filter( pk=assignment.pk).filter(assignment_reaches(model)).exists() direct = any( authorized_queryset(user=actor, permission="service.view_servicecase", queryset=model.objects.filter(pk=target.pk)).exists() for target in model.objects.all()) self.assertEqual(bulk_reaches, direct, "{} assignment={} model={}".format( actor.username, assignment.pk, model.__name__)) class QueryBoundTests(BulkFixture): """E. The resolution cost must not grow with the number of permissions asked.""" def test_query_count_is_independent_of_permission_count(self): actor = self.persona(self.permissions_for(CAPABILITY_PERMISSIONS)) counts = [] for size in (1, 4, len(CAPABILITY_PERMISSIONS)): with CaptureQueriesContext(connection) as captured: authorized_capabilities(user=actor, permissions=CAPABILITY_PERMISSIONS[:size]) counts.append(len(captured.captured_queries)) self.assertEqual(counts[0], counts[1], counts) self.assertEqual(counts[1], counts[2], counts) self.assertLessEqual(counts[2], 6) def test_capability_map_costs_no_more_than_resolving_one_mode(self): actor = self.persona(self.permissions_for(CAPABILITY_PERMISSIONS)) with CaptureQueriesContext(connection) as combined: authorized_capability_map(user=actor, permissions=CAPABILITY_PERMISSIONS) with CaptureQueriesContext(connection) as general: authorized_capabilities(user=actor, permissions=CAPABILITY_PERMISSIONS, company_only=False) self.assertLessEqual(len(combined.captured_queries), len(general.captured_queries)) self.assertLessEqual(len(combined.captured_queries), 6) def test_single_capability_api_is_unchanged_and_still_works(self): company_actor = self.persona(self.permissions_for(["service.view_servicecase"])) center_actor = self.persona(self.permissions_for(["service.view_servicecase"]), scope=self.scoped(region=self.region, service_center=self.center)) self.assertTrue(operations_queries.capable(company_actor, "service.view_servicecase")) self.assertTrue(operations_queries.capable(company_actor, "service.view_servicecase", company_only=True)) self.assertTrue(operations_queries.capable(center_actor, "service.view_servicecase")) self.assertFalse(operations_queries.capable(center_actor, "service.view_servicecase", company_only=True)) self.assertFalse(operations_queries.capable(company_actor, "service.no_such_permission"))