===== zenarmor ===== chmod +x /usr/local/etc/rc.syshook.d/start/99-zenarmor-stream /usr/local/etc/rc.syshook.d/start/99-zenarmor-stream & /usr/local/etc/rc.syshook.d/start/99-zenarmor-stream #!/usr/local/bin/python3 import json import time import subprocess import urllib.request import os # Telegram Configuration Settings TELEGRAM_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN" CHAT_ID = "YOUR_TELEGRAM_CHAT_ID" MUTE_KEYWORDS = [ "ad tracker", "ads", "social", "policy drop", "leiniao", ] # Target the master global search mapping for all live local indices ES_URL = "http://127.0.0.1:9200/_search" query = { "size": 15, "query": { "bool": { "must": [ {"term": {"is_blocked": 1}} ], "filter": { "range": { "start_time": { "gte": "now-15s" } } } } }, "sort": [{"start_time": {"order": "desc"}}] } # Use a dictionary to consolidate message details across duplicate connections alert_cache = {} print("[*] Deduplicating Real-Time Zenarmor Database Listener engaged.") def send_telegram_alert(message): url = f"https://telegram.org{TELEGRAM_TOKEN}/sendMessage" payload = { "chat_id": CHAT_ID, "text": message, "parse_mode": "HTML" } try: req = urllib.request.Request( url, data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST' ) with urllib.request.urlopen(req, timeout=5) as response: if response.status != 200: print(f"[-] Telegram API Error status code: {response.status}") except Exception as e: print(f"[-] Failed to dispatch Telegram broadcast payload: {e}") while True: try: req = urllib.request.Request( ES_URL, data=json.dumps(query).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST' ) with urllib.request.urlopen(req) as response: data = json.loads(response.read().decode('utf-8')) hits = data.get('hits', {}).get('hits', []) current_batch = {} for hit in hits: source = hit.get('_source', {}) src_ip = source.get('ip_src_saddr', 'Unknown') dst_ip = source.get('ip_dst_saddr', 'Unknown') dst_port = source.get('ip_dst_port', '') dst_host = source.get('dst_hostname', '') block_msg = source.get('message', '').strip() if not block_msg or block_msg.lower() == "null": block_msg = "Policy Drop" destination = f"{dst_host}:{dst_port}" if dst_host else f"{dst_ip}:{dst_port}" # Create a unique key mapping for this specific socket connection flow flow_key = f"{src_ip}->{destination}" # Parse optional signature details sig_name = "" alert_info_raw = source.get('alertinfo', '{}') if isinstance(alert_info_raw, str): try: alert_info = json.loads(alert_info_raw) sig_name = alert_info.get('signature_name', '') except json.JSONDecodeError: sig_name = "" elif isinstance(alert_info_raw, dict): sig_name = alert_info_raw.get('signature_name', '') reason = sig_name if sig_name else block_msg # Implement Keyword Suppression Filtering Loop should_mute = False for keyword in MUTE_KEYWORDS: if keyword.lower() in reason.lower(): should_mute = True break if should_mute: continue # Store the most relevant, highest priority alert descriptor message details current_batch[flow_key] = { "src_ip": src_ip, "destination": destination, "reason": reason } # Loop through the compiled batch data array to evaluate deduplication caches for flow_key, details in current_batch.items(): current_time = time.time() # Suppress flood triggers if exact matching traffic was announced within past 5 mins if flow_key in alert_cache and (current_time - alert_cache[flow_key]) < 300: continue # Update state anchors and format clean HTML alerts alert_cache[flow_key] = current_time alert_msg = ( f"🛡️ Zenarmor Security Alarm 🛡️\n\n" f"🚨 Action: BLOCKED\n" f"💥 Reason: {details['reason']}\n" f"🖥️ Source IP: {details['src_ip']}\n" f"🌐 Destination: {details['destination']}" ) send_telegram_alert(alert_msg) # Evacuate old stale entries out of cache loop to prevent excessive memory creep stale_keys = [k for k, v in alert_cache.items() if (time.time() - v) > 600] for k in stale_keys: del alert_cache[k] except Exception as e: print(f"[-] Execution tracking error encountered inside main engine: {e}") # Pause execution for 10 seconds before polling Elasticsearch again time.sleep(10)