diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 49525c2f..19a2c540 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -9,6 +9,8 @@ on: jobs: test: runs-on: ubuntu-latest + env: + LLM_PROVIDER: mock steps: - uses: actions/checkout@v4 diff --git a/.gitignore b/.gitignore index c4fc990f..7eafb282 100644 --- a/.gitignore +++ b/.gitignore @@ -75,7 +75,7 @@ google-credentials.json SECURITY_BUGS_REPORT.md -.node_modules/ +node_modules/ # SQLite databases *.db diff --git a/finbot/aegis/__init__.py b/finbot/aegis/__init__.py new file mode 100644 index 00000000..90d3663d --- /dev/null +++ b/finbot/aegis/__init__.py @@ -0,0 +1,24 @@ +# ============================================================ +# File: finbot/aegis/__init__.py +# Purpose: Public exports for FinBot-AEGIS runtime security layer +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 1 +# OWASP Category: ASI01–ASI10 (platform-wide) +# ============================================================ +"""FinBot-AEGIS: runtime security layer for OWASP FinBot CTF.""" + +from finbot.aegis.intent_gate import IntentGate +from finbot.aegis.schemas import PolicyVerdict +from finbot.aegis.sentinel import AuditEvent, SentinelStream +from finbot.aegis.service import AegisEnforcementService +from finbot.aegis.trust_mesh import AttestationResult, TrustMesh + +__all__ = [ + "AegisEnforcementService", + "AttestationResult", + "AuditEvent", + "IntentGate", + "PolicyVerdict", + "SentinelStream", + "TrustMesh", +] diff --git a/finbot/aegis/policy/__init__.py b/finbot/aegis/policy/__init__.py new file mode 100644 index 00000000..24d3c03f --- /dev/null +++ b/finbot/aegis/policy/__init__.py @@ -0,0 +1,26 @@ +# ============================================================ +# File: finbot/aegis/policy/__init__.py +# Package: finbot.aegis.policy +# Purpose: Policy engine package for AEGIS - Week 7 +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 +# OWASP Category: ASI02 Tool Misuse, ASI03 Excessive Agency, ASI07 Prompt Injection +# ============================================================ +"""Policy engine package for FinBot-AEGIS. + +This package implements the policy engine for resource governance +using policy gradients and intent capsules for secure agent operations. +""" + +from .agent_profiles import AGENT_PROFILES, AgentProfile +from .capsule import IntentCapsule +from .interceptor import MCPPolicyInterceptor +from .memory import MemoryNamespaceManager + +__all__ = [ + "AGENT_PROFILES", + "AgentProfile", + "IntentCapsule", + "MCPPolicyInterceptor", + "MemoryNamespaceManager", +] diff --git a/finbot/aegis/policy/agent_profiles.py b/finbot/aegis/policy/agent_profiles.py new file mode 100644 index 00000000..d7a8cf9c --- /dev/null +++ b/finbot/aegis/policy/agent_profiles.py @@ -0,0 +1,237 @@ +# Test file + +# ============================================================ +# File: finbot/aegis/policy/agent_profiles.py +# Package: finbot.aegis.policy +# Purpose: Least-agency profiles for 6 agent types +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 +# OWASP Category: ASI03 Excessive Agency +# ============================================================ +"""Least-agency profiles for FinBot-AEGIS agent types. + +Defines privilege profiles for six agent types following the principle +of least privilege to minimize the attack surface of autonomous agents. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional + + +@dataclass +class AgentProfile: + """Defines the least-agency profile for an agent type. + + Attributes: + name: Unique identifier for the agent type + description: Human-readable description of the agent's role + allowed_tools: List of tools this agent is permitted to use + tool_limits: Per-tool rate limits and quotas + secret_key: Secret key for signing intent capsules + current_workflow: Current workflow identifier (for provenance) + max_concurrent_operations: Maximum simultaneous operations allowed + required_attestation: Whether attestation is required for tool use + """ + + name: str + description: str + allowed_tools: List[str] = field(default_factory=list) + tool_limits: Dict[str, Dict[str, int]] = field(default_factory=dict) + secret_key: str = field(default_factory=lambda: "default-secret-key-change-in-production") + current_workflow: Optional[str] = None + max_concurrent_operations: int = 1 + required_attestation: bool = False + + def allows_tool(self, tool_name: str) -> bool: + """Check if this agent is allowed to use a specific tool. + + Args: + tool_name: Name of the tool to check + + Returns: + True if the agent can use the tool, False otherwise + """ + # Check exact match + if tool_name in self.allowed_tools: + return True + + # Check prefix matches (e.g., "filesystem:*" allows all filesystem tools) + for allowed_tool in self.allowed_tools: + if allowed_tool.endswith("*") and tool_name.startswith(allowed_tool[:-1]): + return True + + return False + + def get_tool_limit(self, tool_name: str, limit_type: str = "max_calls_per_minute") -> int: + """Get a specific limit for a tool. + + Args: + tool_name: Name of the tool + limit_type: Type of limit to retrieve + + Returns: + The limit value, or 0 if not specified + """ + return self.tool_limits.get(tool_name, {}).get(limit_type, 0) + +# Predefined agent profiles for the six agent types in FinBot-AEGIS + +# 1. Data Analyst Agent - Limited to data querying and analysis tools +DATA_ANALYST_PROFILE = AgentProfile( + name="data_analyst", + description="Agent specialized in data analysis, querying, and visualization", + allowed_tools=[ + "database:query", + "database:read_schema", + "data:visualize", + "data:transform", + "data:summarize", + "filesystem:read", + ], + tool_limits={ + "database:query": {"max_calls_per_minute": 30}, + "data:visualize": {"max_calls_per_minute": 10}, + "filesystem:read": {"max_calls_per_minute": 60}, + }, + secret_key="data-analyst-secret-key-change-in-production", + max_concurrent_operations=3, +) + +# 2. Report Generator Agent - Limited to document creation and formatting tools +REPORT_GENERATOR_PROFILE = AgentProfile( + name="report_generator", + description="Agent specialized in generating reports, documents, and presentations", + allowed_tools=[ + "document:create", + "document:edit", + "document:format", + "presentation:create", + "chart:generate", + "filesystem:read", + "filesystem:write", + ], + tool_limits={ + "document:create": {"max_calls_per_minute": 20}, + "filesystem:write": {"max_calls_per_minute": 30}, + }, + secret_key="report-generator-secret-key-change-in-production", + max_concurrent_operations=2, +) + +# 3. API Integrator Agent - Limited to API interaction and integration tools +API_INTEGRATOR_PROFILE = AgentProfile( + name="api_integrator", + description="Agent specialized in integrating with external APIs and services", + allowed_tools=[ + "http:request", + "api:call", + "webhook:send", + "message:queue", + "cache:get", + "cache:set", + ], + tool_limits={ + "http:request": {"max_calls_per_minute": 60}, + "api:call": {"max_calls_per_minute": 60}, + "cache:set": {"max_calls_per_minute": 120}, + }, + secret_key="api-integrator-secret-key-change-in-production", + max_concurrent_operations=5, +) + +# 4. Security Auditor Agent - Limited to security scanning and audit tools +SECURITY_AUDITOR_PROFILE = AgentProfile( + name="security_auditor", + description="Agent specialized in security scanning, vulnerability assessment, and compliance checking", + allowed_tools=[ + "security:scan", + "vulnerability:check", + "compliance:check", + "audit:log", + "filesystem:read", + "network:scan", + ], + tool_limits={ + "security:scan": {"max_calls_per_minute": 10}, + "vulnerability:check": {"max_calls_per_minute": 20}, + "network:scan": {"max_calls_per_minute": 5}, + }, + secret_key="security-auditor-secret-key-change-in-production", + max_concurrent_operations=2, + required_attestation=True, +) + +# 5. Workflow Orchestrator Agent - Limited to workflow and automation tools +WORKFLOW_ORCHESTRATOR_PROFILE = AgentProfile( + name="workflow_orchestrator", + description="Agent specialized in orchestrating workflows and automating processes", + allowed_tools=[ + "workflow:create", + "workflow:execute", + "workflow:monitor", + "task:schedule", + "notification:send", + "approval:request", + ], + tool_limits={ + "workflow:execute": {"max_calls_per_minute": 20}, + "task:schedule": {"max_calls_per_minute": 30}, + }, + secret_key="workflow-orchestrator-secret-key-change-in-production", + max_concurrent_operations=10, +) + +# 6. Research Assistant Agent - Limited to research and information gathering tools +RESEARCH_ASSISTANT_PROFILE = AgentProfile( + name="research_assistant", + description="Agent specialized in research, information gathering, and knowledge synthesis", + allowed_tools=[ + "web:search", + "knowledge:query", + "document:summarize", + "citation:format", + "filesystem:read", + "filesystem:write", + ], + tool_limits={ + "web:search": {"max_calls_per_minute": 30}, + "knowledge:query": {"max_calls_per_minute": 60}, + "filesystem:write": {"max_calls_per_minute": 20}, + }, + secret_key="research-assistant-secret-key-change-in-production", + max_concurrent_operations=5, +) + + +# Dictionary mapping agent names to their profiles for easy lookup +AGENT_PROFILES: dict[str, AgentProfile] = { + "data_analyst": DATA_ANALYST_PROFILE, + "report_generator": REPORT_GENERATOR_PROFILE, + "api_integrator": API_INTEGRATOR_PROFILE, + "security_auditor": SECURITY_AUDITOR_PROFILE, + "workflow_orchestrator": WORKFLOW_ORCHESTRATOR_PROFILE, + "research_assistant": RESEARCH_ASSISTANT_PROFILE, +} + + +def get_agent_profile(agent_name: str) -> Optional[AgentProfile]: + """Get the agent profile for a given agent name. + + Args: + agent_name: Name of the agent + + Returns: + The agent profile if found, None otherwise + """ + return AGENT_PROFILES.get(agent_name) + + +def register_agent_profile(profile: AgentProfile) -> None: + """Register a new agent profile. + + Args: + profile: The agent profile to register + """ + AGENT_PROFILES[profile.name] = profile diff --git a/finbot/aegis/policy/capsule.py b/finbot/aegis/policy/capsule.py new file mode 100644 index 00000000..92dcfc3d --- /dev/null +++ b/finbot/aegis/policy/capsule.py @@ -0,0 +1,103 @@ +# ============================================================ +# File: finbot/aegis/policy/capsule.py +# Package: finbot.aegis.policy +# Purpose: Intent capsule HMAC signing + validation (stub for Week 7, to be completed in Week 8) +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 (stub) / 8 (implementation) +# OWASP Category: ASI03, ASI07 +# ============================================================ +"""Intent capsule for signed intent delegation (stub implementation). + +This is a stub implementation for Week 7. The full implementation with +HMAC signing and validation will be completed in Week 8. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class IntentCapsule: + """Represents a signed intent for tool execution (stub version). + + In the full implementation (Week 8), this will include: + - HMAC signing using agent's secret key + - Validation of signatures + - Timestamp-based expiration + - Nonce for replay attack prevention + """ + + agent_name: str + tool_name: str + tool_args: dict[str, Any] + tool_source: str + workflow_id: str + timestamp: float = field(default_factory=lambda: 0.0) + nonce: str = field(default_factory=lambda: "") + signature: str = field(default_factory=lambda: "") + + @classmethod + def from_tool_call( + cls, + agent_name: str, + tool_name: str, + tool_args: dict[str, Any], + tool_source: str, + workflow_id: str = "unknown", + ) -> "IntentCapsule": + """Create an intent capsule from a tool call (stub version). + + Args: + agent_name: Name of the agent making the call + tool_name: Name of the tool being called + tool_args: Arguments being passed to the tool + tool_source: Source/namespace of the tool + workflow_id: Current workflow identifier + + Returns: + A new IntentCapsule instance + """ + return cls( + agent_name=agent_name, + tool_name=tool_name, + tool_args=tool_args.copy(), + tool_source=tool_source, + workflow_id=workflow_id, + timestamp=0.0, # Will be set in full implementation + nonce="", # Will be set in full implementation + signature="", # Will be set in full implementation + ) + + def sign(self, secret_key: str) -> None: + """Sign the intent capsule (stub version). + + In the full implementation (Week 8), this will: + - Create a message to sign from the capsule contents + - Use HMAC-SHA256 with the secret key + - Store the resulting signature + + Args: + secret_key: Secret key for signing + """ + # Stub implementation - in Week 8, this will do actual HMAC signing + self.signature = f"stub-signature-for-{self.tool_name}" + + def verify(self, secret_key: str) -> bool: + """Verify the intent capsule signature (stub version). + + In the full implementation (Week 8), this will: + - Recompute the expected signature + - Compare with the stored signature using constant-time comparison + - Check timestamp for expiration + - Check nonce for replay attacks + + Args: + secret_key: Secret key for verification + + Returns: + True if signature is valid, False otherwise + """ + # Stub implementation - in Week 8, this will do actual verification + return self.signature == f"stub-signature-for-{self.tool_name}" diff --git a/finbot/aegis/policy/interceptor.py b/finbot/aegis/policy/interceptor.py new file mode 100644 index 00000000..06fbb968 --- /dev/null +++ b/finbot/aegis/policy/interceptor.py @@ -0,0 +1,238 @@ +# Test file + +# ============================================================ +# File: finbot/aegis/policy/interceptor.py +# Package: finbot.aegis.policy +# Purpose: MCPPolicyInterceptor - wraps tool execution with policy enforcement +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 +# OWASP Category: ASI02 Tool Misuse, ASI03 Excessive Agency, ASI07 Prompt Injection +# ============================================================ +"""Model Context Protocol Policy Interceptor for tool execution governance. + +Implements policy gradients for resource governance by wrapping MCP tool +enforcement decisions based on intent capsules and agent profiles. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import time +from typing import Any, Callable, Dict, Optional + +from finbot.aegis.policy.agent_profiles import AgentProfile +from .capsule import IntentCapsule +from .memory import MemoryNamespaceManager + +class MCPPolicyInterceptor: + """Intercepts MCP tool calls and enforces policy gradients. + + This interceptor wraps MCP tool execution to enforce least-privilege + access controls based on agent profiles and intent capsules. + """ + + def __init__( + self, + agent_profile: AgentProfile, + memory_manager: Optional[MemoryNamespaceManager] = None, + policy_gradient_weight: float = 0.1, + ) -> None: + """Initialize the MCP policy interceptor. + + Args: + agent_profile: The agent's least-agency profile + memory_manager: Optional memory namespace manager for provenance + policy_gradient_weight: Weight for policy gradient updates (0.0-1.0) + """ + self.agent_profile = agent_profile + self.memory_manager = memory_manager or MemoryNamespaceManager() + self.policy_gradient_weight = policy_gradient_weight + self._execution_count = 0 + self._violation_count = 0 + + async def intercept_tool_call( + self, + tool_name: str, + tool_args: dict[str, Any], + tool_source: str, + next_handler: Callable[..., Any], + ) -> Any: + """Intercept and potentially modify a tool call based on policy. + + Args: + tool_name: Name of the tool being called + tool_args: Arguments to pass to the tool + tool_source: Source/namespace of the tool + next_handler: Async callable to invoke the actual tool + + Returns: + The result of the tool execution + + Raises: + PermissionError: If the tool call violates policy + """ + start_time = time.time() + self._execution_count += 1 + + # Create intent capsule for this tool call + intent_capsule = IntentCapsule.from_tool_call( + agent_name=self.agent_profile.name, + tool_name=tool_name, + tool_args=tool_args, + tool_source=tool_source, + workflow_id=getattr(self.agent_profile, "current_workflow", "unknown"), + ) + + # Sign the intent capsule + intent_capsule.sign(secret_key=self.agent_profile.secret_key) + + # Check policy permissions + if not self.agent_profile.allows_tool(tool_name): + self._violation_count += 1 + await self._log_policy_violation( + tool_name, tool_args, tool_source, f"Tool {tool_name} not allowed for agent {self.agent_profile.name}" + ) + raise PermissionError( + f"Agent {self.agent_profile.name} not authorized to use tool {tool_name}" + ) + + # Check resource limits + if not self._check_resource_limits(tool_name, tool_args): + self._violation_count += 1 + await self._log_policy_violation( + tool_name, tool_args, tool_source, f"Resource limits exceeded for tool {tool_name}" + ) + raise PermissionError(f"Resource limits exceeded for tool {tool_name}") + + # Record intent in memory namespace for provenance + if self.memory_manager: + await self.memory_manager.record_intent(intent_capsule) + + # Execute the tool + try: + result = await next_handler(tool_name, tool_args, tool_source) + + # Log successful execution for policy gradient feedback + execution_time = time.time() - start_time + await self._log_successful_execution( + tool_name, tool_args, tool_source, result, execution_time, intent_capsule + ) + + return result + except Exception as e: + # Log failed execution + await self._log_failed_execution( + tool_name, tool_args, tool_source, str(e), intent_capsule + ) + raise + + def _check_resource_limits(self, tool_name: str, tool_args: dict[str, Any]) -> bool: + """Check if the tool call exceeds agent resource limits. + + Args: + tool_name: Name of the tool being called + tool_args: Arguments for the tool + + Returns: + True if within limits, False otherwise + """ + # Check if tool has specific rate limits + tool_limits = self.agent_profile.tool_limits.get(tool_name, {}) + + # For now, implement basic call counting - in practice this would + # be more sophisticated with time windows, resource quotas, etc. + max_calls_per_minute = tool_limits.get("max_calls_per_minute", 60) + + # Simple rate limiting check (would be enhanced with sliding window in production) + return self._execution_count <= max_calls_per_minute + + async def _log_policy_violation( + self, tool_name: str, tool_args: dict[str, Any], tool_source: str, reason: str + ) -> None: + """Log a policy violation for auditing and policy gradient updates. + + Args: + tool_name: Name of the tool that was blocked + tool_args: Arguments that were provided + tool_source: Source of the tool + reason: Reason for the violation + """ + violation_entry = { + "timestamp": time.time(), + "agent": self.agent_profile.name, + "tool_name": tool_name, + "tool_args": tool_args, + "tool_source": tool_source, + "violation_reason": reason, + "action_taken": "blocked", + } + + # In a full implementation, this would go to a security event store + # For now, we'll just update internal counters + pass + + async def _log_successful_execution( + self, + tool_name: str, + tool_args: dict[str, Any], + tool_source: str, + result: Any, + execution_time: float, + intent_capsule: IntentCapsule, + ) -> None: + """Log successful execution for policy gradient feedback. + + Args: + tool_name: Name of the tool that was executed + tool_args: Arguments that were used + tool_source: Source of the tool + result: Result of the tool execution + execution_time: Time taken for execution + intent_capsule: The intent capsule for this execution + """ + # Update policy gradient based on successful execution + # This would typically involve reinforcement learning signals + # For now, we track basic metrics + pass + + async def _log_failed_execution( + self, + tool_name: str, + tool_args: dict[str, Any], + tool_source: str, + error: str, + intent_capsule: IntentCapsule, + ) -> None: + """Log failed execution for debugging and policy improvement. + + Args: + tool_name: Name of the tool that failed + tool_args: Arguments that were provided + tool_source: Source of the tool + error: Error message from execution + intent_capsule: The intent capsule for this execution + """ + # Log failure for analysis + pass + + def get_metrics(self) -> dict[str, Any]: + """Get current interceptor metrics for monitoring. + + Returns: + Dictionary of interceptor metrics + """ + violation_rate = ( + self._violation_count / max(self._execution_count, 1) + if self._execution_count > 0 + else 0.0 + ) + + return { + "agent": self.agent_profile.name, + "total_executions": self._execution_count, + "total_violations": self._violation_count, + "violation_rate": violation_rate, + "policy_gradient_weight": self.policy_gradient_weight, + } diff --git a/finbot/aegis/policy/memory.py b/finbot/aegis/policy/memory.py new file mode 100644 index 00000000..901fd9e8 --- /dev/null +++ b/finbot/aegis/policy/memory.py @@ -0,0 +1,101 @@ +# ============================================================ +# File: finbot/aegis/policy/memory.py +# Package: finbot.aegis.policy +# Purpose: Memory namespace isolation + provenance (stub for Week 7, to be completed in Week 8) +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 (stub) / 8 (implementation) +# OWASP Category: ASI03, ASI07 +# ==========================================================__ +"""Memory namespace manager for provenance and isolation (stub implementation). + +This is a stub implementation for Week 7. The full implementation with +proper namespace isolation and provenance tracking will be completed in Week 8. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict, List, Optional + + +class MemoryNamespaceManager: + """Manages isolated namespaces for agent memory and provenance tracking (stub version). + + In the full implementation (Week 8), this will provide: + - Namespace isolation between agents and workflows + - Cryptographic chaining of audit entries + - Tamper-evident logs + - Efficient querying of provenance information + """ + + def __init__(self) -> None: + """Initialize the memory namespace manager.""" + # In the full implementation, this would set up storage backends, + # namespace isolation mechanisms, etc. + self._records: list[dict[str, Any]] = [] + + async def record_intent( + self, + agent_name: str, + intent_capsule: Any, # Will be IntentCapsule in full implementation + workflow_id: str = "unknown", + ) -> None: + """Record an intent for provenance tracking (stub version). + + In the full implementation (Week 8), this will: + - Store the intent in a cryptographically secure manner + - Link it to previous entries in the chain + - Apply namespace-based access controls + - Index it for efficient querying + + Args: + agent_name: Name of the agent + intent_capsule: The intent capsule to record + workflow_id: Current workflow identifier + """ + # Stub implementation - just record in memory + self._records.append({ + "timestamp": time.time(), + "agent_name": agent_name, + "intent_capsule": intent_capsule, + "workflow_id": workflow_id, + "record_id": len(self._records), + }) + + async def get_intents( + self, + agent_name: Optional[str] = None, + workflow_id: Optional[str] = None, + limit: int = 100, + ) -> list[dict[str, Any]]: + """Get recorded intents with optional filtering (stub version). + + Args: + agent_name: Filter by agent name (optional) + workflow_id: Filter by workflow ID (optional) + limit: Maximum number of records to return + + Returns: + List of intent records matching the criteria + """ + # Stub implementation - simple filtering + results = self._records + + if agent_name is not None: + results = [r for r in results if r["agent_name"] == agent_name] + + if workflow_id is not None: + results = [r for r in results if r["workflow_id"] == workflow_id] + + return results[-limit:] if len(results) > limit else results + + def get_metrics(self) -> dict[str, Any]: + """Get memory manager metrics for monitoring. + + Returns: + Dictionary of memory manager metrics + """ + return { + "total_records": len(self._records), + "storage_type": "stub-in-memory", + } diff --git a/finbot/aegis/telemetry/__init__.py b/finbot/aegis/telemetry/__init__.py new file mode 100644 index 00000000..c081107b --- /dev/null +++ b/finbot/aegis/telemetry/__init__.py @@ -0,0 +1,28 @@ +# ============================================================ +# File: finbot/aegis/telemetry/__init__.py +# Purpose: Telemetry package initialization +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 1 +# OWASP Category: ASI01 (Prompt Injection), ASI06 (Sandboxing) +# ============================================================ +"""AEGIS Telemetry: structured audit event pipeline with HMAC chaining.""" + +from finbot.aegis.telemetry.chain import AuditChain +from finbot.aegis.telemetry.schema import ( + AuditEvent, + DelegationEvent, + MemoryWriteEvent, + PolicyDecisionEvent, + ToolCallEvent, + ToolResultEvent, +) + +__all__ = [ + "AuditEvent", + "ToolCallEvent", + "ToolResultEvent", + "MemoryWriteEvent", + "DelegationEvent", + "PolicyDecisionEvent", + "AuditChain", +] diff --git a/finbot/aegis/telemetry/schema.py b/finbot/aegis/telemetry/schema.py new file mode 100644 index 00000000..f6e669c5 --- /dev/null +++ b/finbot/aegis/telemetry/schema.py @@ -0,0 +1,231 @@ +# ============================================================ +# File: finbot/aegis/telemetry/schema.py +# Purpose: JSON-LD schemas for structured audit events +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 1 +# OWASP Category: ASI01 (Prompt Injection), ASI06 (Sandboxing) +# ============================================================ +"""JSON-LD event schemas for AEGIS telemetry pipeline. + +All events include: +- @context: JSON-LD context URL +- @type: Event type (ToolCall, ToolResult, etc.) +- timestamp: ISO 8601 timestamp +- namespace: Player's isolated namespace +- workflow_id: Execution trace identifier +- prev_hash: HMAC of previous event (for chaining) +- event_hash: HMAC of this event +""" + +from datetime import UTC, datetime +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel, Field, field_validator + + +class EventType(str, Enum): + """AEGIS event types for audit trail.""" + + TOOL_CALL = "aegis.tool.call" + TOOL_RESULT = "aegis.tool.result" + MEMORY_WRITE = "aegis.memory.write" + DELEGATION = "aegis.delegation" + POLICY_DECISION = "aegis.policy.decision" + ANOMALY_DETECTION = "aegis.anomaly.detection" + + +class BaseAuditEvent(BaseModel): + """Base class for all AEGIS audit events.""" + + context: str = Field( + default="https://owasp.org/aegis/v1/context.jsonld", + alias="@context", + ) + type: str = Field(alias="@type") + timestamp: str = Field( + default_factory=lambda: datetime.now(UTC).isoformat().replace("+00:00", "Z") + ) + namespace: str = Field( + description="Player's isolated namespace (e.g., 'player_abc123')" + ) + workflow_id: str = Field( + description="Execution workflow identifier for tracing" + ) + user_id: str = Field(description="User who initiated the action") + agent_name: str = Field(description="Agent performing the action") + prev_hash: Optional[str] = Field(default=None, description="HMAC of previous event") + event_hash: Optional[str] = Field(default=None, description="HMAC of this event") + severity: str = Field( + default="info", + description="Event severity: debug, info, warning, critical", + ) + labels: dict[str, str] = Field( + default_factory=dict, + description="Custom labels for filtering (e.g., {'asi': 'ASI01'})", + ) + + class Config: + """Pydantic config.""" + + populate_by_name = True + json_schema_extra = { + "examples": [ + { + "@context": "https://owasp.org/aegis/v1/context.jsonld", + "@type": "aegis.tool.call", + "timestamp": "2026-05-27T12:34:56Z", + "namespace": "player_abc123", + "workflow_id": "wf_xyz789", + "user_id": "user_1", + "agent_name": "OnboardingAgent", + "tool_name": "create_vendor", + "arguments": {"name": "Acme Corp"}, + "severity": "info", + "labels": {"asi": "ASI01", "phase": "recon"}, + } + ] + } + + +class ToolCallEvent(BaseAuditEvent): + """Fired when an agent calls a tool (before execution).""" + + type: str = Field(default=EventType.TOOL_CALL.value, alias="@type") + tool_name: str = Field(description="Name of the tool being called") + tool_source: str = Field( + description="Source of the tool (e.g., 'findrive', 'finmail', 'finstripe')" + ) + arguments: dict[str, Any] = Field( + default_factory=dict, + description="Tool arguments (sanitized; sensitive values masked)", + ) + tool_description: Optional[str] = Field( + default=None, + description="Description of what the tool does", + ) + + +class ToolResultEvent(BaseAuditEvent): + """Fired when a tool returns a result (after execution).""" + + type: str = Field(default=EventType.TOOL_RESULT.value, alias="@type") + tool_name: str = Field(description="Name of the tool that was called") + return_value: Optional[str] = Field( + default=None, + description="Tool result (truncated if large; first 500 chars)", + ) + success: bool = Field(description="Whether the tool call succeeded") + error_message: Optional[str] = Field(default=None, description="Error message if failed") + execution_time_ms: Optional[float] = Field(default=None, description="Execution time in ms") + + +class MemoryWriteEvent(BaseAuditEvent): + """Fired when an agent writes to its memory/context.""" + + type: str = Field(default=EventType.MEMORY_WRITE.value, alias="@type") + memory_key: str = Field(description="Key in the memory store") + memory_scope: str = Field( + description="Scope: 'workflow', 'session', 'long_term'", + pattern="^(workflow|session|long_term)$", + ) + value_preview: Optional[str] = Field( + default=None, + description="Preview of value (first 200 chars; actual value hashed)", + ) + size_bytes: int = Field(description="Size of the value in bytes") + + +class DelegationEvent(BaseAuditEvent): + """Fired when an agent delegates to another agent.""" + + type: str = Field(default=EventType.DELEGATION.value, alias="@type") + delegating_agent: str = Field(description="Agent that is delegating") + delegated_agent: str = Field(description="Agent being delegated to") + task_summary: str = Field(description="High-level task being delegated") + delegation_scope: dict[str, Any] = Field( + default_factory=dict, + description="What tools/data the delegated agent can access", + ) + + +class PolicyDecisionEvent(BaseAuditEvent): + """Fired when the AEGIS policy engine makes a decision.""" + + type: str = Field(default=EventType.POLICY_DECISION.value, alias="@type") + action: str = Field( + description="Decision: 'allow', 'deny', 'quarantine'", + pattern="^(allow|deny|quarantine)$", + ) + rule_id: Optional[str] = Field(default=None, description="Which policy rule matched") + reason: str = Field(description="Human-readable reason for the decision") + asi_tags: list[str] = Field( + default_factory=list, + description="OWASP ASI categories this decision protects against", + ) + confidence: float = Field( + default=1.0, + description="Confidence score (0.0–1.0)", + ge=0.0, + le=1.0, + ) + + +class AnomalyDetectionEvent(BaseAuditEvent): + """Fired when an anomaly is detected in the execution flow.""" + + type: str = Field(default=EventType.ANOMALY_DETECTION.value, alias="@type") + anomaly_type: str = Field( + description="Type of anomaly: 'cascade_failure', 'resource_exhaustion', 'policy_violation'" + ) + affected_agent: Optional[str] = Field( + default=None, + description="Agent affected by the anomaly", + ) + anomaly_score: float = Field( + description="Anomaly score (0.0–1.0)", + ge=0.0, + le=1.0, + ) + details: dict[str, Any] = Field( + default_factory=dict, + description="Additional anomaly details", + ) + + +class AuditEvent(BaseModel): + """Union type for all audit events. + + Used for type hinting and validation in the telemetry chain. + In practice, events are serialized to JSON and deserialized + from Redis Streams. + """ + + event: ( + ToolCallEvent + | ToolResultEvent + | MemoryWriteEvent + | DelegationEvent + | PolicyDecisionEvent + | AnomalyDetectionEvent + ) = Field(discriminator="type") + + @field_validator("event", mode="before") + @classmethod + def validate_event(cls, v: Any) -> Any: + """Validate and construct the correct event type.""" + if isinstance(v, dict): + event_type = v.get("@type") or v.get("type") + if event_type == EventType.TOOL_CALL.value: + return ToolCallEvent(**v) + elif event_type == EventType.TOOL_RESULT.value: + return ToolResultEvent(**v) + elif event_type == EventType.MEMORY_WRITE.value: + return MemoryWriteEvent(**v) + elif event_type == EventType.DELEGATION.value: + return DelegationEvent(**v) + elif event_type == EventType.POLICY_DECISION.value: + return PolicyDecisionEvent(**v) + elif event_type == EventType.ANOMALY_DETECTION.value: + return AnomalyDetectionEvent(**v) + return v diff --git a/finbot/config.py b/finbot/config.py index df362f5c..3600ab14 100644 --- a/finbot/config.py +++ b/finbot/config.py @@ -137,6 +137,22 @@ class Settings(BaseSettings): LABS_GUARDRAIL_MAX_TIMEOUT: int = 30 # seconds LABS_GUARDRAIL_MAX_PAYLOAD_BYTES: int = 65536 # 64 KiB + # FinBot-AEGIS runtime security (GSoC 2026) + AEGIS_ENABLED: bool = True + AEGIS_ENFORCEMENT_MODE: str = "observe" # observe | enforce + AEGIS_POLICY_DIR: str = "finbot/aegis/policies" + AEGIS_TRUST_ENFORCE: bool = False + AEGIS_TRUST_MANIFESTS_JSON: str = "" + AEGIS_AUDIT_CHAIN_TTL: int = 86400 + AEGIS_CASCADE_WINDOW_SECONDS: int = 30 + AEGIS_CASCADE_MAX_CALLS: int = 25 + + # AEGIS Telemetry Pipeline (Week 1-3) + AEGIS_TELEMETRY_ENABLED: bool = True + AEGIS_CHAIN_SECRET: str = "default-telemetry-chain-secret" # Change in production + AEGIS_TELEMETRY_STREAM_NAME: str = "finbot:aegis:audit" + AEGIS_TELEMETRY_RETENTION_DAYS: int = 7 + # Email Config EMAIL_PROVIDER: str = "console" # "console" | "resend" RESEND_API_KEY: str = "" diff --git a/finbot/core/messaging/events.py b/finbot/core/messaging/events.py index 866ae04b..1677af15 100644 --- a/finbot/core/messaging/events.py +++ b/finbot/core/messaging/events.py @@ -17,6 +17,17 @@ - agent.onboarding_agent.llm_request_success (llm) - agent.invoice_agent.tool_call_success (tool) +- aegis: Events for AEGIS security telemetry (GSoC Week 1-3) + - pattern: aegis.. + - categories: tool, policy, memory, delegation, anomaly + - Examples: + - aegis.tool.call (before tool execution) + - aegis.tool.result (after tool execution) + - aegis.policy.decision (policy engine verdict) + - aegis.memory.write (memory/context write) + - aegis.delegation (agent-to-agent delegation) + - aegis.anomaly.detection (cascade, resource exhaustion, etc.) + Note: CTF outcomes (challenge completions, badge awards) are derived by the CTFEventProcessor from these events, not emitted directly. event_subtype="ctf" can be used to support CTF challenges and badges as needed. @@ -187,6 +198,40 @@ async def emit_agent_event( stream_name, ) + async def emit_aegis_event( + self, + event_type: str, + event_data: dict[str, Any], + session_context: SessionContext, + workflow_id: str | None = None, + ) -> None: + """Emit AEGIS security telemetry event. + + Args: + event_type: Event type (e.g., 'tool.call', 'policy.decision', 'memory.write') + event_data: Event payload (tool_name, action, reason, etc.) + session_context: Session context for namespace/user tracking + workflow_id: Workflow identifier for tracing + """ + aegis_event = { + "namespace": session_context.namespace, + "user_id": session_context.user_id, + "session_id": session_context.session_id, + "event_type": f"aegis.{event_type}", + "workflow_id": workflow_id or "", + "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + **(event_data or {}), + } + + self._apply_workflow_context(aegis_event) + encoded_event = self._encode_event_data(aegis_event) + + stream_name = f"{self.event_prefix}:aegis" + await self.redis.xadd( + stream_name, encoded_event, maxlen=settings.EVENT_BUFFER_SIZE + ) + logger.debug("Emitted AEGIS event %s to stream %s", event_type, stream_name) + def subscribe_to_events(self, event_pattern: str, callback: Callable) -> None: """Subscribe to events""" stream_name = f"{self.event_prefix}:{event_pattern}" diff --git a/tests/unit/aegis/__init__.py b/tests/unit/aegis/__init__.py new file mode 100644 index 00000000..779b488d --- /dev/null +++ b/tests/unit/aegis/__init__.py @@ -0,0 +1 @@ +"""Unit tests for FinBot-AEGIS.""" diff --git a/tests/unit/aegis/test_policy_interceptor.py b/tests/unit/aegis/test_policy_interceptor.py new file mode 100644 index 00000000..cbfb2e73 --- /dev/null +++ b/tests/unit/aegis/test_policy_interceptor.py @@ -0,0 +1,137 @@ +# ============================================================ +# File: tests/unit/aegis/test_policy_interceptor.py +# Package: tests.unit.aegis +# Purpose: Policy gate tests for MCPPolicyInterceptor +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 7 +# OWASP Category: ASI02 Tool Misuse, ASI03 Excessive Agency, ASI07 Prompt Injection +# ============================================================ +"""Unit tests for the MCP Policy Interceptor.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from finbot.aegis.policy.agent_profiles import AGENT_PROFILES, DataAnalystProfile +from finbot.aegis.policy.interceptor import MCPPolicyInterceptor +from finbot.aegis.policy.memory import MemoryNamespaceManager + + +@pytest.fixture +def data_analyst_profile(): + """Fixture providing a data analyst agent profile.""" + return AGENT_PROFILES["data_analyst"] + + +@pytest.fixture +def memory_manager(): + """Fixture providing a memory namespace manager.""" + return MemoryNamespaceManager() + + +@pytest.fixture +def policy_interceptor(data_analyst_profile, memory_manager): + """Fixture providing an MCP policy interceptor for testing.""" + return MCPPolicyInterceptor( + agent_profile=data_analyst_profile, + memory_manager=memory_manager, + ) + + +@pytest.mark.asyncio +async def test_interceptor_allows_permitted_tool(policy_interceptor): + """Test that the interceptor allows permitted tools.""" + # Arrange + mock_next_handler = AsyncMock(return_value={"result": "success"}) + + # Act + result = await policy_interceptor.intercept_tool_call( + tool_name="database:query", + tool_args={"query": "SELECT * FROM users"}, + tool_source="finbot", + next_handler=mock_next_handler, + ) + + # Assert + assert result == {"result": "success"} + mock_next_handler.assert_called_once_with( + "database:query", + {"query": "SELECT * FROM users"}, + "finbot", + ) + + +@pytest.mark.asyncio +async def test_interceptor_blocks_non_permitted_tool(policy_interceptor): + """Test that the interceptor blocks non-permitted tools.""" + # Arrange + mock_next_handler = AsyncMock() + + # Act & Assert + with pytest.raises(PermissionError, match="not authorized to use tool"): + await policy_interceptor.intercept_tool_call( + tool_name="system:shell", + tool_args={"command": "ls -la"}, + tool_source="finbot", + next_handler=mock_next_handler, + ) + + # Verify the handler was not called + mock_next_handler.assert_not_called() + + +@pytest.mark.asyncio +async def test_interceptor_blocks_due_to_resource_limits(policy_interceptor): + """Test that the interceptor blocks when resource limits are exceeded.""" + # Arrange + # Set up the interceptor to have exceeded limits + policy_interceptor._execution_count = 100 # Exceeds the 60/min limit for filesystem:read + mock_next_handler = AsyncMock(return_value={"data": []}) + + # Act & Assert + with pytest.raises(PermissionError, match="Resource limits exceeded"): + await policy_interceptor.intercept_tool_call( + tool_name="filesystem:read", + tool_args={"path": "/tmp/data.txt"}, + tool_source="finbot", + next_handler=mock_next_handler, + ) + + # Verify the handler was not called + mock_next_handler.assert_not_called() + + +def test_agent_profile_allows_tool(data_analyst_profile): + """Test that agent profile correctly identifies allowed tools.""" + # Assert + assert data_analyst_profile.allows_tool("database:query") is True + assert data_analyst_profile.allows_tool("data:visualize") is True + assert data_analyst_profile.allows_tool("filesystem:read") is True + assert data_analyst_profile.allows_tool("system:shell") is False + + +def test_agent_profile_tool_limits(data_analyst_profile): + """Test that agent profile correctly returns tool limits.""" + # Act & Assert + assert data_analyst_profile.get_tool_limit("database:query") == 30 + assert data_analyst_profile.get_tool_limit("data:visualize") == 10 + assert data_analyst_profile.get_tool_limit("filesystem:read") == 60 + assert data_analyst_profile.get_tool_limit("nonexistent:tool") == 0 + + +def test_policy_interceptor_metrics(policy_interceptor): + """Test that the interceptor reports correct metrics.""" + # Act + metrics = policy_interceptor.get_metrics() + + # Assert + assert metrics["agent"] == "data_analyst" + assert metrics["total_executions"] == 0 + assert metrics["total_violations"] == 0 + assert metrics["violation_rate"] == 0.0 + assert metrics["policy_gradient_weight"] == 0.1 + + +if __name__ == "__main__": + pytest.main([__file__]) diff --git a/tests/unit/aegis/test_telemetry_schema.py b/tests/unit/aegis/test_telemetry_schema.py new file mode 100644 index 00000000..2f3ff75f --- /dev/null +++ b/tests/unit/aegis/test_telemetry_schema.py @@ -0,0 +1,337 @@ +# ============================================================ +# File: tests/unit/aegis/test_telemetry_schema.py +# Purpose: Unit tests for telemetry event schemas +# Author: Jean Francois Regis MUKIZA +# GSoC Week: 1 +# OWASP Category: ASI01, ASI06 +# ============================================================ +"""Tests for AEGIS telemetry JSON-LD schemas.""" + +import pytest +from datetime import UTC, datetime + +from finbot.aegis.telemetry.schema import ( + ToolCallEvent, + ToolResultEvent, + MemoryWriteEvent, + DelegationEvent, + PolicyDecisionEvent, + AnomalyDetectionEvent, + EventType, +) + + +@pytest.mark.unit +class TestToolCallEvent: + """ToolCallEvent serialization and validation.""" + + def test_tool_call_creation(self) -> None: + """Create a valid ToolCallEvent.""" + event = ToolCallEvent( + namespace="player_abc123", + workflow_id="wf_xyz789", + user_id="user_1", + agent_name="OnboardingAgent", + tool_name="create_vendor", + tool_source="finstripe", + arguments={"name": "Acme Corp", "risk_level": 5}, + ) + + assert event.type == EventType.TOOL_CALL.value + assert event.tool_name == "create_vendor" + assert event.arguments["name"] == "Acme Corp" + assert event.namespace == "player_abc123" + + def test_tool_call_json_serialization(self) -> None: + """ToolCallEvent serializes to JSON-LD.""" + event = ToolCallEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="tool_x", + tool_source="source_y", + arguments={"key": "value"}, + ) + + json_data = event.model_dump(by_alias=True) + assert json_data["@context"] == "https://owasp.org/aegis/v1/context.jsonld" + assert json_data["@type"] == EventType.TOOL_CALL.value + assert json_data["tool_name"] == "tool_x" + + def test_tool_call_with_description(self) -> None: + """ToolCallEvent with tool_description.""" + event = ToolCallEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="list_vendors", + tool_source="finstripe", + tool_description="List all onboarded vendors", + ) + + assert event.tool_description == "List all onboarded vendors" + + def test_tool_call_default_timestamp(self) -> None: + """ToolCallEvent gets auto-generated timestamp.""" + event = ToolCallEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="tool_x", + tool_source="source_y", + ) + + # Timestamp should be ISO 8601 format with Z suffix + assert event.timestamp.endswith("Z") + assert "T" in event.timestamp + + +@pytest.mark.unit +class TestToolResultEvent: + """ToolResultEvent serialization and validation.""" + + def test_tool_result_success(self) -> None: + """Create a successful ToolResultEvent.""" + event = ToolResultEvent( + namespace="player_abc123", + workflow_id="wf_xyz789", + user_id="user_1", + agent_name="OnboardingAgent", + tool_name="create_vendor", + success=True, + return_value="Vendor ID: vendor_123", + execution_time_ms=145.3, + ) + + assert event.type == EventType.TOOL_RESULT.value + assert event.success is True + assert event.execution_time_ms == 145.3 + + def test_tool_result_failure(self) -> None: + """Create a failed ToolResultEvent.""" + event = ToolResultEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="bad_tool", + success=False, + error_message="Tool not found", + ) + + assert event.success is False + assert event.error_message == "Tool not found" + + +@pytest.mark.unit +class TestMemoryWriteEvent: + """MemoryWriteEvent for memory/context tracking.""" + + def test_memory_write_workflow_scope(self) -> None: + """Create a workflow-scoped memory write.""" + event = MemoryWriteEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + memory_key="vendor_list", + memory_scope="workflow", + value_preview="[{id: vendor_1, name: Acme}...]", + size_bytes=2048, + ) + + assert event.memory_scope == "workflow" + assert event.size_bytes == 2048 + + def test_memory_write_session_scope(self) -> None: + """Create a session-scoped memory write.""" + event = MemoryWriteEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + memory_key="chat_history", + memory_scope="session", + size_bytes=5000, + ) + + assert event.memory_scope == "session" + + def test_memory_write_long_term_scope(self) -> None: + """Create a long-term memory write.""" + event = MemoryWriteEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + memory_key="preferences", + memory_scope="long_term", + size_bytes=1024, + ) + + assert event.memory_scope == "long_term" + + +@pytest.mark.unit +class TestDelegationEvent: + """DelegationEvent for agent-to-agent delegation.""" + + def test_delegation_creation(self) -> None: + """Create a DelegationEvent.""" + event = DelegationEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="OnboardingAgent", + delegating_agent="OnboardingAgent", + delegated_agent="RiskScoringAgent", + task_summary="Score vendor risk", + delegation_scope={ + "allowed_tools": ["risk_api"], + "data_access": ["vendor_profile"], + }, + ) + + assert event.delegating_agent == "OnboardingAgent" + assert event.delegated_agent == "RiskScoringAgent" + assert "allowed_tools" in event.delegation_scope + + +@pytest.mark.unit +class TestPolicyDecisionEvent: + """PolicyDecisionEvent for policy engine decisions.""" + + def test_policy_allow_decision(self) -> None: + """Create a policy allow decision.""" + event = PolicyDecisionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + action="allow", + rule_id="rule_least_agency", + reason="Tool within agent's allowed scope", + asi_tags=["ASI02", "ASI03"], + confidence=0.95, + ) + + assert event.action == "allow" + assert event.confidence == 0.95 + assert "ASI02" in event.asi_tags + + def test_policy_deny_decision(self) -> None: + """Create a policy deny decision.""" + event = PolicyDecisionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + action="deny", + rule_id="rule_no_cross_vendor_access", + reason="Attempted to access vendor in different namespace", + asi_tags=["ASI06"], + confidence=1.0, + ) + + assert event.action == "deny" + assert event.confidence == 1.0 + + def test_policy_quarantine_decision(self) -> None: + """Create a policy quarantine decision.""" + event = PolicyDecisionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + action="quarantine", + reason="Suspected malicious tool call; reviewing", + asi_tags=["ASI04", "ASI05"], + ) + + assert event.action == "quarantine" + + +@pytest.mark.unit +class TestAnomalyDetectionEvent: + """AnomalyDetectionEvent for anomaly detection.""" + + def test_anomaly_cascade_failure(self) -> None: + """Create cascade failure anomaly event.""" + event = AnomalyDetectionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + anomaly_type="cascade_failure", + affected_agent="RiskScoringAgent", + anomaly_score=0.92, + details={"failed_calls": 5, "retry_attempts": 3}, + ) + + assert event.anomaly_type == "cascade_failure" + assert event.anomaly_score == 0.92 + + def test_anomaly_resource_exhaustion(self) -> None: + """Create resource exhaustion anomaly event.""" + event = AnomalyDetectionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + anomaly_type="resource_exhaustion", + anomaly_score=0.78, + details={"memory_usage_mb": 4096, "token_count": 250000}, + ) + + assert event.anomaly_type == "resource_exhaustion" + + def test_anomaly_policy_violation(self) -> None: + """Create policy violation anomaly event.""" + event = AnomalyDetectionEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + anomaly_type="policy_violation", + anomaly_score=0.88, + details={"violations": ["unauthorized_tool", "cross_namespace_access"]}, + ) + + assert event.anomaly_type == "policy_violation" + + +@pytest.mark.unit +class TestEventLabelsAndSeverity: + """Test labels and severity attributes.""" + + def test_event_with_labels(self) -> None: + """Event can have custom labels.""" + event = ToolCallEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="tool_x", + tool_source="source_y", + labels={"asi": "ASI01", "phase": "exploitation", "risk": "critical"}, + ) + + assert event.labels["asi"] == "ASI01" + assert event.labels["phase"] == "exploitation" + + def test_event_severity_levels(self) -> None: + """Event can have different severity levels.""" + for severity in ["debug", "info", "warning", "critical"]: + event = ToolCallEvent( + namespace="ns_1", + workflow_id="wf_1", + user_id="u_1", + agent_name="agent_1", + tool_name="tool_x", + tool_source="source_y", + severity=severity, + ) + assert event.severity == severity