66 lines
2.0 KiB
Python
Executable File
66 lines
2.0 KiB
Python
Executable File
#!/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 datetime
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
from zoneinfo import ZoneInfo
|
|
|
|
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 = 5
|
|
LOCAL_TZ = ZoneInfo("America/New_York")
|
|
|
|
|
|
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"]
|
|
when = datetime.datetime.fromtimestamp(t["when"], LOCAL_TZ).strftime("%-m/%-d %-I:%M%p").lower()
|
|
return {
|
|
"link": {
|
|
"text": f"{when} - {status} - {rule_name}",
|
|
"url": f"{NETDATA_URL}/alerts",
|
|
"title": f"{status}: {rule_name}",
|
|
},
|
|
"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()
|