diff --git a/README.md b/README.md index 18c6837..f12d082 100644 --- a/README.md +++ b/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. diff --git a/blacktip.go b/blacktip.go index 14fe679..88b488c 100644 --- a/blacktip.go +++ b/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 { - rep.Stats[id] = model.Stat{Conns: c, BytesIn: in, BytesOut: out} + if st, any := ms.takeStats(); any { + 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.bytesIn.Add(st.BytesIn) - ms.bytesOut.Add(st.BytesOut) + ms.restoreStats(st) } } b.mu.Unlock() + b.statsSince = prevSince // the traffic put back still spans from there return } resp.Body.Close() diff --git a/listener.go b/listener.go index cb1b300..33159f7 100644 --- a/listener.go +++ b/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.bytesIn.Add(in) - ms.bytesOut.Add(out) + ms.record(c.RemoteAddr().String(), 1, in, 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() diff --git a/traefik.go b/traefik.go index 04a5ea6..1bb5ed2 100644 --- a/traefik.go +++ b/traefik.go @@ -544,9 +544,7 @@ func (t *traefikMgr) tailAccessLog(ctx context.Context) { if ms == nil { continue } - ms.conns.Add(1) - ms.bytesIn.Add(tl.RequestSize) - ms.bytesOut.Add(tl.DownstreamSize) + ms.record(tl.ClientHost, 1, tl.RequestSize, 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}) } diff --git a/traffic.go b/traffic.go new file mode 100644 index 0000000..04f1b3e --- /dev/null +++ b/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) + } +} diff --git a/traffic_test.go b/traffic_test.go new file mode 100644 index 0000000..e639725 --- /dev/null +++ b/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) + } +} diff --git a/web.go b/web.go index 8d70ebb..06d692a 100644 --- a/web.go +++ b/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.bytesIn.Add(entry.BytesIn) - ms.bytesOut.Add(entry.BytesOut) + ms.record(r.RemoteAddr, 1, entry.BytesIn, entry.BytesOut) b.logAccess(entry) }()