Count traffic per client address and report the busiest ones with the port totals

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

README.md

@@ -33,6 +33,7 @@ Settings and secrets can live in a `.env` file instead of the command line; see
 - **Ports:** TCP, UDP, HTTP, HTTPS and TLS passthrough, one listener per IP and port, opened and closed live as you change them in the dashboard. Port ranges for TCP and UDP. Round-robin load balancing and failover across several targets, with health checks.
 - **PROXY protocol (v1/v2)** on TCP and TLS ports, so the target sees the real client address.
 - **Web:** automatic Let's Encrypt certificates, HTTP/2, websockets, and logins (account, password, PIN, emailed code, share links, path and IP rules) enforced in front of any site. Use the built-in proxy, or `-traefik` to let Traefik serve http/https (raw TCP, UDP and TLS-passthrough ports always stay with Blacktip, since Traefik can't open ports without a restart).
+- **Traffic analytics:** counts connections and bytes in and out for every port, and the busiest client addresses on each (the ten biggest per report, at most a few thousand tracked between reports), and sends them to Hammerhead with its regular report. Hammerhead keeps the history, draws the charts and raises traffic alerts. Nothing is stored on this machine, and a report that fails to arrive is added to the next one.
 - **Mail egress:** delivers your mail server's outgoing mail straight to the recipient's server from a public IP you choose, with the HELO name you set, opportunistic STARTTLS, optional DKIM signing, MTA-STS enforcement, a per-minute send limit and a deliverability checklist.
 - **Tunnel:** WireGuard through a userspace network stack, so no root, no kernel interface and no driver. It works on Windows, macOS and Linux, and sites behind NAT connect out to it.
 

blacktip.go

@@ -73,6 +73,7 @@ type mapState struct {
 	rr     atomic.Uint64
 
 	conns, bytesIn, bytesOut atomic.Int64
+	sources                  sourceTable  // the same traffic per client address, for the top-sources list
 	epoch                    atomic.Int64 // current login generation; kept out of sig so a bump doesn't rebuild the resource
 
 	relayLn   net.Listener // traefik mode: loopback port that reaches the targets through the tunnel
@@ -113,6 +114,7 @@ type Blacktip struct {
 
 	logq       chan model.LogEntry
 	reportMu   sync.Mutex
+	statsSince time.Time // when the traffic counters were last taken; guarded by reportMu
 	updating   atomic.Bool
 	addrSig    string // machine addresses a wildcard website was spread across, "" when none was
 	updateMsg  string // outcome of the update Hammerhead asked for, reported back
@@ -610,11 +612,17 @@ func (b *Blacktip) report() {
 		st.Targets = append([]model.TargetHealth(nil), ms.health...)
 		ms.mu.Unlock()
 		rep.Status[id] = st
-		if c, in, out := ms.conns.Swap(0), ms.bytesIn.Swap(0), ms.bytesOut.Swap(0); c|in|out != 0 {
+		if st, any := ms.takeStats(); any {
-			rep.Stats[id] = model.Stat{Conns: c, BytesIn: in, BytesOut: out}
+			rep.Stats[id] = st
 		}
 	}
 	b.mu.Unlock()
+	now := time.Now()
+	if !b.statsSince.IsZero() {
+		rep.StatsSeconds = now.Sub(b.statsSince).Seconds()
+	}
+	prevSince := b.statsSince
+	b.statsSince = now
 	body, _ := json.Marshal(rep)
 	req, _ := http.NewRequest("POST", b.cfg.Hammerhead+"/edge/report", bytes.NewReader(body))
 	req.Header.Set("Authorization", "Bearer "+b.cfg.Token)
@@ -626,12 +634,11 @@ func (b *Blacktip) report() {
 		b.mu.Lock()
 		for id, st := range rep.Stats {
 			if ms := b.maps[id]; ms != nil {
-				ms.conns.Add(st.Conns)
+				ms.restoreStats(st)
-				ms.bytesIn.Add(st.BytesIn)
-				ms.bytesOut.Add(st.BytesOut)
 			}
 		}
 		b.mu.Unlock()
+		b.statsSince = prevSince // the traffic put back still spans from there
 		return
 	}
 	resp.Body.Close()

listener.go

@@ -227,9 +227,7 @@ func (l *listener) handleTCP(c net.Conn) {
 	}
 	in, out := tunnel.Proxy(c, c, up, up)
 	in += int64(len(first))
-	ms.conns.Add(1)
+	ms.record(c.RemoteAddr().String(), 1, in, out)
-	ms.bytesIn.Add(in)
-	ms.bytesOut.Add(out)
 	entry.BytesIn, entry.BytesOut, entry.Millis = in, out, time.Since(start).Milliseconds()
 	l.b.logAccess(entry)
 }
@@ -363,13 +361,13 @@ func (l *listener) serveUDP() {
 			l.mu.Lock()
 			l.sessions[key] = s
 			l.mu.Unlock()
-			ms.conns.Add(1)
+			ms.record(addr.String(), 1, 0, 0)
 			go l.pumpBack(ms, key, addr, s)
 		}
 		l.mu.Lock()
 		s.last = time.Now()
 		l.mu.Unlock()
-		ms.bytesIn.Add(int64(n))
+		ms.record(addr.String(), 0, int64(n), 0)
 		s.conn.Write(buf[:n])
 	}
 }
@@ -394,7 +392,7 @@ func (l *listener) pumpBack(ms *mapState, key string, addr net.Addr, s *udpSessi
 		if _, err := l.udp.WriteTo(p, addr); err != nil {
 			return
 		}
-		ms.bytesOut.Add(int64(len(p)))
+		ms.record(addr.String(), 0, 0, int64(len(p)))
 		l.mu.Lock()
 		s.last = time.Now()
 		l.mu.Unlock()

traefik.go

@@ -544,9 +544,7 @@ func (t *traefikMgr) tailAccessLog(ctx context.Context) {
 			if ms == nil {
 				continue
 			}
-			ms.conns.Add(1)
+			ms.record(tl.ClientHost, 1, tl.RequestSize, tl.DownstreamSize)
-			ms.bytesIn.Add(tl.RequestSize)
-			ms.bytesOut.Add(tl.DownstreamSize)
 			t.b.logAccess(model.LogEntry{Time: tl.StartUTC, Mapping: id, Remote: tl.ClientHost, Method: tl.RequestMethod, Host: hostOnly(tl.RequestHost),
 				Path: tl.RequestPath, Status: tl.DownstreamStatus, Result: "allowed", BytesIn: tl.RequestSize, BytesOut: tl.DownstreamSize, Millis: tl.DurationNanos / 1e6})
 		}

traffic.go

@@ -0,0 +1,89 @@
+package blacktip
+
+import (
+	"net"
+	"net/netip"
+	"sync"
+
+	"git.veillax.com/Veillax/reef/model"
+)
+
+// maxTrackedSources bounds the per-port table of client addresses between two reports, so a flood
+// from spoofed or scanning addresses can't grow it without limit. Addresses past it still count
+// in the port's totals; they just aren't named.
+const maxTrackedSources = 5000
+
+// sourceTable counts a port's traffic per client address until the next report takes it.
+type sourceTable struct {
+	mu sync.Mutex
+	by map[string]*model.SourceStat
+}
+
+func (t *sourceTable) add(ip string, conns, in, out int64) {
+	if ip == "" {
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	s := t.by[ip]
+	if s == nil {
+		if len(t.by) >= maxTrackedSources {
+			return
+		}
+		if t.by == nil {
+			t.by = map[string]*model.SourceStat{}
+		}
+		s = &model.SourceStat{IP: ip}
+		t.by[ip] = s
+	}
+	s.Conns += conns
+	s.BytesIn += in
+	s.BytesOut += out
+}
+
+func (t *sourceTable) take() map[string]*model.SourceStat {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	by := t.by
+	t.by = nil
+	return by
+}
+
+// clientIP reduces a "host:port" remote address to the client's IP (IPv4-mapped IPv6 as plain IPv4).
+// Anything that isn't an address gives "".
+func clientIP(remote string) string {
+	host, _, err := net.SplitHostPort(remote)
+	if err != nil {
+		host = remote
+	}
+	a, err := netip.ParseAddr(host)
+	if err != nil {
+		return ""
+	}
+	return a.WithZone("").Unmap().String()
+}
+
+// record counts traffic on a port, and names the client it came from when that is known.
+func (ms *mapState) record(remote string, conns, in, out int64) {
+	ms.conns.Add(conns)
+	ms.bytesIn.Add(in)
+	ms.bytesOut.Add(out)
+	ms.sources.add(clientIP(remote), conns, in, out)
+}
+
+// takeStats hands over (and resets) the traffic counted since the last call.
+func (ms *mapState) takeStats() (model.Stat, bool) {
+	st := model.Stat{Conns: ms.conns.Swap(0), BytesIn: ms.bytesIn.Swap(0), BytesOut: ms.bytesOut.Swap(0)}
+	st.Sources = model.TopSources(ms.sources.take(), model.MaxSources)
+	return st, st.Conns|st.BytesIn|st.BytesOut != 0
+}
+
+// restoreStats puts taken traffic back after a report that didn't arrive, so none of it is lost.
+func (ms *mapState) restoreStats(st model.Stat) {
+	ms.conns.Add(st.Conns)
+	ms.bytesIn.Add(st.BytesIn)
+	ms.bytesOut.Add(st.BytesOut)
+	for _, s := range st.Sources {
+		ms.sources.add(s.IP, s.Conns, s.BytesIn, s.BytesOut)
+	}
+}

traffic_test.go

@@ -0,0 +1,55 @@
+package blacktip
+
+import "testing"
+
+func TestClientIP(t *testing.T) {
+	for in, want := range map[string]string{
+		"203.0.113.9:5000":     "203.0.113.9",
+		"[2001:db8::1]:443":    "2001:db8::1",
+		"[::ffff:192.0.2.7]:1": "192.0.2.7",
+		"198.51.100.2":         "198.51.100.2", // no port, as Traefik reports it
+		"[fe80::1%eth0]:80":    "fe80::1",
+		"not an address":       "",
+		"":                     "",
+	} {
+		if got := clientIP(in); got != want {
+			t.Errorf("clientIP(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+func TestStatsTakeAndRestore(t *testing.T) {
+	ms := &mapState{}
+	ms.record("203.0.113.1:1000", 1, 100, 400)
+	ms.record("203.0.113.1:1001", 1, 10, 40)
+	ms.record("203.0.113.2:2000", 1, 5, 5)
+	ms.record("garbage", 1, 1, 1) // counted in the totals but not named
+
+	st, any := ms.takeStats()
+	if !any || st.Conns != 4 || st.BytesIn != 116 || st.BytesOut != 446 {
+		t.Fatalf("totals: %+v", st)
+	}
+	if len(st.Sources) != 2 || st.Sources[0].IP != "203.0.113.1" || st.Sources[0].Conns != 2 || st.Sources[0].BytesOut != 440 {
+		t.Fatalf("sources: %+v", st.Sources)
+	}
+	if _, any := ms.takeStats(); any {
+		t.Fatal("counters were not reset")
+	}
+
+	ms.record("203.0.113.1:1", 1, 1, 1) // traffic that arrived before the failed report was retried
+	ms.restoreStats(st)
+	again, _ := ms.takeStats()
+	if again.Conns != 5 || again.BytesIn != 117 || again.Sources[0].Conns != 3 {
+		t.Fatalf("after restore: %+v", again)
+	}
+}
+
+func TestSourceTableIsBounded(t *testing.T) {
+	var tb sourceTable
+	for i := 0; i < maxTrackedSources+50; i++ {
+		tb.add(string(rune('a'+i%26))+"."+string(rune('0'+i/26%10))+"."+string(rune(i)), 1, 1, 1)
+	}
+	if n := len(tb.take()); n > maxTrackedSources {
+		t.Fatalf("tracked %d addresses", n)
+	}
+}

web.go

@@ -258,9 +258,7 @@ func (l *listener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		entry.BytesOut = rec.bytes
 		entry.Millis = time.Since(start).Milliseconds()
-		ms.conns.Add(1)
+		ms.record(r.RemoteAddr, 1, entry.BytesIn, entry.BytesOut)
-		ms.bytesIn.Add(entry.BytesIn)
-		ms.bytesOut.Add(entry.BytesOut)
 		b.logAccess(entry)
 	}()