Spaces:
Configuration error
Configuration error
File size: 2,758 Bytes
d5c8c48 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
# src/llmguardian/agency/executor.py
from typing import Dict, Any, Optional
from dataclasses import dataclass
from ..core.logger import SecurityLogger
from .action_validator import Action, ActionValidator
from .permission_manager import PermissionManager
from .scope_limiter import ScopeLimiter
@dataclass
class ExecutionResult:
success: bool
output: Optional[Any] = None
error: Optional[str] = None
class SafeExecutor:
def __init__(self,
security_logger: Optional[SecurityLogger] = None,
permission_manager: Optional[PermissionManager] = None,
action_validator: Optional[ActionValidator] = None,
scope_limiter: Optional[ScopeLimiter] = None):
self.security_logger = security_logger
self.permission_manager = permission_manager or PermissionManager()
self.action_validator = action_validator or ActionValidator()
self.scope_limiter = scope_limiter or ScopeLimiter()
async def execute(self,
action: Action,
user_id: str,
context: Dict[str, Any]) -> ExecutionResult:
try:
# Validate permissions
if not self.permission_manager.check_permission(
user_id, action.resource, action.type
):
return ExecutionResult(
success=False,
error="Permission denied"
)
# Validate action
if not self.action_validator.validate_action(action, context):
return ExecutionResult(
success=False,
error="Invalid action"
)
# Check scope
if not self.scope_limiter.check_scope(
user_id, action.type, action.resource
):
return ExecutionResult(
success=False,
error="Out of scope"
)
# Execute action safely
result = await self._execute_action(action, context)
return ExecutionResult(success=True, output=result)
except Exception as e:
if self.security_logger:
self.security_logger.log_security_event(
"execution_error",
action=action.__dict__,
error=str(e)
)
return ExecutionResult(
success=False,
error=f"Execution failed: {str(e)}"
)
async def _execute_action(self,
action: Action,
context: Dict[str, Any]) -> Any:
# Implementation of safe action execution
pass |