The Complete Production SOP: How to Deploy and Run Suricata High-Performance Network IDS/IPS in Autonomous AI Agents

---
name: "suricata-ids-ips-network-rules"
description: "Deploy multi-threaded network intrusion detection and prevention with Suricata."
metadata:
  author: "OISF"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "configure suricata ids"
  - "update suricata rules"
  - "parse eve.json suricata"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

%YAML 1.1
---
suricata-version: "7.0"
default-log-dir: /var/log/suricata

af-packet:
  - interface: eth0
    threads: auto
    cluster-id: 99
    cluster-type: cluster_flow
    defrag: yes
    use-mmap: yes
    mmsg-size: 16
    tpacket-v3: yes
    ring-size: 2048
    checksum-checks: no

outputs:
  - fast:
      enabled: no
  - eve-log:
      enabled: yes
      filetype: regular
      filename: eve.json
      types:
        - alert:
            payload: yes
            payload-buffer-size: 4kb
            payload-printable: yes
            metadata: yes
            tagged-packets: yes
        - http: {extended: yes}
        - dns: {query: yes, answer: yes}
        - tls: {extended: yes}
        - flow

default-rule-path: /var/lib/suricata/rules
rule-files:
  - suricata.rules

import json
import sys
from pathlib import Path

def stream_eve_alerts(log_path: Path):
    """Memory-efficient generator yielding only high-severity Suricata alerts."""
    if not log_path.exists():
        raise FileNotFoundError(f"EVE telemetry log not found: {log_path}")
        
    with log_path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
                if record.get("event_type") == "alert":
                    alert_data = record.get("alert", {})
                    yield {
                        "timestamp": record.get("timestamp"),
                        "src_ip": record.get("src_ip"),
                        "src_port": record.get("src_port"),
                        "dest_ip": record.get("dest_ip"),
                        "dest_port": record.get("dest_port"),
                        "signature": alert_data.get("signature"),
                        "severity": alert_data.get("severity"),
                        "action": alert_data.get("action", "allowed")
                    }
            except json.JSONDecodeError:
                continue

if __name__ == "__main__":
    target = Path(sys.argv[1] if len(sys.argv) > 1 else "/var/log/suricata/eve.json")
    for alert in stream_eve_alerts(target):
        print(json.dumps(alert))

import unittest
import json
import tempfile
from pathlib import Path
from parse_eve_alerts import stream_eve_alerts

class TestEveParser(unittest.TestCase):
    def test_alert_filtering(self):
        mock_eve = [
            {"event_type": "flow", "src_ip": "10.0.0.1"},
            {"event_type": "alert", "timestamp": "2026-03-30T12:00:00Z", "src_ip": "192.168.1.5", "alert": {"signature": "ET MALWARE Suspicious User-Agent", "severity": 1, "action": "blocked"}}
        ]
        with tempfile.NamedTemporaryFile("w+", delete=False) as tmp:
            for entry in mock_eve:
                tmp.write(json.dumps(entry) + "\n")
            tmp_path = Path(tmp.name)
            
        try:
            alerts = list(stream_eve_alerts(tmp_path))
            self.assertEqual(len(alerts), 1)
            self.assertEqual(alerts[0]["signature"], "ET MALWARE Suspicious User-Agent")
            self.assertEqual(alerts[0]["action"], "blocked")
        finally:
            tmp_path.unlink()

if __name__ == "__main__":
    unittest.main()