Ensure all IPv4 addresses use only 4 bytes. (#30)

* Ensure all IPv4 addresses use only 4 bytes.
* convert the eventsocket flag to the one provided in tcpinfo.eventsocket

demuxer/flowkey.go

@@ -40,7 +40,13 @@ func fromPacket(p gopacket.Packet) FlowKey {
 	return FlowKeyFrom4Tuple(ip1, ip1P, ip2, ip2P)
 }
 
-// FlowKeyFrom4Tuple creates a FlowKey (suitable for use as a map key) from a TCP 4-tuple.
+// FlowKeyFrom4Tuple creates a FlowKey (suitable for use as a map key) from a
+// TCP 4-tuple. This function is called once per packet, by the demuxer, and
+// once per flow, by the tcpeventsocket handler. Because it is called once per
+// packet, it should be as efficient as possible.
+//
+// IPv4 addresses passed in must be 4 bytes, because we do byte-based
+// comparisons with sub-slices of packets retrieved from the wire.
 func FlowKeyFrom4Tuple(srcIP net.IP, srcPort uint16, dstIP net.IP, dstPort uint16) FlowKey {
 	srcIPS := string(srcIP)
 	dstIPS := string(dstIP)

main.go

@@ -29,7 +29,6 @@ import (
 
 var (
 	dir              = flag.String("datadir", ".", "The directory to which data is written")
-	eventSocket      = flag.String("eventsocket", "", "The absolute pathname of the unix-domain socket to which events will be posted.")
 	captureDuration  = flag.Duration("captureduration", 30*time.Second, "Only save the first captureduration of each flow, to prevent long-lived flows from spamming the hard drive.")
 	uuidWaitDuration = flag.Duration("uuidwaitduration", 5*time.Second, "Wait up to uuidwaitduration for each flow before either assigning a UUID or discarding all future packets. This prevents buffering unsaveable packets.")
 	flowTimeout      = flag.Duration("flowtimeout", 30*time.Second, "Once there have been no packets for a flow for at least flowtimeout, the flow can be assumed to be closed.")
@@ -45,6 +44,7 @@ var (
 
 func init() {
 	flag.Var(&interfaces, "interface", "The interface on which to capture traffic. May be repeated. If unset, will capture on all available interfaces.")
+	log.SetFlags(log.Lshortfile | log.LstdFlags)
 }
 
 func catch(sig os.Signal) {
@@ -124,7 +124,7 @@ func main() {
 	h := tcpinfohandler.New(mainCtx, tcpdm.UUIDChan)
 	cleanupWG.Add(1)
 	go func() {
-		eventsocket.MustRun(mainCtx, *eventSocket, h)
+		eventsocket.MustRun(mainCtx, *eventsocket.Filename, h)
 		mainCancel()
 		cleanupWG.Done()
 	}()

main_test.go

@@ -76,13 +76,13 @@ func TestMainSmokeTest(t *testing.T) {
 	// server.
 	tcpiCtx, tcpiCancel := context.WithCancel(context.Background())
 	defer tcpiCancel()
-	*eventSocket = dir + "/tcpevents.sock"
-	tcpi := eventsocket.New(*eventSocket)
+	*eventsocket.Filename = dir + "/tcpevents.sock"
+	tcpi := eventsocket.New(*eventsocket.Filename)
 	tcpi.Listen()
 	go tcpi.Serve(tcpiCtx)
 
 	// Wait until the eventsocket appears.
-	for _, err := os.Stat(*eventSocket); err != nil; _, err = os.Stat(*eventSocket) {
+	for _, err := os.Stat(*eventsocket.Filename); err != nil; _, err = os.Stat(*eventsocket.Filename) {
 	}
 
 	// Tests are unlikely to have enough privileges to open packet captures, so

saver/tcp.go

@@ -155,6 +155,8 @@ func (t *TCP) savePackets(ctx context.Context, uuidDelay, duration time.Duration
 	// Read the first packet to determine the TCP+IP header size (as IPv6 is variable in size)
 	p, ok := t.readPacket(derivedCtx)
 	if !ok {
+		// This error should never occur in production. It indicates a
+		// configuration error or a bug in packet-headers.
 		log.Println("PCAP capture cancelled with no packets for flow", t.id)
 		t.error("nopackets")
 		return
@@ -200,6 +202,13 @@ func (t *TCP) savePackets(ctx context.Context, uuidDelay, duration time.Duration
 	// Read the UUID to determine the filename
 	t.state.Set("uuidwait")
 	var uuidEvent UUIDEvent
+	// The error conditions below are expected to occur in production. In
+	// particular, every flow that existed prior to the start of the start of
+	// the packet-headers binary will cause this error at least once.
+	//
+	// This error will also occur for long-lived flows that send packets so
+	// infrequently that the flow gets garbage-collected between packet
+	// arrivals.
 	select {
 	case uuidEvent, ok = <-t.uuidchanRead:
 		if !ok {

tcpinfohandler/handler.go

@@ -39,6 +39,14 @@ func (h *handler) Open(ctx context.Context, timestamp time.Time, uuid string, id
 		metrics.BadEventsFromTCPInfo.WithLabelValues("badip").Inc()
 		return
 	}
+	// Convert all IPv4 addresses to a 4-byte representation, as required by
+	// FlowKeyFrom4Tuple.
+	if srcIP.To4() != nil {
+		srcIP = srcIP.To4()
+	}
+	if dstIP.To4() != nil {
+		dstIP = dstIP.To4()
+	}
 	// Can't use a struct literal here due to embedding.
 	ev := demuxer.UUIDEvent{}
 	ev.Flow = demuxer.FlowKeyFrom4Tuple(srcIP, id.SPort, dstIP, id.DPort)