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"🛡️ <b>Zenarmor Security Alarm</b> 🛡️\n\n"
                    f"🚨 <b>Action:</b> BLOCKED\n"
                    f"💥 <b>Reason:</b> <code>{details['reason']}

\n“

                  f"🖥️ <b>Source IP:</b> <code>{details['src_ip']}</code>\n"
                  f"🌐 <b>Destination:</b> <code>{details['destination']}</code>"
              )

              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)

</code>