102 lines
4.6 KiB
Python
102 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
DEFAULT_CONFIG: dict[str, object] = {
|
|
"log_source": "local_syslog",
|
|
"graylog_mcp_url": "",
|
|
"graylog_stream": "",
|
|
"graylog_streams": [],
|
|
"graylog_stream_profiles": [],
|
|
"graylog_query": "*",
|
|
"graylog_fetch_mode": "auto",
|
|
"graylog_tls_verify": True,
|
|
"graylog_range_seconds": 300,
|
|
"graylog_max_events_per_stream": 5000,
|
|
"graylog_raw_sample_events": 5000,
|
|
"graylog_mcp_call_timeout_seconds": 8,
|
|
"graylog_mcp_poll_timeout_seconds": 120,
|
|
"baseline_retention_days": 7,
|
|
"baseline_value_retention_days": 3,
|
|
"baseline_max_values_per_field": 500,
|
|
"baseline_training_days": 7,
|
|
"graylog_field_mapping": "",
|
|
"llm_enabled": False,
|
|
"llm_model": "",
|
|
"profile_advisor_enabled": False,
|
|
"profile_advisor_model": "qwen3:8b",
|
|
"profile_advisor_timeout": 240,
|
|
"threat_intel_enabled": False,
|
|
"threat_intel_provider": "auto",
|
|
"abuseipdb_api_key": "",
|
|
"virustotal_api_key": "",
|
|
"threat_intel_daily_limit": 100,
|
|
"threat_intel_ttl_seconds": 604800,
|
|
"threat_intel_error_ttl_seconds": 3600,
|
|
"abuseipdb_max_age_days": 90,
|
|
}
|
|
|
|
EDITABLE_FIELDS = set(DEFAULT_CONFIG) | {"graylog_mcp_token"}
|
|
|
|
|
|
class ConfigStore:
|
|
def __init__(self, path: str) -> None:
|
|
self.path = Path(path)
|
|
|
|
def read(self) -> dict[str, object]:
|
|
try:
|
|
stored = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
|
stored = {}
|
|
return {**DEFAULT_CONFIG, **{key: value for key, value in stored.items() if key in EDITABLE_FIELDS}}
|
|
|
|
def public(self) -> dict[str, object]:
|
|
config = self.read()
|
|
config["graylog_mcp_token_configured"] = bool(config.pop("graylog_mcp_token", ""))
|
|
config["abuseipdb_api_key_configured"] = bool(config.pop("abuseipdb_api_key", ""))
|
|
config["virustotal_api_key_configured"] = bool(config.pop("virustotal_api_key", ""))
|
|
return config
|
|
|
|
def update(self, values: dict[str, object]) -> dict[str, object]:
|
|
current = self.read()
|
|
for key, value in values.items():
|
|
if key not in EDITABLE_FIELDS:
|
|
continue
|
|
if key in {"graylog_mcp_token", "abuseipdb_api_key", "virustotal_api_key"} and value == "":
|
|
continue
|
|
if key in {"llm_enabled", "profile_advisor_enabled", "threat_intel_enabled", "graylog_tls_verify"}:
|
|
current[key] = bool(value)
|
|
elif key == "log_source" and value in {"local_syslog", "graylog_mcp"}:
|
|
current[key] = value
|
|
elif key == "graylog_fetch_mode" and value in {"auto", "raw", "aggregate"}:
|
|
current[key] = value
|
|
elif key in {"graylog_range_seconds", "graylog_max_events_per_stream", "graylog_raw_sample_events", "graylog_mcp_call_timeout_seconds", "graylog_mcp_poll_timeout_seconds", "baseline_retention_days", "baseline_value_retention_days", "baseline_max_values_per_field", "baseline_training_days", "profile_advisor_timeout", "threat_intel_daily_limit", "threat_intel_ttl_seconds", "threat_intel_error_ttl_seconds", "abuseipdb_max_age_days"}:
|
|
try:
|
|
minimum = 60 if key in {"graylog_range_seconds", "graylog_mcp_poll_timeout_seconds"} else 1
|
|
current[key] = max(minimum, int(value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
elif key == "threat_intel_provider" and value in {"auto", "abuseipdb", "virustotal"}:
|
|
current[key] = value
|
|
elif key == "graylog_streams" and isinstance(value, list):
|
|
streams = [
|
|
{"id": str(item.get("id", "")), "title": str(item.get("title", "")), "enabled": bool(item.get("enabled"))}
|
|
for item in value if isinstance(item, dict) and item.get("id")
|
|
]
|
|
if not streams and current.get("graylog_streams"):
|
|
continue
|
|
current[key] = streams
|
|
elif key == "graylog_stream_profiles" and isinstance(value, list):
|
|
current[key] = [item for item in value if isinstance(item, dict) and item.get("stream_id")]
|
|
elif isinstance(value, str):
|
|
current[key] = value.strip()
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(current, indent=2, sort_keys=True), encoding="utf-8")
|
|
os.chmod(temporary, 0o600)
|
|
temporary.replace(self.path)
|
|
return self.public()
|