from __future__ import annotations import json import time 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 from .graylog_mcp import GraylogMcpClient from .graylog_source import GraylogStreamSource from .history import HistoryStore from .incidents import build_incidents from .data_quality import assess_data_quality from .llm import ollama_dashboard_assessment 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 .recommendations import build_recommendations from .threat_intel import ThreatIntelClient, enrich_ips, is_public_ip from .stream_profiles import parse_profiles 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 _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 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, ) -> 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 = config_store.public() 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", "")) 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: stream_configs = [{"id": str(runtime_values.get("graylog_stream", "")), "title": "Graylog"}] stream_statuses = [] events = [] for stream_config in stream_configs: stream_id = str(stream_config["id"]) profile = stream_profiles.get(stream_id) profile_fields = ( str(getattr(profile, "entity_field", "")), 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", ())), ) if profile else () stream_name = str(stream_config.get("title", "") or stream_titles.get(stream_id) or stream_id) stream_events, stream_status = GraylogStreamSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", "")), stream_name, profile_fields).fetch() events.extend(stream_events) stream_statuses.append({"stream_id": stream_id, "stream_name": stream_name, **stream_status}) mcp_status = {"status": "connected", "streams": stream_statuses, "events_fetched": len(events)} 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 {} field_deviations = baseline.profile_deviations(events, stream_profiles) if baseline else {} 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: match = next(( item for item in feedback if item.get("entity") == entity and item.get("stream_id") == deviation.get("stream_id") and item.get("field") == deviation.get("field") and (not item.get("value") or item.get("value") == deviation.get("value", "")) ), None) if match: deviation["feedback"] = match["status"] if match["status"] in {"false_positive", "expected"}: deviation["score"] = 0 anomalies = detect_source_anomalies(events, limit=anomaly_limit, baselines=profiles, field_deviations=field_deviations) baseline_events = baseline.ingest(events) if baseline else 0 profile_baseline_fields = baseline.ingest_profile_fields(events, stream_profiles) if baseline else 0 profile_readiness = baseline.profile_readiness(stream_profiles) 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 ] 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) threat_intel_status = ThreatIntelClient(enabled=threat_enabled).status() recommendations = build_recommendations(events, anomalies, reputation) correlations = correlate_source_ips(events) 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) status = { "generated_at": int(time.time()), "log_path": log_path, "policy_path": policy_path, "summary": summarize_events(events), "anomaly_summary": anomaly_summary(anomalies), "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields}, "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_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, "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields), "detectors": item.detectors} for item in stream_profiles.values()], "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, "feedback": feedback, "cross_source_correlations": correlations, "incidents": build_incidents(anomalies, field_deviations, correlations), "data_quality": assess_data_quality(events, mcp_status), "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() return status 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": "error", "generated_at": int(time.time()), "error": str(exc), "text": previous or "", } 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 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, ) -> None: print(f"Monitoring {log_path}") print(f"Writing status to {output}") last_llm_at = 0 last_llm_text: str | None = None 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 "") 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, ) if effective_llm: 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=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 else: status["llm_assessment"] = { "enabled": True, "status": "cached", "generated_at": last_llm_at, "text": last_llm_text or "", } else: status["llm_assessment"] = {"enabled": False, "status": "disabled", "text": ""} write_status(status, output) time.sleep(interval)