Security Engineering & Cloud Security Agents
Problem Statement
Writing a Sigma rule is only half the battle. Security engineering teams often deploy untested rules directly into production SIEMs, resulting in broken syntax or query performance bottlenecks that impact SIEM stability. An agentic loop is required to write the rule, test syntax against a backend engine, validate against sample log payloads, and refine the rule iteratively until it produces clean detections without error.
Goal
Build a self-correcting agentic loop using LangGraph that synthesizes a Sigma rule, validates it against sample JSON telemetry logs, and iteratively refines the detection logic.
Outcome Measurement
- Rule Quality: 100% of generated Sigma rules compile error-free when translated into target SIEM languages.
- False Positive Reduction during Drafting: Iterative verification eliminates 80% of obvious syntax and field mapping errors prior to production deployment.
How the Code was Built
The state machine implements a loop:
GenerateRule: Drafts initial rule.ValidateAgainstLogs: Tests rule logic against sample logs.RefineRule: If validation fails or false positives occur, updates detection block and re-validates.
Executable Code
from typing import TypedDict, Listfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIclass SigmaValidationState(TypedDict): threat_description: str sample_logs: str sigma_rule: str is_valid: bool validation_feedback: str iteration_count: intllm = ChatOpenAI(model="gpt-4o", temperature=0)def generate_rule_node(state: SigmaValidationState) -> dict: feedback = state.get("validation_feedback", "") prompt = f"""Generate a Sigma detection rule for: {state['threat_description']}Previous Feedback/Errors: {feedback}Current Rule: {state.get('sigma_rule','')}Output ONLY valid YAML text for the Sigma rule.""" res = llm.invoke(prompt) return { "sigma_rule": res.content, "iteration_count": state.get("iteration_count", 0) + 1 }def validate_rule_node(state: SigmaValidationState) -> dict: prompt = f"""Evaluate this Sigma Rule against these Sample Logs:SIGMA RULE:{state['sigma_rule']}SAMPLE LOGS:{state['sample_logs']}Does the rule match the malicious activity in the logs without syntax errors?Respond with JSON: {{"valid": true/false, "feedback": "reasoning..."}}""" res = llm.invoke(prompt) is_valid = "true" in res.content.lower() and '"valid": true' in res.content.lower() return {"is_valid": is_valid, "validation_feedback": res.content}def decide_next_step(state: SigmaValidationState) -> str: if state["is_valid"] or state["iteration_count"] >= 3: return "approved" return "refine"workflow = StateGraph(SigmaValidationState)workflow.add_node("generate", generate_rule_node)workflow.add_node("validate", validate_rule_node)workflow.set_entry_point("generate")workflow.add_edge("generate", "validate")workflow.add_conditional_edges("validate", decide_next_step, {"approved": END, "refine": "generate"})app = workflow.compile()if __name__ == "__main__": test_logs = """{"EventID":4688, "NewProcessName": "C:\\Windows\\System32\\cmd.exe", "CommandLine": "cmd.exe /c powershell.exe -nop -w hidden -encodedcommand ExecutionData="}""" inputs = { "threat_description": "Detect obfuscated PowerShell invocation from CMD shell", "sample_logs": test_logs, "sigma_rule": "", "is_valid": False, "validation_feedback": "", "iteration_count": 0 } result = app.invoke(inputs) print("Final Validated Sigma Rule:\n") print(result["sigma_rule"])
Leave a Reply