diff --git a/README.md b/README.md index 48f235a..57739c0 100644 --- a/README.md +++ b/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. + diff --git a/soar_framework/__init__.py b/soar_framework/__init__.py new file mode 100644 index 0000000..6705b44 --- /dev/null +++ b/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. +""" + diff --git a/soar_framework/documentation.py b/soar_framework/documentation.py new file mode 100644 index 0000000..28c9ca2 --- /dev/null +++ b/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) diff --git a/soar_framework/evidence.py b/soar_framework/evidence.py new file mode 100644 index 0000000..12783b3 --- /dev/null +++ b/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)}) diff --git a/soar_framework/integrations/__init__.py b/soar_framework/integrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/soar_framework/integrations/cloud.py b/soar_framework/integrations/cloud.py new file mode 100644 index 0000000..d7e395a --- /dev/null +++ b/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() diff --git a/soar_framework/integrations/edr.py b/soar_framework/integrations/edr.py new file mode 100644 index 0000000..6e6255b --- /dev/null +++ b/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() diff --git a/soar_framework/integrations/siem.py b/soar_framework/integrations/siem.py new file mode 100644 index 0000000..880ef27 --- /dev/null +++ b/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() diff --git a/soar_framework/llm.py b/soar_framework/llm.py new file mode 100644 index 0000000..f7f3ed5 --- /dev/null +++ b/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) diff --git a/soar_framework/main.py b/soar_framework/main.py new file mode 100644 index 0000000..34ee5d0 --- /dev/null +++ b/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") diff --git a/soar_framework/playbook.py b/soar_framework/playbook.py new file mode 100644 index 0000000..75a02b8 --- /dev/null +++ b/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}") diff --git a/soar_framework/playbooks/sample_playbook.yaml b/soar_framework/playbooks/sample_playbook.yaml new file mode 100644 index 0000000..0db46d4 --- /dev/null +++ b/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"