dashy: add recent Falco alerts widget
Cron script (every 5m) pulls falco_rule_match raise transitions from netdata's existing alert_transitions API - the same data already driving the Telegram pipeline - and writes them as a Dashy custom-list JSON file. No new containers/Redis/Falcosidekick needed for a quick-glance security event feed on the dashboard. conf.yml and the crontab entry live outside the repo (per existing pattern) - conf.yml got a new custom-list widget block in the System Widgets section, and crontab got a matching */5 entry alongside the other host-cron jobs.
This commit is contained in:
Executable
+65
@@ -0,0 +1,65 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Pull recent Falco alerts from netdata and write them as a Dashy
|
||||
custom-list JSON file, so the homelab dashboard shows a live feed of
|
||||
security events without needing Falcosidekick/Redis/a new UI.
|
||||
|
||||
Run periodically via cron (see repo README/crontab entry). Reads from
|
||||
netdata's existing alert-transitions API - the same data already driving
|
||||
the Telegram pipeline - so this adds zero new attack surface or
|
||||
dependencies.
|
||||
"""
|
||||
import json
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
NETDATA_URL = "http://192.168.1.67:19999"
|
||||
OUTPUT_PATH = "/srv/dashy/json-data/falco-alerts.json"
|
||||
LOOKBACK_SECONDS = 7 * 86400 # 7 days
|
||||
MAX_ENTRIES = 20
|
||||
|
||||
|
||||
def fetch_transitions():
|
||||
now = int(time.time())
|
||||
after = now - LOOKBACK_SECONDS
|
||||
url = (
|
||||
f"{NETDATA_URL}/api/v2/alert_transitions"
|
||||
f"?after={after}&before={now}&last=500"
|
||||
)
|
||||
with urllib.request.urlopen(url, timeout=10) as resp:
|
||||
return json.load(resp).get("transitions", [])
|
||||
|
||||
|
||||
def to_dashy_entry(t):
|
||||
rule_name = t["summary"].removeprefix("Falco rule fired - ")
|
||||
status = t["new"]["status"]
|
||||
return {
|
||||
"link": {
|
||||
"text": "View",
|
||||
"url": f"{NETDATA_URL}/",
|
||||
"title": f"{status}: {rule_name}",
|
||||
},
|
||||
"value": {
|
||||
"text": f"{status} - {rule_name}",
|
||||
"title": "Falco",
|
||||
},
|
||||
"date": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t["when"])),
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
transitions = fetch_transitions()
|
||||
raised = [
|
||||
t
|
||||
for t in transitions
|
||||
if t.get("alert") == "falco_rule_match"
|
||||
and t.get("new", {}).get("status") in ("WARNING", "CRITICAL")
|
||||
]
|
||||
raised.sort(key=lambda t: t["when"], reverse=True)
|
||||
entries = [to_dashy_entry(t) for t in raised[:MAX_ENTRIES]]
|
||||
|
||||
with open(OUTPUT_PATH, "w") as f:
|
||||
json.dump(entries, f, indent=2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user