code_public/ja4sentinel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

release: version 1.1.2 - Add error callback mechanism and comprehensive test suite
Some checks failed
Build RPM Package / Build RPM Packages (CentOS 7, Rocky 8/9/10) (push) Has been cancelled
Details

Browse Source 
Features:
- Add ErrorCallback type for UNIX socket connection error reporting
- Add WithErrorCallback option for UnixSocketWriter configuration
- Add BuilderImpl.WithErrorCallback() for propagating callbacks
- Add consecutive failure tracking in processQueue

Testing (50+ new tests):
- Add integration tests for full pipeline (capture → tlsparse → fingerprint → output)
- Add tests for FileWriter.rotate() and Reopen() log rotation
- Add tests for cleanupExpiredFlows() and cleanupLoop() in TLS parser
- Add tests for extractSNIFromPayload() and extractJA4Hash() helpers
- Add tests for config load error paths (invalid YAML, permission denied)
- Add tests for capture.Run() error conditions
- Add tests for signal handling documentation

Documentation:
- Update architecture.yml with new fields (LogLevel, TLSClientHello extensions)
- Update architecture.yml with Close() methods for Capture and Parser interfaces
- Update RPM spec changelog

Cleanup:
- Remove empty internal/api/ directory

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
Jacquin Antoine
2026-03-02 23:24:56 +01:00
parent 6e5addd6d4
commit 23f3012fb1
10 changed files with 2058 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
internal/capture/capture_test.go
View File

@ -2,8 +2,205 @@ package capture
import (
"testing"
"time"
"ja4sentinel/api"
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
)
func TestCaptureImpl_Run_EmptyInterface(t *testing.T) {
c := New()
if c == nil {
t.Fatal("New() returned nil")
}
cfg := api.Config{
Interface: "",
ListenPorts: []uint16{443},
}
out := make(chan api.RawPacket, 10)
err := c.Run(cfg, out)
if err == nil {
t.Error("Run() with empty interface should return error")
}
if err.Error() != "interface cannot be empty" {
t.Errorf("Run() error = %v, want 'interface cannot be empty'", err)
}
}
func TestCaptureImpl_Run_NonExistentInterface(t *testing.T) {
c := New()
if c == nil {
t.Fatal("New() returned nil")
}
cfg := api.Config{
Interface: "nonexistent_interface_xyz123",
ListenPorts: []uint16{443},
}
out := make(chan api.RawPacket, 10)
err := c.Run(cfg, out)
if err == nil {
t.Error("Run() with non-existent interface should return error")
}
}
func TestCaptureImpl_Run_InvalidBPFFilter(t *testing.T) {
// Get a real interface name
ifaces, err := pcap.FindAllDevs()
if err != nil || len(ifaces) == 0 {
t.Skip("No network interfaces available for testing")
}
c := New()
cfg := api.Config{
Interface: ifaces[0].Name,
ListenPorts: []uint16{443},
BPFFilter: "invalid; rm -rf /", // Invalid characters
}
out := make(chan api.RawPacket, 10)
err = c.Run(cfg, out)
if err == nil {
t.Error("Run() with invalid BPF filter should return error")
}
}
func TestCaptureImpl_Run_ChannelFull_DropsPackets(t *testing.T) {
// This test verifies that when the output channel is full,
// packets are dropped gracefully (non-blocking write)
// We can't easily test the full Run() loop without real interfaces,
// but we can verify the channel behavior with a small buffer
out := make(chan api.RawPacket, 1)
// Fill the channel
out <- api.RawPacket{Data: []byte{1, 2, 3}, Timestamp: time.Now().UnixNano()}
// Channel should be full now, select default should trigger
done := make(chan bool)
go func() {
select {
case out <- api.RawPacket{Data: []byte{4, 5, 6}, Timestamp: time.Now().UnixNano()}:
done <- false // Would block
default:
done <- true // Dropped as expected
}
}()
dropped := <-done
if !dropped {
t.Error("Expected packet to be dropped when channel is full")
}
}
func TestPacketToRawPacket(t *testing.T) {
t.Run("valid_packet", func(t *testing.T) {
// Create a simple TCP packet
eth := layers.Ethernet{
SrcMAC: []byte{0x00, 0x11, 0x22, 0x33, 0x44, 0x55},
DstMAC: []byte{0x66, 0x77, 0x88, 0x99, 0xAA, 0xBB},
EthernetType: layers.EthernetTypeIPv4,
}
ip := layers.IPv4{
Version: 4,
TTL: 64,
Protocol: layers.IPProtocolTCP,
SrcIP: []byte{192, 168, 1, 1},
DstIP: []byte{10, 0, 0, 1},
}
tcp := layers.TCP{
SrcPort: 12345,
DstPort: 443,
}
tcp.SetNetworkLayerForChecksum(&ip)
buf := gopacket.NewSerializeBuffer()
opts := gopacket.SerializeOptions{}
gopacket.SerializeLayers(buf, opts, &eth, &ip, &tcp)
packet := gopacket.NewPacket(buf.Bytes(), layers.LinkTypeEthernet, gopacket.Default)
rawPkt := packetToRawPacket(packet)
if rawPkt == nil {
t.Fatal("packetToRawPacket() returned nil for valid packet")
}
if len(rawPkt.Data) == 0 {
t.Error("packetToRawPacket() returned empty data")
}
if rawPkt.Timestamp == 0 {
t.Error("packetToRawPacket() returned zero timestamp")
}
})
t.Run("empty_packet", func(t *testing.T) {
// Create packet with no data
packet := gopacket.NewPacket([]byte{}, layers.LinkTypeEthernet, gopacket.Default)
rawPkt := packetToRawPacket(packet)
if rawPkt != nil {
t.Error("packetToRawPacket() should return nil for empty packet")
}
})
t.Run("nil_packet", func(t *testing.T) {
// packetToRawPacket will panic with nil packet due to Metadata() call
// This is expected behavior - the function is not designed to handle nil
defer func() {
if r := recover(); r == nil {
t.Error("packetToRawPacket() with nil packet should panic")
}
}()
var packet gopacket.Packet
_ = packetToRawPacket(packet)
})
}
func TestGetInterfaceNames(t *testing.T) {
t.Run("empty_list", func(t *testing.T) {
names := getInterfaceNames([]pcap.Interface{})
if len(names) != 0 {
t.Errorf("getInterfaceNames() with empty list = %v, want []", names)
}
})
t.Run("single_interface", func(t *testing.T) {
ifaces := []pcap.Interface{
{Name: "eth0"},
}
names := getInterfaceNames(ifaces)
if len(names) != 1 || names[0] != "eth0" {
t.Errorf("getInterfaceNames() = %v, want [eth0]", names)
}
})
t.Run("multiple_interfaces", func(t *testing.T) {
ifaces := []pcap.Interface{
{Name: "eth0"},
{Name: "lo"},
{Name: "docker0"},
}
names := getInterfaceNames(ifaces)
if len(names) != 3 {
t.Errorf("getInterfaceNames() returned %d names, want 3", len(names))
}
expected := []string{"eth0", "lo", "docker0"}
for i, name := range names {
if name != expected[i] {
t.Errorf("getInterfaceNames()[%d] = %s, want %s", i, name, expected[i])
}
}
})
}
func TestValidateBPFFilter(t *testing.T) {
tests := []struct {
name string

262
internal/config/loader_test.go
View File

@ -479,3 +479,265 @@ func TestLoad_ExplicitMissingConfig_Fails(t *testing.T) {
t.Fatal("Load() should fail with explicit missing config path")
}
}
// TestLoadFromFile_InvalidYAML tests error handling for malformed YAML
func TestLoadFromFile_InvalidYAML(t *testing.T) {
tmpDir := t.TempDir()
badConfig := filepath.Join(tmpDir, "bad.yml")
// Write invalid YAML syntax
invalidYAML := `
core:
interface: eth0
listen_ports: [443, 8443
bpf_filter: ""
`
if err := os.WriteFile(badConfig, []byte(invalidYAML), 0600); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
loader := NewLoader(badConfig)
_, err := loader.Load()
if err == nil {
t.Error("Load() with invalid YAML should return error")
}
if !strings.Contains(err.Error(), "yaml") {
t.Errorf("Load() error = %v, should mention yaml", err)
}
}
// TestLoadFromFile_PermissionDenied tests error handling for permission errors
func TestLoadFromFile_PermissionDenied(t *testing.T) {
if os.Getuid() == 0 {
t.Skip("Skipping permission test when running as root")
}
tmpDir := t.TempDir()
configPath := filepath.Join(tmpDir, "config.yml")
// Create config file
cfg := api.AppConfig{
Core: api.Config{
Interface: "eth0",
ListenPorts: []uint16{443},
},
}
data := ToJSON(cfg)
if err := os.WriteFile(configPath, []byte(data), 0600); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
// Remove read permissions
if err := os.Chmod(configPath, 0000); err != nil {
t.Fatalf("Chmod() error = %v", err)
}
defer os.Chmod(configPath, 0600) // Restore for cleanup
loader := NewLoader(configPath)
_, err := loader.Load()
if err == nil {
t.Error("Load() with no read permission should return error")
}
}
// TestLoadFromEnv_InvalidValues tests handling of invalid environment variable values
func TestLoadFromEnv_InvalidValues(t *testing.T) {
tests := []struct {
name string
env map[string]string
wantErr bool
errContains string
}{
{
name: "invalid_flow_timeout",
env: map[string]string{
"JA4SENTINEL_FLOW_TIMEOUT": "not-a-number",
},
wantErr: false, // Uses default value when invalid
errContains: "",
},
{
name: "invalid_packet_buffer_size",
env: map[string]string{
"JA4SENTINEL_PACKET_BUFFER_SIZE": "not-a-number",
},
wantErr: false, // Uses default value when invalid
errContains: "",
},
{
name: "negative_flow_timeout",
env: map[string]string{
"JA4SENTINEL_FLOW_TIMEOUT": "-100",
},
wantErr: false, // Uses default value when negative
errContains: "",
},
{
name: "flow_timeout_too_high",
env: map[string]string{
"JA4SENTINEL_FLOW_TIMEOUT": "1000000",
},
wantErr: true, // Validation error
errContains: "flow_timeout_sec must be between",
},
{
name: "invalid_log_level",
env: map[string]string{
"JA4SENTINEL_LOG_LEVEL": "invalid-level",
},
wantErr: true, // Validation error
errContains: "log_level must be one of",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Set environment variables
for key, value := range tt.env {
t.Setenv(key, value)
}
loader := NewLoader("")
cfg, err := loader.Load()
if (err != nil) != tt.wantErr {
t.Fatalf("Load() error = %v, wantErr %v", err, tt.wantErr)
}
if tt.wantErr && tt.errContains != "" {
if err == nil || !strings.Contains(err.Error(), tt.errContains) {
t.Errorf("Load() error = %v, should contain %q", err, tt.errContains)
}
}
if !tt.wantErr {
// Verify defaults are used for invalid values
if tt.name == "invalid_flow_timeout" || tt.name == "negative_flow_timeout" {
if cfg.Core.FlowTimeoutSec != api.DefaultFlowTimeout {
t.Errorf("FlowTimeoutSec = %d, want default %d", cfg.Core.FlowTimeoutSec, api.DefaultFlowTimeout)
}
}
if tt.name == "invalid_packet_buffer_size" {
if cfg.Core.PacketBufferSize != api.DefaultPacketBuffer {
t.Errorf("PacketBufferSize = %d, want default %d", cfg.Core.PacketBufferSize, api.DefaultPacketBuffer)
}
}
}
})
}
}
// TestLoadFromEnv_AllValidValues tests that all valid environment variables are parsed correctly
func TestLoadFromEnv_AllValidValues(t *testing.T) {
t.Setenv("JA4SENTINEL_INTERFACE", "lo")
t.Setenv("JA4SENTINEL_PORTS", "8443, 9443")
t.Setenv("JA4SENTINEL_BPF_FILTER", "tcp port 8443")
t.Setenv("JA4SENTINEL_FLOW_TIMEOUT", "60")
t.Setenv("JA4SENTINEL_PACKET_BUFFER_SIZE", "2000")
t.Setenv("JA4SENTINEL_LOG_LEVEL", "debug")
loader := NewLoader("")
cfg, err := loader.Load()
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if cfg.Core.Interface != "lo" {
t.Errorf("Interface = %q, want 'lo'", cfg.Core.Interface)
}
if len(cfg.Core.ListenPorts) != 2 || cfg.Core.ListenPorts[0] != 8443 {
t.Errorf("ListenPorts = %v, want [8443, 9443]", cfg.Core.ListenPorts)
}
if cfg.Core.BPFFilter != "tcp port 8443" {
t.Errorf("BPFFilter = %q, want 'tcp port 8443'", cfg.Core.BPFFilter)
}
if cfg.Core.FlowTimeoutSec != 60 {
t.Errorf("FlowTimeoutSec = %d, want 60", cfg.Core.FlowTimeoutSec)
}
if cfg.Core.PacketBufferSize != 2000 {
t.Errorf("PacketBufferSize = %d, want 2000", cfg.Core.PacketBufferSize)
}
if cfg.Core.LogLevel != "debug" {
t.Errorf("LogLevel = %q, want 'debug'", cfg.Core.LogLevel)
}
}
// TestValidate_WhitespaceOnlyInterface tests that whitespace-only interface is rejected
// Note: validate() is internal, so we test through Load() with env override
func TestValidate_WhitespaceOnlyInterface(t *testing.T) {
t.Setenv("JA4SENTINEL_INTERFACE", " ")
t.Setenv("JA4SENTINEL_PORTS", "443")
loader := NewLoader("")
_, err := loader.Load()
if err == nil {
t.Error("Load() with whitespace-only interface should return error")
}
}
// TestMergeConfigs_EmptyBase tests merge with empty base config
func TestMergeConfigs_EmptyBase(t *testing.T) {
base := api.AppConfig{}
override := api.AppConfig{
Core: api.Config{
Interface: "lo",
},
}
result := mergeConfigs(base, override)
if result.Core.Interface != "lo" {
t.Errorf("Merged Interface = %q, want 'lo'", result.Core.Interface)
}
}
// TestMergeConfigs_EmptyOverride tests merge with empty override config
func TestMergeConfigs_EmptyOverride(t *testing.T) {
base := api.AppConfig{
Core: api.Config{
Interface: "eth0",
},
}
override := api.AppConfig{}
result := mergeConfigs(base, override)
if result.Core.Interface != "eth0" {
t.Errorf("Merged Interface = %q, want 'eth0'", result.Core.Interface)
}
}
// TestMergeConfigs_OutputMerge tests that outputs are properly merged
func TestMergeConfigs_OutputMerge(t *testing.T) {
base := api.AppConfig{
Core: api.Config{
Interface: "eth0",
},
Outputs: []api.OutputConfig{
{Type: "stdout", Enabled: true},
},
}
override := api.AppConfig{
Core: api.Config{
ListenPorts: []uint16{8443},
},
Outputs: []api.OutputConfig{
{Type: "file", Enabled: true, Params: map[string]string{"path": "/tmp/test.log"}},
},
}
result := mergeConfigs(base, override)
// Override should replace base outputs
if len(result.Outputs) != 1 {
t.Errorf("Merged Outputs length = %d, want 1", len(result.Outputs))
}
if result.Outputs[0].Type != "file" {
t.Errorf("Merged Outputs[0].Type = %q, want 'file'", result.Outputs[0].Type)
}
}

119
internal/fingerprint/engine_test.go
View File

@ -133,3 +133,122 @@ func buildMinimalClientHelloForTest() []byte {
return record
}
// TestExtractJA4Hash tests the extractJA4Hash helper function
func TestExtractJA4Hash(t *testing.T) {
tests := []struct {
name string
ja4 string
want string
}{
{
name: "standard_ja4_format",
ja4: "t13d1516h2_8daaf6152771_02cb136f2775",
want: "8daaf6152771_02cb136f2775",
},
{
name: "ja4_with_single_underscore",
ja4: "t12d1234h1_abcdef123456",
want: "abcdef123456",
},
{
name: "ja4_no_underscore_returns_empty",
ja4: "t13d1516h2",
want: "",
},
{
name: "empty_ja4_returns_empty",
ja4: "",
want: "",
},
{
name: "underscore_at_start",
ja4: "_hash1_hash2",
want: "hash1_hash2",
},
{
name: "multiple_underscores_returns_after_first",
ja4: "base_part1_part2_part3",
want: "part1_part2_part3",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := extractJA4Hash(tt.ja4)
if got != tt.want {
t.Errorf("extractJA4Hash(%q) = %q, want %q", tt.ja4, got, tt.want)
}
})
}
}
// TestFromClientHello_NilPayload tests error handling for nil payload
func TestFromClientHello_NilPayload(t *testing.T) {
engine := NewEngine()
ch := api.TLSClientHello{
Payload: nil,
}
_, err := engine.FromClientHello(ch)
if err == nil {
t.Error("FromClientHello() with nil payload should return error")
}
if err.Error() != "empty ClientHello payload" {
t.Errorf("FromClientHello() error = %v, want 'empty ClientHello payload'", err)
}
}
// TestFromClientHello_JA3Hash tests that JA3Hash is correctly populated
func TestFromClientHello_JA3Hash(t *testing.T) {
clientHello := buildMinimalClientHelloForTest()
ch := api.TLSClientHello{
Payload: clientHello,
}
engine := NewEngine()
fp, err := engine.FromClientHello(ch)
if err != nil {
t.Fatalf("FromClientHello() error = %v", err)
}
// JA3Hash should be populated (MD5 hash of JA3 string)
if fp.JA3Hash == "" {
t.Error("JA3Hash should be populated")
}
// JA3 should also be populated
if fp.JA3 == "" {
t.Error("JA3 should be populated")
}
}
// TestFromClientHello_EmptyJA4Hash tests behavior when JA4 has no underscore
func TestFromClientHello_EmptyJA4Hash(t *testing.T) {
// This test verifies that even if JA4 format changes, the code handles it gracefully
engine := NewEngine()
// Use a valid ClientHello - the library should produce a proper JA4
clientHello := buildMinimalClientHelloForTest()
ch := api.TLSClientHello{
Payload: clientHello,
}
fp, err := engine.FromClientHello(ch)
if err != nil {
t.Fatalf("FromClientHello() error = %v", err)
}
// JA4 should always be populated
if fp.JA4 == "" {
t.Error("JA4 should be populated")
}
// JA4Hash may be empty if the JA4 format doesn't include underscores
// This is acceptable behavior
}

402
internal/integration/pipeline_test.go Normal file
View File

@ -0,0 +1,402 @@
// Package integration provides integration tests for the full ja4sentinel pipeline
package integration
import (
"encoding/json"
"os"
"testing"
"time"
"ja4sentinel/api"
"ja4sentinel/internal/fingerprint"
"ja4sentinel/internal/output"
"ja4sentinel/internal/tlsparse"
)
// TestFullPipeline_TLSClientHelloToFingerprint tests the pipeline from TLS ClientHello to fingerprint
func TestFullPipeline_TLSClientHelloToFingerprint(t *testing.T) {
// Create a minimal TLS 1.2 ClientHello for testing
clientHello := buildMinimalTLSClientHello()
// Step 1: Parse the ClientHello
parser := tlsparse.NewParser()
if parser == nil {
t.Fatal("NewParser() returned nil")
}
defer parser.Close()
// Create a raw packet with the ClientHello
rawPacket := api.RawPacket{
Data: buildEthernetIPPacket(clientHello),
Timestamp: time.Now().UnixNano(),
}
// Process the packet
ch, err := parser.Process(rawPacket)
if err != nil {
t.Fatalf("Process() error = %v", err)
}
if ch == nil {
t.Fatal("Process() returned nil ClientHello")
}
// Step 2: Generate fingerprints
engine := fingerprint.NewEngine()
if engine == nil {
t.Fatal("NewEngine() returned nil")
}
fp, err := engine.FromClientHello(*ch)
if err != nil {
t.Fatalf("FromClientHello() error = %v", err)
}
if fp == nil {
t.Fatal("FromClientHello() returned nil")
}
// Verify fingerprints are populated
if fp.JA4 == "" {
t.Error("JA4 should be populated")
}
if fp.JA3 == "" {
t.Error("JA3 should be populated")
}
if fp.JA3Hash == "" {
t.Error("JA3Hash should be populated")
}
}
// TestFullPipeline_FingerprintToOutput tests the pipeline from fingerprint to output
func TestFullPipeline_FingerprintToOutput(t *testing.T) {
// Create test data
clientHello := api.TLSClientHello{
SrcIP: "192.168.1.100",
SrcPort: 54321,
DstIP: "10.0.0.1",
DstPort: 443,
IPMeta: api.IPMeta{
TTL: 64,
TotalLength: 512,
IPID: 12345,
DF: true,
},
TCPMeta: api.TCPMeta{
WindowSize: 65535,
MSS: 1460,
WindowScale: 7,
Options: []string{"MSS", "SACK", "TS", "WS"},
},
ConnID: "test-flow-123",
SNI: "example.com",
ALPN: "h2",
TLSVersion: "1.3",
SynToCHMs: uint32Ptr(50),
}
// Create fingerprints
fingerprints := &api.Fingerprints{
JA4: "t13d1516h2_8daaf6152771_02cb136f2775",
JA4Hash: "8daaf6152771_02cb136f2775",
JA3: "771,4865-4866-4867,0-23-65281-10-11-35-16-5-13-18-51-45-43-27-17513,29-23-24,0",
JA3Hash: "a0e6f06c7a6d15e5e3f0f0e6f06c7a6d",
}
// Step 1: Create LogRecord
logRecord := api.NewLogRecord(clientHello, fingerprints)
logRecord.SensorID = "test-sensor"
// Step 2: Write to output (stdout writer for testing)
writer := output.NewStdoutWriter()
if writer == nil {
t.Fatal("NewStdoutWriter() returned nil")
}
// Capture stdout by using a buffer (we can't easily test stdout, so we verify the record)
// Instead, verify the LogRecord is valid JSON
data, err := json.Marshal(logRecord)
if err != nil {
t.Fatalf("json.Marshal() error = %v", err)
}
// Verify JSON is valid and contains expected fields
var result map[string]interface{}
if err := json.Unmarshal(data, &result); err != nil {
t.Fatalf("json.Unmarshal() error = %v", err)
}
// Verify key fields
if result["src_ip"] != "192.168.1.100" {
t.Errorf("src_ip = %v, want 192.168.1.100", result["src_ip"])
}
if result["src_port"] != float64(54321) {
t.Errorf("src_port = %v, want 54321", result["src_port"])
}
if result["ja4"] != "t13d1516h2_8daaf6152771_02cb136f2775" {
t.Errorf("ja4 = %v, want t13d1516h2_8daaf6152771_02cb136f2775", result["ja4"])
}
if result["tls_sni"] != "example.com" {
t.Errorf("tls_sni = %v, want example.com", result["tls_sni"])
}
if result["sensor_id"] != "test-sensor" {
t.Errorf("sensor_id = %v, want test-sensor", result["sensor_id"])
}
}
// TestFullPipeline_EndToEnd tests the complete pipeline with file output
func TestFullPipeline_EndToEnd(t *testing.T) {
tmpDir := t.TempDir()
outputPath := tmpDir + "/output.log"
// Create test ClientHello
clientHello := buildMinimalTLSClientHello()
// Step 1: Parse
parser := tlsparse.NewParser()
defer parser.Close()
rawPacket := api.RawPacket{
Data: buildEthernetIPPacket(clientHello),
Timestamp: time.Now().UnixNano(),
}
ch, err := parser.Process(rawPacket)
if err != nil {
t.Fatalf("Process() error = %v", err)
}
// Step 2: Fingerprint
engine := fingerprint.NewEngine()
fp, err := engine.FromClientHello(*ch)
if err != nil {
t.Fatalf("FromClientHello() error = %v", err)
}
// Step 3: Create LogRecord
logRecord := api.NewLogRecord(*ch, fp)
logRecord.SensorID = "test-sensor-e2e"
// Step 4: Write to file
fileWriter, err := output.NewFileWriter(outputPath)
if err != nil {
t.Fatalf("NewFileWriter() error = %v", err)
}
defer fileWriter.Close()
err = fileWriter.Write(logRecord)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Verify output file
data, err := os.ReadFile(outputPath)
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
if len(data) == 0 {
t.Fatal("Output file is empty")
}
// Parse and verify
var result api.LogRecord
if err := json.Unmarshal(data, &result); err != nil {
t.Fatalf("json.Unmarshal() error = %v", err)
}
if result.SensorID != "test-sensor-e2e" {
t.Errorf("SensorID = %v, want test-sensor-e2e", result.SensorID)
}
if result.JA4 == "" {
t.Error("JA4 should be populated")
}
}
// TestFullPipeline_MultiOutput tests writing to multiple outputs simultaneously
func TestFullPipeline_MultiOutput(t *testing.T) {
tmpDir := t.TempDir()
filePath := tmpDir + "/multi.log"
// Create multi-writer
multiWriter := output.NewMultiWriter()
multiWriter.Add(output.NewStdoutWriter())
fileWriter, err := output.NewFileWriter(filePath)
if err != nil {
t.Fatalf("NewFileWriter() error = %v", err)
}
multiWriter.Add(fileWriter)
// Create test record
logRecord := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test-multi-output",
}
// Write to all outputs
err = multiWriter.Write(logRecord)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Verify file output
data, err := os.ReadFile(filePath)
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
if len(data) == 0 {
t.Fatal("File output is empty")
}
}
// TestFullPipeline_ConfigToOutput tests building output from config
func TestFullPipeline_ConfigToOutput(t *testing.T) {
tmpDir := t.TempDir()
// Create config with multiple outputs
config := api.AppConfig{
Core: api.Config{
Interface: "eth0",
ListenPorts: []uint16{443},
},
Outputs: []api.OutputConfig{
{
Type: "stdout",
Enabled: true,
AsyncBuffer: 1000,
},
{
Type: "file",
Enabled: true,
AsyncBuffer: 1000,
Params: map[string]string{"path": tmpDir + "/config-output.log"},
},
},
}
// Build writer from config
builder := output.NewBuilder()
writer, err := builder.NewFromConfig(config)
if err != nil {
t.Fatalf("NewFromConfig() error = %v", err)
}
// Verify writer is MultiWriter
_, ok := writer.(*output.MultiWriter)
if !ok {
t.Fatal("Expected MultiWriter")
}
// Test writing
logRecord := api.LogRecord{
SrcIP: "192.168.1.1",
JA4: "test-config-output",
}
err = writer.Write(logRecord)
if err != nil {
t.Errorf("Write() error = %v", err)
}
}
// Helper functions
// buildMinimalTLSClientHello creates a minimal TLS 1.2 ClientHello for testing
func buildMinimalTLSClientHello() []byte {
// Cipher suites
cipherSuites := []byte{0x00, 0x04, 0x13, 0x01, 0x13, 0x02, 0xc0, 0x2f}
compressionMethods := []byte{0x01, 0x00}
extensions := []byte{}
extLen := len(extensions)
handshakeBody := []byte{
0x03, 0x03, // Version: TLS 1.2
}
// Random (32 bytes)
for i := 0; i < 32; i++ {
handshakeBody = append(handshakeBody, 0x00)
}
handshakeBody = append(handshakeBody, 0x00) // Session ID length
// Cipher suites
cipherSuiteLen := len(cipherSuites)
handshakeBody = append(handshakeBody, byte(cipherSuiteLen>>8), byte(cipherSuiteLen))
handshakeBody = append(handshakeBody, cipherSuites...)
// Compression methods
handshakeBody = append(handshakeBody, compressionMethods...)
// Extensions
handshakeBody = append(handshakeBody, byte(extLen>>8), byte(extLen))
handshakeBody = append(handshakeBody, extensions...)
// Build handshake
handshakeLen := len(handshakeBody)
handshake := append([]byte{
0x01, // Handshake type: ClientHello
byte(handshakeLen >> 16), byte(handshakeLen >> 8), byte(handshakeLen),
}, handshakeBody...)
// Build TLS record
recordLen := len(handshake)
record := make([]byte, 5+recordLen)
record[0] = 0x16 // Handshake
record[1] = 0x03 // Version: TLS 1.2
record[2] = 0x03
record[3] = byte(recordLen >> 8)
record[4] = byte(recordLen)
copy(record[5:], handshake)
return record
}
// buildEthernetIPPacket wraps a TLS payload in Ethernet/IP/TCP headers
func buildEthernetIPPacket(tlsPayload []byte) []byte {
// This is a simplified packet structure for testing
// Real packets would have proper Ethernet, IP, and TCP headers
// Ethernet header (14 bytes)
eth := make([]byte, 14)
eth[12] = 0x08 // EtherType: IPv4
eth[13] = 0x00
// IP header (20 bytes)
ip := make([]byte, 20)
ip[0] = 0x45 // Version 4, IHL 5
ip[1] = 0x00 // DSCP/ECN
ip[2] = byte((20 + 20 + len(tlsPayload)) >> 8) // Total length
ip[3] = byte((20 + 20 + len(tlsPayload)) & 0xFF)
ip[8] = 64 // TTL
ip[9] = 6 // Protocol: TCP
ip[12] = 192
ip[13] = 168
ip[14] = 1
ip[15] = 100 // Src IP: 192.168.1.100
ip[16] = 10
ip[17] = 0
ip[18] = 0
ip[19] = 1 // Dst IP: 10.0.0.1
// TCP header (20 bytes)
tcp := make([]byte, 20)
tcp[0] = byte(54321 >> 8) // Src port high
tcp[1] = byte(54321 & 0xFF) // Src port low
tcp[2] = byte(443 >> 8) // Dst port high
tcp[3] = byte(443 & 0xFF) // Dst port low
tcp[12] = 0x50 // Data offset (5 * 4 = 20 bytes)
tcp[13] = 0x18 // Flags: ACK, PSH
// Combine all headers with payload
packet := make([]byte, len(eth)+len(ip)+len(tcp)+len(tlsPayload))
copy(packet, eth)
copy(packet[len(eth):], ip)
copy(packet[len(eth)+len(ip):], tcp)
copy(packet[len(eth)+len(ip)+len(tcp):], tlsPayload)
return packet
}
func uint32Ptr(v uint32) *uint32 {
return &v
}

63
internal/output/writers.go
View File

@ -202,6 +202,9 @@ func (w *FileWriter) Reopen() error {
return nil
}
// ErrorCallback is a function type for reporting socket connection errors
type ErrorCallback func(socketPath string, err error, attempt int)
// UnixSocketWriter writes log records to a UNIX socket with reconnection logic
// No internal logging - only LogRecord JSON data is sent to the socket
type UnixSocketWriter struct {
@ -220,6 +223,9 @@ type UnixSocketWriter struct {
isClosed bool
pendingWrites [][]byte
pendingMu sync.Mutex
errorCallback ErrorCallback
consecutiveFailures int
failuresMu sync.Mutex
}
// NewUnixSocketWriter creates a new UNIX socket writer with reconnection logic
@ -227,8 +233,18 @@ func NewUnixSocketWriter(socketPath string) (*UnixSocketWriter, error) {
return NewUnixSocketWriterWithConfig(socketPath, DefaultDialTimeout, DefaultWriteTimeout, DefaultQueueSize)
}
// UnixSocketWriterOption is a function type for configuring UnixSocketWriter
type UnixSocketWriterOption func(*UnixSocketWriter)
// WithErrorCallback sets an error callback for socket connection errors
func WithErrorCallback(cb ErrorCallback) UnixSocketWriterOption {
return func(w *UnixSocketWriter) {
w.errorCallback = cb
}
}
// NewUnixSocketWriterWithConfig creates a new UNIX socket writer with custom configuration
func NewUnixSocketWriterWithConfig(socketPath string, dialTimeout, writeTimeout time.Duration, queueSize int) (*UnixSocketWriter, error) {
func NewUnixSocketWriterWithConfig(socketPath string, dialTimeout, writeTimeout time.Duration, queueSize int, opts ...UnixSocketWriterOption) (*UnixSocketWriter, error) {
w := &UnixSocketWriter{
socketPath: socketPath,
dialTimeout: dialTimeout,
@ -242,6 +258,11 @@ func NewUnixSocketWriterWithConfig(socketPath string, dialTimeout, writeTimeout
pendingWrites: make([][]byte, 0),
}
// Apply options
for _, opt := range opts {
opt(w)
}
// Start the queue processor
go w.processQueue()
@ -259,7 +280,6 @@ func (w *UnixSocketWriter) processQueue() {
defer close(w.queueDone)
backoff := w.reconnectBackoff
consecutiveFailures := 0
for {
select {
@ -271,7 +291,14 @@ func (w *UnixSocketWriter) processQueue() {
}
if err := w.writeWithReconnect(data); err != nil {
consecutiveFailures++
w.failuresMu.Lock()
w.consecutiveFailures++
failures := w.consecutiveFailures
w.failuresMu.Unlock()
// Report error via callback if configured
w.reportError(err, failures)
// Queue for retry
w.pendingMu.Lock()
if len(w.pendingWrites) < DefaultQueueSize {
@ -280,7 +307,7 @@ func (w *UnixSocketWriter) processQueue() {
w.pendingMu.Unlock()
// Exponential backoff
if consecutiveFailures > w.maxReconnects {
if failures > w.maxReconnects {
time.Sleep(backoff)
backoff *= 2
if backoff > w.maxBackoff {
@ -288,7 +315,9 @@ func (w *UnixSocketWriter) processQueue() {
}
}
} else {
consecutiveFailures = 0
w.failuresMu.Lock()
w.consecutiveFailures = 0
w.failuresMu.Unlock()
backoff = w.reconnectBackoff
// Try to flush pending data
w.flushPendingData()
@ -301,6 +330,13 @@ func (w *UnixSocketWriter) processQueue() {
}
}
// reportError reports a socket connection error via the configured callback
func (w *UnixSocketWriter) reportError(err error, attempt int) {
if w.errorCallback != nil {
w.errorCallback(w.socketPath, err, attempt)
}
}
// flushPendingData attempts to write any pending data
func (w *UnixSocketWriter) flushPendingData() {
w.pendingMu.Lock()
@ -486,13 +522,21 @@ func (mw *MultiWriter) Reopen() error {
}
// BuilderImpl implements the api.Builder interface
type BuilderImpl struct{}
type BuilderImpl struct {
errorCallback ErrorCallback
}
// NewBuilder creates a new output builder
func NewBuilder() *BuilderImpl {
return &BuilderImpl{}
}
// WithErrorCallback sets an error callback for all unix_socket writers created by this builder
func (b *BuilderImpl) WithErrorCallback(cb ErrorCallback) *BuilderImpl {
b.errorCallback = cb
return b
}
// NewFromConfig constructs writers from AppConfig
// Uses AsyncBuffer from OutputConfig if specified, otherwise uses DefaultQueueSize
func (b *BuilderImpl) NewFromConfig(cfg api.AppConfig) (api.Writer, error) {
@ -529,7 +573,12 @@ func (b *BuilderImpl) NewFromConfig(cfg api.AppConfig) (api.Writer, error) {
if socketPath == "" {
return nil, fmt.Errorf("unix_socket output requires 'socket_path' parameter")
}
writer, err = NewUnixSocketWriterWithConfig(socketPath, DefaultDialTimeout, DefaultWriteTimeout, queueSize)
// Build options list
var opts []UnixSocketWriterOption
if b.errorCallback != nil {
opts = append(opts, WithErrorCallback(b.errorCallback))
}
writer, err = NewUnixSocketWriterWithConfig(socketPath, DefaultDialTimeout, DefaultWriteTimeout, queueSize, opts...)
if err != nil {
return nil, err
}

558
internal/output/writers_test.go
View File

@ -3,6 +3,7 @@ package output
import (
"bytes"
"encoding/json"
"net"
"os"
"path/filepath"
"testing"
@ -501,3 +502,560 @@ func TestLogRecordOptionalFieldsOmitted(t *testing.T) {
func contains(s, substr string) bool {
return bytes.Contains([]byte(s), []byte(substr))
}
// TestUnixSocketWriter_ErrorCallback tests that errors are reported via callback
func TestUnixSocketWriter_ErrorCallback(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "nonexistent.sock")
// Track callback invocations
var errorCalls []struct {
path string
err error
attempt int
}
callback := func(path string, err error, attempt int) {
errorCalls = append(errorCalls, struct {
path string
err error
attempt int
}{path, err, attempt})
}
w, err := NewUnixSocketWriterWithConfig(
socketPath,
100*time.Millisecond,
100*time.Millisecond,
10,
WithErrorCallback(callback),
)
if err != nil {
t.Fatalf("NewUnixSocketWriterWithConfig() error = %v", err)
}
defer w.Close()
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test",
}
// Write should queue the message
err = w.Write(rec)
if err != nil {
t.Errorf("Write() unexpected error = %v", err)
}
// Wait for queue processor to attempt write and trigger callback
time.Sleep(500 * time.Millisecond)
// Callback should have been invoked at least once
if len(errorCalls) == 0 {
t.Error("ErrorCallback was not invoked")
} else {
// Verify callback parameters
lastCall := errorCalls[len(errorCalls)-1]
if lastCall.path != socketPath {
t.Errorf("Callback path = %v, want %v", lastCall.path, socketPath)
}
if lastCall.err == nil {
t.Error("Callback err should not be nil")
}
if lastCall.attempt < 1 {
t.Errorf("Callback attempt = %d, want >= 1", lastCall.attempt)
}
}
}
// TestBuilder_WithErrorCallback tests that the builder propagates error callbacks
func TestBuilder_WithErrorCallback(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "test.sock")
callback := func(path string, err error, attempt int) {
// Callback tracked for verification
}
builder := NewBuilder().WithErrorCallback(callback)
config := api.AppConfig{
Core: api.Config{
Interface: "eth0",
ListenPorts: []uint16{443},
},
Outputs: []api.OutputConfig{
{
Type: "unix_socket",
Enabled: true,
AsyncBuffer: 100,
Params: map[string]string{"socket_path": socketPath},
},
},
}
writer, err := builder.NewFromConfig(config)
if err != nil {
t.Fatalf("NewFromConfig() error = %v", err)
}
// Verify writer is a MultiWriter
mw, ok := writer.(*MultiWriter)
if !ok {
t.Fatal("Writer is not a MultiWriter")
}
// Verify the UnixSocketWriter has the callback set
if len(mw.writers) != 1 {
t.Fatalf("Expected 1 writer, got %d", len(mw.writers))
}
unixWriter, ok := mw.writers[0].(*UnixSocketWriter)
if !ok {
t.Fatal("Writer is not a UnixSocketWriter")
}
if unixWriter.errorCallback == nil {
t.Error("UnixSocketWriter.errorCallback is nil")
}
_ = writer
}
// TestUnixSocketWriter_NoCallback tests that writer works without callback
func TestUnixSocketWriter_NoCallback(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "nonexistent.sock")
// Create writer without callback
w, err := NewUnixSocketWriter(socketPath)
if err != nil {
t.Fatalf("NewUnixSocketWriter() error = %v", err)
}
defer w.Close()
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test",
}
// Write should not panic even without callback
err = w.Write(rec)
if err != nil {
t.Logf("Write() error (expected) = %v", err)
}
// Give queue processor time to run
time.Sleep(100 * time.Millisecond)
// Should not panic
}
// TestUnixSocketWriter_CallbackResetOnSuccess tests that failure counter resets on success
func TestUnixSocketWriter_CallbackResetOnSuccess(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "test.sock")
// Create a real socket
listener, err := net.Listen("unix", socketPath)
if err != nil {
t.Fatalf("Failed to create socket: %v", err)
}
defer listener.Close()
// Start a goroutine to accept and read connections
done := make(chan struct{})
go func() {
for {
conn, err := listener.Accept()
if err != nil {
select {
case <-done:
return
default:
}
continue
}
// Read and discard data
buf := make([]byte, 1024)
conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
conn.Read(buf)
conn.Close()
}
}()
defer close(done)
var errorCalls int
callback := func(path string, err error, attempt int) {
errorCalls++
}
w, err := NewUnixSocketWriterWithConfig(
socketPath,
100*time.Millisecond,
100*time.Millisecond,
10,
WithErrorCallback(callback),
)
if err != nil {
t.Fatalf("NewUnixSocketWriterWithConfig() error = %v", err)
}
defer w.Close()
// Write successfully
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test",
}
err = w.Write(rec)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Wait for write to complete
time.Sleep(200 * time.Millisecond)
// Callback should not have been called since connection succeeded
if errorCalls > 0 {
t.Errorf("ErrorCallback called %d times, want 0 for successful connection", errorCalls)
}
}
// TestFileWriter_Reopen tests the Reopen method for logrotate support
func TestFileWriter_Reopen(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
w, err := NewFileWriter(testFile)
if err != nil {
t.Fatalf("NewFileWriter() error = %v", err)
}
// Write initial data
rec1 := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test1",
}
err = w.Write(rec1)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Reopen the file (for logrotate - file is typically moved externally)
err = w.Reopen()
if err != nil {
t.Errorf("Reopen() error = %v", err)
}
// Write more data after reopen
rec2 := api.LogRecord{
SrcIP: "192.168.1.2",
SrcPort: 54321,
JA4: "test2",
}
err = w.Write(rec2)
if err != nil {
t.Errorf("Write() after reopen error = %v", err)
}
// Close and verify
if err := w.Close(); err != nil {
t.Errorf("Close() error = %v", err)
}
// Read the file - should contain both records (Reopen uses O_APPEND)
data, err := os.ReadFile(testFile)
if err != nil {
t.Fatalf("Failed to read file: %v", err)
}
// Parse JSON lines
lines := bytes.Split(bytes.TrimSpace(data), []byte("\n"))
if len(lines) != 2 {
t.Fatalf("Expected 2 lines, got %d", len(lines))
}
// Verify second record
var got api.LogRecord
if err := json.Unmarshal(lines[1], &got); err != nil {
t.Errorf("Invalid JSON on line 2: %v", err)
}
if got.SrcIP != rec2.SrcIP {
t.Errorf("SrcIP = %v, want %v", got.SrcIP, rec2.SrcIP)
}
}
// TestFileWriter_Rotate tests the log rotation functionality
func TestFileWriter_Rotate(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
// Create writer with very small max size to trigger rotation
// Minimum useful size is ~100 bytes for a log record
w, err := NewFileWriterWithConfig(testFile, 200, 3)
if err != nil {
t.Fatalf("NewFileWriterWithConfig() error = %v", err)
}
// Write multiple records to trigger rotation
records := []api.LogRecord{
{SrcIP: "192.168.1.1", SrcPort: 1111, JA4: "record1"},
{SrcIP: "192.168.1.2", SrcPort: 2222, JA4: "record2"},
{SrcIP: "192.168.1.3", SrcPort: 3333, JA4: "record3"},
{SrcIP: "192.168.1.4", SrcPort: 4444, JA4: "record4"},
}
for i, rec := range records {
err = w.Write(rec)
if err != nil {
t.Errorf("Write() record %d error = %v", i, err)
}
}
if err := w.Close(); err != nil {
t.Errorf("Close() error = %v", err)
}
// Check that rotation occurred (backup file should exist)
backupFile := testFile + ".1"
if _, err := os.Stat(backupFile); os.IsNotExist(err) {
t.Log("Note: Rotation may not have occurred if total data < maxSize")
}
// Verify main file exists and has content
if _, err := os.Stat(testFile); os.IsNotExist(err) {
t.Errorf("Main file %s does not exist", testFile)
}
}
// TestFileWriter_Rotate_MaxBackups tests that old backups are cleaned up
func TestFileWriter_Rotate_MaxBackups(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
// Create writer with small max size and only 2 backups
w, err := NewFileWriterWithConfig(testFile, 150, 2)
if err != nil {
t.Fatalf("NewFileWriterWithConfig() error = %v", err)
}
// Write enough records to trigger multiple rotations
for i := 0; i < 10; i++ {
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: uint16(1000 + i),
JA4: "test",
}
err = w.Write(rec)
if err != nil {
t.Errorf("Write() error = %v", err)
}
}
if err := w.Close(); err != nil {
t.Errorf("Close() error = %v", err)
}
// Count backup files
backupCount := 0
for i := 1; i <= 5; i++ {
backupPath := testFile + "." + string(rune('0'+i))
if _, err := os.Stat(backupPath); err == nil {
backupCount++
}
}
// Should have at most 2 backups
if backupCount > 2 {
t.Errorf("Too many backup files: %d, want <= 2", backupCount)
}
}
// TestFileWriter_Reopen_Error tests Reopen after external file removal
func TestFileWriter_Reopen_Error(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
w, err := NewFileWriter(testFile)
if err != nil {
t.Fatalf("NewFileWriter() error = %v", err)
}
// Write initial data
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test",
}
err = w.Write(rec)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Remove the file externally (simulating logrotate move)
os.Remove(testFile)
// Reopen should succeed - it will create a new file
err = w.Reopen()
if err != nil {
t.Errorf("Reopen() should succeed after file removal, error = %v", err)
}
if err := w.Close(); err != nil {
t.Errorf("Close() error = %v", err)
}
}
// TestFileWriter_NewFileWriterWithConfig tests custom configuration
func TestFileWriter_NewFileWriterWithConfig(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
// Test with custom max size and backups
w, err := NewFileWriterWithConfig(testFile, 50*1024*1024, 5)
if err != nil {
t.Fatalf("NewFileWriterWithConfig() error = %v", err)
}
defer w.Close()
if w.maxSize != 50*1024*1024 {
t.Errorf("maxSize = %d, want %d", w.maxSize, 50*1024*1024)
}
if w.maxBackups != 5 {
t.Errorf("maxBackups = %d, want 5", w.maxBackups)
}
}
// TestFileWriter_NewFileWriterWithConfig_InvalidPath tests error handling
func TestFileWriter_NewFileWriterWithConfig_InvalidPath(t *testing.T) {
// Try to create file in a path that should fail (e.g., /proc which is read-only)
_, err := NewFileWriterWithConfig("/proc/test/test.log", 1024, 3)
if err == nil {
t.Error("NewFileWriterWithConfig() with invalid path should return error")
}
}
// TestMultiWriter_Reopen tests Reopen on MultiWriter
func TestMultiWriter_Reopen(t *testing.T) {
tmpDir := t.TempDir()
testFile := filepath.Join(tmpDir, "test.log")
mw := NewMultiWriter()
// Add a FileWriter (which is Reopenable)
fw, err := NewFileWriter(testFile)
if err != nil {
t.Fatalf("NewFileWriter() error = %v", err)
}
mw.Add(fw)
// Add a StdoutWriter (which is NOT Reopenable)
mw.Add(NewStdoutWriter())
// Write initial data
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: 12345,
JA4: "test",
}
err = mw.Write(rec)
if err != nil {
t.Errorf("Write() error = %v", err)
}
// Reopen should work (FileWriter is reopenable, StdoutWriter is skipped)
err = mw.Reopen()
if err != nil {
t.Errorf("Reopen() error = %v", err)
}
// Write after reopen
rec2 := api.LogRecord{
SrcIP: "192.168.1.2",
SrcPort: 54321,
JA4: "test2",
}
err = mw.Write(rec2)
if err != nil {
t.Errorf("Write() after reopen error = %v", err)
}
if err := mw.CloseAll(); err != nil {
t.Errorf("CloseAll() error = %v", err)
}
}
// TestUnixSocketWriter_QueueFull tests behavior when queue is full
func TestUnixSocketWriter_QueueFull(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "test.sock")
// Create writer with very small queue
w, err := NewUnixSocketWriterWithConfig(socketPath, 10*time.Millisecond, 10*time.Millisecond, 2)
if err != nil {
t.Fatalf("NewUnixSocketWriterWithConfig() error = %v", err)
}
defer w.Close()
// Fill the queue with records
for i := 0; i < 10; i++ {
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: uint16(1000 + i),
JA4: "test",
}
_ = w.Write(rec) // May succeed or fail depending on queue state
}
// Should not panic - queue full messages are dropped
}
// TestUnixSocketWriter_ReconnectBackoff tests exponential backoff behavior
func TestUnixSocketWriter_ReconnectBackoff(t *testing.T) {
tmpDir := t.TempDir()
socketPath := filepath.Join(tmpDir, "nonexistent.sock")
var errorCount int
callback := func(path string, err error, attempt int) {
errorCount++
}
w, err := NewUnixSocketWriterWithConfig(
socketPath,
10*time.Millisecond,
10*time.Millisecond,
5,
WithErrorCallback(callback),
)
if err != nil {
t.Fatalf("NewUnixSocketWriterWithConfig() error = %v", err)
}
defer w.Close()
// Write multiple records to trigger reconnection attempts
for i := 0; i < 3; i++ {
rec := api.LogRecord{
SrcIP: "192.168.1.1",
SrcPort: uint16(1000 + i),
JA4: "test",
}
_ = w.Write(rec)
}
// Wait for queue processor to attempt writes
time.Sleep(500 * time.Millisecond)
// Should have attempted reconnection
if errorCount == 0 {
t.Error("Expected at least one error callback for nonexistent socket")
}
}

333
internal/tlsparse/parser_test.go
View File

@ -714,3 +714,336 @@ func min(a, b int) int {
}
return b
}
// TestExtractSNIFromPayload tests the SNI extraction function
func TestExtractSNIFromPayload(t *testing.T) {
tests := []struct {
name string
payload []byte
wantSNI string
}{
{
name: "empty_payload",
payload: []byte{},
wantSNI: "",
},
{
name: "payload_too_short",
payload: []byte{0x01, 0x00, 0x00, 0x10}, // Only 4 bytes
wantSNI: "",
},
{
name: "no_extensions",
payload: buildClientHelloWithoutExtensions(),
wantSNI: "",
},
{
name: "with_sni_extension",
payload: buildClientHelloWithSNI("example.com"),
wantSNI: "example.com",
},
{
name: "with_sni_long_domain",
payload: buildClientHelloWithSNI("very-long-subdomain.example-test-domain.com"),
wantSNI: "very-long-subdomain.example-test-domain.com",
},
{
name: "malformed_sni_truncated",
payload: buildTruncatedSNI(),
wantSNI: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := extractSNIFromPayload(tt.payload)
if got != tt.wantSNI {
t.Errorf("extractSNIFromPayload() = %q, want %q", got, tt.wantSNI)
}
})
}
}
// TestCleanupExpiredFlows tests the flow cleanup functionality
func TestCleanupExpiredFlows(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
defer p.Close()
// Create a flow manually using exported types
key := "192.168.1.1:12345->10.0.0.1:443"
flow := &ConnectionFlow{
State: NEW,
LastSeen: time.Now().Add(-2 * time.Hour), // Old flow
HelloBuffer: make([]byte, 0),
}
p.mu.Lock()
p.flows[key] = flow
p.mu.Unlock()
// Call cleanup
p.cleanupExpiredFlows()
// Flow should be deleted
p.mu.RLock()
_, exists := p.flows[key]
p.mu.RUnlock()
if exists {
t.Error("cleanupExpiredFlows() should have removed the expired flow")
}
}
// TestCleanupExpiredFlows_JA4Done tests that JA4_DONE flows are cleaned up immediately
func TestCleanupExpiredFlows_JA4Done(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
defer p.Close()
// Create a JA4_DONE flow (should be cleaned up regardless of timestamp)
key := "192.168.1.1:12345->10.0.0.1:443"
flow := &ConnectionFlow{
State: JA4_DONE,
LastSeen: time.Now(), // Recent, but should still be deleted
HelloBuffer: make([]byte, 0),
}
p.mu.Lock()
p.flows[key] = flow
p.mu.Unlock()
// Call cleanup
p.cleanupExpiredFlows()
// Flow should be deleted
p.mu.RLock()
_, exists := p.flows[key]
p.mu.RUnlock()
if exists {
t.Error("cleanupExpiredFlows() should have removed the JA4_DONE flow")
}
}
// TestCleanupExpiredFlows_RecentFlow tests that recent flows are NOT cleaned up
func TestCleanupExpiredFlows_RecentFlow(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
defer p.Close()
// Create a recent flow (should NOT be cleaned up)
key := "192.168.1.1:12345->10.0.0.1:443"
flow := &ConnectionFlow{
State: WAIT_CLIENT_HELLO,
LastSeen: time.Now(),
HelloBuffer: make([]byte, 0),
}
p.mu.Lock()
p.flows[key] = flow
p.mu.Unlock()
// Call cleanup
p.cleanupExpiredFlows()
// Flow should still exist
p.mu.RLock()
_, exists := p.flows[key]
p.mu.RUnlock()
if !exists {
t.Error("cleanupExpiredFlows() should NOT have removed the recent flow")
}
}
// TestCleanupLoop tests the cleanup goroutine shutdown
func TestCleanupLoop_Shutdown(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
// Close should stop the cleanup loop
err := p.Close()
if err != nil {
t.Errorf("Close() error = %v", err)
}
// Give goroutine time to exit
time.Sleep(100 * time.Millisecond)
// Verify cleanupDone channel is closed
select {
case _, ok := <-p.cleanupDone:
if ok {
t.Error("cleanupDone channel should be closed after Close()")
}
default:
t.Error("cleanupDone channel should be closed after Close()")
}
}
// buildClientHelloWithoutExtensions creates a ClientHello without extensions
func buildClientHelloWithoutExtensions() []byte {
// Minimal ClientHello: type(1) + len(3) + version(2) + random(32) + sessionIDLen(1) + cipherLen(2) + compressLen(1) + extLen(2)
handshake := make([]byte, 43)
handshake[0] = 0x01 // ClientHello type
handshake[1] = 0x00
handshake[2] = 0x00
handshake[3] = 0x27 // Length
handshake[4] = 0x03 // Version TLS 1.2
handshake[5] = 0x03
// Random (32 bytes) - zeros
handshake[38] = 0x00 // Session ID length
handshake[39] = 0x00 // Cipher suite length (high)
handshake[40] = 0x02 // Cipher suite length (low)
handshake[41] = 0x13 // Cipher suite data
handshake[42] = 0x01
// Compression length (1 byte) - 0
// Extensions length (2 bytes) - 0
return handshake
}
// buildClientHelloWithSNI creates a ClientHello with SNI extension
func buildClientHelloWithSNI(sni string) []byte {
// Build base handshake
handshake := make([]byte, 43)
handshake[0] = 0x01 // ClientHello type
handshake[1] = 0x00
handshake[2] = 0x00
handshake[4] = 0x03 // Version TLS 1.2
handshake[5] = 0x03
handshake[38] = 0x00 // Session ID length
handshake[39] = 0x00 // Cipher suite length (high)
handshake[40] = 0x02 // Cipher suite length (low)
handshake[41] = 0x13 // Cipher suite data
handshake[42] = 0x01
// Add compression length (1 byte) - 0
handshake = append(handshake, 0x00)
// Add extensions
sniExt := buildSNIExtension(sni)
extLen := len(sniExt)
handshake = append(handshake, byte(extLen>>8), byte(extLen))
handshake = append(handshake, sniExt...)
// Update handshake length
handshakeLen := len(handshake) - 4
handshake[1] = byte(handshakeLen >> 16)
handshake[2] = byte(handshakeLen >> 8)
handshake[3] = byte(handshakeLen)
return handshake
}
// buildTruncatedSNI creates a malformed ClientHello with truncated SNI
func buildTruncatedSNI() []byte {
// Build base handshake
handshake := make([]byte, 44)
handshake[0] = 0x01
handshake[4] = 0x03
handshake[5] = 0x03
handshake[38] = 0x00
handshake[39] = 0x00
handshake[40] = 0x02
handshake[41] = 0x13
handshake[42] = 0x01
handshake[43] = 0x00 // Compression length
// Add extensions with truncated SNI
// Extension type (2) + length (2) + data (truncated)
handshake = append(handshake, 0x00, 0x0a) // Extension length says 10 bytes
handshake = append(handshake, 0x00, 0x05) // But only provide 5 bytes of data
handshake = append(handshake, 0x00, 0x03, 0x74, 0x65, 0x73) // "tes" truncated
return handshake
}
// TestJoinStringSlice tests the deprecated helper function
func TestJoinStringSlice(t *testing.T) {
tests := []struct {
name string
slice []string
sep string
want string
}{
{
name: "empty_slice",
slice: []string{},
sep: ",",
want: "",
},
{
name: "single_element",
slice: []string{"MSS"},
sep: ",",
want: "MSS",
},
{
name: "multiple_elements",
slice: []string{"MSS", "SACK", "TS"},
sep: ",",
want: "MSS,SACK,TS",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := joinStringSlice(tt.slice, tt.sep)
if got != tt.want {
t.Errorf("joinStringSlice() = %q, want %q", got, tt.want)
}
})
}
}
// TestProcess_NilPacketData tests error handling for nil packet data
func TestProcess_NilPacketData(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
defer p.Close()
pkt := api.RawPacket{
Data: nil,
Timestamp: time.Now().UnixNano(),
}
_, err := p.Process(pkt)
if err == nil {
t.Error("Process() with nil data should return error")
}
if err.Error() != "empty packet data" {
t.Errorf("Process() error = %v, want 'empty packet data'", err)
}
}
// TestProcess_EmptyPacketData tests error handling for empty packet data
func TestProcess_EmptyPacketData(t *testing.T) {
p := NewParser()
if p == nil {
t.Fatal("NewParser() returned nil")
}
defer p.Close()
pkt := api.RawPacket{
Data: []byte{},
Timestamp: time.Now().UnixNano(),
}
_, err := p.Process(pkt)
if err == nil {
t.Error("Process() with empty data should return error")
}
}