add filebeat sned test logs

README.md

@@ -223,6 +223,23 @@ OLLAMA_MODEL=llama3.1 OLLAMA_TIMEOUT=300 fgai analyze-logs --logs logs/fg_syslog
 
 ## Optional FortiGate Input
 
+## Synthetic Windows Test Input
+
+For testing a Graylog Beats input without a Windows host, generate Windows
+Security-style JSONL events locally, then use Filebeat to ship them over TCP:
+
+```bash
+python scripts/generate_windows_events.py --interval 0.5
+filebeat -e -c examples/filebeat-windows-synthetic.yml
+```
+
+Update the absolute JSONL path and Graylog host in the Filebeat template first.
+Route `stream_hint: Windows` to a dedicated Graylog stream, then enable that
+stream in SignalScope and configure a profile such as entity `user` or
+`source_ip`, categorical `event_id`, `status`, `logon_type`, and numeric fields
+when present. Filebeat uses its Logstash output to communicate with Graylog's
+Beats input on TCP `5044`. [Graylog Beats input documentation](https://go2docs.graylog.org/current/getting_in_log_data/beats_input.html)
+
 For logs, configure FortiGate syslog to write into a local file such as `logs/fg_syslog.jsonl`. The parser supports common key/value syslog lines and JSONL.
 
 For policies, export a FortiOS config backup and pass it to `audit-policies`.

examples/filebeat-windows-synthetic.yml

@@ -0,0 +1,16 @@
+filebeat.inputs:
+  - type: filestream
+    id: signalscope-windows-synthetic
+    enabled: true
+    paths: ["/absolute/path/to/SignalScope/logs/windows-synthetic.jsonl"]
+    parsers:
+      - ndjson:
+          target: ""
+          add_error_key: true
+    fields_under_root: true
+    fields:
+      log_source: windows_synthetic
+      stream_hint: Windows
+
+output.logstash:
+  hosts: ["192.168.1.25:5044"]

scripts/generate_windows_events.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""Emit synthetic Windows Security-style JSONL events for Graylog/Filebeat testing."""
+from __future__ import annotations
+
+import argparse
+import json
+import random
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+EVENTS = [4624, 4625, 4634, 4672, 4688]
+USERS = ["alice", "bob", "svc_backup", "administrator"]
+HOSTS = ["WIN-CLIENT-01", "WIN-CLIENT-02", "WIN-SRV-01"]
+
+
+def event() -> dict[str, object]:
+    event_id = random.choices(EVENTS, weights=[45, 25, 15, 5, 10])[0]
+    failed = event_id == 4625
+    return {
+        "timestamp": datetime.now(UTC).isoformat(), "event_id": event_id,
+        "user": random.choice(USERS), "hostname": random.choice(HOSTS),
+        "source_ip": f"192.168.1.{random.randint(20, 90)}", "logon_type": random.choice([2, 3, 10]),
+        "status": "failure" if failed else "success", "action": "deny" if failed else "accept",
+        "event_provider": "Microsoft-Windows-Security-Auditing", "stream_hint": "Windows", "synthetic": True,
+        "message": "Synthetic Windows Security event generated by SignalScope test harness",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", default="logs/windows-synthetic.jsonl")
+    parser.add_argument("--interval", type=float, default=1.0)
+    parser.add_argument("--count", type=int, default=0, help="0 runs until interrupted")
+    args = parser.parse_args()
+    path = Path(args.output); path.parent.mkdir(parents=True, exist_ok=True)
+    emitted = 0
+    with path.open("a", encoding="utf-8", buffering=1) as handle:
+        while not args.count or emitted < args.count:
+            handle.write(json.dumps(event()) + "\n"); emitted += 1; time.sleep(args.interval)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())