"""Permission catalog: auto-derivation + sync. Every concrete model contributes ``view``/``add``/``change``/``delete`` permissions (Django-style codenames ``{app_label}.{verb}_{model_name}``). Apps add custom permissions through :class:`infrasynth.security.registry.PermissionRegistry`. Both are merged into the :class:`infrasynth.security.models.Permission` catalog so a UI can list and assign them, and so codenames can be validated. Enforcement itself does not require the catalog to be populated: the codename is derived from the model + action at request time. The catalog is metadata. """ from __future__ import annotations import logging from django.apps import apps from infrasynth.shared.settings_utils import get_setting from .registry import PermissionDefinition, PermissionRegistry logger = logging.getLogger(__name__) __all__ = [ "DRF_ACTION_VERBS", "permission_for", "build_catalog", "sync_permissions", ] # DRF viewset action -> Django permission verb. DRF_ACTION_VERBS: dict[str, str] = { "list": "view", "retrieve": "view", "create": "add", "update": "change", "partial_update": "change", "destroy": "delete", } # Apps/models that never expose a permission (framework internals, logs, the # catalog itself). Everything else — kit and consumer models — is included. DEFAULT_EXCLUDED_MODELS: frozenset[str] = frozenset( { "sessions.Session", "admin.LogEntry", "contenttypes.ContentType", "auth.Permission", "rest_framework.authtoken.Token", "token_blacklist.OutstandingToken", "token_blacklist.BlacklistedToken", "django_celery_results.TaskResult", "django_celery_results.GroupResult", "django_celery_beat.PeriodicTask", "django_celery_beat.IntervalSchedule", "django_celery_beat.CrontabSchedule", "django_celery_beat.SolarSchedule", "django_celery_beat.ClockedSchedule", "infrasynth_audit.ModelChangeLog", "infrasynth_audit.APIInteractionLog", "infrasynth_audit.SecurityEvent", "infrasynth_security.Permission", "infrasynth_security.TwoFactorConfig", "infrasynth_security.ALTCHAChallenge", } ) _VERBS = ("view", "add", "change", "delete") def permission_for(model, action: str) -> str: """Django-style codename for ``model`` and a DRF/verb ``action``.""" verb = DRF_ACTION_VERBS.get(action, action) opts = model._meta return f"{opts.app_label}.{verb}_{opts.model_name}" def _excluded_models() -> frozenset[str]: configured = get_setting("INFRASYNTH_SECURITY", "PERMISSION_EXCLUDE_MODELS", None) if not configured: return DEFAULT_EXCLUDED_MODELS return DEFAULT_EXCLUDED_MODELS | frozenset(configured) def _allowed_apps() -> list[str] | None: return list(get_setting("INFRASYNTH_SECURITY", "PERMISSION_APPS", None) or []) or None def _model_definitions() -> dict[str, PermissionDefinition]: excluded = _excluded_models() allowed_apps = _allowed_apps() definitions: dict[str, PermissionDefinition] = {} for model in apps.get_models(): opts = model._meta if opts.abstract or opts.proxy or opts.auto_created or not opts.managed: continue if opts.label in excluded: continue if allowed_apps is not None and opts.app_label not in allowed_apps: continue group = opts.app_label.replace("_", " ").title() model_name = opts.model_name or "" for verb in _VERBS: codename = f"{opts.app_label}.{verb}_{model_name}" definitions[codename] = PermissionDefinition( codename=codename, name=f"Can {verb} {opts.verbose_name}", app=opts.app_label, model=model_name, action=verb, group=group, is_custom=False, ) return definitions def build_catalog() -> dict[str, PermissionDefinition]: """Merged model-derived + custom-registered definitions (registry wins).""" catalog = _model_definitions() catalog.update(PermissionRegistry.all()) return catalog def sync_permissions(*, deactivate_missing: bool = True) -> dict[str, int]: """Upserts the catalog into the ``Permission`` table. Idempotent.""" from .models import Permission catalog = build_catalog() existing = {permission.codename: permission for permission in Permission.objects.all()} created = updated = reactivated = 0 for codename, definition in catalog.items(): fields = { "name": definition.name, "app_label": definition.app, "model": definition.model, "action": definition.action, "group": definition.group, "description": definition.description, "is_custom": definition.is_custom, } obj = existing.get(codename) if obj is None: Permission.objects.create(codename=codename, **fields) created += 1 continue changed = {key: value for key, value in fields.items() if getattr(obj, key) != value} if not obj.is_active: changed["is_active"] = True reactivated += 1 if changed: for key, value in changed.items(): setattr(obj, key, value) obj.save(update_fields=list(changed)) updated += 1 deactivated = 0 if deactivate_missing: for codename in set(existing) - set(catalog): permission = existing[codename] if permission.is_active: permission.is_active = False permission.save(update_fields=["is_active"]) deactivated += 1 return { "created": created, "updated": updated, "reactivated": reactivated, "deactivated": deactivated, "total": len(catalog), }