Agent Zero
201 lines
11 KiB
Python
201 lines
11 KiB
Python
"""Permission Registry — central catalog of all valid permissions.
|
|
|
|
Built at startup from core SYSTEM_PERMISSIONS and active plugin manifests.
|
|
Validates permission assignments and provides metadata for UI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Core system permissions ──
|
|
CORE_PERMISSIONS: list[dict[str, str]] = [
|
|
{"key": "companies:read", "label": "Companies: Read", "category": "core", "module": "companies"},
|
|
{"key": "companies:write", "label": "Companies: Write", "category": "core", "module": "companies"},
|
|
{"key": "companies:delete", "label": "Companies: Delete", "category": "core", "module": "companies"},
|
|
{"key": "contacts:read", "label": "Contacts: Read", "category": "core", "module": "contacts"},
|
|
{"key": "contacts:write", "label": "Contacts: Write", "category": "core", "module": "contacts"},
|
|
{"key": "contacts:delete", "label": "Contacts: Delete", "category": "core", "module": "contacts"},
|
|
{"key": "users:read", "label": "Users: Read", "category": "core", "module": "users"},
|
|
{"key": "users:write", "label": "Users: Write", "category": "core", "module": "users"},
|
|
{"key": "users:delete", "label": "Users: Delete", "category": "core", "module": "users"},
|
|
{"key": "roles:read", "label": "Roles: Read", "category": "core", "module": "roles"},
|
|
{"key": "roles:write", "label": "Roles: Write", "category": "core", "module": "roles"},
|
|
{"key": "roles:delete", "label": "Roles: Delete", "category": "core", "module": "roles"},
|
|
{"key": "groups:read", "label": "Groups: Read", "category": "core", "module": "groups"},
|
|
{"key": "groups:write", "label": "Groups: Write", "category": "core", "module": "groups"},
|
|
{"key": "groups:delete", "label": "Groups: Delete", "category": "core", "module": "groups"},
|
|
{"key": "audit:read", "label": "Audit Log: Read", "category": "core", "module": "audit"},
|
|
{"key": "settings:read", "label": "Settings: Read", "category": "core", "module": "settings"},
|
|
{"key": "settings:write", "label": "Settings: Write", "category": "core", "module": "settings"},
|
|
{"key": "plugins:read", "label": "Plugins: Read", "category": "core", "module": "plugins"},
|
|
{"key": "plugins:install", "label": "Plugins: Install", "category": "core", "module": "plugins"},
|
|
{"key": "plugins:configure", "label": "Plugins: Configure", "category": "core", "module": "plugins"},
|
|
{"key": "tenants:read", "label": "Tenants: Read", "category": "core", "module": "tenants"},
|
|
{"key": "tenants:write", "label": "Tenants: Write", "category": "core", "module": "tenants"},
|
|
{"key": "tenants:delete", "label": "Tenants: Delete", "category": "core", "module": "tenants"},
|
|
{"key": "notifications:read", "label": "Notifications: Read", "category": "core", "module": "notifications"},
|
|
{"key": "notifications:write", "label": "Notifications: Write", "category": "core", "module": "notifications"},
|
|
{"key": "attachments:read", "label": "Attachments: Read", "category": "core", "module": "attachments"},
|
|
{"key": "attachments:write", "label": "Attachments: Write", "category": "core", "module": "attachments"},
|
|
{"key": "attachments:delete", "label": "Attachments: Delete", "category": "core", "module": "attachments"},
|
|
{"key": "workflows:read", "label": "Workflows: Read", "category": "core", "module": "workflows"},
|
|
{"key": "workflows:write", "label": "Workflows: Write", "category": "core", "module": "workflows"},
|
|
{"key": "sequences:read", "label": "Sequences: Read", "category": "core", "module": "sequences"},
|
|
{"key": "sequences:write", "label": "Sequences: Write", "category": "core", "module": "sequences"},
|
|
{"key": "addresses:read", "label": "Addresses: Read", "category": "core", "module": "addresses"},
|
|
{"key": "addresses:write", "label": "Addresses: Write", "category": "core", "module": "addresses"},
|
|
{"key": "addresses:delete", "label": "Addresses: Delete", "category": "core", "module": "addresses"},
|
|
{"key": "taxes:read", "label": "Taxes: Read", "category": "core", "module": "taxes"},
|
|
{"key": "taxes:write", "label": "Taxes: Write", "category": "core", "module": "taxes"},
|
|
{"key": "currencies:read", "label": "Currencies: Read", "category": "core", "module": "currencies"},
|
|
{"key": "currencies:write", "label": "Currencies: Write", "category": "core", "module": "currencies"},
|
|
{"key": "import_export:read", "label": "Import/Export: Read", "category": "core", "module": "import_export"},
|
|
{"key": "import_export:write", "label": "Import/Export: Write", "category": "core", "module": "import_export"},
|
|
{"key": "system:admin", "label": "System: Admin (cross-tenant)", "category": "system", "module": "system"},
|
|
]
|
|
|
|
|
|
# ── Core field definitions for field-level permissions ──
|
|
CORE_FIELD_DEFINITIONS: list[dict[str, str]] = [
|
|
{"module": "companies", "field": "name", "label": "Name", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "account_number", "label": "Account Number", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "industry", "label": "Industry", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "phone", "label": "Phone", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "email", "label": "Email", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "website", "label": "Website", "sensitivity": "normal"},
|
|
{"module": "companies", "field": "description", "label": "Description", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "first_name", "label": "First Name", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "last_name", "label": "Last Name", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "email", "label": "Email", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "phone", "label": "Phone", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "mobile", "label": "Mobile", "sensitivity": "sensitive"},
|
|
{"module": "contacts", "field": "position", "label": "Position", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "department", "label": "Department", "sensitivity": "normal"},
|
|
{"module": "contacts", "field": "linkedin_url", "label": "LinkedIn URL", "sensitivity": "sensitive"},
|
|
{"module": "contacts", "field": "notes", "label": "Notes", "sensitivity": "sensitive"},
|
|
{"module": "users", "field": "email", "label": "Email", "sensitivity": "normal"},
|
|
{"module": "users", "field": "name", "label": "Name", "sensitivity": "normal"},
|
|
{"module": "users", "field": "role", "label": "Role", "sensitivity": "normal"},
|
|
{"module": "users", "field": "is_active", "label": "Active", "sensitivity": "normal"},
|
|
]
|
|
|
|
|
|
class PermissionRegistry:
|
|
"""Central registry of all valid permissions (core + active plugins)."""
|
|
|
|
def __init__(self) -> None:
|
|
self._permissions: dict[str, dict[str, str]] = {}
|
|
self._plugin_permissions: dict[str, list[dict[str, str]]] = {} # plugin_name → perms
|
|
self._active_plugins: set[str] = set()
|
|
self._initialized = False
|
|
self._field_definitions: dict[str, list[dict[str, str]]] = {} # plugin_name → field_defs
|
|
self._core_field_definitions: list[dict[str, str]] = list(CORE_FIELD_DEFINITIONS)
|
|
|
|
def initialize(self, active_plugin_names: set[str] | None = None) -> None:
|
|
"""Build the registry from core permissions and active plugin manifests."""
|
|
self._permissions = {}
|
|
self._plugin_permissions = {}
|
|
self._active_plugins = active_plugin_names or set()
|
|
|
|
# Register core permissions
|
|
for perm in CORE_PERMISSIONS:
|
|
self._permissions[perm["key"]] = perm
|
|
|
|
self._initialized = True
|
|
logger.info("Permission registry initialized with %d core permissions", len(CORE_PERMISSIONS))
|
|
|
|
def register_plugin_permissions(self, plugin_name: str, permissions: list[str]) -> None:
|
|
"""Register permissions from a plugin manifest."""
|
|
plugin_perms: list[dict[str, str]] = []
|
|
for perm in permissions:
|
|
# Normalize: replace dots with colons
|
|
normalized = perm.replace(".", ":")
|
|
entry = {
|
|
"key": normalized,
|
|
"label": normalized.replace(":", ": ").title(),
|
|
"category": "plugins",
|
|
"plugin_name": plugin_name,
|
|
"module": normalized.split(":")[0] if ":" in normalized else normalized,
|
|
}
|
|
plugin_perms.append(entry)
|
|
self._permissions[normalized] = entry
|
|
|
|
self._plugin_permissions[plugin_name] = plugin_perms
|
|
logger.info("Registered %d permissions for plugin '%s'", len(plugin_perms), plugin_name)
|
|
|
|
def unregister_plugin_permissions(self, plugin_name: str) -> None:
|
|
"""Remove permissions for a deactivated/uninstalled plugin."""
|
|
perms = self._plugin_permissions.pop(plugin_name, [])
|
|
for p in perms:
|
|
self._permissions.pop(p["key"], None)
|
|
logger.info("Unregistered permissions for plugin '%s'", plugin_name)
|
|
|
|
def is_valid(self, permission: str) -> bool:
|
|
"""Check if a permission key is known to the registry."""
|
|
return permission in self._permissions
|
|
|
|
def is_plugin_active(self, plugin_name: str) -> bool:
|
|
"""Check if a plugin is currently active."""
|
|
return plugin_name in self._active_plugins
|
|
|
|
def get_all(self) -> list[dict[str, str]]:
|
|
"""Return all registered permissions."""
|
|
return list(self._permissions.values())
|
|
|
|
def get_core(self) -> list[dict[str, str]]:
|
|
"""Return only core permissions."""
|
|
return [p for p in self._permissions.values() if p.get("category") == "core"]
|
|
|
|
def get_plugin_permissions(self) -> list[dict[str, str]]:
|
|
"""Return only plugin permissions."""
|
|
return [p for p in self._permissions.values() if p.get("category") == "plugins"]
|
|
|
|
def get_grouped(self) -> dict[str, list[dict[str, str]]]:
|
|
"""Return permissions grouped by category."""
|
|
groups: dict[str, list[dict[str, str]]] = {}
|
|
for perm in self._permissions.values():
|
|
cat = perm.get("category", "core")
|
|
if cat not in groups:
|
|
groups[cat] = []
|
|
groups[cat].append(perm)
|
|
return groups
|
|
|
|
def register_field_definitions(self, plugin_name: str, field_defs: list[dict[str, str]]) -> None:
|
|
"""Register field definitions from a plugin."""
|
|
self._field_definitions[plugin_name] = field_defs
|
|
logger.info("Registered %d field definitions for plugin '%s'", len(field_defs), plugin_name)
|
|
|
|
def get_all_field_definitions(self) -> list[dict[str, str]]:
|
|
"""Return all registered field definitions."""
|
|
result = list(self._core_field_definitions)
|
|
for defs in self._field_definitions.values():
|
|
result.extend(defs)
|
|
return result
|
|
|
|
|
|
# Global instance
|
|
_registry = PermissionRegistry()
|
|
|
|
|
|
def get_permission_registry() -> PermissionRegistry:
|
|
return _registry
|
|
|
|
|
|
def init_permission_registry(active_plugin_names: set[str] | None = None) -> None:
|
|
"""Initialize the global permission registry."""
|
|
_registry.initialize(active_plugin_names)
|
|
|
|
|
|
def register_plugin_permissions(plugin_name: str, permissions: list[str]) -> None:
|
|
"""Convenience: register plugin permissions on the global registry."""
|
|
_registry.register_plugin_permissions(plugin_name, permissions)
|
|
|
|
|
|
def unregister_plugin_permissions(plugin_name: str) -> None:
|
|
"""Convenience: unregister plugin permissions on the global registry."""
|
|
_registry.unregister_plugin_permissions(plugin_name)
|