wiki/appdoc/doc_workflow_actions_rbac_concept.md

Workflow Actions RBAC Integration Concept

Übersicht

Dieses Dokument beschreibt das Konzept für die Umstrukturierung der Workflow Actions, um:

1. RBAC-Integration zu ermöglichen (Schutz von Actions über RESOURCE-Context)
2. Strukturierte Parameter-Definitionen statt Docstrings zu verwenden
3. UI-Rendering-Typen für Parameter zu definieren
4. Keine Duplikation von Parameter-Definitionen zu haben
5. Plug-and-Play Funktionalität beizubehalten

Architektur-Konzept

Grundprinzip: Deklarative Action-Definition

Ähnlich wie bei aicore Models, wo eine Struktur definiert wird und die Funktion ein Attribut ist, werden Actions jetzt deklarativ definiert:

class MethodOutlook(MethodBase):
    def __init__(self, services):
        super().__init__(services)
        self.name = "outlook"
        self.description = "Handle Microsoft Outlook email operations"
        
        # Actions werden deklarativ definiert
        self._actions = {
            "readEmails": ActionDefinition(
                actionId="outlook.readEmails",  # Für RBAC: RESOURCE context
                description="Read emails from Outlook mailbox",
                parameters={
                    "connectionReference": ActionParameter(
                        name="connectionReference",
                        type=str,
                        frontendType="select",
                        frontendOptions="user.connection",
                        required=True,
                        description="Microsoft connection label"
                    ),
                    "query": ActionParameter(
                        name="query",
                        type=str,
                        frontendType="text",
                        required=False,
                        description="Search query for emails"
                    ),
                    "folder": ActionParameter(
                        name="folder",
                        type=str,
                        frontendType="select",
                        frontendOptions="outlook.folder",
                        required=False,
                        description="Folder name (e.g., 'Inbox', 'Drafts')"
                    ),
                    "limit": ActionParameter(
                        name="limit",
                        type=int,
                        frontendType="number",
                        required=False,
                        default=50,
                        description="Maximum number of emails to return"
                    )
                },
                execute=self._executeReadEmails  # Funktion als Attribut
            ),
            "sendEmail": ActionDefinition(
                actionId="outlook.sendEmail",
                description="Send email via Outlook",
                parameters={...},
                execute=self._executeSendEmail
            )
        }
    
    async def _executeReadEmails(self, parameters: Dict[str, Any]) -> ActionResult:
        """Execute function - keine Parameter-Definition mehr hier"""
        # Implementation...

Globale Frontend-Type-Definition

WICHTIG: Frontend-Types werden zentral in modules/shared/frontendTypes.py definiert, nicht redundant pro Action.

Die globale FrontendType Enum enthält:

• Standard Types: text, textarea, number, select, multiselect, checkbox, date, datetime, email, timestamp, json, multilingual, file
• Custom Types für Actions: userConnection, documentReference, workflowAction

Custom-Types unterstützen dynamische Option-Listen über API-Endpoints:

• userConnection → /api/options/user.connection (Connections des aktuellen Users)
• documentReference → /api/options/workflow.documentReference (Document-Referenzen aus Workflow-Context)
• workflowAction → /api/options/workflow.action (Verfügbare Actions aus Workflow-Context)

Datenmodelle

ActionParameter

WICHTIG: Frontend-Types werden global definiert in modules/shared/frontendTypes.py und nicht redundant in Actions.

from typing import Optional, Any, Union, List, Dict
from pydantic import BaseModel, Field
from modules.shared.frontendTypes import FrontendType  # Globale Definition

class ActionParameter(BaseModel):
    """Parameter definition for an action"""
    name: str = Field(description="Parameter name")
    type: str = Field(description="Python type as string (e.g., 'str', 'int', 'bool', 'List[str]')")
    frontendType: FrontendType = Field(description="UI rendering type (from global FrontendType enum)")
    frontendOptions: Optional[Union[str, List[Dict[str, Any]]]] = Field(
        None,
        description="Options for select/multiselect/custom types. String reference (e.g., 'user.connection') or static list. For custom types like userConnection, this is automatically set to the API endpoint."
    )
    required: bool = Field(False, description="Whether parameter is required")
    default: Optional[Any] = Field(None, description="Default value")
    description: str = Field("", description="Parameter description")
    validation: Optional[Dict[str, Any]] = Field(
        None,
        description="Validation rules (e.g., {'min': 1, 'max': 100})"
    )

Custom Frontend Types:

• FrontendType.USER_CONNECTION: User connection selector - dynamische Options von /api/options/user.connection
• FrontendType.DOCUMENT_REFERENCE: Document reference selector - dynamische Options aus Workflow-Context
• FrontendType.WORKFLOW_ACTION: Workflow action selector - dynamische Options aus verfügbaren Actions

Für Custom-Types wird frontendOptions automatisch auf den entsprechenden API-Endpoint gesetzt (z.B. "user.connection").

ActionDefinition

from typing import Dict, Callable, Awaitable
from pydantic import BaseModel, Field

class ActionDefinition(BaseModel):
    """Complete definition of an action"""
    actionId: str = Field(
        description="Unique action identifier for RBAC (format: 'module.actionName', e.g., 'outlook.readEmails')"
    )
    description: str = Field(description="Action description")
    parameters: Dict[str, ActionParameter] = Field(
        default_factory=dict,
        description="Parameter definitions"
    )
    execute: Callable[[Dict[str, Any]], Awaitable[ActionResult]] = Field(
        description="Execution function - async function that takes parameters dict and returns ActionResult"
    )
    # Optional metadata
    category: Optional[str] = Field(None, description="Action category for grouping")
    tags: List[str] = Field(default_factory=list, description="Tags for search/filtering")

MethodBase Erweiterung

Neue MethodBase Struktur

class MethodBase:
    """Base class for all methods"""
    
    def __init__(self, services: Any):
        self.services = services
        self.name: str
        self.description: str
        self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")
        
        # Actions werden als Dictionary definiert
        self._actions: Dict[str, ActionDefinition] = {}
        
        # Nach Initialisierung: Actions registrieren
        self._registerActions()
    
    def _registerActions(self):
        """Register all actions defined in _actions"""
        # Kann überschrieben werden für dynamische Registrierung
        pass
    
    @property
    def actions(self) -> Dict[str, Dict[str, Any]]:
        """
        Dynamically collect all actions from _actions dictionary.
        Returns format compatible with existing system.
        """
        result = {}
        for actionName, actionDef in self._actions.items():
            # RBAC-Check: Prüfe ob Action für aktuellen User verfügbar ist
            if not self._checkActionPermission(actionDef.actionId):
                continue  # Skip if user doesn't have permission
            
            # Konvertiere ActionDefinition zu altem Format für Kompatibilität
            result[actionName] = {
                'description': actionDef.description,
                'parameters': self._convertParametersToOldFormat(actionDef.parameters),
                'method': self._createActionWrapper(actionDef)
            }
        return result
    
    def _checkActionPermission(self, actionId: str) -> bool:
        """
        Check if current user has permission to execute this action.
        Uses RBAC RESOURCE context.
        """
        if not hasattr(self.services, 'rbac') or not self.services.rbac:
            # Fallback: Allow if RBAC not available (backward compatibility)
            return True
        
        currentUser = self.services.chat.getCurrentUser()
        if not currentUser:
            return False
        
        # RBAC-Check: RESOURCE context, item = actionId
        permissions = self.services.rbac.getUserPermissions(
            user=currentUser,
            context=AccessRuleContext.RESOURCE,
            item=actionId
        )
        
        return permissions.view
    
    def _convertParametersToOldFormat(self, parameters: Dict[str, ActionParameter]) -> Dict[str, Dict[str, Any]]:
        """Convert ActionParameter dict to old format for compatibility"""
        result = {}
        for paramName, param in parameters.items():
            result[paramName] = {
                'type': param.type,
                'required': param.required,
                'description': param.description,
                'default': param.default,
                'frontendType': param.frontendType.value,
                'frontendOptions': param.frontendOptions
            }
        return result
    
    def _createActionWrapper(self, actionDef: ActionDefinition):
        """Create wrapper function that matches old action signature"""
        async def wrapper(parameters: Dict[str, Any], *args, **kwargs):
            # Parameter-Validierung basierend auf ActionParameter definitions
            validatedParams = self._validateParameters(parameters, actionDef.parameters)
            
            # Execute action
            return await actionDef.execute(validatedParams, *args, **kwargs)
        
        wrapper.is_action = True
        return wrapper
    
    def _validateParameters(self, parameters: Dict[str, Any], paramDefs: Dict[str, ActionParameter]) -> Dict[str, Any]:
        """Validate parameters against definitions"""
        validated = {}
        
        for paramName, paramDef in paramDefs.items():
            value = parameters.get(paramName)
            
            # Check required
            if paramDef.required and value is None:
                raise ValueError(f"Required parameter '{paramName}' is missing")
            
            # Use default if not provided
            if value is None and paramDef.default is not None:
                value = paramDef.default
            
            # Type validation
            if value is not None:
                value = self._validateType(value, paramDef.type)
            
            # Custom validation rules
            if paramDef.validation and value is not None:
                self._applyValidationRules(value, paramDef.validation)
            
            validated[paramName] = value
        
        return validated
    
    def _validateType(self, value: Any, expectedType: type) -> Any:
        """Validate and convert value to expected type"""
        # Type validation logic...
        if expectedType == int:
            return int(value)
        elif expectedType == str:
            return str(value)
        # ... weitere Typen
        return value
    
    def _applyValidationRules(self, value: Any, rules: Dict[str, Any]):
        """Apply custom validation rules"""
        if 'min' in rules and value < rules['min']:
            raise ValueError(f"Value must be >= {rules['min']}")
        if 'max' in rules and value > rules['max']:
            raise ValueError(f"Value must be <= {rules['max']}")
        # ... weitere Validierungsregeln

Migrationsstrategie

Schritt 1: Neue Datenmodelle erstellen

Datei: gateway/modules/datamodels/datamodelWorkflowActions.py

from typing import Optional, Any, Union, List, Dict, Callable, Awaitable
from enum import Enum
from pydantic import BaseModel, Field
from modules.datamodels.datamodelChat import ActionResult
from modules.shared.frontendTypes import FrontendType  # Globale Definition verwenden

class ActionParameter(BaseModel):
    """Parameter definition for an action"""
    name: str
    type: str  # String representation of type: "str", "int", "bool", "List[str]", etc.
    frontendType: FrontendType
    frontendOptions: Optional[Union[str, List[Dict[str, Any]]]] = None
    required: bool = False
    default: Optional[Any] = None
    description: str = ""
    validation: Optional[Dict[str, Any]] = None

class ActionDefinition(BaseModel):
    """Complete definition of an action"""
    actionId: str  # Format: "module.actionName" (e.g., "outlook.readEmails")
    description: str
    parameters: Dict[str, ActionParameter] = Field(default_factory=dict)
    execute: Optional[Callable] = None  # Will be set dynamically
    category: Optional[str] = None
    tags: List[str] = Field(default_factory=list)

Schritt 2: MethodBase erweitern

Datei: gateway/modules/workflows/methods/methodBase.py

• Neue _actions Dictionary Property
• RBAC-Check Integration
• Parameter-Validierung
• Kompatibilität mit bestehendem System

Schritt 3: Beispiel-Migration

Vorher (methodOutlook.py):

@action
async def readEmails(self, parameters: Dict[str, Any]) -> ActionResult:
    """
    GENERAL:
    - Purpose: Read emails from Outlook mailbox
    
    Parameters:
    - connectionReference (str, required): Microsoft connection label.
    - query (str, optional): Search query for emails.
    - folder (str, optional): Folder name.
    - limit (int, optional): Maximum number of emails. Default: 50.
    """
    # Implementation...

Nachher (methodOutlook.py):

def __init__(self, services):
    super().__init__(services)
    self.name = "outlook"
    self.description = "Handle Microsoft Outlook email operations"
    
    self._actions = {
        "readEmails": ActionDefinition(
            actionId="outlook.readEmails",
            description="Read emails from Outlook mailbox",
            parameters={
                "connectionReference": ActionParameter(
                    name="connectionReference",
                    type="str",
                    frontendType=FrontendType.USER_CONNECTION,  # Custom type - automatisch API-Endpoint
                    required=True,
                    description="Microsoft connection label"
                ),
                "query": ActionParameter(
                    name="query",
                    type="str",
                    frontendType=FrontendType.TEXT,
                    required=False,
                    description="Search query for emails"
                ),
                "folder": ActionParameter(
                    name="folder",
                    type="str",
                    frontendType=FrontendType.SELECT,
                    frontendOptions="outlook.folder",
                    required=False,
                    description="Folder name (e.g., 'Inbox', 'Drafts')"
                ),
                "limit": ActionParameter(
                    name="limit",
                    type="int",
                    frontendType=FrontendType.NUMBER,
                    required=False,
                    default=50,
                    description="Maximum number of emails to return",
                    validation={"min": 1, "max": 1000}
                )
            },
            execute=self._executeReadEmails
        )
    }

async def _executeReadEmails(self, parameters: Dict[str, Any]) -> ActionResult:
    """Execute function - keine Parameter-Definition mehr hier"""
    # Implementation bleibt gleich...

RBAC-Integration

Action-IDs Format

Actions werden im RBAC-System als RESOURCE-Context Items behandelt:

• Format: {moduleName}.{actionName}
• Beispiele:
  • outlook.readEmails
  • outlook.sendEmail
  • sharepoint.uploadDocument
  • ai.process

RBAC-Regeln für Actions

{
  "roleLabel": "user",
  "context": "RESOURCE",
  "item": "outlook.readEmails",
  "view": true
}

{
  "roleLabel": "admin",
  "context": "RESOURCE",
  "item": "outlook",
  "view": true
}

Hierarchie: Spezifische Action-Regeln überschreiben generische Module-Regeln.

Bootstrap: Default RBAC Rules für Actions

In interfaceBootstrap.py:

def initRbacRules(db: DatabaseConnector) -> None:
    # ... existing rules ...
    
    # Action Rules (RESOURCE context)
    createActionRules(db)

def createActionRules(db: DatabaseConnector):
    """Create default RBAC rules for workflow actions"""
    
    # SysAdmin: Access to all actions
    db.recordCreate(AccessRule(
        roleLabel="sysadmin",
        context=AccessRuleContext.RESOURCE,
        item=None,  # All resources
        view=True
    ))
    
    # Admin: Access to all actions
    db.recordCreate(AccessRule(
        roleLabel="admin",
        context=AccessRuleContext.RESOURCE,
        item=None,
        view=True
    ))
    
    # User: Access to specific actions only
    userActions = [
        "outlook.readEmails",
        "outlook.sendEmail",
        "sharepoint.readDocuments",
        "ai.process"
    ]
    
    for actionId in userActions:
        db.recordCreate(AccessRule(
            roleLabel="user",
            context=AccessRuleContext.RESOURCE,
            item=actionId,
            view=True
        ))
    
    # Viewer: Read-only actions
    viewerActions = [
        "outlook.readEmails",
        "sharepoint.readDocuments"
    ]
    
    for actionId in viewerActions:
        db.recordCreate(AccessRule(
            roleLabel="viewer",
            context=AccessRuleContext.RESOURCE,
            item=actionId,
            view=True
        ))

Vorteile

1. Keine Duplikation

• Parameter werden nur einmal definiert (in ActionDefinition)
• Keine Docstring-Parsing mehr nötig
• Type-Safety durch Pydantic Models

2. RBAC-Integration

• Jede Action hat eine eindeutige ID für RBAC
• Granulare Kontrolle pro Action möglich
• Hierarchische Regeln (Module → Action)

3. UI-Rendering

• Frontend-Typen explizit definiert
• Options-Referenzen für dynamische Optionen
• Validierung auf Backend-Ebene

4. Plug-and-Play

• Actions bleiben als separate Method-Klassen
• Einfache Erweiterung durch neue Method-Klassen
• Kompatibilität mit bestehendem System

5. Type Safety

• Pydantic Models für Validierung
• Type-Hints für bessere IDE-Unterstützung
• Runtime-Validierung

Migration Timeline

Phase 1: Foundation (Woche 1)

• ✅ Datenmodelle erstellen (datamodelWorkflowActions.py)
• ✅ MethodBase erweitern
• ✅ RBAC-Integration in MethodBase

Phase 2: Beispiel-Migration (Woche 2)

• 📝 Ein Method-Beispiel migrieren (z.B. methodAi.py)
• 📝 Tests schreiben
• 📝 Dokumentation aktualisieren

Phase 3: Vollständige Migration (Woche 3-4)

• 📝 Alle Methods migrieren
• 📝 RBAC-Regeln in Bootstrap erstellen
• 📝 Frontend-Integration

Phase 4: Testing & Cleanup (Woche 5)

• 📝 Unit Tests
• 📝 Integration Tests
• 📝 Performance Tests
• 📝 Alte Docstring-Parsing-Logik entfernen

Offene Fragen

1. Backward Compatibility: Sollen alte Actions ohne _actions Dictionary weiterhin funktionieren?

   • Antwort: Ja, MethodBase prüft zuerst _actions, dann fallback auf @action Decorator

2. Parameter-Validierung: Soll Validierung strikt sein oder tolerant?

   • Antwort: Konfigurierbar pro Action

3. Action-Discovery: Sollen Actions zur Laufzeit registriert werden können?

   • Antwort: Ja, über _registerActions() Methode

4. Frontend-Integration: Wie werden Actions im Frontend angezeigt?

   • Antwort: API-Endpoint /api/workflows/actions liefert strukturierte Action-Definitionen

API-Endpunkte

GET /api/workflows/actions

Liefert alle verfügbaren Actions für den aktuellen User (gefiltert nach RBAC):

{
  "actions": [
    {
      "module": "outlook",
      "actionId": "outlook.readEmails",
      "name": "readEmails",
      "description": "Read emails from Outlook mailbox",
      "parameters": {
        "connectionReference": {
          "type": "str",
          "frontendType": "userConnection",
          "frontendOptions": "user.connection",  # Automatisch für Custom-Types
          "required": true,
          "description": "Microsoft connection label"
        },
        "documentList": {
          "type": "List[str]",
          "frontendType": "documentReference",
          "frontendOptions": "workflow.documentReference",  # Automatisch für Custom-Types
          "required": false,
          "description": "Document list reference(s) from previous actions"
        },
        ...
      }
    },
    ...
  ]
}

GET /api/workflows/actions/{module}

Liefert Actions für ein spezifisches Modul.

POST /api/workflows/actions/{module}/{action}/execute

Führt eine Action aus (mit RBAC-Check).

Custom Frontend Types für Actions

Verfügbare Custom Types

1. FrontendType.USER_CONNECTION

   • API-Endpoint: /api/options/user.connection
   • Beschreibung: Zeigt alle aktiven Connections des aktuellen Users
   • Verwendung: Für Parameter wie connectionReference in Outlook/SharePoint Actions
   • Beispiel:

     ActionParameter(
         name="connectionReference",
         type="str",
         frontendType=FrontendType.USER_CONNECTION,
         required=True,
         description="Microsoft connection label"
     )
     

2. FrontendType.DOCUMENT_REFERENCE

   • API-Endpoint: /api/options/workflow.documentReference (zu implementieren)
   • Beschreibung: Zeigt verfügbare Document-Referenzen aus dem aktuellen Workflow-Context
   • Verwendung: Für Parameter wie documentList in Actions, die auf vorherige Action-Ergebnisse verweisen
   • Beispiel:

     ActionParameter(
         name="documentList",
         type="List[str]",
         frontendType=FrontendType.DOCUMENT_REFERENCE,
         required=False,
         description="Document list reference(s) from previous actions"
     )
     

3. FrontendType.WORKFLOW_ACTION

   • API-Endpoint: /api/options/workflow.action (zu implementieren)
   • Beschreibung: Zeigt verfügbare Actions aus dem Workflow-Context
   • Verwendung: Für Parameter, die auf andere Actions verweisen

Custom Types erweitern

Neue Custom-Types können über frontendTypes.py registriert werden:

from modules.shared.frontendTypes import FrontendType, registerCustomType

# Neuer Custom-Type hinzufügen
FrontendType.SHAREPOINT_FOLDER = "sharepointFolder"

# Registrieren
registerCustomType(
    frontendType=FrontendType.SHAREPOINT_FOLDER,
    optionsApiEndpoint="sharepoint.folder",
    description={
        "en": "SharePoint Folder",
        "fr": "Dossier SharePoint",
        "de": "SharePoint-Ordner"
    }
)

Frontend-Integration

Das Frontend muss:

1. Custom-Types erkennen (z.B. frontendType === "userConnection")
2. Automatisch Options von /api/options/{optionsName} laden
3. Die Options als Select/Multiselect rendern

Beispiel Frontend-Logik:

if (param.frontendType === 'userConnection') {
  // Automatisch Options von /api/options/user.connection laden
  const options = await fetch(`/api/options/${param.frontendOptions}`);
  // Als Select rendern
}