"""Automatic permission management: catalog, registry, auto-enforcement, role assignments, and global grants/revokes. """ import pytest from django.conf import settings from django.test import override_settings from django.urls import include, path from rest_framework import status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.routers import DefaultRouter from infrasynth.security.catalog import build_catalog, permission_for, sync_permissions from infrasynth.security.models import Grant, Permission, Revoke, Role, RoleAssignment from infrasynth.security.permissions import AutoPermission, automatic_permissions from infrasynth.security.registry import PermissionRegistry from infrasynth.security.serializers import RoleSerializer from infrasynth.security.services import AuthorizationService from infrasynth.security.viewsets import InfraSynthModelViewSet from infrasynth.tenancy.context import tenant_context from infrasynth.tenancy.models import Tenant def sec_settings(**overrides): return {**settings.INFRASYNTH_SECURITY, **overrides} @pytest.fixture def clean_permission_registry(): snapshot = dict(PermissionRegistry._permissions) yield PermissionRegistry._permissions = snapshot # --- an end-to-end consumer of the kit base viewset -------------------------- class _RoleViewSet(InfraSynthModelViewSet): from infrasynth.security.models import Role as _Role queryset = _Role.objects.all() serializer_class = RoleSerializer action_permissions = {"custom": "custom.role_action"} @action(detail=False, methods=["get"], url_path="custom") def custom(self, request): return Response({"ok": True}) router = DefaultRouter() router.register("roles", _RoleViewSet, basename="auto-test-roles") urlpatterns = [path("auto/", include(router.urls))] AUTO_URL = "/auto/roles/" class TestPermissionDerivation: def test_permission_for_drf_actions(self, db): assert permission_for(Role, "list") == "infrasynth_security.view_role" assert permission_for(Role, "retrieve") == "infrasynth_security.view_role" assert permission_for(Role, "create") == "infrasynth_security.add_role" assert permission_for(Role, "update") == "infrasynth_security.change_role" assert permission_for(Role, "partial_update") == "infrasynth_security.change_role" assert permission_for(Role, "destroy") == "infrasynth_security.delete_role" def test_automatic_permissions_priority(self, db): view = _RoleViewSet() view.action = "list" assert automatic_permissions(view) == ["infrasynth_security.view_role"] view.action = "custom" assert automatic_permissions(view) == ["custom.role_action"] view.required_permissions = ["explicit.perm"] assert automatic_permissions(view) == ["explicit.perm"] class TestCatalog: def test_build_includes_model_and_custom(self, db, clean_permission_registry): PermissionRegistry.register("helpdesk.resolve_ticket", name="Resolve", group="Helpdesk") catalog = build_catalog() assert "infrasynth_security.view_role" in catalog assert catalog["infrasynth_security.view_role"].is_custom is False assert "helpdesk.resolve_ticket" in catalog assert catalog["helpdesk.resolve_ticket"].is_custom is True def test_kit_custom_permissions_registered(self, db): catalog = build_catalog() for codename in ("configs.manage", "platform.tenants.delete", "audit.view_api_logs"): assert codename in catalog def test_sync_is_idempotent_and_deactivates_missing(self, db, clean_permission_registry): first = sync_permissions() assert first["total"] > 0 second = sync_permissions() assert second["created"] == 0 assert second["updated"] == 0 # A stale entry is deactivated, not deleted. Permission.objects.create(codename="stale.perm", name="Stale", app_label="stale", is_custom=True) summary = sync_permissions() assert summary["deactivated"] == 1 stale = Permission.objects.get(codename="stale.perm") assert stale.is_active is False # Re-registering reactivates. PermissionRegistry.register("stale.perm", name="Stale") summary = sync_permissions() assert summary["reactivated"] == 1 assert Permission.objects.get(codename="stale.perm").is_active is True def test_sync_command_runs(self, db): from django.core.management import call_command call_command("sync_permissions") class TestAutoPermissionIntegration: @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_owner_bypasses(self, authenticated_client, db): assert authenticated_client.get(AUTO_URL).status_code == status.HTTP_200_OK @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_member_denied_without_permission(self, member_client, db): assert member_client.get(AUTO_URL).status_code == status.HTTP_403_FORBIDDEN @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_member_allowed_with_grant(self, member_client, member_user, db): Grant.objects.create(user=member_user, codename="infrasynth_security.view_role") assert member_client.get(AUTO_URL).status_code == status.HTTP_200_OK @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_member_allowed_with_tenant_role_assignment(self, member_client, member_user, tenant, db): role = Role.objects.create( tenant=tenant, name="Viewer", slug="viewer", permissions=["infrasynth_security.view_role"] ) RoleAssignment.objects.create(tenant=tenant, user=member_user, role=role) assert member_client.get(AUTO_URL).status_code == status.HTTP_200_OK @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_custom_action_codename(self, member_client, member_user, db): assert member_client.get(f"{AUTO_URL}custom/").status_code == status.HTTP_403_FORBIDDEN Grant.objects.create(user=member_user, codename="custom.role_action") assert member_client.get(f"{AUTO_URL}custom/").status_code == status.HTTP_200_OK @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_create_requires_add_permission(self, member_client, member_user, db): response = member_client.post(AUTO_URL, {"name": "New", "slug": "new-role", "permissions": []}, format="json") assert response.status_code == status.HTTP_403_FORBIDDEN Grant.objects.create(user=member_user, codename="infrasynth_security.add_role") response = member_client.post(AUTO_URL, {"name": "New", "slug": "new-role", "permissions": []}, format="json") assert response.status_code == status.HTTP_201_CREATED @override_settings(ROOT_URLCONF="tests.test_security.test_permissions") def test_off_mode_abstains(self, member_client, db): with override_settings(INFRASYNTH_SECURITY=sec_settings(AUTO_PERMISSIONS="off")): assert member_client.get(AUTO_URL).status_code == status.HTTP_200_OK def test_auto_permission_abstains_without_model(self, db, user): class _Plain: action = "list" request = type("R", (), {"user": user})() assert AutoPermission().has_permission(request, _Plain()) is True class TestGlobalRolesAndOverrides: def test_global_role_applies_in_every_tenant(self, user, tenant, db): other = Tenant.objects.create(slug="other-global", name="Other") global_role = Role.objects.create(name="Global", slug="global", permissions=["x.perm"]) global_role.users.add(user) authz = AuthorizationService() with tenant_context(tenant): assert authz.has_permission(user, "x.perm") is True with tenant_context(other): assert authz.has_permission(user, "x.perm") is True def test_role_assignment_is_tenant_scoped(self, user, tenant, db): other = Tenant.objects.create(slug="other-role", name="Other") role = Role.objects.create(tenant=tenant, name="Scoped", slug="scoped", permissions=["y.perm"]) RoleAssignment.objects.create(tenant=tenant, user=user, role=role) authz = AuthorizationService() with tenant_context(tenant): assert authz.has_permission(user, "y.perm") is True with tenant_context(other): assert authz.has_permission(user, "y.perm") is False def test_global_grant_applies_everywhere(self, user, tenant, db): other = Tenant.objects.create(slug="other-grant", name="Other") Grant(user=user, codename="z.perm").save(force_global=True) authz = AuthorizationService() with tenant_context(tenant): assert authz.has_permission(user, "z.perm") is True with tenant_context(other): assert authz.has_permission(user, "z.perm") is True def test_global_revoke_blocks_everywhere(self, user, tenant, db): other = Tenant.objects.create(slug="other-revoke", name="Other") role = Role.objects.create(name="R", slug="r", permissions=["z.perm"]) role.users.add(user) Revoke(user=user, codename="z.perm").save(force_global=True) authz = AuthorizationService() with tenant_context(tenant): assert authz.has_permission(user, "z.perm") is False with tenant_context(other): assert authz.has_permission(user, "z.perm") is False def test_context_created_grant_stays_tenant_scoped(self, user, tenant, db): Grant.objects.create(user=user, codename="t.perm") assert Grant.objects.get(codename="t.perm").tenant_id == tenant.pk class TestRoleAndGrantApi: def test_tenant_role_created_in_tenant(self, authenticated_client, tenant, db): response = authenticated_client.post( "/api/v1/auth/roles/", {"name": "Tenant Role", "slug": "tenant-role", "permissions": []}, format="json" ) assert response.status_code == status.HTTP_201_CREATED assert Role.objects.get(slug="tenant-role").tenant_id == tenant.pk def test_global_role_requires_platform_permission(self, member_client, member_user, db): response = member_client.post( "/api/v1/auth/roles/", {"name": "Global", "slug": "global-role", "permissions": []}, format="json" ) assert response.status_code == status.HTTP_403_FORBIDDEN def test_strict_role_validation(self, authenticated_client, db): with override_settings(INFRASYNTH_SECURITY=sec_settings(STRICT_PERMISSION_VALIDATION=True)): response = authenticated_client.post( "/api/v1/auth/roles/", {"name": "Bad", "slug": "bad-role", "permissions": ["does.not.exist"]}, format="json", ) assert response.status_code == status.HTTP_400_BAD_REQUEST def test_global_grant_scope_requires_platform(self, authenticated_client, user, db): response = authenticated_client.post( "/api/v1/auth/grants/", {"user": user.pk, "codename": "p.perm", "scope": "global"}, format="json", ) assert response.status_code == status.HTTP_403_FORBIDDEN def test_global_grant_scope_allowed_with_platform_role(self, member_client, member_user, db): role = Role.objects.create( name="Platform", slug="platform", permissions=["security.manage_grants", "platform.roles.manage"] ) role.users.add(member_user) response = member_client.post( "/api/v1/auth/grants/", {"user": member_user.pk, "codename": "p.perm", "scope": "global"}, format="json", ) assert response.status_code == status.HTTP_201_CREATED assert Grant.all_objects.get(codename="p.perm").tenant_id is None class TestPermissionApi: def test_catalog_endpoint(self, authenticated_client, db): response = authenticated_client.get("/api/v1/auth/permissions/") assert response.status_code == status.HTTP_200_OK codenames = {entry["codename"] for entry in response.json()["results"]} assert "configs.manage" in codenames def test_catalog_requires_permission(self, member_client, db): assert member_client.get("/api/v1/auth/permissions/").status_code == status.HTTP_403_FORBIDDEN def test_catalog_filter_by_app(self, authenticated_client, db): response = authenticated_client.get("/api/v1/auth/permissions/?app_label=configs") assert response.status_code == status.HTTP_200_OK assert all(entry["app_label"] == "configs" for entry in response.json()["results"])