feat(rbac): auto-sync permission catalog from declared route/plugin guards

backend/src/api/routes/admin.py

@@ -22,8 +22,12 @@ from ...schemas.auth import (
     ADGroupMappingSchema, ADGroupMappingCreate
 )
 from ...models.auth import User, Role, ADGroupMapping
-from ...dependencies import has_permission
+from ...dependencies import has_permission, get_plugin_loader
 from ...core.logger import logger, belief_scope
+from ...services.rbac_permission_catalog import (
+    discover_declared_permissions,
+    sync_permission_catalog,
+)
 # [/SECTION]
 
 # [DEF:router:Variable]
@@ -270,9 +274,18 @@ async def delete_role(
 @router.get("/permissions", response_model=List[PermissionSchema])
 async def list_permissions(
     db: Session = Depends(get_auth_db),
+    plugin_loader = Depends(get_plugin_loader),
     _ = Depends(has_permission("admin:roles", "READ"))
 ):
     with belief_scope("api.admin.list_permissions"):
+        declared_permissions = discover_declared_permissions(plugin_loader=plugin_loader)
+        inserted_count = sync_permission_catalog(db=db, declared_permissions=declared_permissions)
+        if inserted_count > 0:
+            logger.info(
+                "[api.admin.list_permissions][Action] Synchronized %s missing RBAC permissions into auth catalog",
+                inserted_count,
+            )
+
         repo = AuthRepository(db)
         return repo.list_permissions()
 # [/DEF:list_permissions:Function]

backend/src/services/__tests__/test_rbac_permission_catalog.py

@@ -0,0 +1,140 @@
+# [DEF:backend.src.services.__tests__.test_rbac_permission_catalog:Module]
+# @TIER: STANDARD
+# @SEMANTICS: tests, rbac, permissions, catalog, discovery, sync
+# @PURPOSE: Verifies RBAC permission catalog discovery and idempotent synchronization behavior.
+# @LAYER: Service Tests
+# @RELATION: TESTS -> backend.src.services.rbac_permission_catalog
+# @INVARIANT: Synchronization adds only missing normalized permission pairs.
+
+# [SECTION: IMPORTS]
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import src.services.rbac_permission_catalog as catalog
+# [/SECTION: IMPORTS]
+
+
+# [DEF:test_discover_route_permissions_extracts_declared_pairs_and_ignores_tests:Function]
+# @PURPOSE: Ensures route-scanner extracts has_permission pairs from route files and skips __tests__.
+# @PRE: Temporary route directory contains route and test files.
+# @POST: Returned set includes production route permissions and excludes test-only declarations.
+def test_discover_route_permissions_extracts_declared_pairs_and_ignores_tests(tmp_path, monkeypatch):
+    routes_dir = tmp_path / "routes"
+    routes_dir.mkdir(parents=True, exist_ok=True)
+
+    (routes_dir / "dashboards.py").write_text(
+        '\n'.join(
+            [
+                '_ = Depends(has_permission("plugin:migration", "READ"))',
+                '_ = Depends(has_permission("plugin:migration", "EXECUTE"))',
+                '_ = Depends(has_permission("tasks", "WRITE"))',
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    tests_dir = routes_dir / "__tests__"
+    tests_dir.mkdir(parents=True, exist_ok=True)
+    (tests_dir / "test_fake.py").write_text(
+        '_ = Depends(has_permission("plugin:ignored", "READ"))',
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(catalog, "ROUTES_DIR", routes_dir)
+
+    discovered = catalog._discover_route_permissions()
+
+    assert ("plugin:migration", "READ") in discovered
+    assert ("plugin:migration", "EXECUTE") in discovered
+    assert ("tasks", "WRITE") in discovered
+    assert ("plugin:ignored", "READ") not in discovered
+# [/DEF:test_discover_route_permissions_extracts_declared_pairs_and_ignores_tests:Function]
+
+
+# [DEF:test_discover_declared_permissions_unions_route_and_plugin_permissions:Function]
+# @PURPOSE: Ensures full catalog includes route-level permissions plus dynamic plugin EXECUTE rights.
+# @PRE: Route discovery and plugin loader both return permission sources.
+# @POST: Result set contains union of both sources.
+def test_discover_declared_permissions_unions_route_and_plugin_permissions(monkeypatch):
+    monkeypatch.setattr(
+        catalog,
+        "_discover_route_permissions",
+        lambda: {("tasks", "READ"), ("plugin:migration", "READ")},
+    )
+
+    plugin_loader = MagicMock()
+    plugin_loader.get_all_plugin_configs.return_value = [
+        SimpleNamespace(id="superset-backup"),
+        SimpleNamespace(id="llm_dashboard_validation"),
+    ]
+
+    discovered = catalog.discover_declared_permissions(plugin_loader=plugin_loader)
+
+    assert ("tasks", "READ") in discovered
+    assert ("plugin:migration", "READ") in discovered
+    assert ("plugin:superset-backup", "EXECUTE") in discovered
+    assert ("plugin:llm_dashboard_validation", "EXECUTE") in discovered
+# [/DEF:test_discover_declared_permissions_unions_route_and_plugin_permissions:Function]
+
+
+# [DEF:test_sync_permission_catalog_inserts_only_missing_normalized_pairs:Function]
+# @PURPOSE: Ensures synchronization inserts only missing pairs and normalizes action/resource tokens.
+# @PRE: DB already contains subset of permissions.
+# @POST: Only missing normalized pairs are inserted and commit is executed once.
+def test_sync_permission_catalog_inserts_only_missing_normalized_pairs():
+    db = MagicMock()
+    db.query.return_value.all.return_value = [
+        SimpleNamespace(resource="tasks", action="READ"),
+        SimpleNamespace(resource="plugin:migration", action="EXECUTE"),
+    ]
+
+    declared_permissions = {
+        ("tasks", "read"),
+        ("plugin:migration", "execute"),
+        ("plugin:migration", "READ"),
+        ("", "WRITE"),
+        ("plugin:migration", ""),
+    }
+
+    inserted_count = catalog.sync_permission_catalog(
+        db=db,
+        declared_permissions=declared_permissions,
+    )
+
+    assert inserted_count == 1
+    assert db.add.call_count == 1
+    inserted_permission = db.add.call_args[0][0]
+    assert inserted_permission.resource == "plugin:migration"
+    assert inserted_permission.action == "READ"
+    db.commit.assert_called_once()
+# [/DEF:test_sync_permission_catalog_inserts_only_missing_normalized_pairs:Function]
+
+
+# [DEF:test_sync_permission_catalog_is_noop_when_all_permissions_exist:Function]
+# @PURPOSE: Ensures synchronization is idempotent when all declared pairs already exist.
+# @PRE: DB contains full declared permission set.
+# @POST: No inserts are added and commit is not called.
+def test_sync_permission_catalog_is_noop_when_all_permissions_exist():
+    db = MagicMock()
+    db.query.return_value.all.return_value = [
+        SimpleNamespace(resource="tasks", action="READ"),
+        SimpleNamespace(resource="plugin:migration", action="READ"),
+    ]
+
+    declared_permissions = {
+        ("tasks", "READ"),
+        ("plugin:migration", "READ"),
+    }
+
+    inserted_count = catalog.sync_permission_catalog(
+        db=db,
+        declared_permissions=declared_permissions,
+    )
+
+    assert inserted_count == 0
+    db.add.assert_not_called()
+    db.commit.assert_not_called()
+# [/DEF:test_sync_permission_catalog_is_noop_when_all_permissions_exist:Function]
+
+
+# [/DEF:backend.src.services.__tests__.test_rbac_permission_catalog:Module]

backend/src/services/rbac_permission_catalog.py

@@ -0,0 +1,156 @@
+# [DEF:backend.src.services.rbac_permission_catalog:Module]
+#
+# @TIER: STANDARD
+# @SEMANTICS: rbac, permissions, catalog, sync, discovery
+# @PURPOSE:   Discovers declared RBAC permissions from API routes/plugins and synchronizes them into auth database.
+# @LAYER:     Service
+# @RELATION:  CALLS -> backend.src.core.plugin_loader.PluginLoader.get_all_plugin_configs
+# @RELATION:  DEPENDS_ON -> backend.src.models.auth.Permission
+# @INVARIANT: Synchronization is idempotent for existing (resource, action) permission pairs.
+
+# [SECTION: IMPORTS]
+import re
+from pathlib import Path
+from typing import Iterable, Set, Tuple
+
+from sqlalchemy.orm import Session
+
+from ..core.logger import belief_scope, logger
+from ..models.auth import Permission
+# [/SECTION: IMPORTS]
+
+# [DEF:HAS_PERMISSION_PATTERN:Constant]
+# @PURPOSE: Regex pattern for extracting has_permission("resource", "ACTION") declarations.
+HAS_PERMISSION_PATTERN = re.compile(
+    r"""has_permission\(\s*['"]([^'"]+)['"]\s*,\s*['"]([A-Z]+)['"]\s*\)"""
+)
+# [/DEF:HAS_PERMISSION_PATTERN:Constant]
+
+# [DEF:ROUTES_DIR:Constant]
+# @PURPOSE: Absolute directory path where API route RBAC declarations are defined.
+ROUTES_DIR = Path(__file__).resolve().parent.parent / "api" / "routes"
+# [/DEF:ROUTES_DIR:Constant]
+
+
+# [DEF:_iter_route_files:Function]
+# @PURPOSE: Iterates API route files that may contain RBAC declarations.
+# @PRE: ROUTES_DIR points to backend/src/api/routes.
+# @POST: Yields Python files excluding test and cache directories.
+# @RETURN: Iterable[Path] - Route file paths for permission extraction.
+def _iter_route_files() -> Iterable[Path]:
+    with belief_scope("rbac_permission_catalog._iter_route_files"):
+        if not ROUTES_DIR.exists():
+            return []
+
+        files = []
+        for file_path in ROUTES_DIR.rglob("*.py"):
+            path_parts = set(file_path.parts)
+            if "__tests__" in path_parts or "__pycache__" in path_parts:
+                continue
+            files.append(file_path)
+        return files
+# [/DEF:_iter_route_files:Function]
+
+
+# [DEF:_discover_route_permissions:Function]
+# @PURPOSE: Extracts explicit has_permission declarations from API route source code.
+# @PRE: Route files are readable UTF-8 text files.
+# @POST: Returns unique set of (resource, action) pairs declared in route guards.
+# @RETURN: Set[Tuple[str, str]] - Permission pairs from route-level RBAC declarations.
+def _discover_route_permissions() -> Set[Tuple[str, str]]:
+    with belief_scope("rbac_permission_catalog._discover_route_permissions"):
+        discovered: Set[Tuple[str, str]] = set()
+        for route_file in _iter_route_files():
+            try:
+                source = route_file.read_text(encoding="utf-8")
+            except OSError as read_error:
+                logger.warning(
+                    "[rbac_permission_catalog][EXPLORE] Failed to read route file %s: %s",
+                    route_file,
+                    read_error,
+                )
+                continue
+
+            for resource, action in HAS_PERMISSION_PATTERN.findall(source):
+                normalized_resource = str(resource or "").strip()
+                normalized_action = str(action or "").strip().upper()
+                if normalized_resource and normalized_action:
+                    discovered.add((normalized_resource, normalized_action))
+        return discovered
+# [/DEF:_discover_route_permissions:Function]
+
+
+# [DEF:_discover_plugin_execute_permissions:Function]
+# @PURPOSE: Derives dynamic task permissions of form plugin:{plugin_id}:EXECUTE from plugin registry.
+# @PRE: plugin_loader is optional and may expose get_all_plugin_configs.
+# @POST: Returns unique plugin EXECUTE permissions if loader is available.
+# @RETURN: Set[Tuple[str, str]] - Permission pairs derived from loaded plugin IDs.
+def _discover_plugin_execute_permissions(plugin_loader=None) -> Set[Tuple[str, str]]:
+    with belief_scope("rbac_permission_catalog._discover_plugin_execute_permissions"):
+        discovered: Set[Tuple[str, str]] = set()
+        if plugin_loader is None:
+            return discovered
+
+        try:
+            plugin_configs = plugin_loader.get_all_plugin_configs()
+        except Exception as plugin_error:
+            logger.warning(
+                "[rbac_permission_catalog][EXPLORE] Failed to read plugin configs for RBAC discovery: %s",
+                plugin_error,
+            )
+            return discovered
+
+        for plugin_config in plugin_configs:
+            plugin_id = str(getattr(plugin_config, "id", "") or "").strip()
+            if plugin_id:
+                discovered.add((f"plugin:{plugin_id}", "EXECUTE"))
+        return discovered
+# [/DEF:_discover_plugin_execute_permissions:Function]
+
+
+# [DEF:discover_declared_permissions:Function]
+# @PURPOSE: Builds canonical RBAC permission catalog from routes and plugin registry.
+# @PRE: plugin_loader may be provided for dynamic task plugin permission discovery.
+# @POST: Returns union of route-declared and dynamic plugin EXECUTE permissions.
+# @RETURN: Set[Tuple[str, str]] - Complete discovered permission set.
+def discover_declared_permissions(plugin_loader=None) -> Set[Tuple[str, str]]:
+    with belief_scope("rbac_permission_catalog.discover_declared_permissions"):
+        permissions = _discover_route_permissions()
+        permissions.update(_discover_plugin_execute_permissions(plugin_loader))
+        return permissions
+# [/DEF:discover_declared_permissions:Function]
+
+
+# [DEF:sync_permission_catalog:Function]
+# @PURPOSE: Persists missing RBAC permission pairs into auth database.
+# @PRE: db is a valid SQLAlchemy session bound to auth database.
+# @PRE: declared_permissions is an iterable of (resource, action) tuples.
+# @POST: Missing permissions are inserted; existing permissions remain untouched.
+# @SIDE_EFFECT: Commits auth database transaction when new permissions are added.
+# @RETURN: int - Number of inserted permission records.
+def sync_permission_catalog(
+    db: Session,
+    declared_permissions: Iterable[Tuple[str, str]],
+) -> int:
+    with belief_scope("rbac_permission_catalog.sync_permission_catalog"):
+        normalized_declared: Set[Tuple[str, str]] = set()
+        for resource, action in declared_permissions:
+            normalized_resource = str(resource or "").strip()
+            normalized_action = str(action or "").strip().upper()
+            if normalized_resource and normalized_action:
+                normalized_declared.add((normalized_resource, normalized_action))
+
+        existing_permissions = db.query(Permission).all()
+        existing_pairs = {(perm.resource, perm.action.upper()) for perm in existing_permissions}
+
+        missing_pairs = sorted(normalized_declared - existing_pairs)
+        for resource, action in missing_pairs:
+            db.add(Permission(resource=resource, action=action))
+
+        if missing_pairs:
+            db.commit()
+
+        return len(missing_pairs)
+# [/DEF:sync_permission_catalog:Function]
+
+# [/DEF:backend.src.services.rbac_permission_catalog:Module]

specs/016-multi-user-auth/tasks.md

@@ -105,3 +105,10 @@
 - [x] D057 Implement RBAC-aware sidebar navigation builder and integrate permission-filtered categories in `frontend/src/lib/components/layout/sidebarNavigation.js` and `frontend/src/lib/components/layout/Sidebar.svelte`
 - [x] D058 Add automated frontend tests for permission normalization/checking and sidebar visibility matrix in `frontend/src/lib/auth/__tests__/permissions.test.js` and `frontend/src/lib/components/layout/__tests__/sidebarNavigation.test.js`
 - [x] D059 Execute targeted frontend test verification for RBAC navigation filtering (`npm run test -- src/lib/auth/__tests__/permissions.test.js src/lib/components/layout/__tests__/sidebarNavigation.test.js`)
+
+## Post-Delivery RBAC Permission Catalog Sync (2026-03-06)
+
+- [x] D060 Implement automatic RBAC permission discovery/synchronization service for admin role settings permission catalog in `backend/src/services/rbac_permission_catalog.py`
+- [x] D061 Integrate permission catalog synchronization into `GET /api/admin/permissions` flow in `backend/src/api/routes/admin.py`
+- [x] D062 Add unit tests for permission discovery and idempotent sync behavior in `backend/src/services/__tests__/test_rbac_permission_catalog.py`
+- [x] D063 Execute targeted backend test verification for RBAC catalog sync (`cd backend && .venv/bin/python3 -m pytest src/services/__tests__/test_rbac_permission_catalog.py`)