1037 lines
57 KiB
Python
1037 lines
57 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import signal
|
|
import threading
|
|
import time
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from .anomaly import anomaly_summary, detect_source_anomalies
|
|
from .baseline import BaselineStore
|
|
from .config import ConfigStore
|
|
from .correlation import correlate_source_ips
|
|
from .event_context import build_event_context
|
|
from .feedback import FeedbackStore, SUPPRESSING_STATUSES, feedback_matches, select_match, suggest_generalizations
|
|
from .graylog_aggregate import GraylogAggregateSource
|
|
from .graylog_mcp import GraylogMcpClient
|
|
from .graylog_source import GraylogStreamSource
|
|
from .history import FieldDiscoveryStore, HistoryStore, StatusSnapshotStore
|
|
from .incidents import IncidentStore, build_incidents
|
|
from .data_quality import assess_data_quality
|
|
from .llm import ollama_dashboard_assessment, ollama_profile_advice
|
|
from .logs import local_in_failures, read_events, summarize_events, top_field_values
|
|
from .mitigation import parse_allowlist, suggest_block_candidates
|
|
from .policies import audit_policies, read_policies
|
|
from .profile_suggestions import apply_profile_advice, suggest_stream_profiles
|
|
from .recommendations import build_recommendations
|
|
from .sequences import detect_sequences
|
|
from .threat_intel import ThreatIntelClient, enrich_ips, is_public_ip
|
|
from .triage import build_triage_queue
|
|
from .stream_profiles import parse_profiles
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _cycle_timeout(seconds: int):
|
|
if seconds <= 0 or threading.current_thread() is not threading.main_thread() or not hasattr(signal, "SIGALRM"):
|
|
yield
|
|
return
|
|
previous_handler = signal.getsignal(signal.SIGALRM)
|
|
previous_timer = signal.getitimer(signal.ITIMER_REAL)
|
|
started = time.monotonic()
|
|
|
|
def _raise_timeout(_signum, _frame):
|
|
raise TimeoutError(f"status_cycle_timeout_{seconds}s")
|
|
|
|
signal.signal(signal.SIGALRM, _raise_timeout)
|
|
signal.setitimer(signal.ITIMER_REAL, seconds)
|
|
try:
|
|
yield
|
|
finally:
|
|
elapsed = time.monotonic() - started
|
|
remaining = max(0.0, previous_timer[0] - elapsed) if previous_timer[0] else 0.0
|
|
signal.setitimer(signal.ITIMER_REAL, remaining, previous_timer[1])
|
|
signal.signal(signal.SIGALRM, previous_handler)
|
|
|
|
|
|
def _stream_titles(config: dict[str, object]) -> dict[str, str]:
|
|
return {
|
|
str(item.get("id", "")): str(item.get("title", "") or item.get("id", ""))
|
|
for item in config.get("graylog_streams", [])
|
|
if isinstance(item, dict) and item.get("id")
|
|
}
|
|
|
|
|
|
def _public_runtime_config(config: dict[str, object]) -> dict[str, object]:
|
|
public = {
|
|
key: value
|
|
for key, value in config.items()
|
|
if key not in {"graylog_mcp_token", "abuseipdb_api_key", "virustotal_api_key"}
|
|
}
|
|
public["graylog_mcp_token_configured"] = bool(config.get("graylog_mcp_token"))
|
|
public["abuseipdb_api_key_configured"] = bool(config.get("abuseipdb_api_key"))
|
|
public["virustotal_api_key_configured"] = bool(config.get("virustotal_api_key"))
|
|
public["enabled_streams"] = sum(
|
|
1
|
|
for item in config.get("graylog_streams", [])
|
|
if isinstance(item, dict) and item.get("enabled")
|
|
)
|
|
return public
|
|
|
|
|
|
def _profile_field_set(profile: object | None) -> set[str]:
|
|
if profile is None:
|
|
return set()
|
|
fields = {
|
|
str(getattr(profile, "entity_field", "") or ""),
|
|
str(getattr(profile, "timestamp_field", "") or ""),
|
|
}
|
|
fields.update(str(field) for field in getattr(profile, "entity_fields", ()) if field)
|
|
fields.update(str(field) for field in getattr(profile, "categorical_fields", ()) if field)
|
|
fields.update(str(field) for field in getattr(profile, "numeric_fields", ()) if field)
|
|
for relation in getattr(profile, "relationship_fields", ()):
|
|
fields.add(str(getattr(relation, "left", "") or ""))
|
|
fields.add(str(getattr(relation, "right", "") or ""))
|
|
return {field for field in fields if field}
|
|
|
|
|
|
def _suggestion_field_set(suggestion: dict[str, object]) -> set[str]:
|
|
profile = suggestion.get("profile", {}) if isinstance(suggestion.get("profile"), dict) else {}
|
|
fields = {
|
|
str(profile.get("entity_field", "") or ""),
|
|
str(profile.get("timestamp_field", "") or ""),
|
|
}
|
|
for key in ("entity_fields", "categorical_fields", "numeric_fields"):
|
|
value = profile.get(key, [])
|
|
if isinstance(value, list):
|
|
fields.update(str(field) for field in value if field)
|
|
relationships = profile.get("relationship_fields", [])
|
|
if isinstance(relationships, list):
|
|
for relation in relationships:
|
|
if isinstance(relation, dict):
|
|
fields.add(str(relation.get("left", "") or ""))
|
|
fields.add(str(relation.get("right", "") or ""))
|
|
return {field for field in fields if field}
|
|
|
|
|
|
def _ambiguous_fields(suggestion: dict[str, object], threshold: int) -> list[str]:
|
|
field_confidence = suggestion.get("field_confidence", {})
|
|
if not isinstance(field_confidence, dict):
|
|
return []
|
|
selected = _suggestion_field_set(suggestion)
|
|
# Only fields that were actually observed and shape-classified are "ambiguous" —
|
|
# a field with no field_confidence entry (e.g. a placeholder timestamp field name
|
|
# used when no time-like field was discovered at all) has nothing to disambiguate.
|
|
return [field for field in selected if field in field_confidence and int(field_confidence[field] or 0) < threshold]
|
|
|
|
|
|
def _needs_profile_advisor(suggestion: dict[str, object], stream_profiles: dict[str, object], *, confidence_threshold: int = 60) -> bool:
|
|
stream_id = str(suggestion.get("stream_id", ""))
|
|
suggested_fields = _suggestion_field_set(suggestion)
|
|
if not suggestion.get("profile_exists"):
|
|
return True
|
|
existing_fields = _profile_field_set(stream_profiles.get(stream_id))
|
|
if suggested_fields - existing_fields:
|
|
return True
|
|
return bool(_ambiguous_fields(suggestion, confidence_threshold))
|
|
|
|
|
|
def _suggestion_is_high_confidence(suggestion: dict[str, object], threshold: int) -> bool:
|
|
field_confidence = suggestion.get("field_confidence", {})
|
|
if not isinstance(field_confidence, dict):
|
|
return False
|
|
observed = [field for field in _suggestion_field_set(suggestion) if field in field_confidence]
|
|
if not observed:
|
|
return False
|
|
return all(int(field_confidence[field] or 0) >= threshold for field in observed)
|
|
|
|
|
|
def _profile_is_auto_apply_safe(existing: object) -> bool:
|
|
"""A profile with any manual-origin field is left alone entirely by auto-apply."""
|
|
if existing is None:
|
|
return True
|
|
field_origin = getattr(existing, "field_origin", None) or {}
|
|
return all(origin != "manual" for origin in field_origin.values())
|
|
|
|
|
|
def _auto_apply_profile_suggestions(
|
|
profile_suggestions: list[dict[str, object]],
|
|
stream_profiles: dict[str, object],
|
|
runtime_values: dict[str, object],
|
|
config_path: str | None,
|
|
) -> list[str]:
|
|
"""Write high-confidence, provenance-safe suggestions straight to config.
|
|
|
|
Mirrors the dashboard's own POST /api/config write path (ConfigStore.update),
|
|
just invoked here unattended each cycle instead of from a human clicking Apply.
|
|
"""
|
|
if not config_path or not runtime_values.get("auto_apply_profile_suggestions"):
|
|
return []
|
|
threshold = int(runtime_values.get("field_confidence_threshold", 60) or 60)
|
|
updated_profiles = [item for item in runtime_values.get("graylog_stream_profiles", []) if isinstance(item, dict)]
|
|
applied: list[str] = []
|
|
now = int(time.time())
|
|
for suggestion in profile_suggestions:
|
|
if not isinstance(suggestion, dict):
|
|
continue
|
|
stream_id = str(suggestion.get("stream_id", ""))
|
|
profile = suggestion.get("profile", {})
|
|
if not stream_id or not isinstance(profile, dict):
|
|
continue
|
|
if not _suggestion_is_high_confidence(suggestion, threshold):
|
|
continue
|
|
if not _profile_is_auto_apply_safe(stream_profiles.get(stream_id)):
|
|
continue
|
|
all_fields = {*profile.get("entity_fields", []), *profile.get("categorical_fields", []), *profile.get("numeric_fields", [])}
|
|
new_profile = dict(profile)
|
|
new_profile["field_origin"] = {str(field): "auto" for field in all_fields if field}
|
|
new_profile["updated_at"] = now
|
|
updated_profiles = [item for item in updated_profiles if item.get("stream_id") != stream_id]
|
|
updated_profiles.append(new_profile)
|
|
applied.append(stream_id)
|
|
if applied:
|
|
ConfigStore(config_path).update({"graylog_stream_profiles": updated_profiles})
|
|
return applied
|
|
|
|
|
|
DRIFT_CONFIDENCE_DROP_THRESHOLD = 20
|
|
|
|
|
|
def _detect_profile_drift(
|
|
profile_suggestions: list[dict[str, object]],
|
|
stream_profiles: dict[str, object],
|
|
discovery_store: object | None,
|
|
) -> list[dict[str, object]]:
|
|
"""Flag auto-applied fields whose classification has drifted since they were applied.
|
|
|
|
Flag-only: this never touches the live profile, only surfaces a diagnostic for a
|
|
human to review. A silently "self-corrected" auto-applied field could suppress or
|
|
misdirect real findings with nobody noticing, which is worse than a stale mapping
|
|
that gets flagged and reviewed.
|
|
"""
|
|
if discovery_store is None:
|
|
return []
|
|
drift: list[dict[str, object]] = []
|
|
for suggestion in profile_suggestions:
|
|
if not isinstance(suggestion, dict):
|
|
continue
|
|
stream_id = str(suggestion.get("stream_id", ""))
|
|
if not stream_id:
|
|
continue
|
|
existing = stream_profiles.get(stream_id)
|
|
field_origin = getattr(existing, "field_origin", None) or {}
|
|
auto_fields = [field for field, origin in field_origin.items() if origin == "auto"]
|
|
if not auto_fields:
|
|
continue
|
|
applied = discovery_store.applied_confidence(stream_id)
|
|
if not applied:
|
|
continue
|
|
field_confidence = suggestion.get("field_confidence", {})
|
|
field_shapes = suggestion.get("field_shapes", {})
|
|
if not isinstance(field_confidence, dict) or not isinstance(field_shapes, dict):
|
|
continue
|
|
for field in auto_fields:
|
|
baseline = applied.get(field)
|
|
if not baseline or field not in field_confidence:
|
|
continue
|
|
current_confidence = int(field_confidence.get(field, 0) or 0)
|
|
current_shape = str(field_shapes.get(field, ""))
|
|
confidence_drop = int(baseline["applied_confidence"]) - current_confidence
|
|
shape_changed = bool(current_shape) and current_shape != str(baseline["applied_shape"])
|
|
if confidence_drop > DRIFT_CONFIDENCE_DROP_THRESHOLD or shape_changed:
|
|
drift.append({
|
|
"stream_id": stream_id,
|
|
"field": field,
|
|
"applied_confidence": baseline["applied_confidence"],
|
|
"current_confidence": current_confidence,
|
|
"applied_shape": baseline["applied_shape"],
|
|
"current_shape": current_shape,
|
|
"applied_at": baseline["applied_at"],
|
|
})
|
|
return drift
|
|
|
|
|
|
CLUSTER_SIZE_DAMPEN_THRESHOLD = 5
|
|
CLUSTER_DAMPENED_SCORE_CAP = 25
|
|
|
|
|
|
def _cluster_field_deviations(field_deviations: dict[str, list[dict[str, object]]]) -> None:
|
|
"""Recognize the same (detector, field) signature repeating across many
|
|
entities as one systemic pattern instead of N independent findings.
|
|
|
|
Mutates deviations in place: tags cluster_size, and caps score for large
|
|
clusters — a signature hitting many entities identically usually means a
|
|
detector/profile problem, not N separate critical events. Signature
|
|
deliberately excludes stream_id: the incident this guards against (a
|
|
misconfigured relationship field) recurred identically across multiple
|
|
streams, and that cross-stream repetition is itself part of the pattern.
|
|
Already-suppressed deviations don't count toward another entity's cluster
|
|
size, so a false-positive rule doesn't inflate the appearance of a pattern.
|
|
"""
|
|
signature_entities: dict[tuple[str, str], set[str]] = defaultdict(set)
|
|
for entity, deviations in field_deviations.items():
|
|
for deviation in deviations:
|
|
if deviation.get("feedback") in SUPPRESSING_STATUSES:
|
|
continue
|
|
signature = (str(deviation.get("detector", "")), str(deviation.get("field", "")))
|
|
signature_entities[signature].add(entity)
|
|
for entity, deviations in field_deviations.items():
|
|
for deviation in deviations:
|
|
signature = (str(deviation.get("detector", "")), str(deviation.get("field", "")))
|
|
cluster_size = len(signature_entities.get(signature, ()))
|
|
deviation["cluster_size"] = cluster_size
|
|
if cluster_size >= CLUSTER_SIZE_DAMPEN_THRESHOLD:
|
|
deviation["score"] = min(int(deviation.get("score", 0) or 0), CLUSTER_DAMPENED_SCORE_CAP)
|
|
deviation["cluster_dampened"] = True
|
|
|
|
|
|
def _stream_name(stream_id: str, stream_titles: dict[str, str], profile: object | None = None) -> str:
|
|
return stream_titles.get(stream_id) or getattr(profile, "name", "") or stream_id
|
|
|
|
|
|
def _profile_name(stream_id: str, stream_titles: dict[str, str], profile: object | None = None) -> str:
|
|
name = str(getattr(profile, "name", "") or "").strip()
|
|
if not name or name == stream_id:
|
|
return f"{_stream_name(stream_id, stream_titles, profile)} profile"
|
|
return name
|
|
|
|
|
|
def _range_seconds(value: object) -> int:
|
|
try:
|
|
return max(60, int(value))
|
|
except (TypeError, ValueError):
|
|
return 300
|
|
|
|
|
|
def monitor_cycle_timeout_seconds(mcp_poll_timeout: int, mcp_call_timeout: int) -> int:
|
|
return max(600, mcp_poll_timeout + max(180, mcp_call_timeout * 8))
|
|
|
|
|
|
def _graylog_fields_from_result(result: dict[str, object]) -> list[dict[str, object]]:
|
|
content = result.get("result", {}).get("content", []) if isinstance(result.get("result"), dict) else []
|
|
text = next((item.get("text", "") for item in content if isinstance(item, dict)), "")
|
|
if not text:
|
|
return []
|
|
try:
|
|
payload = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
fields = payload.get("fields", payload) if isinstance(payload, dict) else payload
|
|
if isinstance(fields, dict) and isinstance(fields.get("fields"), list):
|
|
fields = fields["fields"]
|
|
return [item for item in fields if isinstance(item, dict)] if isinstance(fields, list) else []
|
|
|
|
|
|
def _stream_coverage(runtime_values: dict[str, object], stream_profiles: dict[str, object], stream_status: dict[str, object], profile_readiness: list[dict[str, object]], stream_titles: dict[str, str]) -> list[dict[str, object]]:
|
|
configured = [
|
|
item for item in runtime_values.get("graylog_streams", [])
|
|
if isinstance(item, dict) and item.get("id")
|
|
]
|
|
status_by_id = {
|
|
str(item.get("stream_id", "")): item
|
|
for item in stream_status.get("streams", [])
|
|
if isinstance(item, dict) and item.get("stream_id")
|
|
}
|
|
readiness_by_stream: dict[str, list[dict[str, object]]] = {}
|
|
for item in profile_readiness:
|
|
readiness_by_stream.setdefault(str(item.get("stream_id", "")), []).append(item)
|
|
ids = list(dict.fromkeys([str(item.get("id", "")) for item in configured] + list(stream_profiles) + list(status_by_id)))
|
|
rows = []
|
|
for stream_id in ids:
|
|
profile = stream_profiles.get(stream_id)
|
|
readiness = readiness_by_stream.get(stream_id, [])
|
|
ready_fields = sum(1 for item in readiness if item.get("ready"))
|
|
total_fields = len(readiness)
|
|
status = status_by_id.get(stream_id, {})
|
|
enabled = next((bool(item.get("enabled")) for item in configured if str(item.get("id", "")) == stream_id), False)
|
|
events_fetched = int(status.get("events_fetched", 0) or 0)
|
|
health = "not_enabled" if not enabled else "poll_budget_skipped" if status.get("error") == "skipped_poll_budget" else "partial_fetch" if status.get("partial") else "missing_profile" if not profile else "no_events" if events_fetched == 0 else "learning" if total_fields and ready_fields < total_fields else "ready" if total_fields else "profile_needs_fields"
|
|
rows.append({
|
|
"stream_id": stream_id,
|
|
"stream_name": _stream_name(stream_id, stream_titles, profile),
|
|
"enabled": enabled,
|
|
"profile": _profile_name(stream_id, stream_titles, profile) if profile else "",
|
|
"profile_ready": bool(profile),
|
|
"entity_field": ", ".join(getattr(profile, "entity_fields", ()) or (str(getattr(profile, "entity_field", "")),)) if profile else "",
|
|
"tracked_fields": len(getattr(profile, "categorical_fields", ())) + len(getattr(profile, "numeric_fields", ())) + len(getattr(profile, "relationship_fields", ())) if profile else 0,
|
|
"ready_fields": ready_fields,
|
|
"total_fields": total_fields,
|
|
"readiness": f"{ready_fields}/{total_fields}" if total_fields else "0/0",
|
|
"events_fetched": events_fetched,
|
|
"aggregate_events": int(status.get("aggregate_events", 0) or 0),
|
|
"aggregate_status": str(status.get("aggregate_status", "")),
|
|
"aggregate_schema_properties": ", ".join(str(item) for item in status.get("aggregate_schema_properties", []) if item),
|
|
"latest_event_time": str(status.get("latest_event_time", "")),
|
|
"truncated": bool(status.get("truncated")),
|
|
"partial": bool(status.get("partial")),
|
|
"raw_error": str(status.get("error", "")),
|
|
"aggregate_error": str(status.get("aggregate_error", "")),
|
|
"error": str(status.get("aggregate_error", "") or status.get("error", "")),
|
|
"health": health,
|
|
"health_detail": str(status.get("health_detail", "")) or ("No raw events returned for this stream in the current MCP poll window." if enabled and events_fetched == 0 else ""),
|
|
})
|
|
return rows
|
|
|
|
|
|
def build_status(
|
|
log_path: str,
|
|
*,
|
|
policy_path: str | None = None,
|
|
min_block_events: int = 3,
|
|
min_block_score: int = 7,
|
|
anomaly_limit: int = 20,
|
|
baseline_path: str | None = None,
|
|
config_path: str | None = None,
|
|
history_path: str | None = None,
|
|
incident_path: str | None = None,
|
|
status_cache_path: str | None = None,
|
|
run_profile_advisor: bool = True,
|
|
) -> dict[str, object]:
|
|
config_store = ConfigStore(config_path) if config_path else None
|
|
config_exists = bool(config_store and config_store.path.exists())
|
|
runtime_values = config_store.read() if config_exists and config_store else {}
|
|
runtime_config = _public_runtime_config(runtime_values) if config_store else {}
|
|
stream_profiles = parse_profiles(runtime_values.get("graylog_stream_profiles", []))
|
|
stream_titles = _stream_titles(runtime_values)
|
|
events = read_events(log_path) if Path(log_path).exists() else []
|
|
mcp_status: dict[str, object] = {"status": "not_configured"}
|
|
if runtime_values.get("log_source") == "graylog_mcp":
|
|
url, token = str(runtime_values.get("graylog_mcp_url", "")), str(runtime_values.get("graylog_mcp_token", ""))
|
|
verify_tls = bool(runtime_values.get("graylog_tls_verify", True))
|
|
if not url or not token:
|
|
mcp_status = {"status": "missing_configuration"}
|
|
events = []
|
|
else:
|
|
try:
|
|
configured_streams = runtime_values.get("graylog_streams", [])
|
|
stream_configs = [item for item in configured_streams if isinstance(item, dict) and item.get("enabled") and item.get("id")]
|
|
stream_ids = [str(item.get("id")) for item in stream_configs]
|
|
if not stream_ids:
|
|
legacy_stream = str(runtime_values.get("graylog_stream", "") or "")
|
|
if legacy_stream:
|
|
stream_configs = [{"id": legacy_stream, "title": "Graylog"}]
|
|
else:
|
|
mcp_status = {"status": "no_streams_enabled", "streams": [], "events_fetched": 0, "coverage_status": "no_streams_enabled"}
|
|
events = []
|
|
raise StopIteration
|
|
stream_statuses = []
|
|
events = []
|
|
range_seconds = _range_seconds(runtime_values.get("graylog_range_seconds", 300))
|
|
max_events_per_stream = max(1, int(runtime_values.get("graylog_max_events_per_stream", 5000) or 5000))
|
|
raw_sample_events = max(1, int(runtime_values.get("graylog_raw_sample_events", 5000) or 5000))
|
|
mcp_call_timeout = max(1, int(runtime_values.get("graylog_mcp_call_timeout_seconds", 8) or 8))
|
|
mcp_poll_timeout = max(60, int(runtime_values.get("graylog_mcp_poll_timeout_seconds", 240) or 240))
|
|
fetch_mode = str(runtime_values.get("graylog_fetch_mode", "auto") or "auto")
|
|
use_aggregate = fetch_mode == "aggregate" or (fetch_mode == "auto" and max_events_per_stream > raw_sample_events)
|
|
aggregate_events_total = 0
|
|
poll_started_monotonic = time.monotonic()
|
|
poll_deadline = poll_started_monotonic + mcp_poll_timeout
|
|
client = GraylogMcpClient(url, token, timeout=mcp_call_timeout, verify_tls=verify_tls)
|
|
probe_status = client.probe()
|
|
discovery_store = FieldDiscoveryStore(history_path) if history_path else None
|
|
catalog_fields_total = 0
|
|
def budget_exceeded() -> bool:
|
|
return time.monotonic() >= poll_deadline
|
|
|
|
def skipped_status(stream_id: str, stream_name: str) -> dict[str, object]:
|
|
return {
|
|
"stream_id": stream_id,
|
|
"stream_name": stream_name,
|
|
"source": "graylog_mcp",
|
|
"events_fetched": 0,
|
|
"aggregate_events": 0,
|
|
"pages": 0,
|
|
"partial": True,
|
|
"error": "skipped_poll_budget",
|
|
"truncated": False,
|
|
"latest_event_time": "",
|
|
"raw_sample_limit": min(raw_sample_events, 10_000) if use_aggregate else max_events_per_stream,
|
|
"health_detail": "Skipped because the MCP poll time budget was reached before this stream.",
|
|
}
|
|
|
|
for stream_config in stream_configs:
|
|
stream_id = str(stream_config["id"])
|
|
stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id)
|
|
if budget_exceeded():
|
|
stream_statuses.append(skipped_status(stream_id, stream_name))
|
|
continue
|
|
profile = stream_profiles.get(stream_id)
|
|
profile_fields = (
|
|
str(getattr(profile, "entity_field", "")),
|
|
*tuple(str(field) for field in getattr(profile, "entity_fields", ())),
|
|
str(getattr(profile, "timestamp_field", "")),
|
|
*tuple(str(field) for field in getattr(profile, "categorical_fields", ())),
|
|
*tuple(str(field) for field 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 profile else ()
|
|
if discovery_store and not budget_exceeded():
|
|
try:
|
|
catalog_fields_total += discovery_store.ingest_catalog(stream_id, stream_name, _graylog_fields_from_result(client.call_tool("list_fields", {"streams": [stream_id]})))
|
|
except RuntimeError:
|
|
pass
|
|
aggregate_status: dict[str, object] = {}
|
|
if use_aggregate and not budget_exceeded():
|
|
aggregate_status = GraylogAggregateSource(client, stream_id, str(runtime_values.get("graylog_query", "*"))).fetch_count(range_seconds=range_seconds, probe_status=probe_status, deadline_monotonic=poll_deadline)
|
|
aggregate_events_total += int(aggregate_status.get("aggregate_events", 0) or 0)
|
|
raw_limit = min(raw_sample_events, 10_000) if use_aggregate else max_events_per_stream
|
|
if budget_exceeded():
|
|
stream_statuses.append({**skipped_status(stream_id, stream_name), **aggregate_status})
|
|
continue
|
|
stream_events, stream_status = GraylogStreamSource(client, stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", "")), stream_name, profile_fields).fetch(max_events=raw_limit, range_seconds=range_seconds, probe_status=probe_status, deadline_monotonic=poll_deadline)
|
|
events.extend(stream_events)
|
|
stream_statuses.append({"stream_id": stream_id, "stream_name": stream_name, **aggregate_status, **stream_status, "raw_sample_limit": raw_limit})
|
|
sample_limited_streams = [item for item in stream_statuses if item.get("truncated") and use_aggregate]
|
|
truncated_streams = [item for item in stream_statuses if item.get("truncated") and not use_aggregate]
|
|
partial_streams = [item for item in stream_statuses if item.get("partial")]
|
|
skipped_streams = [item for item in stream_statuses if item.get("error") == "skipped_poll_budget"]
|
|
aggregate_errors = [item for item in stream_statuses if item.get("aggregate_status") == "error"]
|
|
warnings = []
|
|
if skipped_streams:
|
|
warnings.append(f"{len(skipped_streams)} stream(s) skipped because the MCP poll time budget was reached.")
|
|
if aggregate_errors:
|
|
warnings.append(f"{len(aggregate_errors)} stream(s) returned aggregate MCP errors.")
|
|
if partial_streams:
|
|
warnings.append(f"{len(partial_streams)} stream(s) returned a partial MCP fetch; Graylog likely timed out or rejected a large paged query.")
|
|
if truncated_streams:
|
|
warnings.append(f"{len(truncated_streams)} stream(s) hit max_events_per_stream; high EPS means the analysis window is only partially sampled.")
|
|
mcp_status = {
|
|
"status": "partial" if partial_streams or aggregate_errors else "connected",
|
|
"streams": stream_statuses,
|
|
"events_fetched": len(events),
|
|
"raw_events_fetched": len(events),
|
|
"aggregate_events": aggregate_events_total,
|
|
"poll_completed_at": int(time.time()),
|
|
"poll_duration_seconds": round(time.monotonic() - poll_started_monotonic, 2),
|
|
"fetch_mode": "aggregate" if use_aggregate else "raw",
|
|
"range_seconds": range_seconds,
|
|
"max_events_per_stream": max_events_per_stream,
|
|
"raw_sample_events": raw_sample_events,
|
|
"call_timeout_seconds": mcp_call_timeout,
|
|
"poll_timeout_seconds": mcp_poll_timeout,
|
|
"partial_streams": len(partial_streams),
|
|
"skipped_streams": len(skipped_streams),
|
|
"sample_limited_streams": len(sample_limited_streams),
|
|
"truncated_streams": len(truncated_streams),
|
|
"aggregate_error_streams": len(aggregate_errors),
|
|
"catalog_fields": catalog_fields_total,
|
|
"coverage_status": "partial" if partial_streams else "truncated" if truncated_streams else "complete_window",
|
|
"coverage_warning": " ".join(warnings),
|
|
}
|
|
except StopIteration:
|
|
pass
|
|
except RuntimeError as exc:
|
|
mcp_status = {"status": "error", "error": str(exc)}
|
|
events = []
|
|
baseline = BaselineStore(baseline_path) if baseline_path else None
|
|
profiles = baseline.profiles({event.src_ip for event in events if event.src_ip}) if baseline else {}
|
|
baseline_training_days = int(runtime_values.get("baseline_training_days", 7) or 7)
|
|
field_deviations = baseline.profile_deviations(events, stream_profiles, min_training_days=baseline_training_days) if baseline else {}
|
|
sequence_findings = detect_sequences(events)
|
|
for entity, findings in sequence_findings.items():
|
|
field_deviations.setdefault(entity, []).extend(findings)
|
|
for deviations in field_deviations.values():
|
|
for deviation in deviations:
|
|
stream_id = str(deviation.get("stream_id", ""))
|
|
profile = stream_profiles.get(stream_id)
|
|
name = _stream_name(stream_id, stream_titles, profile)
|
|
deviation["stream_name"] = name
|
|
deviation["stream_title"] = name
|
|
deviation["profile_name"] = _profile_name(stream_id, stream_titles, profile)
|
|
deviation["sample_events"] = [
|
|
{"stream": name, **sample} if isinstance(sample, dict) and not sample.get("stream") else sample
|
|
for sample in deviation.get("sample_events", [])
|
|
]
|
|
feedback = FeedbackStore().entries()
|
|
for entity, deviations in field_deviations.items():
|
|
for deviation in deviations:
|
|
stream_id = str(deviation.get("stream_id", ""))
|
|
field = str(deviation.get("field", ""))
|
|
value = str(deviation.get("value", ""))
|
|
candidates = [
|
|
item for item in feedback
|
|
if feedback_matches(item, entity=entity, stream_id=stream_id, field=field, value=value)
|
|
]
|
|
if candidates:
|
|
match = select_match(candidates)
|
|
deviation["feedback"] = match["status"]
|
|
deviation["feedback_rule_id"] = match.get("id", "")
|
|
if match["status"] in SUPPRESSING_STATUSES:
|
|
deviation["score"] = 0
|
|
_cluster_field_deviations(field_deviations)
|
|
generalization_min_values = int(runtime_values.get("feedback_generalization_min_values", 3) or 3)
|
|
feedback_generalization_suggestions = suggest_generalizations(feedback, min_values=generalization_min_values)
|
|
anomalies = detect_source_anomalies(events, limit=anomaly_limit, baselines=profiles, field_deviations=field_deviations, feedback=feedback)
|
|
baseline_events = baseline.ingest(events) if baseline else 0
|
|
profile_baseline_fields = (
|
|
baseline.ingest_profile_fields(
|
|
events,
|
|
stream_profiles,
|
|
max_values_per_field=int(runtime_values.get("baseline_max_values_per_field", 500) or 500),
|
|
)
|
|
if baseline else 0
|
|
)
|
|
baseline_maintenance = (
|
|
baseline.maintenance(
|
|
retention_days=int(runtime_values.get("baseline_retention_days", 7) or 7),
|
|
value_retention_days=int(runtime_values.get("baseline_value_retention_days", 3) or 3),
|
|
max_values_per_field=int(runtime_values.get("baseline_max_values_per_field", 500) or 500),
|
|
)
|
|
if baseline
|
|
else {}
|
|
)
|
|
profile_readiness = baseline.profile_readiness(stream_profiles, min_training_days=baseline_training_days) if baseline else []
|
|
profile_readiness = [
|
|
{
|
|
**item,
|
|
"profile_name": _profile_name(str(item.get("stream_id", "")), stream_titles, stream_profiles.get(str(item.get("stream_id", "")))),
|
|
"stream_name": _stream_name(str(item.get("stream_id", "")), stream_titles, stream_profiles.get(str(item.get("stream_id", "")))),
|
|
}
|
|
for item in profile_readiness
|
|
]
|
|
stream_coverage = _stream_coverage(runtime_values, stream_profiles, mcp_status, profile_readiness, stream_titles)
|
|
discovery_cache_events = []
|
|
discovered_profile_fields = 0
|
|
discovery_store = None
|
|
if history_path:
|
|
discovery_store = FieldDiscoveryStore(history_path)
|
|
discovered_profile_fields = discovery_store.ingest(events)
|
|
discovery_cache_events = discovery_store.synthetic_events()
|
|
profile_suggestions = suggest_stream_profiles([*discovery_cache_events, *events], existing_profiles=stream_profiles)
|
|
confidence_threshold = int(runtime_values.get("field_confidence_threshold", 60) or 60)
|
|
profile_advisor_status = {"enabled": bool(runtime_values.get("profile_advisor_enabled")), "status": "disabled"}
|
|
advisor_candidates = [item for item in profile_suggestions if _needs_profile_advisor(item, stream_profiles, confidence_threshold=confidence_threshold)]
|
|
ambiguous_fields_by_stream = {
|
|
str(item.get("stream_id", "")): _ambiguous_fields(item, confidence_threshold)
|
|
for item in advisor_candidates
|
|
}
|
|
if runtime_values.get("profile_advisor_enabled") and not run_profile_advisor:
|
|
profile_advisor_status = {"enabled": True, "status": "pending" if advisor_candidates else "skipped_no_profile_changes", "candidates": len(advisor_candidates)}
|
|
elif runtime_values.get("profile_advisor_enabled") and profile_suggestions and not advisor_candidates:
|
|
profile_advisor_status = {"enabled": True, "status": "skipped_no_profile_changes", "reason": "No missing profiles or newly discovered profile fields need advisor review."}
|
|
elif runtime_values.get("profile_advisor_enabled") and profile_suggestions:
|
|
try:
|
|
advice = ollama_profile_advice(
|
|
advisor_candidates[:10],
|
|
model=str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b"),
|
|
timeout=int(runtime_values.get("profile_advisor_timeout", 240) or 240),
|
|
ambiguous_fields_by_stream=ambiguous_fields_by_stream,
|
|
)
|
|
profile_suggestions = apply_profile_advice(profile_suggestions, advice)
|
|
profile_advisor_status = {"enabled": True, "status": "ok" if advice else "empty", "profiles_returned": len(advice), "model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b")}
|
|
except Exception as exc:
|
|
profile_advisor_status = {
|
|
"enabled": True,
|
|
"status": "heuristic_fallback",
|
|
"error": str(exc),
|
|
"model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b"),
|
|
"candidates": len(advisor_candidates),
|
|
}
|
|
for suggestion in profile_suggestions:
|
|
suggestion.setdefault("profile_advisor", {"status": "heuristic", "error": str(exc)})
|
|
auto_applied_streams = _auto_apply_profile_suggestions(profile_suggestions, stream_profiles, runtime_values, config_path)
|
|
if auto_applied_streams:
|
|
if discovery_store:
|
|
suggestions_by_stream = {str(item.get("stream_id", "")): item for item in profile_suggestions if isinstance(item, dict)}
|
|
for stream_id in auto_applied_streams:
|
|
suggestion = suggestions_by_stream.get(stream_id)
|
|
if not suggestion:
|
|
continue
|
|
profile = suggestion.get("profile", {})
|
|
if not isinstance(profile, dict):
|
|
continue
|
|
applied_fields = [*profile.get("entity_fields", []), *profile.get("categorical_fields", []), *profile.get("numeric_fields", [])]
|
|
discovery_store.record_applied_confidence(
|
|
stream_id,
|
|
applied_fields,
|
|
suggestion.get("field_confidence", {}) if isinstance(suggestion.get("field_confidence"), dict) else {},
|
|
suggestion.get("field_shapes", {}) if isinstance(suggestion.get("field_shapes"), dict) else {},
|
|
)
|
|
if config_store:
|
|
stream_profiles = parse_profiles(config_store.read().get("graylog_stream_profiles", []))
|
|
confidence_history_maintenance = discovery_store.confidence_history_maintenance(retention_days=int(runtime_values.get("field_confidence_retention_days", 90) or 90)) if discovery_store else {}
|
|
profile_drift = _detect_profile_drift(profile_suggestions, stream_profiles, discovery_store)
|
|
intel_ips = sorted(
|
|
{
|
|
ip
|
|
for event in events
|
|
for ip in (event.src_ip, event.dst_ip)
|
|
if is_public_ip(ip)
|
|
}
|
|
)
|
|
threat_enabled = bool(runtime_values.get("threat_intel_enabled")) if runtime_values else None
|
|
reputation = enrich_ips(intel_ips, limit=25, enabled=threat_enabled, config=runtime_values)
|
|
threat_intel_status = ThreatIntelClient(
|
|
enabled=threat_enabled,
|
|
provider=str(runtime_values.get("threat_intel_provider", "auto")),
|
|
abuseipdb_key=str(runtime_values.get("abuseipdb_api_key", "") or "") or None,
|
|
virustotal_key=str(runtime_values.get("virustotal_api_key", "") or "") or None,
|
|
daily_limit=int(runtime_values.get("threat_intel_daily_limit", 100) or 100),
|
|
ttl_seconds=int(runtime_values.get("threat_intel_ttl_seconds", 604800) or 604800),
|
|
error_ttl_seconds=int(runtime_values.get("threat_intel_error_ttl_seconds", 3600) or 3600),
|
|
abuseipdb_max_age_days=int(runtime_values.get("abuseipdb_max_age_days", 90) or 90),
|
|
).status()
|
|
recommendations = build_recommendations(events, anomalies, reputation)
|
|
correlations = correlate_source_ips(events)
|
|
incidents = IncidentStore(incident_path or "state/signalscope-incidents.json").apply(build_incidents(anomalies, field_deviations, correlations))
|
|
triage_queue = build_triage_queue(incidents, field_deviations, correlations, recommendations)
|
|
block_candidates = suggest_block_candidates(
|
|
events,
|
|
min_events=min_block_events,
|
|
min_score=min_block_score,
|
|
allowlist=parse_allowlist(),
|
|
)
|
|
|
|
policy_findings: list[dict[str, str | None]] = []
|
|
policy_error: str | None = None
|
|
if policy_path and Path(policy_path).exists():
|
|
try:
|
|
policy_findings = [finding.__dict__ for finding in audit_policies(read_policies(policy_path))]
|
|
except Exception as exc:
|
|
policy_error = str(exc)
|
|
|
|
summary = summarize_events(events)
|
|
if isinstance(mcp_status, dict) and int(mcp_status.get("aggregate_events", 0) or 0) > summary.get("total", 0):
|
|
summary["total"] = int(mcp_status.get("aggregate_events", 0) or 0)
|
|
summary["raw_sample_total"] = len(events)
|
|
summary["aggregate_backed"] = True
|
|
profile_shape_warnings = [
|
|
{"stream_id": suggestion.get("stream_id", ""), "stream_name": suggestion.get("stream_name", ""), **warning}
|
|
for suggestion in profile_suggestions
|
|
for warning in suggestion.get("shape_warnings", [])
|
|
]
|
|
status = {
|
|
"status_schema": 2,
|
|
"stale": False,
|
|
"generated_at": int(time.time()),
|
|
"log_path": log_path,
|
|
"policy_path": policy_path,
|
|
"summary": summary,
|
|
"anomaly_summary": anomaly_summary(anomalies),
|
|
"baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "training_days": baseline_training_days, "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields, "discovery_fields_recorded": discovered_profile_fields, "discovery_cache_events": len(discovery_cache_events), "maintenance": baseline_maintenance, "size_bytes": baseline_maintenance.get("size_bytes", 0) if isinstance(baseline_maintenance, dict) else 0},
|
|
"capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status, "profile_advisor": profile_advisor_status},
|
|
"configuration": runtime_config,
|
|
"stream_profiles": [{"stream_id": item.stream_id, "name": _profile_name(item.stream_id, stream_titles, item), "stream_name": _stream_name(item.stream_id, stream_titles, item), "entity_field": item.entity_field, "entity_fields": list(item.entity_fields), "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors, "field_weights": item.field_weights, "relationship_fields": [{"left": relation.left, "right": relation.right, "name": relation.name} for relation in item.relationship_fields], "field_origin": item.field_origin, "updated_at": item.updated_at} for item in stream_profiles.values()],
|
|
"profile_auto_apply": {"applied_streams": auto_applied_streams, "confidence_history_maintenance": confidence_history_maintenance},
|
|
"profile_drift": profile_drift,
|
|
"stream_coverage": stream_coverage,
|
|
"profile_suggestions": profile_suggestions,
|
|
"profile_shape_warnings": profile_shape_warnings,
|
|
"profile_readiness": profile_readiness,
|
|
"diagnostics": {
|
|
"top_source_ips": top_field_values(events, "srcip", limit=10),
|
|
"top_destination_ips": top_field_values(events, "dstip", limit=10),
|
|
"top_policy_ids": top_field_values(events, "policyid", limit=10),
|
|
"top_destination_ports": top_field_values(events, "dstport", limit=10),
|
|
"top_source_ports": top_field_values(events, "srcport", 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),
|
|
},
|
|
"event_context": build_event_context(events),
|
|
"field_deviations": field_deviations,
|
|
"triage_queue": triage_queue,
|
|
"sequence_findings": sequence_findings,
|
|
"feedback": feedback,
|
|
"feedback_generalization_suggestions": feedback_generalization_suggestions,
|
|
"cross_source_correlations": correlations,
|
|
"incidents": incidents,
|
|
"data_quality": assess_data_quality(events, mcp_status),
|
|
"status_cache": {"served_from_cache": False, "reason": ""},
|
|
"anomalies": [
|
|
{
|
|
"subject": finding.subject,
|
|
"score": finding.score,
|
|
"severity": finding.severity,
|
|
"confidence": finding.confidence,
|
|
"reasons": finding.reasons,
|
|
"evidence": finding.evidence,
|
|
}
|
|
for finding in anomalies
|
|
],
|
|
"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
|
|
],
|
|
"reputation": reputation,
|
|
"block_candidates": [
|
|
{"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons}
|
|
for candidate in block_candidates
|
|
],
|
|
"policy_findings": policy_findings,
|
|
"policy_error": policy_error,
|
|
}
|
|
if history_path:
|
|
history = HistoryStore(history_path)
|
|
history.record(status)
|
|
status["history"] = history.recent()
|
|
if status_cache_path and not (isinstance(mcp_status, dict) and mcp_status.get("status") == "error"):
|
|
StatusSnapshotStore(status_cache_path).save("last_good", status)
|
|
return status
|
|
|
|
|
|
def cached_status_with_error(cache_path: str, error_status: dict[str, object]) -> dict[str, object] | None:
|
|
cached = StatusSnapshotStore(cache_path).load("last_good")
|
|
if not cached:
|
|
return None
|
|
for row in cached.get("stream_coverage", []) if isinstance(cached.get("stream_coverage"), list) else []:
|
|
if isinstance(row, dict):
|
|
row.setdefault("raw_error", "")
|
|
row.setdefault("aggregate_error", "")
|
|
row.setdefault("error", row.get("aggregate_error") or row.get("raw_error") or "")
|
|
if row.get("aggregate_status") == "error" and not row.get("aggregate_error") and row.get("error"):
|
|
row["aggregate_error"] = str(row.get("error", ""))
|
|
if row.get("partial") and not row.get("raw_error") and row.get("error"):
|
|
row["raw_error"] = str(row.get("error", ""))
|
|
cached["status_schema"] = 2
|
|
cached["generated_at"] = int(time.time())
|
|
cached["stale"] = True
|
|
cached["stale_reason"] = "live_mcp_error"
|
|
capabilities = cached.setdefault("capabilities", {})
|
|
if isinstance(capabilities, dict):
|
|
capabilities["graylog_mcp"] = error_status
|
|
cached.setdefault("status_cache", {})
|
|
if isinstance(cached["status_cache"], dict):
|
|
cached["status_cache"].update({"served_from_cache": True, "reason": "live_mcp_error"})
|
|
return cached
|
|
|
|
|
|
def add_llm_assessment(status: dict[str, object], *, previous: str | None = None, model: str | None = None, timeout: int | None = None) -> None:
|
|
try:
|
|
status["llm_assessment"] = {
|
|
"enabled": True,
|
|
"status": "ok",
|
|
"generated_at": int(time.time()),
|
|
"text": ollama_dashboard_assessment(status, model=model, timeout=timeout),
|
|
}
|
|
except Exception as exc:
|
|
status["llm_assessment"] = {
|
|
"enabled": True,
|
|
"status": "cached_assessment_error" if previous else "assessment_unavailable",
|
|
"generated_at": int(time.time()),
|
|
"error": str(exc),
|
|
"text": previous or "",
|
|
}
|
|
|
|
|
|
def add_profile_advisor(status: dict[str, object], runtime_values: dict[str, object]) -> None:
|
|
profile_suggestions = status.get("profile_suggestions", [])
|
|
if not isinstance(profile_suggestions, list):
|
|
return
|
|
stream_profiles = parse_profiles(runtime_values.get("graylog_stream_profiles", []))
|
|
confidence_threshold = int(runtime_values.get("field_confidence_threshold", 60) or 60)
|
|
advisor_candidates = [item for item in profile_suggestions if isinstance(item, dict) and _needs_profile_advisor(item, stream_profiles, confidence_threshold=confidence_threshold)]
|
|
ambiguous_fields_by_stream = {
|
|
str(item.get("stream_id", "")): _ambiguous_fields(item, confidence_threshold)
|
|
for item in advisor_candidates
|
|
}
|
|
capabilities = status.setdefault("capabilities", {})
|
|
if not isinstance(capabilities, dict):
|
|
return
|
|
if not runtime_values.get("profile_advisor_enabled"):
|
|
capabilities["profile_advisor"] = {"enabled": False, "status": "disabled"}
|
|
return
|
|
if not profile_suggestions or not advisor_candidates:
|
|
capabilities["profile_advisor"] = {"enabled": True, "status": "skipped_no_profile_changes", "reason": "No missing profiles or newly discovered profile fields need advisor review."}
|
|
return
|
|
try:
|
|
advice = ollama_profile_advice(
|
|
advisor_candidates[:10],
|
|
model=str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b"),
|
|
timeout=int(runtime_values.get("profile_advisor_timeout", 240) or 240),
|
|
ambiguous_fields_by_stream=ambiguous_fields_by_stream,
|
|
)
|
|
status["profile_suggestions"] = apply_profile_advice(profile_suggestions, advice)
|
|
capabilities["profile_advisor"] = {"enabled": True, "status": "ok" if advice else "empty", "profiles_returned": len(advice), "model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b")}
|
|
except Exception as exc:
|
|
capabilities["profile_advisor"] = {
|
|
"enabled": True,
|
|
"status": "heuristic_fallback",
|
|
"error": str(exc),
|
|
"model": str(runtime_values.get("profile_advisor_model", "") or "qwen3:8b"),
|
|
"candidates": len(advisor_candidates),
|
|
}
|
|
for suggestion in profile_suggestions:
|
|
if isinstance(suggestion, dict):
|
|
suggestion.setdefault("profile_advisor", {"status": "heuristic", "error": str(exc)})
|
|
|
|
|
|
def write_status(status: dict[str, object], output: str) -> None:
|
|
output_path = Path(output)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = output_path.with_suffix(f"{output_path.suffix}.tmp")
|
|
tmp_path.write_text(json.dumps(status, indent=2, sort_keys=True), encoding="utf-8")
|
|
tmp_path.replace(output_path)
|
|
|
|
|
|
def write_refreshing_status(output: str, *, cache_path: str | None = None, call_timeout_seconds: int = 0, poll_timeout_seconds: int = 0, runtime_values: dict[str, object] | None = None) -> None:
|
|
output_path = Path(output)
|
|
now = int(time.time())
|
|
try:
|
|
current = json.loads(output_path.read_text(encoding="utf-8")) if output_path.exists() else {}
|
|
except (json.JSONDecodeError, OSError):
|
|
current = {}
|
|
if not isinstance(current, dict):
|
|
current = {}
|
|
used_cache = False
|
|
if cache_path:
|
|
cached = StatusSnapshotStore(cache_path).load("last_good")
|
|
if cached:
|
|
current = cached
|
|
used_cache = True
|
|
previous_mcp = current.get("capabilities", {}).get("graylog_mcp", {}) if isinstance(current.get("capabilities"), dict) else {}
|
|
previous_completed_at = previous_mcp.get("poll_completed_at", current.get("generated_at", 0)) if isinstance(previous_mcp, dict) else current.get("generated_at", 0)
|
|
current.setdefault("status_schema", 2)
|
|
current["generated_at"] = now
|
|
current["stale"] = False
|
|
current["stale_reason"] = ""
|
|
if runtime_values is not None:
|
|
stream_profiles = parse_profiles(runtime_values.get("graylog_stream_profiles", []))
|
|
current["configuration"] = _public_runtime_config(runtime_values)
|
|
current["stream_coverage"] = _stream_coverage(runtime_values, stream_profiles, {"streams": []}, [], _stream_titles(runtime_values))
|
|
capabilities = current.setdefault("capabilities", {})
|
|
if isinstance(capabilities, dict):
|
|
previous_events = previous_mcp.get("events_fetched", 0) if isinstance(previous_mcp, dict) else 0
|
|
previous_raw_events = previous_mcp.get("raw_events_fetched", 0) if isinstance(previous_mcp, dict) else 0
|
|
previous_aggregate_events = previous_mcp.get("aggregate_events", 0) if isinstance(previous_mcp, dict) else 0
|
|
previous_fetch_mode = previous_mcp.get("fetch_mode", "") if isinstance(previous_mcp, dict) else ""
|
|
previous_coverage_status = previous_mcp.get("coverage_status", "") if isinstance(previous_mcp, dict) else ""
|
|
capabilities["graylog_mcp"] = {
|
|
"status": "refreshing",
|
|
"previous_status": previous_mcp.get("status", "") if isinstance(previous_mcp, dict) else "",
|
|
"previous_error": previous_mcp.get("error", "") if isinstance(previous_mcp, dict) else "",
|
|
"previous_events_fetched": previous_events,
|
|
"previous_raw_events_fetched": previous_raw_events,
|
|
"previous_aggregate_events": previous_aggregate_events,
|
|
"previous_fetch_mode": previous_fetch_mode,
|
|
"previous_coverage_status": previous_coverage_status,
|
|
"events_fetched": previous_events,
|
|
"raw_events_fetched": previous_raw_events,
|
|
"aggregate_events": previous_aggregate_events,
|
|
"poll_started_at": now,
|
|
"previous_poll_completed_at": previous_completed_at,
|
|
"fetch_mode": previous_fetch_mode,
|
|
"coverage_status": previous_coverage_status or "refreshing",
|
|
"call_timeout_seconds": call_timeout_seconds or previous_mcp.get("call_timeout_seconds", 0),
|
|
"poll_timeout_seconds": poll_timeout_seconds or previous_mcp.get("poll_timeout_seconds", 0),
|
|
}
|
|
current["status_cache"] = {"served_from_cache": used_cache, "reason": "refreshing"}
|
|
write_status(current, output)
|
|
|
|
|
|
def monitor_loop(
|
|
log_path: str,
|
|
output: str,
|
|
*,
|
|
policy_path: str | None = None,
|
|
interval: int = 10,
|
|
anomaly_limit: int = 20,
|
|
llm: bool = False,
|
|
llm_interval: int = 300,
|
|
llm_model: str | None = None,
|
|
llm_timeout: int | None = None,
|
|
baseline_path: str | None = None,
|
|
config_path: str | None = None,
|
|
history_path: str | None = None,
|
|
status_cache_path: str | None = None,
|
|
) -> None:
|
|
print(f"Monitoring {log_path}", flush=True)
|
|
print(f"Writing status to {output}", flush=True)
|
|
last_llm_at = 0
|
|
last_llm_text: str | None = None
|
|
last_advisor_at = 0
|
|
while True:
|
|
runtime = ConfigStore(config_path).read() if config_path and Path(config_path).exists() else {}
|
|
effective_llm = bool(runtime.get("llm_enabled")) if runtime else llm
|
|
effective_model = str(runtime.get("llm_model") or llm_model or "")
|
|
effective_llm_timeout = int(runtime.get("llm_timeout") or llm_timeout or 180)
|
|
mcp_call_timeout = max(1, int(runtime.get("graylog_mcp_call_timeout_seconds", 8) or 8))
|
|
mcp_poll_timeout = max(60, int(runtime.get("graylog_mcp_poll_timeout_seconds", 240) or 240))
|
|
if runtime.get("log_source") == "graylog_mcp":
|
|
write_refreshing_status(output, cache_path=status_cache_path, call_timeout_seconds=mcp_call_timeout, poll_timeout_seconds=mcp_poll_timeout, runtime_values=runtime)
|
|
try:
|
|
status_timeout = monitor_cycle_timeout_seconds(mcp_poll_timeout, mcp_call_timeout)
|
|
with _cycle_timeout(status_timeout if runtime.get("log_source") == "graylog_mcp" else 0):
|
|
status = build_status(
|
|
log_path, policy_path=policy_path, anomaly_limit=anomaly_limit,
|
|
baseline_path=baseline_path, config_path=config_path, history_path=history_path,
|
|
status_cache_path=status_cache_path,
|
|
run_profile_advisor=False,
|
|
)
|
|
except Exception as exc:
|
|
error_status = {"status": "error", "error": f"monitor_error: {exc}", "call_timeout_seconds": mcp_call_timeout, "poll_timeout_seconds": mcp_poll_timeout}
|
|
status = cached_status_with_error(status_cache_path, error_status) if status_cache_path else None
|
|
if not status:
|
|
status = {
|
|
"status_schema": 2,
|
|
"generated_at": int(time.time()),
|
|
"stale": True,
|
|
"stale_reason": "monitor_error",
|
|
"capabilities": {"graylog_mcp": error_status},
|
|
"summary": {"total": 0},
|
|
"status_cache": {"served_from_cache": False, "reason": "monitor_error"},
|
|
}
|
|
mcp = status.get("capabilities", {}).get("graylog_mcp", {}) if isinstance(status.get("capabilities"), dict) else {}
|
|
if status_cache_path and isinstance(mcp, dict) and mcp.get("status") == "error":
|
|
cached = cached_status_with_error(status_cache_path, mcp)
|
|
if cached:
|
|
status = cached
|
|
status["llm_assessment"] = {
|
|
"enabled": bool(effective_llm),
|
|
"status": "cached" if last_llm_text else ("pending" if effective_llm else "disabled"),
|
|
"generated_at": last_llm_at,
|
|
"started_at": int(time.time()) if effective_llm and not last_llm_text else 0,
|
|
"model": effective_model or "",
|
|
"timeout_seconds": effective_llm_timeout if effective_llm else 0,
|
|
"text": last_llm_text or "",
|
|
}
|
|
if effective_llm:
|
|
if status_cache_path and isinstance(mcp, dict) and mcp.get("status") not in {"error", "refreshing"}:
|
|
StatusSnapshotStore(status_cache_path).save("last_good", status)
|
|
write_status(status, output)
|
|
now = int(time.time())
|
|
if now - last_llm_at >= llm_interval:
|
|
add_llm_assessment(status, previous=last_llm_text, model=effective_model or None, timeout=effective_llm_timeout)
|
|
assessment = status.get("llm_assessment", {})
|
|
if isinstance(assessment, dict):
|
|
last_llm_text = str(assessment.get("text", "") or last_llm_text or "")
|
|
last_llm_at = now
|
|
write_status(status, output)
|
|
else:
|
|
status["llm_assessment"] = {
|
|
"enabled": True,
|
|
"status": "cached",
|
|
"generated_at": last_llm_at,
|
|
"text": last_llm_text or "",
|
|
}
|
|
advisor_interval = int(runtime.get("profile_advisor_interval_seconds", 300) or 300)
|
|
if runtime.get("profile_advisor_enabled") and now - last_advisor_at >= advisor_interval:
|
|
add_profile_advisor(status, runtime)
|
|
last_advisor_at = now
|
|
write_status(status, output)
|
|
else:
|
|
status["llm_assessment"] = {"enabled": False, "status": "disabled", "text": ""}
|
|
advisor_interval = int(runtime.get("profile_advisor_interval_seconds", 300) or 300)
|
|
now = int(time.time())
|
|
if runtime.get("profile_advisor_enabled") and now - last_advisor_at >= advisor_interval:
|
|
if status_cache_path and isinstance(mcp, dict) and mcp.get("status") not in {"error", "refreshing"}:
|
|
StatusSnapshotStore(status_cache_path).save("last_good", status)
|
|
write_status(status, output)
|
|
add_profile_advisor(status, runtime)
|
|
last_advisor_at = now
|
|
if status_cache_path and isinstance(mcp, dict) and mcp.get("status") not in {"error", "refreshing"}:
|
|
StatusSnapshotStore(status_cache_path).save("last_good", status)
|
|
write_status(status, output)
|
|
time.sleep(interval)
|