136 lines
5.9 KiB
Python
136 lines
5.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def _list(value: object) -> list[object]:
|
|
return value if isinstance(value, list) else []
|
|
|
|
|
|
def _dict(value: object) -> dict[str, object]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _text(value: object, default: str = "-") -> str:
|
|
text = str(value or "").strip()
|
|
return text or default
|
|
|
|
|
|
def _incident_matches(incident: dict[str, object], incident_id: str | None) -> bool:
|
|
return not incident_id or str(incident.get("id", "")) == incident_id
|
|
|
|
|
|
def investigation_report(status: dict[str, object], incident_id: str | None = None) -> dict[str, object]:
|
|
incidents = [item for item in _list(status.get("incidents")) if isinstance(item, dict) and _incident_matches(item, incident_id)]
|
|
recommendations = [item for item in _list(status.get("recommendations")) if isinstance(item, dict)]
|
|
entities = {str(item.get("entity", "")) for item in incidents}
|
|
return {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"incident_id": incident_id or "",
|
|
"summary": _dict(status.get("summary")),
|
|
"anomaly_summary": _dict(status.get("anomaly_summary")),
|
|
"baseline": _dict(status.get("baseline")),
|
|
"data_quality": _dict(status.get("data_quality")),
|
|
"stream_coverage": _list(status.get("stream_coverage")),
|
|
"incidents": incidents,
|
|
"recommendations": [
|
|
item for item in recommendations if not entities or str(item.get("subject", "")) in entities
|
|
],
|
|
}
|
|
|
|
|
|
def investigation_report_markdown(report: dict[str, object]) -> str:
|
|
lines = [
|
|
"# SignalScope Investigation Report",
|
|
"",
|
|
f"Generated: {_text(report.get('generated_at'))}",
|
|
]
|
|
if report.get("incident_id"):
|
|
lines.append(f"Incident filter: `{_text(report.get('incident_id'))}`")
|
|
lines += ["", "## Summary"]
|
|
summary = _dict(report.get("summary"))
|
|
anomaly_summary = _dict(report.get("anomaly_summary"))
|
|
baseline = _dict(report.get("baseline"))
|
|
lines += [
|
|
f"- Events: {_text(summary.get('total'), '0')}",
|
|
f"- High anomalies: {_text(anomaly_summary.get('high'), '0')}",
|
|
f"- Critical anomalies: {_text(anomaly_summary.get('critical'), '0')}",
|
|
f"- Baseline sources ready: {_text(baseline.get('sources_ready'), '0')}",
|
|
]
|
|
|
|
coverage = [item for item in _list(report.get("stream_coverage")) if isinstance(item, dict)]
|
|
if coverage:
|
|
lines += ["", "## Stream Coverage", "", "| Stream | Enabled | Profile | Events | Health |", "| --- | --- | --- | ---: | --- |"]
|
|
for item in coverage[:20]:
|
|
lines.append(
|
|
"| "
|
|
+ " | ".join(
|
|
[
|
|
_text(item.get("stream_name") or item.get("stream_id")),
|
|
_text(item.get("enabled")),
|
|
_text(item.get("profile_name") or item.get("profile")),
|
|
_text(item.get("events_fetched"), "0"),
|
|
_text(item.get("health")),
|
|
]
|
|
)
|
|
+ " |"
|
|
)
|
|
|
|
incidents = [item for item in _list(report.get("incidents")) if isinstance(item, dict)]
|
|
lines += ["", "## Incidents"]
|
|
if not incidents:
|
|
lines.append("No incidents matched the report scope.")
|
|
for incident in incidents:
|
|
lines += [
|
|
"",
|
|
f"### {_text(incident.get('entity'))} ({_text(incident.get('severity'))})",
|
|
"",
|
|
f"- ID: `{_text(incident.get('id'))}`",
|
|
f"- Type: {_text(incident.get('entity_type'))}",
|
|
f"- Score: {_text(incident.get('score'), '0')}",
|
|
f"- State: {_text(incident.get('lifecycle_status'), 'open')}",
|
|
f"- Streams: {_text(', '.join(str(item) for item in _list(incident.get('correlated_streams'))))}",
|
|
f"- First seen: {_text(incident.get('first_seen'))}",
|
|
f"- Last seen: {_text(incident.get('last_seen'))}",
|
|
]
|
|
if incident.get("note"):
|
|
lines.append(f"- Analyst note: {_text(incident.get('note'))}")
|
|
evidence = [str(item) for item in _list(incident.get("evidence")) if item]
|
|
if evidence:
|
|
lines += ["", "Evidence:"]
|
|
lines.extend(f"- {item}" for item in evidence[:10])
|
|
timeline = [item for item in _list(incident.get("timeline")) if isinstance(item, dict)]
|
|
if timeline:
|
|
lines += ["", "Timeline:", "", "| Time | Stream | Action | Destination | Service | Context | Graylog query |", "| --- | --- | --- | --- | --- | --- | --- |"]
|
|
for item in timeline[:20]:
|
|
query = _dict(item.get("query_details")).get("query") if isinstance(item.get("query_details"), dict) else item.get("graylog_query")
|
|
lines.append(
|
|
"| "
|
|
+ " | ".join(
|
|
_text(value).replace("|", "\\|")
|
|
for value in (
|
|
item.get("timestamp"),
|
|
item.get("stream_name") or item.get("stream_title") or item.get("stream_id"),
|
|
item.get("action"),
|
|
item.get("destination"),
|
|
item.get("service"),
|
|
item.get("context") or item.get("message"),
|
|
query,
|
|
)
|
|
)
|
|
+ " |"
|
|
)
|
|
|
|
recommendations = [item for item in _list(report.get("recommendations")) if isinstance(item, dict)]
|
|
if recommendations:
|
|
lines += ["", "## Recommendations"]
|
|
for item in recommendations[:20]:
|
|
lines += [
|
|
"",
|
|
f"- {_text(item.get('subject'))}: {_text(item.get('title'))} ({_text(item.get('severity'))}, score {_text(item.get('score'), '0')})",
|
|
f" Recommendation: {_text(item.get('recommendation'))}",
|
|
]
|
|
|
|
return "\n".join(lines).rstrip() + "\n"
|