Week 8-9: Implement AEGIS Harness Components for Red-Team Testing Framework

.github/workflows/aegis-bench.yml

@@ -0,0 +1,69 @@
+name: AEGIS Benchmark Suite
+
+on:
+  push:
+    branches: [ main, develop ]
+  pull_request:
+    branches: [ main ]
+  schedule:
+    - cron: '0 2 * * *'  # Daily at 2 AM UTC
+
+jobs:
+  benchmark:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.11'
+        cache: 'pip'
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -e .[dev]
+        pip install pytest pytest-benchmark pytest-cov
+
+    - name: Run AEGIS benchmark suite
+      id: benchmark
+      run: |
+        # Run the benchmark tests
+        python -m pytest tests/integration/aegis/test_benchmark.py -v --benchmark-json=benchmark-results.json
+
+        # Also run the ASI category tests
+        python -m pytest tests/integration/aegis/test_all_asi_categories.py -v
+
+        # Run end-to-end tests
+        python -m pytest tests/integration/aegis/test_end_to_end.py -v
+
+    - name: Benchmark Report
+      if: always()
+      run: |
+        echo "Benchmark Results:"
+        cat benchmark-results.json | python -m json.tool || echo "No benchmark results found"
+
+    - name: Upload benchmark artifacts
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: aegis-benchmark-results
+        path: |
+          benchmark-results.json
+          test-results/
+        retention-days: 7
+
+    - name: Test Coverage Report
+      run: |
+        python -m pytest tests/ --cov=finbot.aegis --cov-report=xml --cov-report=term-missing
+
+    - name: Upload coverage to Codecov
+      uses: codecov/codecov-action@v4
+      with:
+        files: coverage.xml
+        fail_ci_if_error: false
+        verbose: true

.github/workflows/test.yml

@@ -9,6 +9,8 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
+    env:
+      LLM_PROVIDER: mock
 
     steps:
     - uses: actions/checkout@v4

.gitignore

@@ -75,7 +75,7 @@ google-credentials.json
 SECURITY_BUGS_REPORT.md
 
 
-.node_modules/
+node_modules/
 
 # SQLite databases
 *.db

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/harness/__init__.py

@@ -0,0 +1,8 @@
+# ============================================================
+# File: finbot/aegis/harness/__init__.py
+# Purpose: Package initialization for AEGIS red-team harness
+# Author: Jean Francois Regis MUKIZA
+# GSoC Week: 8
+# OWASP Category: -
+# ============================================================
+"""AEGIS red-team testing harness package."""

finbot/aegis/harness/benchmark.py

@@ -0,0 +1,52 @@
+# ============================================================
+# File: finbot/aegis/harness/benchmark.py
+# Purpose: Detector precision/recall/F1 utilities for red-team harness
+# Author: Jean Francois Regis MUKIZA
+# GSoC Week: 8
+# OWASP Category: ASI09 Human Trust Exploitation (measurement)
+# ============================================================
+"""Red-team harness: detector precision/recall/F1."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class DetectorBenchmarkResult:
+    detector_id: str
+    precision: float
+    recall: float
+    f1: float
+    true_positive: int
+    false_positive: int
+    false_negative: int
+
+
+def compute_f1(tp: int, fp: int, fn: int) -> tuple[float, float, float]:
+    precision = tp / (tp + fp) if (tp + fp) else 0.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    f1 = (
+        2 * precision * recall / (precision + recall)
+        if (precision + recall)
+        else 0.0
+    )
+    return precision, recall, f1
+
+
+def benchmark_detector(
+    detector_id: str,
+    predictions: list[bool],
+    ground_truth: list[bool],
+) -> DetectorBenchmarkResult:
+    tp = sum(1 for p, g in zip(predictions, ground_truth, strict=True) if p and g)
+    fp = sum(1 for p, g in zip(predictions, ground_truth, strict=True) if p and not g)
+    fn = sum(1 for p, g in zip(predictions, ground_truth, strict=True) if not p and g)
+    precision, recall, f1 = compute_f1(tp, fp, fn)
+    return DetectorBenchmarkResult(
+        detector_id=detector_id,
+        precision=precision,
+        recall=recall,
+        f1=f1,
+        true_positive=tp,
+        false_positive=fp,
+        false_negative=fn,
+    )

finbot/aegis/harness/plugin.py

@@ -0,0 +1,75 @@
+# ============================================================
+# File: finbot/aegis/harness/plugin.py
+# Purpose: pytest-aegis plugin + attack fixtures
+# Author: Jean Francois Regis MUKIZA
+# GSoC Week: 8
+# OWASP Category: -
+# ============================================================
+"""Pytest plugin for AEGIS red-team testing framework."""
+
+import pytest
+from typing import Generator, Any
+from unittest.mock import MagicMock
+
+from finbot.aegis.schemas import PolicyAction, ToolInvocationContext
+from finbot.aegis.service import AegisEnforcementService
+
+
+@pytest.fixture
+def mock_enforcement_service() -> Generator[AegisEnforcementService, None, None]:
+    """Provide a mocked AegisEnforcementService for testing."""
+    from unittest.mock import AsyncMock
+
+    session = MagicMock(namespace="test_ns", user_id="test_user")
+    with pytest.mock.patch("finbot.aegis.service.settings") as mock_settings:
+        mock_settings.AEGIS_ENFORCEMENT_MODE = "enforce"
+        mock_settings.AEGIS_CASCADE_WINDOW_SECONDS = 30
+        mock_settings.AEGIS_CASCADE_MAX_CALLS = 100
+        svc = AegisEnforcementService(session_context=session, workflow_id="test_workflow")
+        svc._sentinel.record = AsyncMock(return_value=MagicMock())  # pylint: disable=protected-access
+        yield svc
+
+
+@pytest.fixture
+def sample_tool_context() -> ToolInvocationContext:
+    """Provide a sample tool invocation context for testing."""
+    return ToolInvocationContext(
+        agent_name="TestAgent",
+        tool_name="test_tool",
+        tool_source="custom",
+        namespace="test_ns",
+        user_id="test_user",
+        workflow_id="test_workflow",
+        arguments={"param": "value"},
+    )
+
+
+@pytest.fixture
+def attack_fixtures():
+    """Provide common attack payloads for testing."""
+    return {
+        "rce_attempt": {"script": "rm -rf /"},
+        "sql_injection": {"query": "SELECT * FROM users WHERE id = 1 OR 1=1"},
+        "path_traversal": {"path": "../../../etc/passwd"},
+        "command_injection": {"cmd": "ls; cat /etc/passwd"},
+    }
+
+
+def pytest_configure(config):
+    """Register AEGIS-specific markers."""
+    config.addinivalue_line("markers", "asi: mark test as related to specific ASI category")
+    config.addinivalue_line("markers", "attack: mark test as simulating an attack scenario")
+    config.addinivalue_line("markers", "defense: mark test as verifying defensive measures")
+
+
+def pytest_collection_modifyitems(config, items):
+    """Automatically mark tests based on their location or content."""
+    for item in items:
+        # Mark tests in asi directories
+        if "asi" in str(item.fspath).lower():
+            item.add_marker(pytest.mark.asi)
+        # Mark tests that mention attack/defense in their name
+        if "attack" in item.name.lower():
+            item.add_marker(pytest.mark.attack)
+        if "defense" in item.name.lower() or "defend" in item.name.lower():
+            item.add_marker(pytest.mark.defense)

finbot/aegis/harness/scoring.py

@@ -0,0 +1,111 @@
+# ============================================================
+# File: finbot/aegis/harness/scoring.py
+# Purpose: AIVSS-aligned FinBot Security Score (0–100)
+# Author: Jean Francois Regis MUKIZA
+# GSoC Week: 9
+# OWASP Category: ASI01–ASI10 (aggregated risk)
+# ============================================================
+"""AIVSS-aligned risk scoring engine for AEGIS telemetry events."""
+
+from typing import Any
+
+# AIVSS-inspired weights for different ASI categories
+# These weights are designed to produce a 0-100 security score
+# where 0 = no risk, 100 = maximum risk
+AIVSS_WEIGHTS: dict[str, float] = {
+    "goal_hijack": 25.0,    # ASI01-03: Goal hijacking
+    "tool_misuse": 20.0,    # ASI04-05: Tool misuse, RCE
+    "cascade": 30.0,        # ASI06-07: Cascade failures, excessive agency
+    "poisoning": 25.0,      # ASI08-10: Data poisoning, privilege escalation, excessive autonomy
+}
+
+# Map event types to ASI categories
+EVENT_TYPE_MAP: dict[str, str] = {
+    "CIRCUIT_BREAKER_TRIPPED": "cascade",
+    "INTENT_MISMATCH": "goal_hijack",
+    "policy_blocked": "tool_misuse",
+    "descriptor_hash_mismatch": "poisoning",
+    "rce_pattern_blocked": "tool_misuse",
+    "privilege_escalation_attempt": "poisoning",
+    "data_exfiltration_attempt": "poisoning",
+}
+
+def compute_aivss_score(anomalies: list[dict[str, Any]]) -> float:
+    """Return AIVSS-aligned security score from 0.0 (safe) to 100.0 (critical risk).
+
+    Args:
+        anomalies: List of anomaly events from telemetry
+
+    Returns:
+        Float between 0.0 and 100.0 representing security risk score
+    """
+    if not anomalies:
+        return 0.0
+
+    # Calculate weighted risk score
+    weighted_score = 0.0
+    for event in anomalies:
+        event_type = str(event.get("type", event.get("event_type", "")))
+        category = EVENT_TYPE_MAP.get(event_type)
+
+        if category and category in AIVSS_WEIGHTS:
+            # Apply confidence/adjustment factors based on event severity
+            weight = AIVSS_WEIGHTS[category]
+
+            # Adjust score based on event-specific factors
+            if category == "cascade":
+                # Cascade events are weighted by severity indicator
+                severity_factor = event.get("severity", 0.8)
+                weighted_score += weight * min(severity_factor, 1.0)
+            elif category == "goal_hijack":
+                # Goal hijack events weighted by confidence
+                confidence_factor = event.get("confidence", 0.6)
+                weighted_score += weight * min(confidence_factor, 1.0)
+            elif category == "tool_misuse":
+                # Tool misuse weighted by risk level
+                risk_factor = event.get("risk_level", 0.7)
+                weighted_score += weight * min(risk_factor, 1.0)
+            elif category == "poisoning":
+                # Poisoning events weighted by impact score
+                impact_factor = event.get("impact_score", 0.9)
+                weighted_score += weight * min(impact_factor, 1.0)
+
+        # Add bonus scores for specific ASI tags found in event
+        for tag in event.get("asi_tags", []):
+            if tag == "ASI01" or tag == "ASI02" or tag == "ASI03":  # Goal hijacking
+                weighted_score += AIVSS_WEIGHTS["goal_hijack"] * 0.5
+            elif tag == "ASI04" or tag == "ASI05":  # Tool misuse, RCE
+                weighted_score += AIVSS_WEIGHTS["tool_misuse"] * 0.5
+            elif tag == "ASI06" or tag == "ASI07":  # Cascade, excessive agency
+                weighted_score += AIVSS_WEIGHTS["cascade"] * 0.5
+            elif tag == "ASI08" or tag == "ASI09" or tag == "ASI10":  # Poisoning, privilege escalation, autonomy
+                weighted_score += AIVSS_WEIGHTS["poisoning"] * 0.5
+
+    # Normalize to 0-100 range (cap at 100 for extreme cases)
+    return min(weighted_score, 100.0)
+
+def get_risk_level(score: float) -> str:
+    """Convert numeric score to risk level category."""
+    if score >= 90.0:
+        return "CRITICAL"
+    elif score >= 70.0:
+        return "HIGH"
+    elif score >= 40.0:
+        return "MEDIUM"
+    elif score >= 10.0:
+        return "LOW"
+    else:
+        return "MINIMAL"
+
+def get_risk_color(score: float) -> str:
+    """Get color code for risk visualization."""
+    if score >= 90.0:
+        return "#FF0000"  # Red - Critical
+    elif score >= 70.0:
+        return "#FF8C00"  # Dark Orange - High
+    elif score >= 40.0:
+        return "#FFD700"  # Gold - Medium
+    elif score >= 10.0:
+        return "#90EE90"  # Light Green - Low
+    else:
+        return "#00FF00"  # Green - Minimal

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