"""Final foundation audit: real HTTP boundaries and additional scope/race attacks.""" import uuid from django.conf import settings from django.contrib.auth import authenticate, get_user_model, SESSION_KEY from django.contrib.auth.models import Group from django.test import Client, TestCase, TransactionTestCase from django.urls import reverse from apps.organization import assignment_services, services as organization_services from apps.organization.models import UserOrganizationAssignment as Assignment from apps.organization import test_assignment_concurrency as concurrency from . import services, test_authorization as fixtures from .authorization import authorized_queryset, is_authorized class AuthenticationSecurityTests(TestCase): @classmethod def setUpTestData(cls): cls.password = "Audit-only-Password-913!" cls.user = get_user_model().objects.create_user(username="audit-user", password=cls.password) cls.staff = get_user_model().objects.create_user(username="audit-staff", password=cls.password, is_staff=True) cls.admin = get_user_model().objects.create_superuser(username="audit-admin", password=cls.password) def test_native_authentication_hashing_identity_and_inactive_denial(self): original_id = self.user.pk self.assertIsInstance(original_id, uuid.UUID) self.assertEqual(authenticate(username=self.user.username, password=self.password), self.user) self.assertIsNone(authenticate(username=self.user.username, password="wrong")) self.assertNotEqual(self.user.password, self.password) self.assertNotIn(self.password, str(self.user)) self.user.is_active = False self.user.save(update_fields=["is_active"]) self.user.refresh_from_db() self.assertEqual(self.user.pk, original_id) self.assertIsNone(authenticate(username=self.user.username, password=self.password)) def test_admin_login_staff_rules(self): for user, expected in [(self.user, 200), (self.staff, 302), (self.admin, 302)]: with self.subTest(user=user.username): client = Client() response = client.post(reverse("admin:login"), { "username": user.username, "password": self.password, "next": reverse("admin:index"), }) self.assertEqual(response.status_code, expected) self.assertEqual(SESSION_KEY in client.session, expected == 302) self.admin.is_active = False self.admin.save(update_fields=["is_active"]) self.assertIsNone(authenticate(username=self.admin.username, password=self.password)) def test_staff_without_model_permissions_cannot_manage_access(self): self.client.force_login(self.staff) self.assertEqual(self.client.get(reverse("admin:index")).status_code, 200) self.assertEqual(self.client.get(reverse("admin:access_role_changelist")).status_code, 403) def test_deactivation_invalidates_authenticated_admin_access(self): self.client.force_login(self.admin) self.admin.is_active = False self.admin.save(update_fields=["is_active"]) self.assertEqual(self.client.get(reverse("admin:index")).status_code, 302) def test_logout_flushes_session_and_get_cannot_logout(self): self.client.force_login(self.admin) self.assertEqual(self.client.get(reverse("admin:logout")).status_code, 405) self.assertIn(SESSION_KEY, self.client.session) self.assertEqual(self.client.post(reverse("admin:logout")).status_code, 200) self.assertNotIn(SESSION_KEY, self.client.session) self.assertEqual(self.client.get(reverse("admin:index")).status_code, 302) def test_admin_password_change_hashes_and_invalidates_other_session(self): self.client.force_login(self.admin) other = Client() other.force_login(self.user) new_password = "Audit-new-Password-724!" response = self.client.post(reverse("admin:auth_user_password_change", args=[self.user.pk]), { "password1": new_password, "password2": new_password, }) self.assertEqual(response.status_code, 302) self.user.refresh_from_db() self.assertTrue(self.user.check_password(new_password)) self.assertFalse(self.user.check_password(self.password)) self.assertNotEqual(self.user.password, new_password) other.get(reverse("admin:index")) self.assertNotIn(SESSION_KEY, other.session) response = self.client.get(reverse("admin:accounts_user_change", args=[self.user.pk])) self.assertNotContains(response, new_password) def test_csrf_required_for_login_and_authenticated_mutations(self): client = Client(enforce_csrf_checks=True) self.assertEqual(client.post(reverse("admin:login"), { "username": self.admin.username, "password": self.password, }).status_code, 403) client.force_login(self.admin) self.assertEqual(client.post(reverse("admin:access_role_add"), { "code": "CSRF", "name": "Rejected", "is_active": "on", }).status_code, 403) self.assertEqual(client.post(reverse("admin:logout")).status_code, 403) client.get(reverse("admin:index")) token = client.cookies[settings.CSRF_COOKIE_NAME].value self.assertEqual(client.post(reverse("admin:access_role_add"), { "code": "CSRF", "name": "Accepted", "is_active": "on", "csrfmiddlewaretoken": token, }).status_code, 302) class FinalAuthorizationAuditTests(TestCase): setUpTestData = classmethod(fixtures.AuthorizationTests.setUpTestData.__func__) grant = fixtures.AuthorizationTests.grant def visible(self, user, permission=None, queryset=None): return list(authorized_queryset(user=user, permission=permission or self.permission, queryset=Assignment.objects.all() if queryset is None else queryset)) def targets(self): local = assignment_services.create_assignment(user=self.other_user, company=self.company, region=self.region, service_center=self.center) foreign = assignment_services.create_assignment(user=self.other_user, company=self.foreign_company, region=self.foreign_region, service_center=self.foreign_center) return local, foreign def test_assignment_queryset_cross_company_mixing_filters_and_order(self): local, foreign = self.targets() self.grant(role=self.change_role) self.grant(company=self.foreign_company) queryset = Assignment.objects.filter(user=self.other_user).order_by("-created_at", "id") self.assertEqual(self.visible(self.user, "organization.change_servicecenter", queryset), [local]) self.assertEqual(self.visible(self.user, queryset=queryset), [foreign]) self.assertFalse(is_authorized(user=self.user, permission="organization.change_servicecenter", target=foreign)) self.assertFalse(is_authorized(user=self.user, permission=self.permission, target=local)) self.assertEqual(self.visible(self.superuser, queryset=queryset), list(queryset)) def test_assignment_queryset_staff_direct_group_and_no_roles(self): self.targets() assignment_services.create_assignment(user=self.user, company=self.company) self.assertEqual(self.visible(self.user), []) self.user.is_staff = True self.user.save(update_fields=["is_staff"]) self.assertEqual(self.visible(self.user), []) self.user.user_permissions.add(self.view) self.assertTrue(get_user_model().objects.get(pk=self.user.pk).has_perm(self.permission)) self.assertEqual(self.visible(self.user), []) self.user.user_permissions.clear() group = Group.objects.create(name="Native audit") group.permissions.add(self.view) self.user.groups.add(group) self.assertTrue(get_user_model().objects.get(pk=self.user.pk).has_perm(self.permission)) self.assertEqual(self.visible(self.user), []) def test_assignment_queryset_each_revoked_path_and_inactive_superuser(self): local, foreign = self.targets() source, path = self.grant() queryset = Assignment.objects.filter(user=self.other_user) self.assertEqual(self.visible(self.user, queryset=queryset), [local]) for parent in [self.role, path, source, self.user]: with self.subTest(layer=type(parent).__name__): # Deliberately bypass cascades to attack each authorization predicate. type(parent).objects.filter(pk=parent.pk).update(is_active=False) self.assertEqual(self.visible(self.user, queryset=queryset), []) type(parent).objects.filter(pk=parent.pk).update(is_active=True) self.assertEqual(set(self.visible(self.superuser, queryset=queryset)), {local, foreign}) self.superuser.is_active = False self.superuser.save(update_fields=["is_active"]) self.assertEqual(self.visible(self.superuser), []) def test_assignment_queryset_unsupported_shapes_fail_closed(self): self.targets() self.grant() queryset = Assignment.objects.all() for user in [self.user, self.superuser]: for invalid in [queryset[:1], queryset.union(queryset), queryset.intersection(queryset), queryset.difference(queryset)]: self.assertEqual(self.visible(user, queryset=invalid), []) def test_admin_reactivation_does_not_restore_descendants_or_paths(self): source, path = self.grant(region=self.region, service_center=self.center, department=self.department) organization_services.deactivate_company(company=self.company) self.client.force_login(self.superuser) for obj in [self.company, self.region, self.center, self.department, source]: response = self.client.post(reverse(f"admin:{obj._meta.app_label}_{obj._meta.model_name}_changelist"), { "action": "reactivate_selected", "_selected_action": [str(obj.pk)], }) self.assertEqual(response.status_code, 302) obj.refresh_from_db() self.assertTrue(obj.is_active) path.refresh_from_db() self.assertFalse(path.is_active) self.assertFalse(is_authorized(user=self.user, permission=self.permission, target=self.center)) self.assertFalse(Assignment.objects.get(pk=source.pk).is_primary) class FinalConcurrencyAuditTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent setUp = concurrency.AssignmentConcurrencyTests.setUp def test_region_deactivation_then_assignment_creation(self): self.run_concurrent(lambda: organization_services.deactivate_region(region=self.region), lambda: assignment_services.create_assignment(user=self.user, company=self.company, region=self.region, service_center=self.center), expected="validation") self.assertFalse(Assignment.objects.exists()) def test_assignment_creation_then_region_deactivation(self): self.run_concurrent(lambda: assignment_services.create_assignment(user=self.user, company=self.company, region=self.region, service_center=self.center, is_primary=True), lambda: organization_services.deactivate_region(region=self.region), expected="success") assignment = Assignment.objects.get() self.assertFalse(assignment.is_active) self.assertFalse(assignment.is_primary)