490 lines
24 KiB
Python
490 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from .anomaly import anomaly_summary, detect_source_anomalies
|
|
from .dashboard import serve_dashboard
|
|
from .llm import ollama_summary
|
|
from .logs import local_in_failures, read_events, summarize_events, top_field_values
|
|
from .mitigation import FortiGateClient, parse_allowlist, suggest_block_candidates
|
|
from .policies import audit_policies, read_policies
|
|
from .recommendations import build_recommendations
|
|
from .syslog_server import listen_udp_syslog
|
|
from .monitor import monitor_loop
|
|
from .threat_intel import enrich_ips, is_public_ip
|
|
from .config import ConfigStore
|
|
from .baseline import BaselineStore, DEFAULT_MAX_VALUES_PER_FIELD, DEFAULT_RETENTION_DAYS, DEFAULT_VALUE_RETENTION_DAYS
|
|
from .exports import investigation_report, investigation_report_markdown
|
|
from .graylog_mcp import GraylogMcpClient
|
|
from .graylog_source import GraylogStreamSource
|
|
from .replay import replay_comparison, replay_events
|
|
from .stream_profiles import parse_profiles
|
|
|
|
|
|
def _print_json(data: object) -> None:
|
|
print(json.dumps(data, indent=2, sort_keys=True))
|
|
|
|
|
|
def analyze_logs(args: argparse.Namespace) -> int:
|
|
events = read_events(args.logs)
|
|
candidates = suggest_block_candidates(
|
|
events,
|
|
min_events=args.min_events,
|
|
min_score=args.min_score,
|
|
allowlist=parse_allowlist(args.allowlist),
|
|
)
|
|
anomalies = detect_source_anomalies(events, limit=args.anomaly_limit)
|
|
analysis = {
|
|
"summary": summarize_events(events),
|
|
"anomaly_summary": anomaly_summary(anomalies),
|
|
"diagnostics": {
|
|
"top_source_ips": top_field_values(events, "srcip", limit=10),
|
|
"top_services": top_field_values(events, "service", limit=10),
|
|
"top_actions": top_field_values(events, "action", limit=10),
|
|
"top_subtypes": top_field_values(events, "subtype", limit=10),
|
|
"local_in_failures": local_in_failures(events, limit=10),
|
|
},
|
|
"block_candidates": [
|
|
{"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons}
|
|
for candidate in candidates
|
|
],
|
|
"anomalies": [
|
|
{
|
|
"subject": finding.subject,
|
|
"score": finding.score,
|
|
"severity": finding.severity,
|
|
"confidence": finding.confidence,
|
|
"reasons": finding.reasons,
|
|
"evidence": finding.evidence,
|
|
}
|
|
for finding in anomalies
|
|
],
|
|
}
|
|
_print_json(analysis)
|
|
if args.llm:
|
|
print("\nLLM summary:")
|
|
print(ollama_summary([], candidates, args.model, analysis=analysis, timeout=args.llm_timeout))
|
|
return 0
|
|
|
|
|
|
def audit_policy_file(args: argparse.Namespace) -> int:
|
|
findings = audit_policies(read_policies(args.config))
|
|
_print_json([finding.__dict__ for finding in findings])
|
|
if args.llm:
|
|
print("\nLLM summary:")
|
|
print(ollama_summary(findings, [], args.model, timeout=args.llm_timeout))
|
|
return 0
|
|
|
|
|
|
def suggest_blocks(args: argparse.Namespace) -> int:
|
|
events = read_events(args.logs)
|
|
candidates = suggest_block_candidates(
|
|
events,
|
|
min_events=args.min_events,
|
|
min_score=args.min_score,
|
|
allowlist=parse_allowlist(args.allowlist),
|
|
)
|
|
if not candidates:
|
|
print("No block candidates met the current thresholds.")
|
|
return 0
|
|
|
|
for candidate in candidates:
|
|
print(f"{candidate.src_ip} score={candidate.score} reasons={', '.join(candidate.reasons)}")
|
|
if not args.execute:
|
|
print(" dry-run: not blocked")
|
|
continue
|
|
client = FortiGateClient.from_env()
|
|
reason = f"fgAI UTM mitigation score={candidate.score}"
|
|
response = client.quarantine_ip(candidate.src_ip, args.expiry_minutes, reason)
|
|
print(f" blocked: {response}")
|
|
return 0
|
|
|
|
|
|
def detect_anomalies(args: argparse.Namespace) -> int:
|
|
events = read_events(args.logs)
|
|
anomalies = detect_source_anomalies(events, limit=args.limit)
|
|
output = {
|
|
"summary": anomaly_summary(anomalies),
|
|
"anomalies": [
|
|
{
|
|
"subject": finding.subject,
|
|
"score": finding.score,
|
|
"severity": finding.severity,
|
|
"confidence": finding.confidence,
|
|
"reasons": finding.reasons,
|
|
"evidence": finding.evidence,
|
|
}
|
|
for finding in anomalies
|
|
if finding.score >= args.min_score
|
|
],
|
|
}
|
|
_print_json(output)
|
|
if args.llm:
|
|
print("\nLLM summary:")
|
|
print(ollama_summary([], [], args.model, analysis=output, timeout=args.llm_timeout))
|
|
return 0
|
|
|
|
|
|
def recommend(args: argparse.Namespace) -> int:
|
|
events = read_events(args.logs)
|
|
anomalies = detect_source_anomalies(events, limit=args.limit)
|
|
intel_ips = sorted(
|
|
{
|
|
ip
|
|
for event in events
|
|
for ip in (event.src_ip, event.dst_ip)
|
|
if is_public_ip(ip)
|
|
}
|
|
)
|
|
reputation = enrich_ips(intel_ips, limit=args.intel_limit) if args.threat_intel else {}
|
|
recommendations = build_recommendations(events, anomalies, reputation)
|
|
output = {
|
|
"recommendations": [
|
|
{
|
|
"subject": item.subject,
|
|
"score": item.score,
|
|
"severity": item.severity,
|
|
"title": item.title,
|
|
"recommendation": item.recommendation,
|
|
"reasons": item.reasons,
|
|
"related_policy_ids": item.related_policy_ids,
|
|
"related_services": item.related_services,
|
|
}
|
|
for item in recommendations
|
|
if item.score >= args.min_score
|
|
],
|
|
"reputation": reputation,
|
|
}
|
|
_print_json(output)
|
|
if args.llm:
|
|
print("\nLLM summary:")
|
|
print(ollama_summary([], [], args.model, analysis=output, timeout=args.llm_timeout))
|
|
return 0
|
|
|
|
|
|
def test_connection(args: argparse.Namespace) -> int:
|
|
client = FortiGateClient.from_env()
|
|
status = client.system_status()
|
|
results = status.get("results", status)
|
|
print("FortiGate API connection OK")
|
|
_print_json(results)
|
|
return 0
|
|
|
|
|
|
def fetch_policies(args: argparse.Namespace) -> int:
|
|
client = FortiGateClient.from_env()
|
|
policies = client.firewall_policies()
|
|
if args.output:
|
|
output = Path(args.output)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(policies, indent=2, sort_keys=True), encoding="utf-8")
|
|
print(f"Wrote {output}")
|
|
else:
|
|
_print_json(policies)
|
|
return 0
|
|
|
|
|
|
def listen_syslog(args: argparse.Namespace) -> int:
|
|
listen_udp_syslog(
|
|
args.host,
|
|
args.port,
|
|
args.output,
|
|
rotate_bytes=args.rotate_bytes,
|
|
rotate_count=args.rotate_count,
|
|
quiet=args.quiet,
|
|
)
|
|
return 0
|
|
|
|
|
|
def run_monitor(args: argparse.Namespace) -> int:
|
|
monitor_loop(
|
|
args.logs,
|
|
args.output,
|
|
policy_path=args.policies,
|
|
interval=args.interval,
|
|
anomaly_limit=args.anomaly_limit,
|
|
llm=args.llm,
|
|
llm_interval=args.llm_interval,
|
|
llm_model=args.model,
|
|
llm_timeout=args.llm_timeout,
|
|
baseline_path=args.baseline_db,
|
|
config_path=args.config_file,
|
|
history_path=args.history_db,
|
|
status_cache_path=args.status_cache_db,
|
|
)
|
|
return 0
|
|
|
|
|
|
def run_dashboard(args: argparse.Namespace) -> int:
|
|
serve_dashboard(args.host, args.port, args.status_file, image_dir=args.image_dir, config_file=args.config_file)
|
|
return 0
|
|
|
|
|
|
def export_investigation(args: argparse.Namespace) -> int:
|
|
status_path = Path(args.status_file)
|
|
status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {}
|
|
report = investigation_report(status, incident_id=args.incident_id or None)
|
|
fmt = "markdown" if args.format == "md" else args.format
|
|
output = investigation_report_markdown(report) if fmt == "markdown" else json.dumps(report, indent=2, sort_keys=True)
|
|
if args.output:
|
|
path = Path(args.output)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(output, encoding="utf-8")
|
|
print(f"Wrote {path}")
|
|
else:
|
|
print(output, end="" if output.endswith("\n") else "\n")
|
|
return 0
|
|
|
|
|
|
def baseline_maintenance(args: argparse.Namespace) -> int:
|
|
result = BaselineStore(args.baseline_db).maintenance(
|
|
retention_days=args.retention_days,
|
|
value_retention_days=args.value_retention_days,
|
|
max_values_per_field=args.max_values_per_field,
|
|
vacuum=args.vacuum,
|
|
include_rows=True,
|
|
)
|
|
_print_json(result)
|
|
return 0
|
|
|
|
|
|
def _stream_name(config: dict[str, object], stream_id: str) -> str:
|
|
return next((str(item.get("title", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") == stream_id), stream_id)
|
|
|
|
|
|
def _profile_fields(profile: object | None) -> tuple[str, ...]:
|
|
if not profile:
|
|
return ()
|
|
return tuple(
|
|
field for field in (
|
|
str(getattr(profile, "entity_field", "")),
|
|
*tuple(str(item) for item in getattr(profile, "entity_fields", ())),
|
|
str(getattr(profile, "timestamp_field", "")),
|
|
*tuple(str(item) for item in getattr(profile, "categorical_fields", ())),
|
|
*tuple(str(item) for item in getattr(profile, "numeric_fields", ())),
|
|
*tuple(str(getattr(relation, "left", "")) for relation in getattr(profile, "relationship_fields", ())),
|
|
*tuple(str(getattr(relation, "right", "")) for relation in getattr(profile, "relationship_fields", ())),
|
|
)
|
|
if field
|
|
)
|
|
|
|
|
|
def _configured_streams(config: dict[str, object], selected: list[str] | None = None) -> list[dict[str, str]]:
|
|
selected_order = list(dict.fromkeys(item for item in selected or [] if item))
|
|
selected_set = set(selected_order)
|
|
configured = [
|
|
{"id": str(item.get("id", "")), "title": str(item.get("title", ""))}
|
|
for item in config.get("graylog_streams", [])
|
|
if isinstance(item, dict) and item.get("id") and (item.get("enabled") or selected_set)
|
|
]
|
|
if selected_set:
|
|
configured = [item for item in configured if item["id"] in selected_set]
|
|
by_id = {item["id"]: item for item in configured}
|
|
configured = [by_id.get(stream_id, {"id": stream_id, "title": _stream_name(config, stream_id)}) for stream_id in selected_order]
|
|
if not configured and str(config.get("graylog_stream", "")):
|
|
stream_id = str(config.get("graylog_stream", ""))
|
|
configured = [{"id": stream_id, "title": _stream_name(config, stream_id)}]
|
|
return configured
|
|
|
|
|
|
def replay_history(args: argparse.Namespace) -> int:
|
|
config = ConfigStore(args.config_file).read()
|
|
profiles = parse_profiles(config.get("graylog_stream_profiles", []))
|
|
stream_name = _stream_name(config, args.stream_id)
|
|
events = read_events(args.logs)
|
|
result = replay_events(events, profiles, stream_id=args.stream_id, stream_name=stream_name, bucket_seconds=args.bucket_seconds)
|
|
if args.compare_config_file:
|
|
candidate_config = ConfigStore(args.compare_config_file).read()
|
|
candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
|
|
candidate = replay_events(events, candidate_profiles, stream_id=args.stream_id, stream_name=_stream_name(candidate_config, args.stream_id), bucket_seconds=args.bucket_seconds)
|
|
result = {"current": result, "candidate": candidate, "comparison": replay_comparison(result, candidate), "candidate_config_file": args.compare_config_file}
|
|
_print_json(result)
|
|
return 0
|
|
|
|
|
|
def replay_graylog(args: argparse.Namespace) -> int:
|
|
config = ConfigStore(args.config_file).read()
|
|
url, token = str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", ""))
|
|
if not url or not token:
|
|
raise RuntimeError("graylog_mcp_not_configured")
|
|
profiles = parse_profiles(config.get("graylog_stream_profiles", []))
|
|
candidate_profiles = {}
|
|
if args.compare_config_file:
|
|
candidate_config = ConfigStore(args.compare_config_file).read()
|
|
candidate_profiles = parse_profiles(candidate_config.get("graylog_stream_profiles", []))
|
|
streams = _configured_streams(config, args.stream_id)
|
|
if not streams:
|
|
raise RuntimeError("no_graylog_streams_selected")
|
|
events = []
|
|
stream_statuses = []
|
|
for stream in streams:
|
|
stream_id = stream["id"]
|
|
profile_fields = tuple(dict.fromkeys([*_profile_fields(profiles.get(stream_id)), *_profile_fields(candidate_profiles.get(stream_id))]))
|
|
stream_events, status = GraylogStreamSource(
|
|
GraylogMcpClient(url, token),
|
|
stream_id,
|
|
str(config.get("graylog_query", "*")),
|
|
str(config.get("graylog_field_mapping", "")),
|
|
stream.get("title") or stream_id,
|
|
profile_fields,
|
|
).fetch(max_events=args.max_events_per_stream, range_seconds=args.range_seconds)
|
|
events.extend(stream_events)
|
|
stream_statuses.append({"stream_id": stream_id, "stream_name": stream.get("title") or stream_id, **status})
|
|
result = {
|
|
"source": "graylog_mcp",
|
|
"range_seconds": args.range_seconds,
|
|
"streams": stream_statuses,
|
|
"replay": replay_events(events, profiles, bucket_seconds=args.bucket_seconds),
|
|
}
|
|
if args.compare_config_file:
|
|
candidate = replay_events(events, candidate_profiles, bucket_seconds=args.bucket_seconds)
|
|
current = result["replay"]
|
|
result["candidate_config_file"] = args.compare_config_file
|
|
result["candidate_replay"] = candidate
|
|
result["comparison"] = replay_comparison(current, candidate)
|
|
_print_json(result)
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Local FortiGate AI/ML inspection tool")
|
|
subparsers = parser.add_subparsers(required=True)
|
|
|
|
logs = subparsers.add_parser("analyze-logs", help="Analyze local FortiGate logs")
|
|
logs.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file")
|
|
logs.add_argument("--min-events", type=int, default=3)
|
|
logs.add_argument("--min-score", type=int, default=7)
|
|
logs.add_argument("--allowlist", default=None, help="Comma-separated IPs/CIDRs never to block")
|
|
logs.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results")
|
|
logs.add_argument("--model", default=None, help="Ollama model name")
|
|
logs.add_argument("--llm-timeout", type=int, default=None, help="Ollama request timeout in seconds")
|
|
logs.add_argument("--anomaly-limit", type=int, default=20, help="Maximum anomaly findings to include")
|
|
logs.set_defaults(func=analyze_logs)
|
|
|
|
anomalies = subparsers.add_parser("detect-anomalies", help="Score likely traffic anomalies by source IP")
|
|
anomalies.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file")
|
|
anomalies.add_argument("--limit", type=int, default=20, help="Maximum anomaly findings to include")
|
|
anomalies.add_argument("--min-score", type=int, default=1, help="Minimum anomaly score to output")
|
|
anomalies.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results")
|
|
anomalies.add_argument("--model", default=None, help="Ollama model name")
|
|
anomalies.add_argument("--llm-timeout", type=int, default=None, help="Ollama request timeout in seconds")
|
|
anomalies.set_defaults(func=detect_anomalies)
|
|
|
|
recommendations = subparsers.add_parser("recommend", help="Generate policy and response recommendations from anomalies")
|
|
recommendations.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file")
|
|
recommendations.add_argument("--limit", type=int, default=20, help="Maximum anomaly findings to evaluate")
|
|
recommendations.add_argument("--min-score", type=int, default=35, help="Minimum recommendation score to output")
|
|
recommendations.add_argument("--threat-intel", action="store_true", help="Use enabled external threat intelligence lookups")
|
|
recommendations.add_argument("--intel-limit", type=int, default=25, help="Maximum public IPs to enrich")
|
|
recommendations.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results")
|
|
recommendations.add_argument("--model", default=None, help="Ollama model name")
|
|
recommendations.add_argument("--llm-timeout", type=int, default=None, help="Ollama request timeout in seconds")
|
|
recommendations.set_defaults(func=recommend)
|
|
|
|
policies = subparsers.add_parser("audit-policies", help="Audit FortiOS firewall policy config")
|
|
policies.add_argument("--config", required=True, help="Path to FortiOS config backup")
|
|
policies.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results")
|
|
policies.add_argument("--model", default=None, help="Ollama model name")
|
|
policies.add_argument("--llm-timeout", type=int, default=None, help="Ollama request timeout in seconds")
|
|
policies.set_defaults(func=audit_policy_file)
|
|
|
|
blocks = subparsers.add_parser("suggest-blocks", help="Suggest or execute guarded source IP blocks")
|
|
blocks.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file")
|
|
blocks.add_argument("--min-events", type=int, default=3)
|
|
blocks.add_argument("--min-score", type=int, default=7)
|
|
blocks.add_argument("--allowlist", default=None, help="Comma-separated IPs/CIDRs never to block")
|
|
blocks.add_argument("--execute", action="store_true", help="Actually call the FortiGate quarantine API")
|
|
blocks.add_argument("--expiry-minutes", type=int, default=60)
|
|
blocks.set_defaults(func=suggest_blocks)
|
|
|
|
connection = subparsers.add_parser("test-connection", help="Test FortiGate REST API credentials")
|
|
connection.set_defaults(func=test_connection)
|
|
|
|
fetch = subparsers.add_parser("fetch-policies", help="Fetch firewall policies through the FortiGate REST API")
|
|
fetch.add_argument("--output", help="Write JSON response to this file")
|
|
fetch.set_defaults(func=fetch_policies)
|
|
|
|
listener = subparsers.add_parser("listen-syslog", help="Listen for UDP syslog and append to a local log file")
|
|
listener.add_argument("--host", default="0.0.0.0", help="Bind address")
|
|
listener.add_argument("--port", type=int, default=5514, help="UDP port. Use 514 only with sudo/capability.")
|
|
listener.add_argument("--output", default="logs/fg_syslog.jsonl", help="File to append received logs to")
|
|
listener.add_argument("--rotate-bytes", type=int, default=25 * 1024 * 1024, help="Rotate active log at this size; 0 disables rotation")
|
|
listener.add_argument("--rotate-count", type=int, default=14, help="Number of compressed log archives to retain")
|
|
listener.add_argument("--quiet", action="store_true", help="Do not print each received syslog message")
|
|
listener.set_defaults(func=listen_syslog)
|
|
|
|
monitor = subparsers.add_parser("monitor", help="Continuously analyze logs and write dashboard status JSON")
|
|
monitor.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file")
|
|
monitor.add_argument("--output", default="state/fgai-status.json", help="Status JSON written for dashboard")
|
|
monitor.add_argument("--policies", default=None, help="Optional FortiGate policy JSON/config file to audit continuously")
|
|
monitor.add_argument("--interval", type=int, default=10, help="Seconds between analysis runs")
|
|
monitor.add_argument("--anomaly-limit", type=int, default=20, help="Maximum anomaly findings to include")
|
|
monitor.add_argument("--baseline-db", default="state/fgai-baseline.sqlite3", help="SQLite database for historical behavior baselines")
|
|
monitor.add_argument("--config-file", default="state/fgai-config.json", help="Runtime configuration written by dashboard")
|
|
monitor.add_argument("--history-db", default="state/signalscope-history.sqlite3", help="Aggregate history database for dashboard trends")
|
|
monitor.add_argument("--status-cache-db", default="state/signalscope-status-cache.sqlite3", help="SQLite cache for last successful dashboard status")
|
|
monitor.add_argument("--llm", action="store_true", help="Generate cached Ollama analyst note for dashboard")
|
|
monitor.add_argument("--llm-interval", type=int, default=300, help="Seconds between Ollama dashboard assessments")
|
|
monitor.add_argument("--model", default=None, help="Ollama model name")
|
|
monitor.add_argument("--llm-timeout", type=int, default=None, help="Ollama request timeout in seconds")
|
|
monitor.set_defaults(func=run_monitor)
|
|
|
|
dashboard = subparsers.add_parser("dashboard", help="Serve local fgAI dashboard")
|
|
dashboard.add_argument("--host", default="127.0.0.1", help="Dashboard bind address")
|
|
dashboard.add_argument("--port", type=int, default=8088, help="Dashboard TCP port")
|
|
dashboard.add_argument("--status-file", default="state/fgai-status.json", help="Status JSON produced by monitor")
|
|
dashboard.add_argument("--image-dir", default="images", help="Directory containing dashboard images")
|
|
dashboard.add_argument("--config-file", default="state/fgai-config.json", help="Local runtime configuration JSON")
|
|
dashboard.set_defaults(func=run_dashboard)
|
|
|
|
export = subparsers.add_parser("export-investigation", help="Export current investigation incidents as Markdown or JSON")
|
|
export.add_argument("--status-file", default="state/fgai-status.json", help="Status JSON produced by monitor")
|
|
export.add_argument("--incident-id", default="", help="Export only one incident ID")
|
|
export.add_argument("--format", choices=["markdown", "md", "json"], default="markdown", help="Report format")
|
|
export.add_argument("--output", default="", help="Write the report to this file instead of stdout")
|
|
export.set_defaults(func=export_investigation)
|
|
|
|
maintenance = subparsers.add_parser("baseline-maintenance", help="Prune and optionally compact the local SQLite baseline")
|
|
maintenance.add_argument("--baseline-db", default="state/fgai-baseline.sqlite3", help="SQLite baseline database")
|
|
maintenance.add_argument("--retention-days", type=int, default=DEFAULT_RETENTION_DAYS, help="Keep bucket and dedupe history for this many days")
|
|
maintenance.add_argument("--value-retention-days", type=int, default=DEFAULT_VALUE_RETENTION_DAYS, help="Prune stale one-off categorical values after this many days")
|
|
maintenance.add_argument("--max-values-per-field", type=int, default=DEFAULT_MAX_VALUES_PER_FIELD, help="Keep at most this many categorical values per stream/entity/field")
|
|
maintenance.add_argument("--vacuum", action="store_true", help="Run SQLite VACUUM after pruning to return disk space")
|
|
maintenance.set_defaults(func=baseline_maintenance)
|
|
|
|
replay = subparsers.add_parser("replay", help="Replay a historical log export against temporary baselines")
|
|
replay.add_argument("--logs", required=True, help="Historic JSONL or key/value log export")
|
|
replay.add_argument("--config-file", default="state/fgai-config.json", help="Stream profile configuration")
|
|
replay.add_argument("--stream-id", default="", help="Apply this configured Graylog stream profile to exported events")
|
|
replay.add_argument("--bucket-seconds", type=int, default=300, help="Replay baseline bucket size")
|
|
replay.add_argument("--compare-config-file", default="", help="Replay the same export against another saved config and report detector deltas")
|
|
replay.set_defaults(func=replay_history)
|
|
|
|
replay_mcp = subparsers.add_parser("replay-graylog", help="Replay a Graylog MCP time window against temporary baselines")
|
|
replay_mcp.add_argument("--config-file", default="state/fgai-config.json", help="Runtime configuration with Graylog MCP URL, token, streams and profiles")
|
|
replay_mcp.add_argument("--stream-id", action="append", default=[], help="Replay only this stream ID. Repeat for multiple streams. Defaults to enabled streams.")
|
|
replay_mcp.add_argument("--range-seconds", type=int, default=24 * 3600, help="Graylog relative time window to fetch")
|
|
replay_mcp.add_argument("--max-events-per-stream", type=int, default=5_000, help="Maximum events fetched from each selected stream")
|
|
replay_mcp.add_argument("--bucket-seconds", type=int, default=300, help="Replay baseline bucket size")
|
|
replay_mcp.add_argument("--compare-config-file", default="", help="Replay the same Graylog events against another saved config and report detector deltas")
|
|
replay_mcp.set_defaults(func=replay_graylog)
|
|
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
return args.func(args)
|
|
except Exception as exc:
|
|
print(f"fgai: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|