Agent Zero
4a25ac1379
Check Cross-Plugin Imports / check (push) Has been cancelled
Verifikation: Alle 17 Audit-Findings gegen den Code geprueft — alle bestaetigt. Backend-Lifecycle-Fixes umgesetzt; 4 Frontend-Plugin-Architektur-Punkte als Phase Q in die Roadmap eingeplant. - P1 list_workspaces: Module + User-Counts gebuendelt laden (Editor-Overwrite-Bug) - P1 active-manifests: Tenant-Deaktivierung (tenant_plugin_activation) filtern - P1 uninstall: volle Service-Deactivation VOR registry.uninstall() - P1 ContractRegistry: DB-Aktivstatus-Guard (Restart-Edge-Case) + Re-Activate - P1/P2 Field-Definitions: voller Lifecycle (register/unregister) im Service - P1/P2 Contact-Felddefinitionen (39) ins ContactsPlugin-Manifest verschoben - P1 12 fehlende Permission-Keys registriert (AST-Scan: 0 fehlend) - P2 contact_folder -> ContactsPlugin; ENTITY_PLUGIN_OWNERS wird befuellt - P2 Entity-Permission-Fallback fail-closed statt contacts:read - P2 forgejo_error_reporter is_core=False; DMS is_core=True (ADR-020) - P2 Worker: Contacts-Trash-Cleanup ins Plugin (get_job_modules-Discovery) - P1/P2 DSGVO-Export delegiert an DSAR-Collector (kein Core->Contacts) - P2 False-green Tests korrigiert (or True, veraltete Route-Count-Assertion) Verifikation: tests/test_audit_architecture_fixes.py 17/17; Regressionen gruen (contacts_lifecycle, entity_registry, workspace_scopes, rbac, lifecycle_service); Combo-Order-Test 35/35; Cross-Plugin-Checker 497/0; compileall sauber; ruff auf 7-Error-Baseline. Doku: PROGRESS.md Audit-Section, PLATFORM_ROADMAP.md Phase Q (Q1-Q4), plugin-development-guide.md Lifecycle, permissions.md Katalog.
279 lines
15 KiB
Python
279 lines
15 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
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.custom_field_definition import CustomFieldDefinition
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Core system permissions ──
|
|
CORE_PERMISSIONS: list[dict[str, str]] = [
|
|
{"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": "user_preferences:read", "label": "User Preferences: Read", "category": "core", "module": "user_preferences"},
|
|
{"key": "user_preferences:write", "label": "User Preferences: Write", "category": "core", "module": "user_preferences"},
|
|
{"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": "custom_fields:read", "label": "Custom Fields: Read", "category": "core", "module": "custom_fields"},
|
|
{"key": "custom_fields:write", "label": "Custom Fields: Write", "category": "core", "module": "custom_fields"},
|
|
{"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": "workspaces:read", "label": "Workspaces: Read", "category": "core", "module": "workspaces"},
|
|
{"key": "workspaces:create", "label": "Workspaces: Create", "category": "core", "module": "workspaces"},
|
|
{"key": "workspaces:update", "label": "Workspaces: Update", "category": "core", "module": "workspaces"},
|
|
{"key": "workspaces:delete", "label": "Workspaces: Delete", "category": "core", "module": "workspaces"},
|
|
{"key": "workspaces:assign_users", "label": "Workspaces: Assign Users", "category": "core", "module": "workspaces"},
|
|
{"key": "workspaces:configure_modules", "label": "Workspaces: Configure Modules", "category": "core", "module": "workspaces"},
|
|
{"key": "approvals:read", "label": "Approvals: Read", "category": "core", "module": "approvals"},
|
|
{"key": "approvals:write", "label": "Approvals: Write", "category": "core", "module": "approvals"},
|
|
{"key": "approvals:approve", "label": "Approvals: Approve/Reject", "category": "core", "module": "approvals"},
|
|
{"key": "dashboard:read", "label": "Dashboard: Read", "category": "core", "module": "dashboard"},
|
|
{"key": "dashboard:write", "label": "Dashboard: Write", "category": "core", "module": "dashboard"},
|
|
{"key": "system:admin", "label": "System: Admin (cross-tenant)", "category": "system", "module": "system"},
|
|
# Audit P1 (permission catalog): these keys were required by core routes
|
|
# but never registered, so non-admin roles could never be granted them.
|
|
{"key": "automation:admin", "label": "Automation: Admin (backups, self-improvement)", "category": "core", "module": "automation"},
|
|
{"key": "bank-accounts:read", "label": "Bank Accounts: Read", "category": "core", "module": "bank_accounts"},
|
|
{"key": "bank-accounts:write", "label": "Bank Accounts: Write", "category": "core", "module": "bank_accounts"},
|
|
{"key": "delegations:read", "label": "Delegations: Read", "category": "core", "module": "delegations"},
|
|
{"key": "delegations:write", "label": "Delegations: Write", "category": "core", "module": "delegations"},
|
|
{"key": "policies:read", "label": "Policies: Read", "category": "core", "module": "policies"},
|
|
{"key": "policies:write", "label": "Policies: Write", "category": "core", "module": "policies"},
|
|
{"key": "templates:read", "label": "Permission Templates: Read", "category": "core", "module": "templates"},
|
|
{"key": "templates:write", "label": "Permission Templates: Write", "category": "core", "module": "templates"},
|
|
# NOTE: Plugin permissions (calendar, dms, mail, tasks, comm, automation, ai,
|
|
# tags, entity_links, reports, search, mcp, permissions, agents)
|
|
# are registered dynamically via register_plugin_permissions() from plugin
|
|
# manifests at activation time. They are NOT hardcoded here (P0-4 fix).
|
|
]
|
|
|
|
|
|
# ── Core field definitions for field-level permissions ──
|
|
CORE_FIELD_DEFINITIONS: list[dict[str, str]] = [
|
|
# Audit P1/P2 (contact field definitions): all contacts:* field
|
|
# definitions moved to the ContactsPlugin manifest (field_definitions=)
|
|
# so the plugin fully owns its field structure. The core keeps only
|
|
# genuinely core-owned fields (users). Plugin field definitions are
|
|
# registered at activation time via register_field_definitions().
|
|
# ── User fields (core-owned) ──
|
|
{"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.
|
|
|
|
Preserves already-registered plugin permissions (fixes P1.3 bug where
|
|
initialize() would wipe plugin permissions registered before startup).
|
|
"""
|
|
# Preserve existing plugin permissions
|
|
existing_plugin_perms = self._plugin_permissions.copy()
|
|
|
|
# Reset only core permissions, keep plugin permissions
|
|
self._permissions = {}
|
|
self._active_plugins = active_plugin_names or set()
|
|
|
|
# Register core permissions
|
|
for perm in CORE_PERMISSIONS:
|
|
self._permissions[perm["key"]] = perm
|
|
|
|
# Re-apply plugin permissions that were registered before initialize()
|
|
for plugin_name, perms in existing_plugin_perms.items():
|
|
self._plugin_permissions[plugin_name] = perms
|
|
for entry in perms:
|
|
self._permissions[entry["key"]] = entry
|
|
|
|
self._initialized = True
|
|
logger.info("Permission registry initialized with %d core permissions, %d plugin permissions",
|
|
len(CORE_PERMISSIONS), len(existing_plugin_perms))
|
|
|
|
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 unregister_field_definitions(self, plugin_name: str) -> None:
|
|
"""Remove field definitions of a deactivated/uninstalled plugin.
|
|
|
|
Audit P1/P2 (field-definitions lifecycle): the contribution type was
|
|
only half-integrated — register_field_definitions() existed but no
|
|
matching unregister, so a deactivated plugin kept serving its field
|
|
definitions in the permission UI.
|
|
"""
|
|
removed = self._field_definitions.pop(plugin_name, None)
|
|
if removed is not None:
|
|
logger.info("Unregistered %d field definitions for plugin '%s'", len(removed), 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
|
|
|
|
async def get_all_field_definitions_with_custom(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: Any,
|
|
) -> list[dict[str, str]]:
|
|
"""Return all field definitions including custom fields from DB.
|
|
|
|
Merges core field definitions with plugin-provided definitions
|
|
and active custom field definitions from the database.
|
|
"""
|
|
result = list(self._core_field_definitions)
|
|
|
|
# Add plugin field definitions
|
|
for defs in self._field_definitions.values():
|
|
result.extend(defs)
|
|
|
|
# Load custom field definitions from DB
|
|
q = select(CustomFieldDefinition).where(
|
|
CustomFieldDefinition.tenant_id == tenant_id,
|
|
CustomFieldDefinition.is_active.is_(True),
|
|
)
|
|
custom_defs = await db.execute(q)
|
|
for cfd in custom_defs.scalars().all():
|
|
result.append({
|
|
"module": cfd.entity,
|
|
"field": cfd.name,
|
|
"label": cfd.label,
|
|
"sensitivity": cfd.sensitivity,
|
|
"custom": "true",
|
|
})
|
|
|
|
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)
|