Add initial SOAR framework skeleton

README.md

@@ -14,3 +14,14 @@ All scripts are located in the `scripts/` directory.
   outputs.
 
 See [scripts/README.md](scripts/README.md) for more details on each script.
+
+## SOAR Framework
+
+The `soar_framework` package provides a minimal incident response
+automation framework. It demonstrates dynamic playbook execution with
+stub integrations for SIEM, EDR and cloud security APIs. An Ollama-based
+LLM is used to generate forensic documentation such as chain of custody
+records and incident reports.
+
+Run `python -m soar_framework.main` to execute the sample playbook.
+

soar_framework/__init__.py

@@ -0,0 +1,8 @@
+"""Incident Response Automation Framework.
+
+This package provides a simplified Security Orchestration, Automation,
+and Response (SOAR) platform with dynamic playbook execution. It also
+integrates with an Ollama large language model to produce forensically
+sound documentation.
+"""
+

soar_framework/documentation.py

@@ -0,0 +1,27 @@
+"""Automated forensic documentation utilities."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Dict
+
+from .llm import generate_documentation
+
+
+def chain_of_custody_record(evidence: Dict[str, str]) -> str:
+    """Return a chain of custody document for the provided evidence."""
+    prompt = (
+        "Generate a chain of custody record using NIST SP 800-61, ISO 27035, "
+        "and RFC 3227 guidelines. Include timestamps, handler IDs, storage "
+        "locations and digital signature notes. Data: " + str(evidence)
+    )
+    return generate_documentation(prompt)
+
+
+def incident_report(summary: str, findings: str) -> str:
+    """Generate a high-level incident report."""
+    prompt = (
+        "Create an executive summary followed by technical details. "
+        f"Summary: {summary}\nFindings: {findings}"
+    )
+    return generate_documentation(prompt)

soar_framework/evidence.py

@@ -0,0 +1,47 @@
+"""Evidence collection and preservation utilities."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Dict, List
+
+
+def blockchain_hash(data: bytes) -> str:
+    """Return a simple blockchain-style hash for the provided data."""
+    return hashlib.sha256(data).hexdigest()
+
+
+@dataclass
+class EvidenceItem:
+    """Represents a collected piece of evidence."""
+
+    identifier: str
+    path: str
+    timestamp: str
+    handler: str
+    integrity_hash: str
+
+
+class EvidenceLedger:
+    """Maintains an immutable audit trail using hash chaining."""
+
+    def __init__(self) -> None:
+        self.records: List[Dict[str, str]] = []
+
+    def add_record(self, item: EvidenceItem) -> None:
+        previous_hash = self.records[-1]["ledger_hash"] if self.records else ""
+        data = json.dumps(
+            {
+                "id": item.identifier,
+                "path": item.path,
+                "time": item.timestamp,
+                "handler": item.handler,
+                "hash": item.integrity_hash,
+                "prev": previous_hash,
+            },
+            sort_keys=True,
+        ).encode()
+        ledger_hash = blockchain_hash(data)
+        self.records.append({"ledger_hash": ledger_hash, **json.loads(data)})

soar_framework/integrations/cloud.py

@@ -0,0 +1,26 @@
+"""Cloud provider security integration stubs."""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+import requests
+
+
+class CloudSecurityClient:
+    """Example client for a cloud service provider security API."""
+
+    def __init__(self, base_url: str, api_key: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+
+    def quarantine_resource(self, resource_id: str) -> None:
+        url = f"{self.base_url}/resources/{resource_id}/quarantine"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        requests.post(url, headers=headers, timeout=10)
+
+    def fetch_events(self) -> Dict[str, Any]:
+        url = f"{self.base_url}/events"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        resp = requests.get(url, headers=headers, timeout=10)
+        resp.raise_for_status()
+        return resp.json()

soar_framework/integrations/edr.py

@@ -0,0 +1,26 @@
+"""Endpoint Detection and Response integration stubs."""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+import requests
+
+
+class EdrClient:
+    """Example client for an EDR platform."""
+
+    def __init__(self, base_url: str, api_key: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+
+    def isolate_host(self, host_id: str) -> None:
+        url = f"{self.base_url}/hosts/{host_id}/isolate"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        requests.post(url, headers=headers, timeout=10)
+
+    def fetch_incidents(self) -> Dict[str, Any]:
+        url = f"{self.base_url}/incidents"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        resp = requests.get(url, headers=headers, timeout=10)
+        resp.raise_for_status()
+        return resp.json()

soar_framework/integrations/siem.py

@@ -0,0 +1,26 @@
+"""SIEM integration stubs."""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+import requests
+
+
+class SiemClient:
+    """Example client for a SIEM platform."""
+
+    def __init__(self, base_url: str, api_key: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+
+    def send_event(self, event: Dict[str, Any]) -> None:
+        url = f"{self.base_url}/events"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        requests.post(url, json=event, headers=headers, timeout=10)
+
+    def fetch_alerts(self) -> Dict[str, Any]:
+        url = f"{self.base_url}/alerts"
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        resp = requests.get(url, headers=headers, timeout=10)
+        resp.raise_for_status()
+        return resp.json()

soar_framework/llm.py

@@ -0,0 +1,35 @@
+"""Ollama integration utilities.
+
+This module provides helper functions for interacting with an Ollama
+server that hosts incident-response specific models. The LLM is used to
+produce forensically sound reports that meet legal admissibility
+requirements.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Dict, Any
+import requests
+
+
+class OllamaClient:
+    """Simple wrapper around the Ollama REST API."""
+
+    def __init__(self, base_url: str = "http://localhost:11434") -> None:
+        self.base_url = base_url.rstrip("/")
+
+    def generate(self, prompt: str, model: str = "incident-model") -> str:
+        """Generate text using the given model."""
+        url = f"{self.base_url}/api/generate"
+        payload = {"model": model, "prompt": prompt}
+        response = requests.post(url, json=payload, timeout=30)
+        response.raise_for_status()
+        data = response.json()
+        return data.get("response", "")
+
+
+def generate_documentation(prompt: str) -> str:
+    """Generate documentation text for the given prompt."""
+    client = OllamaClient()
+    return client.generate(prompt)

soar_framework/main.py

@@ -0,0 +1,46 @@
+"""Example entry point for the SOAR framework."""
+
+from __future__ import annotations
+
+from .playbook import Playbook
+from .documentation import incident_report
+from .integrations.siem import SiemClient
+from .integrations.edr import EdrClient
+from .integrations.cloud import CloudSecurityClient
+
+
+def load_default_actions(pb: Playbook, siem: SiemClient, edr: EdrClient, cloud: CloudSecurityClient) -> None:
+    def fetch_alerts(_: dict) -> None:
+        alerts = siem.fetch_alerts()
+        print("Fetched alerts", alerts)
+
+    def isolate(params: dict) -> None:
+        host = params.get("host")
+        edr.isolate_host(host)
+        print(f"Host {host} isolated")
+
+    def quarantine(params: dict) -> None:
+        resource = params.get("resource")
+        cloud.quarantine_resource(resource)
+        print(f"Resource {resource} quarantined")
+
+    pb.register_action("fetch_alerts", fetch_alerts)
+    pb.register_action("isolate_host", isolate)
+    pb.register_action("quarantine_resource", quarantine)
+
+
+def run_playbook(path: str) -> None:
+    siem = SiemClient("https://siem.example.com/api", "token")
+    edr = EdrClient("https://edr.example.com/api", "token")
+    cloud = CloudSecurityClient("https://cloud.example.com/api", "token")
+
+    pb = Playbook(path)
+    load_default_actions(pb, siem, edr, cloud)
+    pb.run()
+
+    report = incident_report("Sample incident", "No findings yet")
+    print(report)
+
+
+if __name__ == "__main__":
+    run_playbook("playbooks/sample_playbook.yaml")

soar_framework/playbook.py

@@ -0,0 +1,28 @@
+"""Dynamic playbook execution engine."""
+
+from __future__ import annotations
+
+import yaml
+from typing import Callable, Dict, Any
+
+
+class Playbook:
+    """Loads and executes playbooks defined in YAML."""
+
+    def __init__(self, path: str) -> None:
+        with open(path, "r", encoding="utf-8") as f:
+            self.definition = yaml.safe_load(f)
+        self.actions: Dict[str, Callable[[Dict[str, Any]], None]] = {}
+
+    def register_action(self, name: str, func: Callable[[Dict[str, Any]], None]) -> None:
+        self.actions[name] = func
+
+    def run(self) -> None:
+        for step in self.definition.get("steps", []):
+            action = step.get("action")
+            params = step.get("params", {})
+            handler = self.actions.get(action)
+            if handler:
+                handler(params)
+            else:
+                raise ValueError(f"Unknown action: {action}")

soar_framework/playbooks/sample_playbook.yaml

@@ -0,0 +1,8 @@
+steps:
+  - action: fetch_alerts
+  - action: isolate_host
+    params:
+      host: "host-123"
+  - action: quarantine_resource
+    params:
+      resource: "vm-456"