admin / Synapse-Sonar

#!/usr/bin/env python3
"""
Synapse Sonar host agent — Debian / Ubuntu.

Lightweight, dependency-free (Python 3 standard library only). Periodically reads
this host's established connections via `ss` (iproute2) and pushes them as flow
records to the Synapse Sonar ingestion endpoint.

Config is read from environment variables (see sonar-agent.env.example):
  SONAR_URL          e.g. https://sonar.corp.internal:3000
  SONAR_INGEST_KEY   the "Sensor ingest key" from Integrations -> NetscanXi
  SONAR_INTERVAL     seconds between pushes (default 30)
  SONAR_VERIFY_TLS   "false" to skip TLS verification (self-signed) (default true)
"""

import json
import os
import socket
import ssl
import subprocess
import sys
import re
import time
import urllib.error
import urllib.request

SONAR_URL = os.environ.get("SONAR_URL", "").rstrip("/")
INGEST_KEY = os.environ.get("SONAR_INGEST_KEY", "")
INTERVAL = int(os.environ.get("SONAR_INTERVAL", "30"))
VERIFY_TLS = os.environ.get("SONAR_VERIFY_TLS", "true").lower() != "false"
HOSTNAME = socket.gethostname()

_IPV4 = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$")


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


def local_ipv4s():
    """All IPv4 addresses configured on this host (incl. Docker bridges)."""
    try:
        out = subprocess.run(["hostname", "-I"], capture_output=True, text=True, timeout=5).stdout
        return [t for t in out.split() if _IPV4.match(t)]
    except Exception:
        return []


def is_dockerish(ip: str) -> bool:
    # Docker default bridges live in 172.16.0.0/12; also skip link-local.
    return ip.startswith("172.") or ip.startswith("169.254.")


def host_addresses():
    """Return (primary_ipv4, related_ipv4s).

    The primary is the address used for the default route (the host's "real" IP);
    related are the host's other addresses (Docker bridge, secondary NICs) that
    should be shown against the same node rather than as separate nodes.
    """
    locals_ = local_ipv4s()
    primary = None
    try:
        out = subprocess.run(
            ["ip", "-4", "route", "get", "1.1.1.1"], capture_output=True, text=True, timeout=5
        ).stdout
        m = re.search(r"\bsrc\s+(\d{1,3}(?:\.\d{1,3}){3})", out)
        if m:
            primary = m.group(1)
    except Exception:
        pass
    if not primary:
        non_docker = [ip for ip in locals_ if not is_dockerish(ip) and not ip.startswith("127.")]
        primary = non_docker[0] if non_docker else (locals_[0] if locals_ else "127.0.0.1")
    related = [ip for ip in locals_ if ip != primary and not ip.startswith("127.")]
    return primary, related


def parse_addr(token: str):
    """Split '1.2.3.4:5678' or '[::1]:5678' into (host, port:int|None)."""
    if token.startswith("["):
        host, _, port = token[1:].partition("]:")
    else:
        host, _, port = token.rpartition(":")
    try:
        return host, int(port)
    except ValueError:
        return host, None


def collect_flows():
    """Return a de-duplicated list of flow dicts for established connections.

    Every flow is reported with the host's PRIMARY IPv4 as the source so the
    host appears as a single node, regardless of which local interface (e.g. a
    Docker bridge) actually owns the socket. The host's other addresses are sent
    as sourceRelatedIps and shown against the same node.
    """
    primary, related = host_addresses()
    flows = {}
    try:
        out = subprocess.run(
            ["ss", "-H", "-tu", "-n", "state", "established"],
            capture_output=True, text=True, timeout=10,
        ).stdout
    except Exception as e:  # ss missing or failed
        log(f"ss failed: {e}")
        return []

    for line in out.splitlines():
        parts = line.split()
        if len(parts) < 2:
            continue
        proto = parts[0] if parts[0] in ("tcp", "udp") else "tcp"
        # Local and peer are the last two address tokens on the line.
        local_host, local_port = parse_addr(parts[-2])
        peer_host, peer_port = parse_addr(parts[-1])
        if not peer_host or peer_port is None or local_port is None:
            continue
        # Skip loopback / link-local noise.
        if peer_host.startswith("127.") or peer_host in ("::1", "*") or peer_host.startswith("fe80"):
            continue
        # Use the lower port as the "service" port (well-known service vs ephemeral).
        service_port = min(local_port, peer_port)
        # Source is consolidated onto the primary host IP, so key by peer only.
        key = (peer_host, proto, service_port)
        if key in flows:
            continue
        flows[key] = {
            "sourceIp": primary,
            "destIp": peer_host,
            "sourceHostname": HOSTNAME,
            "sourceRelatedIps": related,
            "protocol": proto,
            "port": service_port,
            "bytes": 0,  # ss does not give reliable cumulative bytes per flow
            "metadata": {"collector": "host-agent", "host": HOSTNAME},
        }
    return list(flows.values())


def push(flows):
    if not flows:
        return
    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)} 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)
    log(f"starting; target={SONAR_URL} interval={INTERVAL}s host={HOSTNAME}")
    while True:
        push(collect_flows())
        time.sleep(INTERVAL)


if __name__ == "__main__":
    main()