feat: multi-distro VM tests, ja4ebpf eBPF improvements, bot-detector scoring
ja4ebpf: - Refactor BPF TC capture with improved SYN offset handling and TCP option parsing - Enhance TLS uprobe SSL hooking for better key extraction - Add ClickHouse writer improvements for HTTP log materialized views - Update RPM spec for Rocky Linux 8/9/10, fix systemd service - Simplify loader with cleaner bpf2go integration bot-detector: - Add H2 SETTINGS per-parameter comparison in browser_matcher - Enhance browser signatures and scoring pipeline - Improve preprocessing and cycle detection infra: - Multi-distro Vagrantfile (centos8, rocky9, rocky10) with per-distro provisioning - New Makefile targets: vm-up-all, test-vm-matrix, test-vm-centos8/rocky10 - Add debug helpers and run-test-from-host.sh for host-driven VM testing - Update run-tests-vm.sh for cross-distro compatibility - Remove accidental binary blob (\004) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jacquin Antoine
@ -10,6 +10,8 @@ import (
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@ -18,7 +20,7 @@ import (
|
||||
"github.com/antitbone/ja4/ja4ebpf/internal/parser"
|
||||
"github.com/antitbone/ja4/ja4ebpf/internal/procutil"
|
||||
"github.com/antitbone/ja4/ja4ebpf/internal/writer"
|
||||
"github.com/cilium/ebpf/ringbuf"
|
||||
"github.com/cilium/ebpf/perf"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
@ -32,6 +34,7 @@ var fdCache = procutil.NewFDCache(5 * time.Second)
|
||||
type Config struct {
|
||||
Interface string `yaml:"interface"` // interface réseau à surveiller (ex: "eth0")
|
||||
SSLLibPath string `yaml:"ssl_lib_path"` // chemin vers libssl (ex: "/usr/lib64/libssl.so.3")
|
||||
Debug bool `yaml:"debug"` // mode debug : dump compteurs BPF, log verbeux, ClickHouse optionnel
|
||||
|
||||
ClickHouse struct {
|
||||
DSN string `yaml:"dsn"` // DSN ClickHouse natif
|
||||
@ -87,6 +90,9 @@ func loadConfig(path string) (*Config, error) {
|
||||
if v := os.Getenv("JA4EBPF_CLICKHOUSE_DSN"); v != "" {
|
||||
cfg.ClickHouse.DSN = v
|
||||
}
|
||||
if v := os.Getenv("JA4EBPF_DEBUG"); v != "" {
|
||||
cfg.Debug = strings.EqualFold(v, "true") || v == "1" || v == "yes"
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
@ -104,7 +110,10 @@ func main() {
|
||||
log.Fatalf("erreur chargement configuration: %v", err)
|
||||
}
|
||||
|
||||
log.Printf("[ja4ebpf] démarrage — interface=%s ssl=%s", cfg.Interface, cfg.SSLLibPath)
|
||||
if cfg.Debug {
|
||||
log.Printf("[ja4ebpf] MODE DEBUG ACTIVÉ")
|
||||
}
|
||||
log.Printf("[ja4ebpf] démarrage — interface=%s ssl=%s debug=%v", cfg.Interface, cfg.SSLLibPath, cfg.Debug)
|
||||
|
||||
// Contexte principal avec annulation sur signal système
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
@ -122,9 +131,11 @@ func main() {
|
||||
defer ldr.Close()
|
||||
|
||||
// --- 2. Attachement TC ingress ---
|
||||
log.Printf("[ja4ebpf] attachement TC ingress sur %s...", cfg.Interface)
|
||||
if err := ldr.AttachTC(cfg.Interface); err != nil {
|
||||
log.Fatalf("erreur attachement TC sur %s: %v", cfg.Interface, err)
|
||||
}
|
||||
log.Printf("[ja4ebpf] TC ingress attaché sur %s", cfg.Interface)
|
||||
|
||||
// --- 3. Attachement uprobes SSL ---
|
||||
if err := ldr.AttachUprobes(cfg.SSLLibPath); err != nil {
|
||||
@ -144,26 +155,46 @@ func main() {
|
||||
defer mgr.Close()
|
||||
|
||||
// --- 6. Writer ClickHouse ---
|
||||
var w *writer.ClickHouseWriter
|
||||
flushInterval := time.Duration(cfg.ClickHouse.FlushSecs) * time.Second
|
||||
w, err := writer.NewClickHouseWriter(cfg.ClickHouse.DSN, cfg.ClickHouse.BatchSize, flushInterval)
|
||||
w, err = writer.NewClickHouseWriter(cfg.ClickHouse.DSN, cfg.ClickHouse.BatchSize, flushInterval)
|
||||
if err != nil {
|
||||
log.Fatalf("erreur initialisation writer ClickHouse: %v", err)
|
||||
if cfg.Debug {
|
||||
log.Printf("[ja4ebpf] DEBUG: writer ClickHouse non disponible: %v (continue sans CH)", err)
|
||||
} else {
|
||||
log.Fatalf("erreur initialisation writer ClickHouse: %v", err)
|
||||
}
|
||||
}
|
||||
if w != nil {
|
||||
w.Start(ctx)
|
||||
}
|
||||
w.Start(ctx)
|
||||
|
||||
// --- 7. Goroutine : écriture des sessions prêtes ---
|
||||
go func() {
|
||||
for s := range mgr.ReadyCh {
|
||||
w.Write(s)
|
||||
if w != nil {
|
||||
w.Write(s)
|
||||
} else if cfg.Debug {
|
||||
log.Printf("[ja4ebpf] DEBUG: session prête (sans CH): has_l3l4=%v has_tls=%v",
|
||||
s.L3L4 != nil, s.TLS != nil)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// --- 8. Goroutines de consommation des ring buffers ---
|
||||
go consumeSynEvents(ctx, ldr.SynReader, mgr)
|
||||
go consumeTLSEvents(ctx, ldr.TLSReader, mgr)
|
||||
go consumeSSLEvents(ctx, ldr.SSLReader, mgr)
|
||||
go consumeAcceptEvents(ctx, ldr.AcceptReader, mgr)
|
||||
go consumeHTTPPlainEvents(ctx, ldr.HTTPPlainReader, mgr)
|
||||
// --- 8. Compteurs d'événements consommés (mode debug) ---
|
||||
consumed := &eventCounters{}
|
||||
|
||||
// --- 9. Goroutines de consommation des ring buffers ---
|
||||
go consumeSynEvents(ctx, ldr.SynReader, mgr, &consumed.syn)
|
||||
go consumeTLSEvents(ctx, ldr.TLSReader, mgr, &consumed.tls)
|
||||
go consumeSSLEvents(ctx, ldr.SSLReader, mgr, &consumed.ssl)
|
||||
go consumeAcceptEvents(ctx, ldr.AcceptReader, mgr, &consumed.accept)
|
||||
go consumeHTTPPlainEvents(ctx, ldr.HTTPPlainReader, mgr, &consumed.httpPlain)
|
||||
|
||||
// --- 10. Stats dumper (mode debug) ---
|
||||
if cfg.Debug {
|
||||
go debugStatsDumper(ctx, ldr, consumed)
|
||||
}
|
||||
|
||||
log.Printf("[ja4ebpf] démon actif — en attente des événements")
|
||||
|
||||
@ -178,6 +209,43 @@ func main() {
|
||||
log.Printf("[ja4ebpf] arrêt terminé")
|
||||
}
|
||||
|
||||
// eventCounters contient les compteurs atomiques pour chaque type d'événement consommé.
|
||||
type eventCounters struct {
|
||||
syn atomic.Uint64
|
||||
tls atomic.Uint64
|
||||
ssl atomic.Uint64
|
||||
accept atomic.Uint64
|
||||
httpPlain atomic.Uint64
|
||||
}
|
||||
|
||||
// debugStatsDumper affiche les compteurs BPF et les événements consommés toutes les 5 secondes.
|
||||
func debugStatsDumper(ctx context.Context, ldr *loader.Loader, consumed *eventCounters) {
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
|
||||
// Compteurs BPF kernel
|
||||
stats, err := ldr.ReadStats()
|
||||
if err != nil {
|
||||
log.Printf("[debug] erreur lecture tc_stats: %v", err)
|
||||
continue
|
||||
}
|
||||
log.Printf("[debug] BPF: TOTAL=%d IPV4=%d TCP=%d SYN=%d SYN_SUB=%d TLS_SUB=%d HTTP_SUB=%d",
|
||||
stats[0], stats[1], stats[2], stats[3], stats[4], stats[5], stats[6])
|
||||
|
||||
// Compteurs userspace
|
||||
log.Printf("[debug] GO: syn=%d tls=%d ssl=%d accept=%d http=%d",
|
||||
consumed.syn.Load(), consumed.tls.Load(), consumed.ssl.Load(),
|
||||
consumed.accept.Load(), consumed.httpPlain.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// parseTCPOptions extrait le MSS et le Window Scale depuis les options TCP brutes.
|
||||
// Les options TCP suivent le format TLV (Type-Length-Value), sauf les options 0 et 1.
|
||||
// Retourne (mss=0, windowScale=0xFF) si les options sont absentes ou mal formées.
|
||||
@ -220,7 +288,7 @@ func parseTCPOptions(opts []byte) (mss uint16, windowScale uint8) {
|
||||
|
||||
// consumeSynEvents lit les événements TCP SYN depuis le ring buffer
|
||||
// et met à jour l'état L3/L4 des sessions.
|
||||
func consumeSynEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.Manager) {
|
||||
func consumeSynEvents(ctx context.Context, rd *perf.Reader, mgr *correlation.Manager, counter *atomic.Uint64) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@ -230,7 +298,7 @@ func consumeSynEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
|
||||
record, err := rd.Read()
|
||||
if err != nil {
|
||||
if err == ringbuf.ErrClosed {
|
||||
if err == os.ErrClosed {
|
||||
return
|
||||
}
|
||||
continue
|
||||
@ -240,7 +308,7 @@ func consumeSynEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
// src_ip(4)+dst_ip(4)+src_port(2)+dst_port(2)+ttl(1)+df_bit(1)+ip_id(2)+
|
||||
// window_size(2)+window_scale(1)+mss(2)+tcp_options_raw[40]+tcp_options_len(1)+timestamp_ns(8)
|
||||
// offsets: 0 4 8 10 12 13 14 16 18 19 21 61 62
|
||||
if len(record.RawSample) < 62 {
|
||||
if len(record.RawSample) < 70 {
|
||||
continue
|
||||
}
|
||||
data := record.RawSample
|
||||
@ -288,12 +356,13 @@ func consumeSynEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
_ = s.TLS // corrélation implicite par présence des deux champs
|
||||
}
|
||||
})
|
||||
counter.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
// consumeTLSEvents lit les événements TLS ClientHello depuis le ring buffer
|
||||
// et calcule l'empreinte JA4 pour chaque session.
|
||||
func consumeTLSEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.Manager) {
|
||||
func consumeTLSEvents(ctx context.Context, rd *perf.Reader, mgr *correlation.Manager, counter *atomic.Uint64) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@ -303,7 +372,7 @@ func consumeTLSEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
|
||||
record, err := rd.Read()
|
||||
if err != nil {
|
||||
if err == ringbuf.ErrClosed {
|
||||
if err == os.ErrClosed {
|
||||
return
|
||||
}
|
||||
continue
|
||||
@ -312,20 +381,20 @@ func consumeTLSEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
// struct tls_hello_event (packed):
|
||||
// src_ip(4) + src_port(2) + payload[2048] + payload_len(2) + timestamp_ns(8)
|
||||
// offsets: 0 4 6 2054 2056
|
||||
if len(record.RawSample) < 2056 {
|
||||
if len(record.RawSample) < 2064 {
|
||||
continue
|
||||
}
|
||||
data := record.RawSample
|
||||
|
||||
srcIPRaw := binary.LittleEndian.Uint32(data[0:4])
|
||||
srcPort := binary.LittleEndian.Uint16(data[4:6])
|
||||
srcIPRaw := binary.LittleEndian.Uint32(data[2048:2052])
|
||||
srcPort := binary.LittleEndian.Uint16(data[2052:2054])
|
||||
payloadLen := binary.LittleEndian.Uint16(data[2054:2056])
|
||||
|
||||
if int(payloadLen) > 2048 {
|
||||
payloadLen = 2048
|
||||
}
|
||||
payload := make([]byte, payloadLen)
|
||||
copy(payload, data[6:6+payloadLen])
|
||||
copy(payload, data[0:payloadLen])
|
||||
|
||||
var key correlation.SessionKey
|
||||
key.SrcIP[0] = byte(srcIPRaw >> 24)
|
||||
@ -366,13 +435,14 @@ func consumeTLSEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
_ = s.L3L4 // corrélation implicite par présence des deux champs
|
||||
}
|
||||
})
|
||||
counter.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
// consumeSSLEvents lit les données SSL déchiffrées depuis le ring buffer.
|
||||
// Parse les requêtes HTTP/1.x et détecte le préambule HTTP/2.
|
||||
// Quand src_ip=0 (accept4 non disponible), tente un lookup /proc pour retrouver l'IP du client.
|
||||
func consumeSSLEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.Manager) {
|
||||
func consumeSSLEvents(ctx context.Context, rd *perf.Reader, mgr *correlation.Manager, counter *atomic.Uint64) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@ -382,7 +452,7 @@ func consumeSSLEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
|
||||
record, err := rd.Read()
|
||||
if err != nil {
|
||||
if err == ringbuf.ErrClosed {
|
||||
if err == os.ErrClosed {
|
||||
return
|
||||
}
|
||||
continue
|
||||
@ -439,6 +509,7 @@ func consumeSSLEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
key.SrcIP[3] = byte(srcIPRaw)
|
||||
key.SrcPort = srcPort
|
||||
|
||||
counter.Add(1)
|
||||
// === Routeur Magic Bytes ===
|
||||
|
||||
if parser.DetectH2Preface(sslData) {
|
||||
@ -517,7 +588,7 @@ func consumeSSLEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.
|
||||
|
||||
// consumeAcceptEvents lit les événements accept4 depuis le ring buffer.
|
||||
// Met à jour les sessions avec les informations de connexion client.
|
||||
func consumeAcceptEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.Manager) {
|
||||
func consumeAcceptEvents(ctx context.Context, rd *perf.Reader, mgr *correlation.Manager, counter *atomic.Uint64) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@ -527,7 +598,7 @@ func consumeAcceptEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlati
|
||||
|
||||
record, err := rd.Read()
|
||||
if err != nil {
|
||||
if err == ringbuf.ErrClosed {
|
||||
if err == os.ErrClosed {
|
||||
return
|
||||
}
|
||||
continue
|
||||
@ -556,13 +627,14 @@ func consumeAcceptEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlati
|
||||
|
||||
// S'assurer que la session existe
|
||||
mgr.GetOrCreate(key)
|
||||
counter.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
// consumeHTTPPlainEvents lit les payloads HTTP en clair depuis le ring buffer XDP.
|
||||
// consumeHTTPPlainEvents lit les payloads HTTP en clair depuis le perf buffer TC.
|
||||
// Parse la requête HTTP/1.x ou détecte la préface HTTP/2 pour les connexions
|
||||
// non-chiffrées sur les ports 80/8080.
|
||||
func consumeHTTPPlainEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correlation.Manager) {
|
||||
func consumeHTTPPlainEvents(ctx context.Context, rd *perf.Reader, mgr *correlation.Manager, counter *atomic.Uint64) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@ -572,21 +644,21 @@ func consumeHTTPPlainEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correl
|
||||
|
||||
record, err := rd.Read()
|
||||
if err != nil {
|
||||
if err == ringbuf.ErrClosed {
|
||||
if err == os.ErrClosed {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
data := record.RawSample
|
||||
// struct http_plain_event: src_ip(4)+dst_ip(4)+src_port(2)+dst_port(2)+payload(4096)+payload_len(2)+timestamp_ns(8)
|
||||
// struct http_plain_event: payload(4096)+src_ip(4)+dst_ip(4)+src_port(2)+dst_port(2)+payload_len(2)+timestamp_ns(8)
|
||||
if len(data) < 14 {
|
||||
continue
|
||||
}
|
||||
|
||||
// src_ip et src_port en host byte order (bpf_ntohl appliqué dans tc_capture.c)
|
||||
srcIPRaw := binary.LittleEndian.Uint32(data[0:4])
|
||||
srcPort := binary.LittleEndian.Uint16(data[8:10])
|
||||
srcIPRaw := binary.LittleEndian.Uint32(data[4096:4100])
|
||||
srcPort := binary.LittleEndian.Uint16(data[4104:4106])
|
||||
|
||||
if srcIPRaw == 0 && srcPort == 0 {
|
||||
continue
|
||||
@ -610,10 +682,10 @@ func consumeHTTPPlainEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correl
|
||||
if payloadLen == 0 {
|
||||
continue
|
||||
}
|
||||
if 12+payloadLen > len(data) {
|
||||
payloadLen = len(data) - 12
|
||||
if 4096+payloadLen > len(data) {
|
||||
payloadLen = len(data) - 4096
|
||||
}
|
||||
httpData := data[12 : 12+payloadLen]
|
||||
httpData := data[0:payloadLen]
|
||||
|
||||
// Routeur Magic Bytes : HTTP/1.x uniquement sur port 80
|
||||
if parser.IsHTTP1Request(httpData) {
|
||||
@ -633,6 +705,7 @@ func consumeHTTPPlainEvents(ctx context.Context, rd *ringbuf.Reader, mgr *correl
|
||||
// Corréler si L3/L4 est déjà présent (TCP SYN capturé)
|
||||
_ = s.L3L4 // corrélation implicite
|
||||
})
|
||||
counter.Add(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user