wiki/appdoc/doc_workflow_actions_rbac_concept.md
2025-12-16 00:27:41 +01:00

763 lines
28 KiB
Markdown

# 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:
```python
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": WorkflowActionDefinition(
actionId="outlook.readEmails", # Für RBAC: RESOURCE context
description="Read emails from Outlook mailbox",
parameters={
"connectionReference": WorkflowActionParameter(
name="connectionReference",
type="str",
frontendType=FrontendType.USER_CONNECTION,
required=True,
description="Microsoft connection label"
),
"query": WorkflowActionParameter(
name="query",
type="str",
frontendType=FrontendType.TEXT,
required=False,
description="Search query for emails"
),
"folder": WorkflowActionParameter(
name="folder",
type="str",
frontendType=FrontendType.SELECT,
frontendOptions="outlook.folder",
required=False,
description="Folder name (e.g., 'Inbox', 'Drafts')"
),
"limit": WorkflowActionParameter(
name="limit",
type="int",
frontendType=FrontendType.NUMBER,
required=False,
default=50,
description="Maximum number of emails to return"
)
},
execute=self._executeReadEmails # Funktion als Attribut
),
"sendEmail": WorkflowActionDefinition(
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
### WorkflowActionParameter
**WICHTIG**:
- Frontend-Types werden global definiert in `modules/shared/frontendTypes.py` und nicht redundant in Actions
- Diese Klasse heißt `WorkflowActionParameter` (nicht `ActionParameter`) um Konflikte mit `ActionParameters` aus `datamodelChat.py` zu vermeiden
```python
from typing import Optional, Any, Union, List, Dict
from pydantic import BaseModel, Field
from modules.shared.frontendTypes import FrontendType # Globale Definition
class WorkflowActionParameter(BaseModel):
"""
Parameter schema definition for a workflow action.
This defines the structure and UI rendering for a single action parameter,
NOT the actual parameter values (those are in ActionDefinition.parameters).
"""
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"`).
### WorkflowActionDefinition
**WICHTIG**: Diese Klasse heißt `WorkflowActionDefinition` (nicht `ActionDefinition`) um Konflikte mit der bestehenden `ActionDefinition` aus `datamodelWorkflow.py` zu vermeiden:
- **Bestehende `ActionDefinition`**: Für Workflow-Execution-Planning (enthält konkrete Werte: `action`, `actionObjective`, `parameters` mit Werten)
- **Neue `WorkflowActionDefinition`**: Für Action-Schema-Definitionen (enthält Metadaten: `actionId`, `description`, `parameters` als Schemas)
```python
from typing import Dict, Callable, Awaitable, Optional, List
from pydantic import BaseModel, Field
from modules.datamodels.datamodelChat import ActionResult
class WorkflowActionDefinition(BaseModel):
"""
Complete schema definition of a workflow action.
This defines the metadata, parameters, and execution function for an action.
This is different from datamodelWorkflow.ActionDefinition which contains
actual execution values (action, actionObjective, parameters with values).
This class defines the ACTION SCHEMA, not the execution plan.
"""
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, WorkflowActionParameter] = Field(
default_factory=dict,
description="Parameter schema definitions"
)
execute: Optional[Callable[[Dict[str, Any]], Awaitable[ActionResult]]] = Field(
None,
description="Execution function - async function that takes parameters dict and returns ActionResult. Set dynamically."
)
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
```python
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, WorkflowActionDefinition] = {}
# 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 WorkflowActionDefinition 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, WorkflowActionParameter]) -> Dict[str, Dict[str, Any]]:
"""Convert WorkflowActionParameter 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: WorkflowActionDefinition):
"""Create wrapper function that matches old action signature"""
async def wrapper(parameters: Dict[str, Any], *args, **kwargs):
# Parameter-Validierung basierend auf WorkflowActionParameter 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, WorkflowActionParameter]) -> 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
**WICHTIG**: Die bestehenden Klassen `ActionDefinition` (in `datamodelWorkflow.py`) und `ActionParameters` (in `datamodelChat.py`) haben einen anderen Zweck:
- `ActionDefinition` (existing): Für Workflow-Execution-Planning (enthält konkrete Werte)
- `ActionParameters` (existing): Einfacher Parameter-Wrapper
**Lösung**: Neue Klassen mit klaren Namen für Action-Schema-Definitionen erstellen.
**Datei**: `gateway/modules/datamodels/datamodelWorkflowActions.py`
```python
from typing import Optional, Any, Union, List, Dict, Callable, Awaitable
from pydantic import BaseModel, Field
from modules.datamodels.datamodelChat import ActionResult
from modules.shared.frontendTypes import FrontendType # Globale Definition verwenden
from modules.shared.attributeUtils import registerModelLabels
class WorkflowActionParameter(BaseModel):
"""
Parameter schema definition for a workflow action.
This defines the structure and UI rendering for a single action parameter,
NOT the actual parameter values (those are in ActionDefinition.parameters).
"""
name: str = Field(description="Parameter name")
type: str = Field(description="Python type as string: 'str', 'int', 'bool', 'List[str]', etc.")
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, 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})"
)
class WorkflowActionDefinition(BaseModel):
"""
Complete schema definition of a workflow action.
This defines the metadata, parameters, and execution function for an action.
This is different from datamodelWorkflow.ActionDefinition which contains
actual execution values (action, actionObjective, parameters with values).
This class defines the ACTION SCHEMA, not the execution plan.
"""
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, WorkflowActionParameter] = Field(
default_factory=dict,
description="Parameter schema definitions"
)
execute: Optional[Callable] = Field(
None,
description="Execution function - async function that takes parameters dict and returns ActionResult. Set dynamically."
)
category: Optional[str] = Field(None, description="Action category for grouping")
tags: List[str] = Field(default_factory=list, description="Tags for search/filtering")
# Register model labels for UI
registerModelLabels(
"WorkflowActionDefinition",
{"en": "Workflow Action Definition", "fr": "Définition d'action de workflow"},
{
"actionId": {"en": "Action ID", "fr": "ID d'action"},
"description": {"en": "Description", "fr": "Description"},
"parameters": {"en": "Parameters", "fr": "Paramètres"},
"category": {"en": "Category", "fr": "Catégorie"},
"tags": {"en": "Tags", "fr": "Étiquettes"},
},
)
registerModelLabels(
"WorkflowActionParameter",
{"en": "Workflow Action Parameter", "fr": "Paramètre d'action de workflow"},
{
"name": {"en": "Name", "fr": "Nom"},
"type": {"en": "Type", "fr": "Type"},
"frontendType": {"en": "Frontend Type", "fr": "Type frontend"},
"frontendOptions": {"en": "Frontend Options", "fr": "Options frontend"},
"required": {"en": "Required", "fr": "Requis"},
"default": {"en": "Default", "fr": "Par défaut"},
"description": {"en": "Description", "fr": "Description"},
"validation": {"en": "Validation", "fr": "Validation"},
},
)
```
### 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):
```python
@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):
```python
def __init__(self, services):
super().__init__(services)
self.name = "outlook"
self.description = "Handle Microsoft Outlook email operations"
self._actions = {
"readEmails": WorkflowActionDefinition(
actionId="outlook.readEmails",
description="Read emails from Outlook mailbox",
parameters={
"connectionReference": WorkflowActionParameter(
name="connectionReference",
type="str",
frontendType=FrontendType.USER_CONNECTION, # Custom type - automatisch API-Endpoint
required=True,
description="Microsoft connection label"
),
"query": WorkflowActionParameter(
name="query",
type="str",
frontendType=FrontendType.TEXT,
required=False,
description="Search query for emails"
),
"folder": WorkflowActionParameter(
name="folder",
type="str",
frontendType=FrontendType.SELECT,
frontendOptions="outlook.folder",
required=False,
description="Folder name (e.g., 'Inbox', 'Drafts')"
),
"limit": WorkflowActionParameter(
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
```json
{
"roleLabel": "user",
"context": "RESOURCE",
"item": "outlook.readEmails",
"view": true
}
```
```json
{
"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`:
```python
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 `WorkflowActionDefinition`)
- 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):
```json
{
"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**:
```python
WorkflowActionParameter(
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**:
```python
WorkflowActionParameter(
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:
```python
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**:
```typescript
if (param.frontendType === 'userConnection') {
// Automatisch Options von /api/options/user.connection laden
const options = await fetch(`/api/options/${param.frontendOptions}`);
// Als Select rendern
}
```