Autonomous Multi-Agent Digital Forensics Pipeline powered by LangGraph: hypothesis planning, Text-to-SQL forensic queries with AST anti-SQLi guardrails, k-way timeline alibi validation, timestomping detection, Human-in-the-Loop approval gates and ISO/IEC 27037 HMAC-SHA256 signed expert reports.
The agent operates as a cyclical StateGraph backed by checkpointer memory (InMemorySaver / SqliteSaver) and strict recursion boundaries.
graph TD
START([START]) --> planner["1. HypothesisPlanner\n(Generates investigative hypotheses)"]
planner --> sql["2. SQLInvestigator\n(AST Guardrail + Parameterized SELECT)"]
sql --> timeline["3. TimelineValidator\n(k-way alibi + timestomping check)"]
timeline --> checker{"4. ConsistencyChecker\n(Anti-DoS check + verdict gate)"}
checker -- "Needs more evidence (iterations < 5)" --> sql
checker -- "Evidence conclusive / Awaiting HITL" --> report["5. ReportGenerator\n(ISO/IEC 27037 + HMAC-SHA256)"]
report --> END([END])
classDef node fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
classDef gate fill:#0f172a,stroke:#f59e0b,stroke-width:2px,color:#f8fafc;
class planner,sql,timeline,report node;
class checker gate;
[START]
│
▼
┌───────────────────────┐
│ 1. HypothesisPlanner │ ──► Parses case & suspects into structured hypotheses
└──────────┬────────────┘
│
▼
┌───────────────────────┐
│ 2. SQLInvestigator │ ◄──────────────────────────────────┐ (Loop if inconclusive)
└──────────┬────────────┘ │
│ │
▼ │
┌───────────────────────┐ │
│ 3. TimelineValidator │ ──► Validates alibis & timestomping │
└──────────┬────────────┘ │
│ │
▼ │
┌───────────────────────┐ │
│ 4. ConsistencyChecker │ ──[ Incomplete & Iterations < 5 ]──┘
└──────────┬────────────┘
│ [ Conclusive OR Awaiting HITL Approval ]
▼
┌───────────────────────┐
│ 5. ReportGenerator │ ──► ISO/IEC 27037 Signed Expert Report (HMAC-SHA256)
└──────────┬────────────┘
│
▼
[END]
| Node | Responsibility | Inputs / State Mutations | Guardrails & Security |
|---|---|---|---|
HypothesisPlanner |
Extracts suspect list, analyzes case narrative, formulates candidate hypotheses |
case_number, case_description, suspects hypotheses
|
Input sanitization, max hypothesis cap (anti-DoS) |
SQLInvestigator |
Formulates read-only Text-to-SQL forensic queries across incident databases |
hypotheses evidences, status_messages
|
AST SQL parser whitelist, blocks DROP/DELETE/UNION/--, parameterized SELECT only |
TimelineValidator |
Cross-examines digital timeline events against suspect alibis; flags anomalous gaps |
evidences, alibis |
Timestomping heuristic detector, deterministic k-way interval validation |
ConsistencyChecker |
Evaluates overall hypothesis convergence and enforces iteration limits |
iterations, hypotheses is_complete, awaiting_human_approval
|
Hard loop bound (max_iterations=5), recursion limit ( |
ReportGenerator |
Generates tamper-evident forensic report conforming to ISO/IEC 27037 |
hypotheses, evidences expert_report
|
Secret HMAC-SHA256 signature via constant-time hmac.compare_digest()
|
- #9 Cryptographic Hygiene: HMAC-SHA256 verification on all forensic findings with constant-time equality validation.
- #14 Anti-SSRF CWE-918: Strictly local database queries; no outbound HTTP requests during query execution.
-
#15 AST Anti-SQLi: Strict regex & AST token analysis ensuring no DDL/DML injection (
UNION,DROP,INSERT, comments blocked). -
#16 Human-in-the-Loop (OWASP LLM06): LangGraph
interrupt_before=["report"]prevents unilateral autonomous expert dictamens. -
#17 Anti-DoS (OWASP LLM10): Bounded cyclic recursion limit (
$=10$ ), maximum 5 iterations per case, and memory-managed checkpointers.
docker compose up --build# Install editable with dev dependencies
pip install -e ".[dev]"
# Run forensic investigation on a case
forensic-investigator CASE-001 "Unauthorized database access and credential dump" "alice,bob"from investigator.graph import compile_graph
from investigator.state import ForensicState
app = compile_graph(":memory:")
config = {"configurable": {"thread_id": "CASE-2026-X"}, "recursion_limit": 10}
state: ForensicState = {
"case_number": "CASE-2026-X",
"case_description": "Data exfiltration from financial DB",
"suspects": ["alice", "bob"],
"max_iterations": 5,
"iterations": 0,
}
for step in app.stream(state, config=config):
node_name, node_state = next(iter(step.items()))
print(f"Executed node: {node_name}")# Unit & Integration Tests with Coverage Gate (>= 90%)
pytest -v --cov=investigator --cov-fail-under=90
# Static Security Analysis (0 findings required)
bandit -r src/ -ll
# Secret Detection Scan
gitleaks detect --no-git --source . -v- Situation: Digital forensics investigations require tamper-evident custody chains and rigorous alibi validation without risk of hallucination or SQL injection.
- Task: Design an autonomous multi-agent LangGraph system meeting ISO/IEC 27037 standards with continuous guardrails and HITL verification.
- Action: Implemented a 5-node cyclical LangGraph pipeline featuring AST SQL parsing, k-way timestomping analysis, and HMAC-SHA256 signing.
- Result: 100% test pass rate across 80 tests, 0 Bandit security vulnerabilities, >90% code coverage, and sub-second deterministic evaluation.