Skip to content

eventbridge-api.md

Latest commit

 

History

History
773 lines (620 loc) · 24.3 KB

eventbridge-api.md

File metadata and controls

773 lines (620 loc) · 24.3 KB
Raw

EventBridge API

Synapse in agent mode captures JA4+ fingerprints for every TCP connection seen on the wire (via XDP/eBPF, AF_PACKET, or PF_RING). The EventBridge streams these fingerprint events over a UNIX socket so that 3rd party applications can enrich their own connections with fingerprint, GeoIP, and threat intelligence data.

Use Case

Your application (web server, reverse proxy, API gateway) receives an incoming connection. You know the client's src_ip and src_port. You want to know:

  • What OS/TCP stack is the client running? (JA4T)
  • What TLS library? (JA4)
  • What TLS server response? (JA4S)
  • Where is the client located? (GeoIP)
  • Is this IP known-malicious? (threat score)
  • Was this connection blocked? (block status)

Synapse provides two sockets:

Socket Default path Purpose
Event socket /var/run/synapse-events.sock Real-time event stream (broadcast)
Control socket /var/run/synapse-control.sock Query-response API (lookup by connection)

For the use case above, use the control socket. It's a simple request-response: send a query, get back fingerprints.

Control Socket (Query-Response)

Connection

Platform Default
Linux / macOS /var/run/synapse-control.sock (UNIX socket)
Windows 127.0.0.1:19198 (TCP)

Protocol

HTTP/1.1 POST over the UNIX socket. One endpoint: POST /query.

POST /query HTTP/1.1
Content-Type: application/json
Content-Length: 52

{"src_ip":"203.0.113.42","src_port":54321}

The connection closes after the response (HTTP Connection: close).

Request body

POST JSON to /query with these fields:

Required:

Field Type Description
src_ip string Client IP address
src_port number Client source port

Optional (narrow your match):

Field Type Description
dst_ip string Your server's IP address
dst_port number Your server's listening port
ja4 string Match specific TLS fingerprint
ja4t string Match specific TCP fingerprint
expect_ja4_prefix string Disambiguate port-reuse aliasing under heavy same-IP load. Returns the most recent stored entry for (src_ip, src_port) whose JA4 starts with this prefix; if no stored entry matches the prefix, returns not_found (no fallback to potentially-aliased event-cache data).
aggregate bool CGNAT mode: returns every entry observed under src_ip across all ports and recent ktimes. src_port is not required when this is true. Combine with expect_ja4_prefix to filter by fingerprint family.

Response

200 OK (found):

{
  "status": "found",
  "event": {
    "src_ip": "203.0.113.42",
    "src_port": 54321,
    "dst_ip": "10.0.0.5",
    "dst_port": 443,
    "ja4":      "t13d311200_e8f1e7e78f70_d339722ba4af",
    "ja4_raw":  "t13d311200_e8f1e7e78f70_d339722ba4af",
    "ja4t":     "t64320_2_1-3-8-nop_nop,sackOK,ts,nop,ws,eol_",
    "ja4t_hash": null,
    "ja4s":     "t301600_c02bc02f002fc030_h2,http/1.1_",
    "ja4ts":    null,
    "ja4l":     "client-fingerprint-string",
    "ja4ls":    "server-fingerprint-string",
    "ja4h":     null,
    "ja4x":     null,
    "sni":      "example.com",
    "alpn":     "h2",
    "tls_version": "TLS 1.3"
  }
}

Only fields that were captured for this connection appear in the response; absent fields are omitted from the JSON object (not serialized as null).

FingerprintQueryResult field reference:

Field Type Description Linux Windows
src_ip string Client IP address ✅ ✅
src_port number Client source port ✅ ✅
dst_ip string Server IP address ✅ ✅
dst_port number Server listening port ✅ ✅
ja4 string TLS ClientHello fingerprint ✅ ✅
ja4_raw string JA4 without hash (full plaintext) ✅ ✅
ja4t string TCP SYN fingerprint ✅ ✅
ja4t_hash string JA4T hash ✅ ✅
ja4s string TLS ServerHello fingerprint ✅ ✅
ja4ts string TCP SYN-ACK fingerprint (egress) ✅ ✅
ja4l string Latency fingerprint (client-side RTT) ✅ ✅
ja4ls string Latency fingerprint (server-side RTT) ✅ ✅
ja4h string HTTP header fingerprint ✅ ❌
ja4x string X.509 certificate fingerprint ✅ ❌
ja4ssh string SSH session fingerprint (encrypted SSH) ✅ ❌
sni string TLS SNI hostname ✅ ✅
alpn string Negotiated ALPN protocol (e.g. "h2") ✅ ✅
tls_version string TLS version string (e.g. "TLS 1.3") ✅ ✅

200 OK (CGNAT aggregate, aggregate: true):

{
  "status": "found_many",
  "events": [
    { "src_ip": "203.0.113.42", "src_port": 54321, "ja4": "t13d311200_...", ... },
    { "src_ip": "203.0.113.42", "src_port": 54322, "ja4": "t13d1516h2_...", ... }
  ]
}

Each entry in events uses the same FingerprintQueryResult schema as the single-event found response.

404 Not Found:

{"status": "not_found"}

400 Bad Request:

{"status": "error", "message": "invalid JSON: ..."}

405 Method Not Allowed (not POST):

{"status": "error", "message": "method GET not allowed, use POST"}

Examples

curl:

curl --unix-socket /var/run/synapse-control.sock \
  -X POST http://localhost/query \
  -H "Content-Type: application/json" \
  -d '{"src_ip":"203.0.113.42","src_port":54321}'

curl with optional filters:

curl --unix-socket /var/run/synapse-control.sock \
  -X POST http://localhost/query \
  -H "Content-Type: application/json" \
  -d '{"src_ip":"203.0.113.42","src_port":54321,"dst_port":443}'

curl, CGNAT aggregate (all ports under one IP):

curl --unix-socket /var/run/synapse-control.sock \
  -X POST http://localhost/query \
  -H "Content-Type: application/json" \
  -d '{"src_ip":"203.0.113.42","aggregate":true}'

curl, port-reuse disambiguation:

curl --unix-socket /var/run/synapse-control.sock \
  -X POST http://localhost/query \
  -H "Content-Type: application/json" \
  -d '{"src_ip":"203.0.113.42","src_port":54321,"expect_ja4_prefix":"t13d1516h2"}'

Python (requests + requests-unixsocket):

import requests_unixsocket

session = requests_unixsocket.Session()
resp = session.post(
    "http+unix://%2Fvar%2Frun%2Fsynapse-control.sock/query",
    json={"src_ip": "203.0.113.42", "src_port": 54321},
)
if resp.status_code == 200:
    event = resp.json()["event"]
    print(f"JA4T: {event.get('ja4t')}")
    print(f"JA4:  {event.get('ja4')}")
    print(f"Geo:  {event.get('country')} AS{event.get('asn')}")

Python (stdlib, no extra deps):

import socket
import json

def lookup_fingerprints(client_ip, client_port, socket_path="/var/run/synapse-control.sock"):
    body = json.dumps({"src_ip": client_ip, "src_port": client_port})
    request = (
        f"POST /query HTTP/1.1\r\n"
        f"Content-Type: application/json\r\n"
        f"Content-Length: {len(body)}\r\n"
        f"\r\n"
        f"{body}"
    )

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.connect(socket_path)
    sock.settimeout(5.0)
    sock.sendall(request.encode())

    data = sock.recv(1024 * 1024).decode()
    sock.close()

    # Skip HTTP headers, parse JSON body after blank line
    _, _, response_body = data.partition("\r\n\r\n")
    result = json.loads(response_body)
    if result["status"] == "found":
        return result["event"]
    return None

# Usage
fp = lookup_fingerprints("203.0.113.42", 54321)
if fp:
    print(f"JA4T: {fp.get('ja4t')}")
    print(f"JA4:  {fp.get('ja4')}")
    print(f"Geo:  {fp.get('country')} AS{fp.get('asn')}")

Go:

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net"
    "net/http"
)

func lookupFingerprints(srcIP string, srcPort int) (map[string]any, error) {
    client := &http.Client{
        Transport: &http.Transport{
            DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
                return net.Dial("unix", "/var/run/synapse-control.sock")
            },
        },
    }

    body, _ := json.Marshal(map[string]any{
        "src_ip":   srcIP,
        "src_port": srcPort,
    })

    resp, err := client.Post("http://localhost/query", "application/json", bytes.NewReader(body))
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    data, _ := io.ReadAll(resp.Body)
    var result map[string]any
    json.Unmarshal(data, &result)

    if result["status"] == "found" {
        return result["event"].(map[string]any), nil
    }
    return nil, fmt.Errorf("not found")
}

Cache behavior

  • Events are cached for 10 minutes after being seen
  • Cache holds up to 50,000 recent connections
  • The most recent event for each (src_ip, src_port) pair is kept
  • Query after your TLS handshake completes for the most complete fingerprint set

In addition to the unified event cache, each JA4+ source is backed by a userspace authoritative store that keeps a per-(ip, port) history up to 1024 entries deep. The expect_ja4_prefix and aggregate query modes consult these stores directly, so they remain accurate even when the unified event cache has been overwritten by port reuse under heavy same-IP load.

Configuration

daemon:
  event_socket: "/var/run/synapse-events.sock"
  control_socket: "/var/run/synapse-control.sock"

CLI:

synapse --daemon --control-socket /var/run/synapse-control.sock

BPF map sizing (bpf_capacity)

Each JA4+ collector keeps state in a BPF map sized at compile time. Override the ceiling at runtime via the bpf_capacity: block in config.yaml. Any field left unset keeps the per-collector default; defaults handle ~1,000 conn/sec at >99% attribution.

bpf_capacity:
  ja4ts: 20000             # JA4T/JA4TS timing maps
  ja4l: 100000             # JA4L latency state tracker
  ja4s: 20000              # JA4S ServerHello cache
  ja4ssh: 20000            # JA4SSH session map (v4 + v6)
  ja4x_cert: 8000          # JA4X TLS 1.2 certificate cache
  ssl_fd: 40960            # SSL uprobe (pid, ssl_ptr) map
  ssl_ringbuf_bytes: 1048576  # SSL uprobe ring buffer (power-of-two)
  ja4_client_hello: 20000  # JA4 ClientHello cache (synapse XDP skel)

Event Socket (Streaming)

For real-time monitoring or when you need all events (not just specific connections), use the event socket.

Linux / macOS (UNIX socket)

/var/run/synapse-events.sock

Windows (TCP)

127.0.0.1:19199

Protocol

  • No handshake. Connect and immediately receive events.
  • No authentication. Access is controlled by filesystem permissions on the socket file.
  • Max 64 concurrent clients.

Configuration

The socket path is configurable via config.yaml:

daemon:
  event_socket: "/var/run/synapse-events.sock"

Or via CLI:

synapse --daemon --event-socket /var/run/synapse-events.sock

Message Format

Newline-delimited JSON (NDJSON). Each line is a self-contained JSON object with a type tag:

{"type":"Packet","src_ip":"203.0.113.42","dst_ip":"10.0.0.5","src_port":54321,"dst_port":443,...}
{"type":"Packet","src_ip":"198.51.100.7","dst_ip":"10.0.0.5","src_port":12345,"dst_port":80,...}
{"type":"Http","client_ip":"203.0.113.42","server_ip":"10.0.0.5","client_port":54321,"server_port":443,...}

Two event types:

Type Description
Packet TCP/TLS connection fingerprint (SYN-level, captured by BPF)
Http HTTP request fingerprint (proxy mode only)

In agent mode, you will primarily see Packet events.

Filtering

Events are broadcast to all connected clients. Filter client-side by matching on your connection tuple.

Required fields

Field Type Description
src_ip string Client IP address (e.g. "203.0.113.42")
src_port number Client source port (e.g. 54321)

Optional fields (narrow your match)

Field Type Description
dst_ip string Your server's IP address
dst_port number Your server's listening port
ja4t object/null TCP SYN fingerprint (match by .fingerprint() string)
ja4 object/null TLS ClientHello fingerprint

Match strategy

For most use cases, match on src_ip + src_port. This uniquely identifies a TCP connection:

event.src_ip == connection.client_ip AND event.src_port == connection.client_port

If your app handles NAT or doesn't know the source port, match on src_ip + dst_port and take the most recent event.

PacketEvent Schema

{
  "type": "Packet",
  "timestamp": "2026-04-16T12:34:56.789Z",
  "src_ip": "203.0.113.42",
  "dst_ip": "10.0.0.5",
  "src_port": 54321,
  "dst_port": 443,
  "direction": "ClientToServer",
  "tcp_flags": { "0": 2 },
  "ttl": 64,
  "window_size": 65535,
  "mss": 1460,
  "window_scale": 8,
  "tcp_options": [2, 4, 8, 1, 3],

  "ja4t": { "window_size": 65535, "tcp_options": [2,4,8,1,3], "mss": 1460, ... },
  "ja4ts": null,
  "ja4":  { "ja4": "t13d311200_e8f1e7e78f70_d339722ba4af", ... },
  "ja4s": null,
  "ja4h": null,
  "ja4x": null,
  "ja4l": null,
  "ja4ssh": null,

  "blocked": false,
  "block_rule": null,
  "block_source": null,
  "interface": "eth0",

  "country": "US",
  "asn": 15169,
  "asn_org": "Google LLC",
  "threat_score": 15,
  "classifier_label": "benign",
  "classifier_score": 0.05
}

Field reference

Connection tuple:

Field Type Always present
timestamp ISO 8601 string yes
src_ip string (IPv4/IPv6) yes
dst_ip string (IPv4/IPv6) yes
src_port number (0-65535) yes
dst_port number (0-65535) yes
direction "ClientToServer" or "ServerToClient" yes

TCP metadata:

Field Type Description
tcp_flags object TCP flags bitfield
ttl number IP TTL
window_size number TCP window size
mss number or null Maximum segment size
window_scale number or null Window scale factor
tcp_options array of numbers TCP option kind bytes in order

JA4+ fingerprints (all nullable, present when the relevant handshake data was captured):

Field Contains Linux Windows
ja4t TCP SYN fingerprint ✅ ✅
ja4ts TCP SYN-ACK fingerprint ✅ ✅
ja4 TLS ClientHello fingerprint ✅ ✅
ja4s TLS ServerHello fingerprint ✅ ✅
ja4l Latency fingerprint (client-side RTT) ✅ ✅
ja4ls Latency fingerprint (server-side RTT) ✅ ✅
ja4h HTTP header fingerprint ✅ ❌
ja4x X.509 certificate fingerprint (TLS 1.2 passive, TLS 1.3 via active prober) ✅ ❌
ja4ssh SSH session fingerprint (works on encrypted SSH) ✅ ❌

Platform note. Linux uses XDP/TC eBPF and SSL uprobes for the full JA4+ suite. Windows uses an NDIS LWF callback for the core TCP/TLS/latency set; ja4h, ja4x, and ja4ssh rely on kernel uprobes / TC programs that have no Windows equivalent yet, so those fields are always absent on Windows agents.

Enrichment (all nullable):

Field Type Description
country string ISO country code ("US", "DE")
asn number AS number
asn_org string AS organization name
threat_score number 0-100, higher = more suspicious
classifier_label string ML classification ("benign", "malicious")
classifier_score number 0.0-1.0 probability
blocked boolean Whether this connection was blocked
block_source string or null What blocked it: "AccessRules", "ThreatIntel", "Classifier", "Waf", "DosProtection", "FingerprintRule", "Manual"
block_rule string or null Human-readable rule description

HttpEvent Schema

Only emitted in proxy mode. Same enrichment fields, different connection naming:

Field Equivalent in PacketEvent
client_ip src_ip
client_port src_port
server_ip dst_ip
server_port dst_port

Additional HTTP-specific fields:

Field Type Description
threat_categories array of strings or null e.g. ["malware", "botnet"]
threat_advice string or null "allow", "block", "challenge"

Examples

Python: Look up fingerprints for a connection

import socket
import json


def lookup_fingerprints(client_ip, client_port, socket_path="/var/run/synapse-events.sock", timeout=5.0):
    """
    Connect to synapse EventBridge, wait for a matching event,
    and return all fingerprints for the given connection.
    """
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.connect(socket_path)
    sock.settimeout(timeout)

    buf = b""
    try:
        while True:
            chunk = sock.recv(65536)
            if not chunk:
                break
            buf += chunk
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                if not line:
                    continue
                event = json.loads(line)
                if event.get("type") != "Packet":
                    continue
                if event["src_ip"] == client_ip and event["src_port"] == client_port:
                    return {
                        "ja4t": event.get("ja4t"),
                        "ja4":  event.get("ja4"),
                        "ja4s": event.get("ja4s"),
                        "ja4l": event.get("ja4l"),
                        "country": event.get("country"),
                        "asn": event.get("asn"),
                        "asn_org": event.get("asn_org"),
                        "threat_score": event.get("threat_score"),
                        "blocked": event.get("blocked"),
                    }
    except socket.timeout:
        return None
    finally:
        sock.close()


# Usage: your app receives a connection from 203.0.113.42:54321
result = lookup_fingerprints("203.0.113.42", 54321)
if result:
    print(f"JA4T: {result['ja4t']}")
    print(f"JA4:  {result['ja4']}")
    print(f"Geo:  {result['country']} AS{result['asn']} {result['asn_org']}")
    print(f"Threat: {result['threat_score']}")

Python: Persistent listener with callback

import socket
import json
import threading


class EventBridgeClient:
    """
    Long-lived EventBridge client that maintains a fingerprint cache.
    Look up any connection's fingerprints instantly from the cache.
    """

    def __init__(self, socket_path="/var/run/synapse-events.sock", cache_size=10000):
        self.socket_path = socket_path
        self.cache = {}         # (src_ip, src_port) -> event
        self.cache_size = cache_size
        self._running = False

    def start(self):
        self._running = True
        self._thread = threading.Thread(target=self._read_loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False

    def _read_loop(self):
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(self.socket_path)
        buf = b""
        while self._running:
            chunk = sock.recv(65536)
            if not chunk:
                break
            buf += chunk
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                if not line:
                    continue
                event = json.loads(line)
                if event.get("type") == "Packet":
                    key = (event["src_ip"], event["src_port"])
                    self.cache[key] = event
                    if len(self.cache) > self.cache_size:
                        oldest = next(iter(self.cache))
                        del self.cache[oldest]
        sock.close()

    def lookup(self, src_ip, src_port):
        """Look up fingerprints for a connection. Returns dict or None."""
        return self.cache.get((src_ip, src_port))

    def lookup_by_ip(self, src_ip):
        """Look up the most recent event from an IP (ignoring port)."""
        matches = [v for k, v in self.cache.items() if k[0] == src_ip]
        return max(matches, key=lambda e: e["timestamp"]) if matches else None


# Usage
bridge = EventBridgeClient()
bridge.start()

# Later, when your app gets a connection:
fp = bridge.lookup("203.0.113.42", 54321)
if fp:
    print(f"JA4: {fp.get('ja4')}")

Go: Read events from socket

package main

import (
    "bufio"
    "encoding/json"
    "fmt"
    "net"
)

type PacketEvent struct {
    Type       string  `json:"type"`
    SrcIP      string  `json:"src_ip"`
    DstIP      string  `json:"dst_ip"`
    SrcPort    uint16  `json:"src_port"`
    DstPort    uint16  `json:"dst_port"`
    JA4T       any     `json:"ja4t"`
    JA4        any     `json:"ja4"`
    JA4S       any     `json:"ja4s"`
    Country    *string `json:"country"`
    ASN        *uint32 `json:"asn"`
    ASNOrg     *string `json:"asn_org"`
    ThreatScore *uint32 `json:"threat_score"`
    Blocked    bool    `json:"blocked"`
}

func main() {
    conn, err := net.Dial("unix", "/var/run/synapse-events.sock")
    if err != nil {
        panic(err)
    }
    defer conn.Close()

    scanner := bufio.NewScanner(conn)
    scanner.Buffer(make([]byte, 1024*1024), 1024*1024)

    for scanner.Scan() {
        var event PacketEvent
        if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
            continue
        }
        if event.Type != "Packet" {
            continue
        }

        // Filter: match your connection
        if event.SrcIP == "203.0.113.42" && event.SrcPort == 54321 {
            fmt.Printf("Found: JA4T=%v JA4=%v Country=%v\n",
                event.JA4T, event.JA4, event.Country)
        }
    }
}

Rust: Using dendrite types directly

use std::io::{BufRead, BufReader};
use std::os::unix::net::UnixStream;
use dendrite::terminal::event::SocketEvent;

fn main() -> anyhow::Result<()> {
    let stream = UnixStream::connect("/var/run/synapse-events.sock")?;
    let reader = BufReader::new(stream);

    for line in reader.lines() {
        let line = line?;
        if line.is_empty() { continue; }

        let event: SocketEvent = serde_json::from_str(&line)?;
        match event {
            SocketEvent::Packet(pkt) => {
                // Filter by your connection
                let src = pkt.src_ip.to_string();
                if src == "203.0.113.42" && pkt.src_port == 54321 {
                    println!("JA4T: {:?}", pkt.ja4t.map(|t| t.fingerprint()));
                    println!("JA4:  {:?}", pkt.ja4.map(|j| j.fingerprint()));
                    println!("Geo:  {:?} AS{:?}", pkt.country, pkt.asn);
                }
            }
            SocketEvent::Http(_http) => {
                // HTTP events (proxy mode only)
            }
        }
    }
    Ok(())
}

Event Timing

Fingerprints arrive in stages as the TCP/TLS handshake progresses:

  1. SYN (~0ms): ja4t available (TCP fingerprint)
  2. SYN-ACK (~1 RTT): ja4ts available (server TCP fingerprint)
  3. ClientHello (~1 RTT): ja4 available (TLS fingerprint)
  4. ServerHello (~2 RTT): ja4s available (TLS server fingerprint)
  5. Latency (~2 RTT): ja4l available (round-trip timing)

A single connection may produce multiple events as fingerprints become available. The most complete event is typically the last one for a given (src_ip, src_port) pair.

For the most complete fingerprint set, either:

  • Wait briefly (~200ms) after seeing the first event for a connection
  • Use the persistent cache pattern (Python example above) and look up after your TLS handshake completes

Troubleshooting

No events received:

  • Verify synapse is running in agent mode: systemctl status synapse
  • Check socket exists: ls -la /var/run/synapse-events.sock
  • Test with: socat - UNIX-CONNECT:/var/run/synapse-events.sock

Events received but no fingerprints (all null):

  • JA4T requires the BPF TCP fingerprint collector (enabled by default)
  • JA4 requires the XDP ClientHello capture (default on Linux)
  • JA4S requires the TC egress ServerHello collector
  • Check capture mode in config: capture_mode: auto

Socket permission denied:

  • The socket inherits filesystem permissions from the synapse process
  • Run your client as the same user, or adjust umask before starting synapse