Backup-bdg/OpenHands

1import asyncio2from typing import Any3from uuid import uuid44 5from fastapi import Request6 7from openhands.core.logger import openhands_logger as logger8from openhands.events.action.action import Action, ActionSecurityRisk9from openhands.events.event import Event10from openhands.events.stream import EventStream, EventStreamSubscriber11 12 13class SecurityAnalyzer:14    """Security analyzer that receives all events and analyzes agent actions for security risks."""15 16    def __init__(self, event_stream: EventStream) -> None:17        """Initializes a new instance of the SecurityAnalyzer class.18 19        Args:20            event_stream: The event stream to listen for events.21        """22        self.event_stream = event_stream23 24        def sync_on_event(event: Event) -> None:25            asyncio.create_task(self.on_event(event))26 27        self.event_stream.subscribe(28            EventStreamSubscriber.SECURITY_ANALYZER, sync_on_event, str(uuid4())29        )30 31    async def on_event(self, event: Event) -> None:32        """Handles the incoming event, and when Action is received, analyzes it for security risks."""33        logger.debug(f'SecurityAnalyzer received event: {event}')34        await self.log_event(event)35        if not isinstance(event, Action):36            return37 38        try:39            # Set the security_risk attribute on the event40            event.security_risk = await self.security_risk(event)  # type: ignore [attr-defined]41            await self.act(event)42        except Exception as e:43            logger.error(f'Error occurred while analyzing the event: {e}')44 45    async def handle_api_request(self, request: Request) -> Any:46        """Handles the incoming API request."""47        raise NotImplementedError(48            'Need to implement handle_api_request method in SecurityAnalyzer subclass'49        )50 51    async def log_event(self, event: Event) -> None:52        """Logs the incoming event."""53        pass54 55    async def act(self, event: Event) -> None:56        """Performs an action based on the analyzed event."""57        pass58 59    async def security_risk(self, event: Action) -> ActionSecurityRisk:60        """Evaluates the Action for security risks and returns the risk level."""61        raise NotImplementedError(62            'Need to implement security_risk method in SecurityAnalyzer subclass'63        )64 65    async def close(self) -> None:66        """Cleanup resources allocated by the SecurityAnalyzer."""67        pass68