diff --git a/tools/desktop_notifier.py b/tools/desktop_notifier.py new file mode 100755 index 0000000..ce82f5a --- /dev/null +++ b/tools/desktop_notifier.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +""" +desktop_notifier.py - LocalObserve Host-Side Desktop Alerting Daemon +Watches the local .data/alerts directory for security alerts and triggers native notifications. +""" +import os +import sys +import time +import json +import subprocess +from pathlib import Path + +# Paths +REPO_ROOT = Path(__file__).resolve().parents[1] +ALERT_DIR = REPO_ROOT / ".data" / "alerts" + +# Severity Styling +SEVERITY_STYLES = { + "critical": { + "urgency": "critical", + "icon": "dialog-error", + "title_prefix": "🚨 CRITICAL SECURITY ALERT: ", + "sound": "bell" + }, + "high": { + "urgency": "critical", + "icon": "security-high", + "title_prefix": "⚠️ High Alert: ", + "sound": "message-new-instant" + }, + "normal": { + "urgency": "normal", + "icon": "security-medium", + "title_prefix": "Security Event: ", + "sound": "message-new-instant" + }, + "low": { + "urgency": "low", + "icon": "security-low", + "title_prefix": "Security Info: ", + "sound": None + } +} + +def play_alert_sound(sound_name): + """Play alert sound if audio utilities exist on host.""" + if not sound_name: + return + for cmd in [ + ["paplay", f"/usr/share/sounds/freedesktop/stereo/{sound_name}.oga"], + ["aplay", f"/usr/share/sounds/freedesktop/stereo/{sound_name}.wav"] + ]: + if Path(cmd[1]).exists(): + try: + subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + break + except OSError: + pass + +def send_notification(alert): + """Trigger the native host notify-send command.""" + alert_name = alert.get("alert_name", "Unknown Security Event") + description = alert.get("description", "No details provided.") + severity = alert.get("severity", "normal").lower() + + style = SEVERITY_STYLES.get(severity, SEVERITY_STYLES["normal"]) + title = f"{style['title_prefix']}{alert_name}" + + cmd = [ + "notify-send", + "--urgency", style["urgency"], + "--icon", style["icon"], + "--category", "transfer.error", + title, + description + ] + + try: + subprocess.run(cmd, check=True) + play_alert_sound(style["sound"]) + except FileNotFoundError: + print("Error: 'notify-send' is not installed or available on this system.", file=sys.stderr) + +def main(): + print(f"[*] LocalObserve Desktop Notifier watching: {ALERT_DIR}") + ALERT_DIR.mkdir(parents=True, exist_ok=True) + + seen_files = set(ALERT_DIR.glob("*.json")) + + while True: + try: + current_files = set(ALERT_DIR.glob("*.json")) + new_files = current_files - seen_files + + for file_path in new_files: + if not file_path.exists(): + continue + try: + time.sleep(0.1) + alert_data = json.loads(file_path.read_text(encoding="utf-8")) + send_notification(alert_data) + file_path.unlink() + except Exception as e: + print(f"[!] Error processing {file_path.name}: {e}", file=sys.stderr) + + seen_files = current_files - {f for f in new_files if not f.exists()} + time.sleep(1) + except KeyboardInterrupt: + print("\n[*] Exiting notifier daemon.") + break + +if __name__ == "__main__": + main()