admin / Synapse-Sonar

#!/usr/bin/env python3
"""
NetscanXi -> Synapse Sonar vulnerability connector.

Reads a JSON file of findings (Synapse Sonar's documented format) and POSTs them
to /api/ingest/vulnerabilities. Use it to test the feed, run it on a schedule
(cron) against a NetscanXi export, or adapt the `load_findings` function to call
NetscanXi v13 directly once its API is available.

Pure Python 3 standard library.

Usage:
  SONAR_URL=https://sonar.corp.internal:3000 \
  SONAR_INGEST_KEY=nsx_sensor_xxxx \
  python3 sonar-netscan-connector.py findings.json

findings.json:
  { "findings": [ { "ip": "10.0.3.30", "cveId": "CVE-2024-1234",
                    "cvssScore": 9.8, "title": "..." } ] }
"""
import json
import os
import ssl
import sys
import urllib.error
import urllib.request

SONAR_URL = os.environ.get("SONAR_URL", "").rstrip("/")
INGEST_KEY = os.environ.get("SONAR_INGEST_KEY", "")
VERIFY_TLS = os.environ.get("SONAR_VERIFY_TLS", "true").lower() != "false"


def load_findings(path: str):
    """Load findings from a JSON file. Adapt this to pull from NetscanXi v13."""
    with open(path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    findings = data.get("findings", data) if isinstance(data, dict) else data
    if not isinstance(findings, list):
        raise ValueError("Expected a list of findings or {\"findings\": [...]}")
    return findings


def push(findings):
    body = json.dumps({"findings": findings[:1000]}).encode()
    req = urllib.request.Request(
        f"{SONAR_URL}/api/ingest/vulnerabilities",
        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
    with urllib.request.urlopen(req, timeout=20, context=ctx) as resp:
        print(f"[netscan-connector] HTTP {resp.status}: {resp.read().decode()}")


def main():
    if len(sys.argv) < 2:
        print("usage: sonar-netscan-connector.py <findings.json>", file=sys.stderr)
        sys.exit(2)
    if not SONAR_URL or not INGEST_KEY:
        print("SONAR_URL and SONAR_INGEST_KEY must be set", file=sys.stderr)
        sys.exit(1)
    findings = load_findings(sys.argv[1])
    if not findings:
        print("[netscan-connector] no findings to send")
        return
    try:
        push(findings)
    except urllib.error.HTTPError as e:
        print(f"[netscan-connector] error HTTP {e.code}: {e.read().decode(errors='ignore')}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()