""" Security event handler for MCP Hangar. Provides dedicated security audit logging for: - Authentication and authorization events - Access control violations - Rate limit violations - Suspicious activity detection - Input validation failures - Command injection attempts A DELIBERATE SECOND EVENT SYSTEM -------------------------------- This module carries its own `SecurityEvent` (not a `DomainEvent`), its own severity enum, and its own delivery mechanism (`SecurityEventSink`, with a composite) beside the real `EventBus`. That reads like duplication, and it was investigated as such. It is not, on three grounds -- recorded here so the next reader does not spend the afternoon re-deriving them: 1. **Severity.** `DomainEvent` has no severity, and adding one would put a security concept on every event in the system. 2. **An intake the bus never sees.** Validation failures, injection attempts and suspicious commands have NO `DomainEvent` counterpart at all. They are raised imperatively from the request path through the `log_*` methods below, by `server/validation.py`. Folding this into the bus would mean inventing domain events for them first, which is a larger design question than deduplication. 3. **Aggregation.** `FAILURE_THRESHOLD` / `TIME_WINDOW_S` detect anomalies ACROSS events. A stateless bus subscriber cannot express that. The one apparent collision is rate limiting, and it is not one: `RateLimitLockout` is an auth-lockout domain event (per source IP, published by `auth/infrastructure/rate_limiter.py`), while `log_rate_limit_exceeded` reports request-rate rejection from the budgets Hangar charges a call to: the tool's, through `server/validation.charge_tool()`, and the command type's, through the command bus's `RateLimitMiddleware` (#1495). Different occurrences, so the two cannot disagree about the same one. What the investigation DID find: of the four sinks, only `LogSecuritySink` is ever wired -- `get_security_handler()` is always called with no argument. `InMemorySecuritySink` appears in two test files; `CallbackSecuritySink` and `CompositeSecuritySink` have zero references anywhere, tests included. They are exported through this package's `__all__`, so removing them is a release decision rather than a cleanup, which is why they are still here. """ from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime, UTC from enum import Enum import hashlib import json import logging import threading import time from typing import Any from ...domain.events import ( DomainEvent, HealthCheckFailed, McpServerDegraded, McpServerStarted, McpServerStopped, ToolInvocationCompleted, ToolInvocationFailed, ) from ...logging_config import get_logger logger = get_logger(__name__) class SecurityEventType(Enum): """Types of security events.""" # Access events ACCESS_GRANTED = "access_granted" ACCESS_DENIED = "access_denied" # Rate limiting RATE_LIMIT_EXCEEDED = "rate_limit_exceeded" RATE_LIMIT_WARNING = "rate_limit_warning" # Validation VALIDATION_FAILED = "validation_failed" INJECTION_ATTEMPT = "injection_attempt" # McpServer security PROVIDER_START_BLOCKED = "mcp_server_start_blocked" SUSPICIOUS_COMMAND = "suspicious_command" UNAUTHORIZED_TOOL = "unauthorized_tool" # Health and availability REPEATED_FAILURES = "repeated_failures" PROVIDER_COMPROMISE_SUSPECTED = "mcp_server_compromise_suspected" # Configuration CONFIG_CHANGE = "config_change" SENSITIVE_DATA_ACCESS = "sensitive_data_access" class SecuritySeverity(Enum): """Severity levels for security events.""" INFO = "info" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" @dataclass class SecurityEvent: """Represents a security-related event.""" event_type: SecurityEventType severity: SecuritySeverity message: str timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) # Context mcp_server_id: str | None = None tool_name: str | None = None source_ip: str | None = None user_id: str | None = None # Details details: dict[str, Any] = field(default_factory=dict) # Tracking event_id: str = field(default_factory=lambda: "") correlation_id: str | None = None def __post_init__(self): """Generate event ID if not provided.""" if not self.event_id: # Generate deterministic ID from content content = f"{self.event_type.value}:{self.timestamp.isoformat()}:{self.message}" self.event_id = hashlib.sha256(content.encode()).hexdigest()[:16] def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "event_id": self.event_id, "event_type": self.event_type.value, "severity": self.severity.value, "message": self.message, "timestamp": self.timestamp.isoformat(), "mcp_server_id": self.mcp_server_id, "tool_name": self.tool_name, "source_ip": self.source_ip, "user_id": self.user_id, "details": self.details, "correlation_id": self.correlation_id, } def to_json(self) -> str: """Convert to JSON string.""" return json.dumps(self.to_dict()) class SecurityEventSink(ABC): """Abstract interface for security event destinations.""" @abstractmethod def emit(self, event: SecurityEvent) -> None: """Emit a security event.""" pass class LogSecuritySink(SecurityEventSink): """Security sink that writes to structured logs.""" def __init__(self, logger_name: str = "security"): self._logger = logging.getLogger(logger_name) def emit(self, event: SecurityEvent) -> None: """Log the security event with appropriate level.""" log_data = {"security_event": event.to_dict()} # Map severity to log level if event.severity == SecuritySeverity.CRITICAL: self._logger.critical(json.dumps(log_data)) elif event.severity == SecuritySeverity.HIGH: self._logger.error(json.dumps(log_data)) elif event.severity == SecuritySeverity.MEDIUM: self._logger.warning(json.dumps(log_data)) elif event.severity == SecuritySeverity.LOW: self._logger.info(json.dumps(log_data)) else: self._logger.debug(json.dumps(log_data)) class InMemorySecuritySink(SecurityEventSink): """In-memory security sink for testing and recent event queries.""" def __init__(self, max_events: int = 10000): self._events: list[SecurityEvent] = [] self._max_events = max_events self._lock = threading.Lock() def emit(self, event: SecurityEvent) -> None: """Store the security event.""" with self._lock: self._events.append(event) if len(self._events) > self._max_events: self._events = self._events[-self._max_events :] def query( self, event_type: SecurityEventType | None = None, severity: SecuritySeverity | None = None, mcp_server_id: str | None = None, since: datetime | None = None, limit: int = 100, ) -> list[SecurityEvent]: """Query stored security events.""" with self._lock: from typing import Any results: list[Any] = [] for event in reversed(self._events): if len(results) >= limit: break if event_type and event.event_type != event_type: continue if severity and event.severity != severity: continue if mcp_server_id and event.mcp_server_id != mcp_server_id: continue if since and event.timestamp < since: continue results.append(event) return results def get_severity_counts(self) -> dict[str, int]: """Get counts by severity level.""" with self._lock: counts = {s.value: 0 for s in SecuritySeverity} for event in self._events: counts[event.severity.value] += 1 return counts def clear(self) -> None: """Clear all stored events.""" with self._lock: self._events.clear() @property def count(self) -> int: """Get total event count.""" with self._lock: return len(self._events) class SecurityEventHandler: """ Handler for domain events that detects and logs security-relevant activity. Monitors for: - Repeated failures (potential attacks) - Unusual patterns - Rate limit violations - Validation failures """ # Thresholds for anomaly detection FAILURE_THRESHOLD = 5 # Failures before warning CRITICAL_FAILURE_THRESHOLD = 10 # Failures before critical alert TIME_WINDOW_S = 300 # 5 minute window for tracking def __init__( self, sink: SecurityEventSink | None = None, enable_anomaly_detection: bool = True, ): """ Initialize the security handler. Args: sink: Where to emit security events (defaults to log sink) enable_anomaly_detection: Whether to detect anomalies in event patterns """ self._sink = sink or LogSecuritySink() self._enable_anomaly_detection = enable_anomaly_detection # Tracking for anomaly detection self._failure_counts: dict[str, list[float]] = {} # mcp_server_id -> timestamps self._lock = threading.Lock() def handle(self, event: DomainEvent) -> None: """ Handle a domain event, checking for security implications. Args: event: The domain event to process """ # Dispatch to specific handlers handlers = { McpServerStarted: self._handle_mcp_server_started, McpServerStopped: self._handle_mcp_server_stopped, McpServerDegraded: self._handle_mcp_server_degraded, ToolInvocationCompleted: self._handle_tool_invocation_completed, ToolInvocationFailed: self._handle_tool_invocation_failed, HealthCheckFailed: self._handle_health_check_failed, } handler = handlers.get(type(event)) if handler and callable(handler): handler(event) # Run anomaly detection if self._enable_anomaly_detection: self._check_anomalies(event) def _handle_mcp_server_started(self, event: McpServerStarted) -> None: """Handle mcp_server start event.""" # Log mcp_server starts for audit trail self._emit( SecurityEvent( event_type=SecurityEventType.ACCESS_GRANTED, severity=SecuritySeverity.INFO, message=f"McpServer started: {event.mcp_server_id}", mcp_server_id=event.mcp_server_id, details={ "mode": event.mode, "tools_count": event.tools_count, "startup_duration_ms": event.startup_duration_ms, }, correlation_id=event.event_id, ) ) def _handle_mcp_server_stopped(self, event: McpServerStopped) -> None: """Handle mcp_server stop event.""" # Clear failure tracking for this mcp_server with self._lock: self._failure_counts.pop(event.mcp_server_id, None) def _handle_mcp_server_degraded(self, event: McpServerDegraded) -> None: """Handle mcp_server degradation event.""" severity = SecuritySeverity.MEDIUM if event.consecutive_failures >= self.CRITICAL_FAILURE_THRESHOLD: severity = SecuritySeverity.HIGH self._emit( SecurityEvent( event_type=SecurityEventType.REPEATED_FAILURES, severity=severity, message=f"McpServer degraded after {event.consecutive_failures} consecutive failures", mcp_server_id=event.mcp_server_id, details={ "consecutive_failures": event.consecutive_failures, "total_failures": event.total_failures, "reason": event.reason, }, correlation_id=event.event_id, ) ) def _handle_tool_invocation_completed(self, event: ToolInvocationCompleted) -> None: """Handle successful tool invocation.""" # Only log if unusually slow (potential DoS or resource exhaustion) if event.duration_ms > 10000: # > 10 seconds self._emit( SecurityEvent( event_type=SecurityEventType.ACCESS_GRANTED, severity=SecuritySeverity.LOW, message=f"Slow tool invocation: {event.tool_name}", mcp_server_id=event.mcp_server_id, tool_name=event.tool_name, details={ "duration_ms": event.duration_ms, }, correlation_id=event.event_id, ) ) def _handle_tool_invocation_failed(self, event: ToolInvocationFailed) -> None: """Handle failed tool invocation. Carries the bounded ``error_type``, never ``error_message``: the message can hold what the tool returned, and the log sink writes this at INFO (GHSA-qwq2-7g49-jxc6). The full message stays on the event itself. """ self._record_failure(event.mcp_server_id) self._emit( SecurityEvent( event_type=SecurityEventType.ACCESS_DENIED, severity=SecuritySeverity.LOW, message=f"Tool invocation failed: {event.tool_name}", mcp_server_id=event.mcp_server_id, tool_name=event.tool_name, details={ "error_type": event.error_type, }, correlation_id=event.event_id, ) ) def _handle_health_check_failed(self, event: HealthCheckFailed) -> None: """Handle health check failure. Carries no error text: ``error_message`` can hold what the upstream answered, the log sink writes this at WARNING, and the event has no bounded error type to log instead (GHSA-qwq2-7g49-jxc6). The full message stays on the event itself. """ self._record_failure(event.mcp_server_id) if event.consecutive_failures >= self.FAILURE_THRESHOLD: self._emit( SecurityEvent( event_type=SecurityEventType.REPEATED_FAILURES, severity=SecuritySeverity.MEDIUM, message="Multiple health check failures for mcp_server", mcp_server_id=event.mcp_server_id, details={ "consecutive_failures": event.consecutive_failures, }, correlation_id=event.event_id, ) ) def _record_failure(self, mcp_server_id: str) -> None: """Record a failure for anomaly detection.""" now = time.time() with self._lock: if mcp_server_id not in self._failure_counts: self._failure_counts[mcp_server_id] = [] # Add current failure self._failure_counts[mcp_server_id].append(now) # Clean old entries cutoff = now - self.TIME_WINDOW_S self._failure_counts[mcp_server_id] = [t for t in self._failure_counts[mcp_server_id] if t > cutoff] def _check_anomalies(self, event: DomainEvent) -> None: """Check for anomalous patterns across events.""" mcp_server_id = getattr(event, "mcp_server_id", None) if not mcp_server_id: return with self._lock: failures = self._failure_counts.get(mcp_server_id, []) if len(failures) >= self.CRITICAL_FAILURE_THRESHOLD: self._emit( SecurityEvent( event_type=SecurityEventType.PROVIDER_COMPROMISE_SUSPECTED, severity=SecuritySeverity.HIGH, message="High failure rate detected for mcp_server (possible attack or compromise)", mcp_server_id=mcp_server_id, details={ "failures_in_window": len(failures), "window_seconds": self.TIME_WINDOW_S, }, ) ) def _emit(self, event: SecurityEvent) -> None: """Emit a security event to the sink.""" try: self._sink.emit(event) except Exception as e: # noqa: BLE001 -- fault-barrier: security event emission failure must not crash handler logger.error(f"Failed to emit security event: {e}") # --- Public API for direct security event emission --- def log_rate_limit_exceeded( self, mcp_server_id: str | None = None, limit: int = 0, window_seconds: int = 0, source_ip: str | None = None, *, scope: str = "", key_kind: str = "", key: str = "", ) -> None: """Log a rate limit violation, from whichever limiter refused the call (#1495). Every refusal is recorded here once: the tool-level check (`server/validation.charge_tool`) and the command bus's limiter alike. `scope` is whose budget was used up (`caller` or `all_callers`), `key_kind` what the budget is named after (`tool` or `command`), and `key` that name. All three are values Hangar chose, so the record stays bounded: no argument value and no caller's own text reaches it. """ details: dict[str, Any] = { "limit": limit, "window_seconds": window_seconds, } if scope: details["scope"] = scope if key_kind: details["key_kind"] = key_kind if key: details["key"] = key self._emit( SecurityEvent( event_type=SecurityEventType.RATE_LIMIT_EXCEEDED, severity=SecuritySeverity.MEDIUM, message="Rate limit exceeded", mcp_server_id=mcp_server_id, source_ip=source_ip, details=details, ) ) def log_validation_failed( self, field: str, message: str, mcp_server_id: str | None = None, value: str | None = None, ) -> None: """Log a validation failure.""" # Determine severity based on field severity = SecuritySeverity.LOW if field in ("command", "image"): severity = SecuritySeverity.MEDIUM details = {"field": field} if value: # Truncate value for safety details["value"] = value[:50] if len(value) > 50 else value self._emit( SecurityEvent( event_type=SecurityEventType.VALIDATION_FAILED, severity=severity, message=f"Validation failed: {message}", mcp_server_id=mcp_server_id, details=details, ) ) def log_injection_attempt( self, field: str, pattern: str, mcp_server_id: str | None = None, source_ip: str | None = None, ) -> None: """Log a potential injection attempt.""" self._emit( SecurityEvent( event_type=SecurityEventType.INJECTION_ATTEMPT, severity=SecuritySeverity.HIGH, message=f"Potential injection attempt detected in {field}", mcp_server_id=mcp_server_id, source_ip=source_ip, details={ "field": field, "pattern_detected": pattern, }, ) ) def log_suspicious_command( self, command: list[str], mcp_server_id: str | None = None, reason: str = "", ) -> None: """Log a suspicious command execution attempt.""" # Sanitize command for logging (don't log full values) safe_command = [c[:20] + "..." if len(c) > 20 else c for c in command[:5]] self._emit( SecurityEvent( event_type=SecurityEventType.SUSPICIOUS_COMMAND, severity=SecuritySeverity.HIGH, message=f"Suspicious command blocked: {reason}", mcp_server_id=mcp_server_id, details={ "command_preview": safe_command, "reason": reason, }, ) ) def log_config_change( self, change_type: str, mcp_server_id: str | None = None, user_id: str | None = None, details: dict[str, Any] | None = None, ) -> None: """Log a configuration change.""" self._emit( SecurityEvent( event_type=SecurityEventType.CONFIG_CHANGE, severity=SecuritySeverity.INFO, message=f"Configuration changed: {change_type}", mcp_server_id=mcp_server_id, user_id=user_id, details=details or {}, ) ) @property def sink(self) -> SecurityEventSink: """Get the security event sink.""" return self._sink # --- Global security handler instance --- _security_handler: SecurityEventHandler | None = None def get_security_handler( sink: SecurityEventSink | None = None, ) -> SecurityEventHandler: """Get or create the global security handler instance.""" global _security_handler if _security_handler is None: _security_handler = SecurityEventHandler(sink=sink) return _security_handler def reset_security_handler() -> None: """Reset the global security handler (for testing).""" global _security_handler _security_handler = None