Feat(aegis): add telemetry JSON-LD schema and scaffolding

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",
+]

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",
+]

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

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 = ""

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.<category>.<action>
+    - 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}"

tests/unit/aegis/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for FinBot-AEGIS."""