Correctifs pipeline L7 (uprobe SSL_read) :
- uprobe_ssl.c : ssl_set_fd ne retourne plus tôt quand fd_conn_map est
vide (accept4 non disponible en Docker). Sauvegarde ssl_ptr→{fd,0,0}
pour permettre le fallback /proc côté Go.
- main.go : consumeSSLEvents reécrit avec routeur magic-bytes complet :
* HTTP/2 preface → extraction SETTINGS + conversion correlation.HTTP2Settings
* HTTP/1.x requête → method, path, query, headers, header_order_sig
* HTTP/1.x réponse → status_code
* Fallback /proc/<tgid>/fd/<fd> quand src_ip=0 (accept4 absent)
- writer/clickhouse.go : export header_order_signature ajouté
Nouveaux packages :
- internal/parser/http1.go : parseur HTTP/1.x (IsHTTP1Request,
ParseHTTP1Request, IsHTTP1Response, ParseHTTP1Response)
- internal/parser/http1_test.go : 11 tests unitaires (28 total passent)
- internal/procutil/proc_lookup.go : résolution fd→IP via /proc avec cache
TTL 5s (FDCache). Supporte /proc/PID/net/tcp et tcp6, IPv4-mappé IPv6.
Infrastructure tests VM (tests/vm/) :
- Vagrantfile : VM Rocky Linux 9 KVM, 4 CPU / 4 GB RAM
- provision.sh : installation toolchain eBPF + Go + Docker + nginx
- run-tests-vm.sh : suite de test complète dans la VM (L3/L4+TLS+L7)
- README.md : guide d'installation et d'utilisation
- Makefile : cibles vm-up, vm-down, vm-ssh, test-vm-nginx, test-vm-all,
vm-rebuild-ja4ebpf
Corrections stack Docker :
- Dockerfiles nginx/apache/nginx-varnish/hitch-varnish : suppression des
références à shared/go/ja4common/ (répertoire supprimé)
- clickhouse-init.sh : restauré depuis git, seed anubis_ua_rules obsolète
supprimé (table REGEXP_TREE supprimée du schéma)
- traffic-gen : ajout HTTP/1.0 (http.client) et HTTP/2 (httpx)
- verify_db.py : script de vérification 35 checks (L3/L4/TLS/L7/corrélation)
- run-stack-tests.sh : phase 6 verify_db ajoutée
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
519 lines
15 KiB
Go
519 lines
15 KiB
Go
// Package loader initialise les programmes eBPF via cilium/ebpf,
|
|
// attache les hooks TC ingress et les uprobes SSL, et expose
|
|
// les readers RingBuffer aux consommateurs Go.
|
|
package loader
|
|
|
|
import (
|
|
"context"
|
|
"encoding/binary"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
|
|
"github.com/cilium/ebpf"
|
|
"github.com/cilium/ebpf/link"
|
|
"github.com/cilium/ebpf/ringbuf"
|
|
"github.com/cilium/ebpf/rlimit"
|
|
)
|
|
|
|
//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cc clang -target amd64 -cflags "-O2 -g -Wall -D__TARGET_ARCH_x86 -Wno-pass-failed" Ja4Tc ../../bpf/tc_capture.c -- -I../../bpf/headers
|
|
//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cc clang -target amd64 -cflags "-O2 -g -Wall -D__TARGET_ARCH_x86 -Wno-pass-failed" Ja4Ssl ../../bpf/uprobe_ssl.c -- -I../../bpf/headers
|
|
|
|
// Loader encapsule les objets eBPF compilés, les liens vers les hooks,
|
|
// et les readers RingBuffer exposés au pipeline de traitement.
|
|
type Loader struct {
|
|
tcObjs *Ja4TcObjects // généré par bpf2go (tc_capture.c)
|
|
sslObjs *Ja4SslObjects // généré par bpf2go (uprobe_ssl.c)
|
|
tcLink link.Link
|
|
uprobeLinks []link.Link
|
|
|
|
// SynReader lit les événements TCP SYN depuis rb_tcp_syn.
|
|
SynReader *ringbuf.Reader
|
|
// TLSReader lit les événements TLS ClientHello depuis rb_tls_hello.
|
|
TLSReader *ringbuf.Reader
|
|
// SSLReader lit les données SSL déchiffrées depuis rb_ssl_data.
|
|
SSLReader *ringbuf.Reader
|
|
// AcceptReader lit les événements accept4 depuis rb_accept.
|
|
AcceptReader *ringbuf.Reader
|
|
// HTTPPlainReader lit les payloads HTTP en clair depuis rb_http_plain.
|
|
HTTPPlainReader *ringbuf.Reader
|
|
}
|
|
|
|
// New charge le bytecode eBPF embarqué, supprime la limite mémoire
|
|
// RLIMIT_MEMLOCK (requise pour les ring buffers et les maps eBPF),
|
|
// et retourne un Loader prêt à être attaché aux hooks.
|
|
//
|
|
// Cible : CentOS 8 / RHEL 8 et supérieur (kernel ≥ 4.18 avec BTF backporté).
|
|
// Le BTF natif est détecté automatiquement par cilium/ebpf via
|
|
// /sys/kernel/btf/vmlinux — aucun fallback manuel n'est requis.
|
|
func New() (*Loader, error) {
|
|
// Supprimer la limite mémoire pour les opérations eBPF
|
|
if err := rlimit.RemoveMemlock(); err != nil {
|
|
return nil, fmt.Errorf("suppression RLIMIT_MEMLOCK: %w", err)
|
|
}
|
|
|
|
// Charger les objets TC (tc_capture.c)
|
|
tcObjs := &Ja4TcObjects{}
|
|
if err := LoadJa4TcObjects(tcObjs, nil); err != nil {
|
|
return nil, fmt.Errorf("chargement objets TC eBPF: %w", err)
|
|
}
|
|
|
|
// Charger les objets SSL/uprobe (uprobe_ssl.c)
|
|
sslObjs := &Ja4SslObjects{}
|
|
if err := LoadJa4SslObjects(sslObjs, nil); err != nil {
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("chargement objets SSL eBPF: %w", err)
|
|
}
|
|
|
|
// Initialiser les readers pour chaque ring buffer
|
|
synReader, err := ringbuf.NewReader(tcObjs.RbTcpSyn)
|
|
if err != nil {
|
|
sslObjs.Close()
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("création reader rb_tcp_syn: %w", err)
|
|
}
|
|
|
|
tlsReader, err := ringbuf.NewReader(tcObjs.RbTlsHello)
|
|
if err != nil {
|
|
synReader.Close()
|
|
sslObjs.Close()
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("création reader rb_tls_hello: %w", err)
|
|
}
|
|
|
|
httpPlainReader, err := ringbuf.NewReader(tcObjs.RbHttpPlain)
|
|
if err != nil {
|
|
tlsReader.Close()
|
|
synReader.Close()
|
|
sslObjs.Close()
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("création reader rb_http_plain: %w", err)
|
|
}
|
|
|
|
sslReader, err := ringbuf.NewReader(sslObjs.RbSslData)
|
|
if err != nil {
|
|
httpPlainReader.Close()
|
|
tlsReader.Close()
|
|
synReader.Close()
|
|
sslObjs.Close()
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("création reader rb_ssl_data: %w", err)
|
|
}
|
|
|
|
acceptReader, err := ringbuf.NewReader(sslObjs.RbAccept)
|
|
if err != nil {
|
|
sslReader.Close()
|
|
httpPlainReader.Close()
|
|
tlsReader.Close()
|
|
synReader.Close()
|
|
sslObjs.Close()
|
|
tcObjs.Close()
|
|
return nil, fmt.Errorf("création reader rb_accept: %w", err)
|
|
}
|
|
|
|
return &Loader{
|
|
tcObjs: tcObjs,
|
|
sslObjs: sslObjs,
|
|
SynReader: synReader,
|
|
TLSReader: tlsReader,
|
|
SSLReader: sslReader,
|
|
AcceptReader: acceptReader,
|
|
HTTPPlainReader: httpPlainReader,
|
|
}, nil
|
|
}
|
|
|
|
// AttachTC attache le programme TC ingress sur l'interface réseau spécifiée.
|
|
// Utilise TCX (TC eXpress) disponible depuis le noyau 6.6+.
|
|
func (l *Loader) AttachTC(iface string) error {
|
|
// Résoudre l'interface réseau par son nom
|
|
netIface, err := net.InterfaceByName(iface)
|
|
if err != nil {
|
|
return fmt.Errorf("interface réseau %q introuvable: %w", iface, err)
|
|
}
|
|
|
|
// Attacher le programme TC en ingress via TCX
|
|
lnk, err := link.AttachTCX(link.TCXOptions{
|
|
Interface: netIface.Index,
|
|
Program: l.tcObjs.CaptureTcIngress,
|
|
Attach: ebpf.AttachTCXIngress,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("attachement TC ingress sur %q: %w", iface, err)
|
|
}
|
|
|
|
l.tcLink = lnk
|
|
return nil
|
|
}
|
|
|
|
// AttachUprobes attache les uprobes SSL_read et SSL_set_fd
|
|
// sur le binaire libssl spécifié (ex: "/usr/lib64/libssl.so.3").
|
|
func (l *Loader) AttachUprobes(sslLibPath string) error {
|
|
// Vérifier que le fichier existe
|
|
if _, err := os.Stat(sslLibPath); err != nil {
|
|
return fmt.Errorf("bibliothèque SSL %q: %w", sslLibPath, err)
|
|
}
|
|
|
|
// Ouvrir le binaire exécutable pour les uprobes
|
|
ex, err := link.OpenExecutable(sslLibPath)
|
|
if err != nil {
|
|
return fmt.Errorf("ouverture exécutable %q pour uprobe: %w", sslLibPath, err)
|
|
}
|
|
|
|
// Uprobe sur SSL_set_fd (entry)
|
|
setFdLink, err := ex.Uprobe("SSL_set_fd", l.sslObjs.UprobeSslSetFd, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("attachement uprobe SSL_set_fd: %w", err)
|
|
}
|
|
l.uprobeLinks = append(l.uprobeLinks, setFdLink)
|
|
|
|
// Uprobe sur SSL_read (entry)
|
|
readEntryLink, err := ex.Uprobe("SSL_read", l.sslObjs.UprobeSslReadEntry, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("attachement uprobe SSL_read (entry): %w", err)
|
|
}
|
|
l.uprobeLinks = append(l.uprobeLinks, readEntryLink)
|
|
|
|
// Uretprobe sur SSL_read (exit)
|
|
readExitLink, err := ex.Uretprobe("SSL_read", l.sslObjs.UretprobeSslReadExit, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("attachement uretprobe SSL_read (exit): %w", err)
|
|
}
|
|
l.uprobeLinks = append(l.uprobeLinks, readExitLink)
|
|
|
|
return nil
|
|
}
|
|
|
|
// AttachAcceptProbe attache les tracepoints syscalls/sys_{enter,exit}_accept4.
|
|
// Les tracepoints sont préférés aux kprobes car ils ne dépendent pas du nom
|
|
// manglé __x64_sys_accept4 qui varie entre les versions du kernel (5.1+).
|
|
func (l *Loader) AttachAcceptProbe() error {
|
|
// Tracepoint à l'entrée de accept4
|
|
kpEntry, err := link.Tracepoint("syscalls", "sys_enter_accept4",
|
|
l.sslObjs.KprobeAccept4Entry, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("attachement tracepoint sys_enter_accept4: %w", err)
|
|
}
|
|
l.uprobeLinks = append(l.uprobeLinks, kpEntry)
|
|
|
|
// Tracepoint à la sortie de accept4
|
|
kpExit, err := link.Tracepoint("syscalls", "sys_exit_accept4",
|
|
l.sslObjs.KretprobeAccept4Exit, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("attachement tracepoint sys_exit_accept4: %w", err)
|
|
}
|
|
l.uprobeLinks = append(l.uprobeLinks, kpExit)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close détache tous les hooks eBPF et libère toutes les ressources associées.
|
|
func (l *Loader) Close() error {
|
|
// Fermer les readers RingBuffer
|
|
if l.HTTPPlainReader != nil {
|
|
l.HTTPPlainReader.Close()
|
|
}
|
|
if l.AcceptReader != nil {
|
|
l.AcceptReader.Close()
|
|
}
|
|
if l.SSLReader != nil {
|
|
l.SSLReader.Close()
|
|
}
|
|
if l.TLSReader != nil {
|
|
l.TLSReader.Close()
|
|
}
|
|
if l.SynReader != nil {
|
|
l.SynReader.Close()
|
|
}
|
|
|
|
// Détacher les uprobes et kprobes
|
|
for _, lnk := range l.uprobeLinks {
|
|
if lnk != nil {
|
|
lnk.Close()
|
|
}
|
|
}
|
|
|
|
// Détacher le hook TC
|
|
if l.tcLink != nil {
|
|
l.tcLink.Close()
|
|
}
|
|
|
|
// Libérer les objets eBPF (maps, programmes)
|
|
if l.sslObjs != nil {
|
|
l.sslObjs.Close()
|
|
}
|
|
if l.tcObjs != nil {
|
|
l.tcObjs.Close()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// =============================================================================
|
|
// Types d'événements : représentations Go des structures C eBPF
|
|
// =============================================================================
|
|
|
|
// TCPSynEvent représente un événement TCP SYN capturé par TC ingress.
|
|
type TCPSynEvent struct {
|
|
SrcIP uint32
|
|
DstIP uint32
|
|
SrcPort uint16
|
|
DstPort uint16
|
|
TTL uint8
|
|
DFBit uint8
|
|
IPID uint16
|
|
WindowSize uint16
|
|
WindowScale uint8
|
|
MSS uint16
|
|
TCPOptions [40]byte
|
|
TCPOptionsLen uint8
|
|
Timestamp uint64
|
|
}
|
|
|
|
// TLSHelloEvent représente un événement TLS ClientHello.
|
|
type TLSHelloEvent struct {
|
|
SrcIP uint32
|
|
SrcPort uint16
|
|
Payload []byte
|
|
PayloadLen uint16
|
|
Timestamp uint64
|
|
}
|
|
|
|
// SSLDataEvent représente un bloc de données SSL déchiffré par uprobe.
|
|
type SSLDataEvent struct {
|
|
PID uint32
|
|
TGID uint32
|
|
FD uint32
|
|
SrcIP uint32
|
|
SrcPort uint16
|
|
Data []byte
|
|
DataLen uint32
|
|
Timestamp uint64
|
|
Direction uint8
|
|
EOF bool
|
|
}
|
|
|
|
// HTTPPlainEvent représente un payload TCP HTTP en clair capturé par TC ingress.
|
|
type HTTPPlainEvent struct {
|
|
SrcIP uint32
|
|
DstIP uint32
|
|
SrcPort uint16
|
|
DstPort uint16
|
|
Payload []byte
|
|
PayloadLen uint16
|
|
Timestamp uint64
|
|
}
|
|
|
|
// AcceptEvent représente une acceptation de connexion TCP (accept4).
|
|
type AcceptEvent struct {
|
|
PID uint32
|
|
TGID uint32
|
|
FD uint32
|
|
SrcIP uint32
|
|
SrcPort uint16
|
|
Timestamp uint64
|
|
}
|
|
|
|
// =============================================================================
|
|
// Méthodes de lecture des RingBuffers
|
|
// =============================================================================
|
|
|
|
// ReadTCPSynEvent lit un événement TCP SYN depuis le RingBuffer.
|
|
// Bloque jusqu'à ce qu'un événement soit disponible ou que ctx soit annulé.
|
|
func (l *Loader) ReadTCPSynEvent(ctx context.Context) (*TCPSynEvent, error) {
|
|
rec, err := readRecord(ctx, l.SynReader)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data := rec.RawSample
|
|
// struct tcp_syn_event packed: src_ip(4)+dst_ip(4)+src_port(2)+dst_port(2)+
|
|
// ttl(1)+df(1)+ip_id(2)+window(2)+wscale(1)+mss(2)+opts(40)+opts_len(1)+_pad(1)+ts(8) = 71
|
|
if len(data) < 64 {
|
|
return nil, fmt.Errorf("tcp_syn_event trop court: %d octets", len(data))
|
|
}
|
|
|
|
ev := &TCPSynEvent{
|
|
SrcIP: binary.LittleEndian.Uint32(data[0:4]),
|
|
DstIP: binary.LittleEndian.Uint32(data[4:8]),
|
|
SrcPort: binary.LittleEndian.Uint16(data[8:10]),
|
|
DstPort: binary.LittleEndian.Uint16(data[10:12]),
|
|
TTL: data[12],
|
|
DFBit: data[13],
|
|
IPID: binary.LittleEndian.Uint16(data[14:16]),
|
|
WindowSize: binary.LittleEndian.Uint16(data[16:18]),
|
|
WindowScale: data[18],
|
|
MSS: binary.LittleEndian.Uint16(data[19:21]),
|
|
}
|
|
copy(ev.TCPOptions[:], data[21:61])
|
|
ev.TCPOptionsLen = data[61]
|
|
if len(data) >= 70 {
|
|
ev.Timestamp = binary.LittleEndian.Uint64(data[62:70])
|
|
}
|
|
return ev, nil
|
|
}
|
|
|
|
// ReadTLSHelloEvent lit un événement TLS ClientHello depuis le RingBuffer.
|
|
func (l *Loader) ReadTLSHelloEvent(ctx context.Context) (*TLSHelloEvent, error) {
|
|
rec, err := readRecord(ctx, l.TLSReader)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data := rec.RawSample
|
|
// struct tls_hello_event: src_ip(4)+src_port(2)+payload(512)+payload_len(2)+ts(8) = 528
|
|
if len(data) < 8 {
|
|
return nil, fmt.Errorf("tls_hello_event trop court: %d octets", len(data))
|
|
}
|
|
|
|
plen := uint16(0)
|
|
if len(data) >= 520 {
|
|
plen = binary.LittleEndian.Uint16(data[518:520])
|
|
}
|
|
payload := make([]byte, plen)
|
|
if int(plen) <= 512 && len(data) >= 6+int(plen) {
|
|
copy(payload, data[6:6+plen])
|
|
}
|
|
|
|
ts := uint64(0)
|
|
if len(data) >= 528 {
|
|
ts = binary.LittleEndian.Uint64(data[520:528])
|
|
}
|
|
|
|
return &TLSHelloEvent{
|
|
SrcIP: binary.LittleEndian.Uint32(data[0:4]),
|
|
SrcPort: binary.LittleEndian.Uint16(data[4:6]),
|
|
Payload: payload,
|
|
PayloadLen: plen,
|
|
Timestamp: ts,
|
|
}, nil
|
|
}
|
|
|
|
// ReadSSLDataEvent lit un bloc de données SSL déchiffrées depuis le RingBuffer.
|
|
func (l *Loader) ReadSSLDataEvent(ctx context.Context) (*SSLDataEvent, error) {
|
|
rec, err := readRecord(ctx, l.SSLReader)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data := rec.RawSample
|
|
// struct ssl_data_event: pid_tgid(8)+fd(4)+src_ip(4)+src_port(2)+data(4096)+data_len(4)+ts(8)+direction(1)
|
|
if len(data) < 27 {
|
|
return nil, fmt.Errorf("ssl_data_event trop court: %d octets", len(data))
|
|
}
|
|
|
|
pidTGID := binary.LittleEndian.Uint64(data[0:8])
|
|
dlen := uint32(0)
|
|
if len(data) >= 4118 {
|
|
dlen = binary.LittleEndian.Uint32(data[4114:4118])
|
|
}
|
|
payload := make([]byte, dlen)
|
|
if int(dlen) <= 4096 && len(data) >= 18+int(dlen) {
|
|
copy(payload, data[18:18+dlen])
|
|
}
|
|
|
|
ts := uint64(0)
|
|
if len(data) >= 4126 {
|
|
ts = binary.LittleEndian.Uint64(data[4118:4126])
|
|
}
|
|
dir := uint8(0)
|
|
if len(data) >= 4127 {
|
|
dir = data[4126]
|
|
}
|
|
|
|
return &SSLDataEvent{
|
|
PID: uint32(pidTGID & 0xFFFFFFFF),
|
|
TGID: uint32(pidTGID >> 32),
|
|
FD: binary.LittleEndian.Uint32(data[8:12]),
|
|
SrcIP: binary.LittleEndian.Uint32(data[12:16]),
|
|
SrcPort: binary.LittleEndian.Uint16(data[16:18]),
|
|
Data: payload,
|
|
DataLen: dlen,
|
|
Timestamp: ts,
|
|
Direction: dir,
|
|
}, nil
|
|
}
|
|
|
|
// ReadHTTPPlainEvent lit un événement HTTP en clair depuis le RingBuffer TC.
|
|
// struct http_plain_event: src_ip(4)+dst_ip(4)+src_port(2)+dst_port(2)+
|
|
//
|
|
// payload(4096)+payload_len(2)+ts(8) = 4118
|
|
func (l *Loader) ReadHTTPPlainEvent(ctx context.Context) (*HTTPPlainEvent, error) {
|
|
rec, err := readRecord(ctx, l.HTTPPlainReader)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data := rec.RawSample
|
|
if len(data) < 12 {
|
|
return nil, fmt.Errorf("http_plain_event trop court: %d octets", len(data))
|
|
}
|
|
|
|
plen := uint16(0)
|
|
if len(data) >= 4110 {
|
|
plen = binary.LittleEndian.Uint16(data[4108:4110])
|
|
}
|
|
payload := make([]byte, plen)
|
|
if int(plen) <= 4096 && len(data) >= 12+int(plen) {
|
|
copy(payload, data[12:12+plen])
|
|
}
|
|
|
|
ts := uint64(0)
|
|
if len(data) >= 4118 {
|
|
ts = binary.LittleEndian.Uint64(data[4110:4118])
|
|
}
|
|
|
|
return &HTTPPlainEvent{
|
|
SrcIP: binary.LittleEndian.Uint32(data[0:4]),
|
|
DstIP: binary.LittleEndian.Uint32(data[4:8]),
|
|
SrcPort: binary.LittleEndian.Uint16(data[8:10]),
|
|
DstPort: binary.LittleEndian.Uint16(data[10:12]),
|
|
Payload: payload,
|
|
PayloadLen: plen,
|
|
Timestamp: ts,
|
|
}, nil
|
|
}
|
|
|
|
// ReadAcceptEvent lit un événement accept4 depuis le RingBuffer.
|
|
func (l *Loader) ReadAcceptEvent(ctx context.Context) (*AcceptEvent, error) {
|
|
rec, err := readRecord(ctx, l.AcceptReader)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data := rec.RawSample
|
|
// struct accept_event: pid_tgid(8)+fd(4)+src_ip(4)+src_port(2)+ts(8) = 26
|
|
if len(data) < 26 {
|
|
return nil, fmt.Errorf("accept_event trop court: %d octets", len(data))
|
|
}
|
|
|
|
pidTGID := binary.LittleEndian.Uint64(data[0:8])
|
|
return &AcceptEvent{
|
|
PID: uint32(pidTGID & 0xFFFFFFFF),
|
|
TGID: uint32(pidTGID >> 32),
|
|
FD: binary.LittleEndian.Uint32(data[8:12]),
|
|
SrcIP: binary.LittleEndian.Uint32(data[12:16]),
|
|
SrcPort: binary.LittleEndian.Uint16(data[16:18]),
|
|
Timestamp: binary.LittleEndian.Uint64(data[18:26]),
|
|
}, nil
|
|
}
|
|
|
|
// readRecord lit un record brut depuis un RingBuffer avec annulation via context.
|
|
func readRecord(ctx context.Context, rd *ringbuf.Reader) (ringbuf.Record, error) {
|
|
type result struct {
|
|
rec ringbuf.Record
|
|
err error
|
|
}
|
|
ch := make(chan result, 1)
|
|
go func() {
|
|
rec, err := rd.Read()
|
|
ch <- result{rec, err}
|
|
}()
|
|
select {
|
|
case <-ctx.Done():
|
|
rd.Close() // débloque le Read() bloquant
|
|
return ringbuf.Record{}, ctx.Err()
|
|
case r := <-ch:
|
|
return r.rec, r.err
|
|
}
|
|
}
|