admin / Synapse-Sonar

#!/usr/bin/env python3
"""
Synapse Sonar passive sensor — agentless flow monitoring.

For devices that cannot run an agent. Deploy this on a host connected to a
switch SPAN / mirror port (or an inline tap / gateway) so it can observe the
traffic to and from those devices. It sniffs IPv4 TCP/UDP packets, aggregates
them into flows over a window, and pushes them to the Synapse Sonar ingestion
endpoint.

Linux-only (uses AF_PACKET). Requires CAP_NET_RAW (+ CAP_NET_ADMIN for
promiscuous mode). Pure Python 3 standard library — no pip dependencies.

Config (see sonar-sensor.env.example):
  SONAR_URL          e.g. https://sonar.corp.internal:3000
  SONAR_INGEST_KEY   the "Sensor ingest key" from Integrations -> NetscanXi
  SONAR_IFACE        interface to sniff, e.g. eth1 (default: all)
  SONAR_FLUSH        seconds per aggregation window (default 30)
  SONAR_VERIFY_TLS   "false" to skip TLS verification (default true)
"""

import json
import os
import socket
import ssl
import struct
import sys
import time
import urllib.error
import urllib.request

SONAR_URL = os.environ.get("SONAR_URL", "").rstrip("/")
INGEST_KEY = os.environ.get("SONAR_INGEST_KEY", "")
IFACE = os.environ.get("SONAR_IFACE", "").strip()
FLUSH = int(os.environ.get("SONAR_FLUSH", "30"))
VERIFY_TLS = os.environ.get("SONAR_VERIFY_TLS", "true").lower() != "false"

ETH_P_ALL = 0x0003
SENSOR_HOST = socket.gethostname()


def log(msg: str) -> None:
    print(f"[sonar-sensor] {msg}", flush=True)


def resolve_sonar_ip():
    """Resolve the Sonar host IP so we can exclude our own push traffic."""
    try:
        netloc = SONAR_URL.split("://", 1)[-1].split("/", 1)[0]
        host = netloc.split("@")[-1].rsplit(":", 1)[0].strip("[]")
        return socket.gethostbyname(host)
    except Exception:
        return None


def push(flows):
    if not flows:
        return
    # Send the heaviest flows first, capped to the endpoint's batch limit.
    flows.sort(key=lambda f: f["bytes"], reverse=True)
    body = json.dumps({"flows": flows[:1000]}).encode()
    req = urllib.request.Request(
        f"{SONAR_URL}/api/ingest/flow-logs",
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {INGEST_KEY}",
        },
    )
    ctx = None
    if SONAR_URL.startswith("https") and not VERIFY_TLS:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
    try:
        with urllib.request.urlopen(req, timeout=15, context=ctx) as resp:
            log(f"pushed {len(flows[:1000])} flows -> HTTP {resp.status}")
    except urllib.error.HTTPError as e:
        log(f"push failed: HTTP {e.code} {e.read().decode(errors='ignore')[:200]}")
    except Exception as e:
        log(f"push error: {e}")


def main():
    if not SONAR_URL or not INGEST_KEY:
        log("SONAR_URL and SONAR_INGEST_KEY must be set. Exiting.")
        sys.exit(1)
    if not hasattr(socket, "AF_PACKET"):
        log("AF_PACKET unavailable — this sensor runs on Linux only. Exiting.")
        sys.exit(1)

    sonar_ip = resolve_sonar_ip()
    s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
    if IFACE:
        s.bind((IFACE, 0))
    s.settimeout(1.0)
    log(f"sniffing iface={IFACE or 'all'} window={FLUSH}s target={SONAR_URL}")

    flows = {}  # (src,dst,proto,port) -> bytes
    last_flush = time.monotonic()

    while True:
        try:
            frame = s.recv(65535)
        except socket.timeout:
            frame = None
        except Exception as e:
            log(f"recv error: {e}")
            frame = None

        if frame and len(frame) >= 34:
            eth_proto = struct.unpack("!H", frame[12:14])[0]
            if eth_proto == 0x0800:  # IPv4
                ihl = (frame[14] & 0x0F) * 4
                proto_num = frame[23]
                total_len = struct.unpack("!H", frame[16:18])[0]
                src = socket.inet_ntoa(frame[26:30])
                dst = socket.inet_ntoa(frame[30:34])
                if proto_num in (6, 17):  # TCP / UDP
                    l4 = 14 + ihl
                    if len(frame) >= l4 + 4:
                        sport = struct.unpack("!H", frame[l4:l4 + 2])[0]
                        dport = struct.unpack("!H", frame[l4 + 2:l4 + 4])[0]
                        proto = "tcp" if proto_num == 6 else "udp"
                        # Skip our own push traffic to avoid a feedback loop.
                        if sonar_ip and (src == sonar_ip or dst == sonar_ip):
                            pass
                        else:
                            port = min(sport, dport)  # service port heuristic
                            key = (src, dst, proto, port)
                            flows[key] = flows.get(key, 0) + total_len

        now = time.monotonic()
        if now - last_flush >= FLUSH:
            batch = [
                {
                    "sourceIp": k[0],
                    "destIp": k[1],
                    "protocol": k[2],
                    "port": k[3],
                    "bytes": v,
                    "metadata": {"collector": "sensor", "sensor": SENSOR_HOST},
                }
                for k, v in flows.items()
            ]
            push(batch)
            flows.clear()
            last_flush = now


if __name__ == "__main__":
    main()