feat(admin): track and filter traffic by client IP

The console now records the client source IP for every connection and DNS query, shows the top clients, and can filter the domain list by one: - WrapConn captures conn.RemoteAddr into the counting conn; RecordDNS takes the client IP from the DNS ResponseWriter. Domain stats keep a per-client breakdown alongside the per-source one. - /api/traffic?client=<ip> restricts domains to that client and reports the client's conns/bytes/lastSeen; it composes with ?source=. - The snapshot now carries clients: top 50 per-IP aggregates (conns, bytes up/down, last seen) for display. - Traffic page shows a client chip row (全部客户端 + IPs with counts); clicking one filters, tooltips show byte totals. Verified live: ?client=192.168.1.1 lists that device's domains with its counts; ?source=dns&client=... composes correctly.
2026-10-19 16:14:33 +00:00 · 2026-08-06 19:54:43 +08:00
parent 6d52cc14c3
commit fbc0d54893
12 changed files with 1626 additions and 1387 deletions
@@ -84,7 +84,7 @@ type dnsStatsHandler struct {

 func (h dnsStatsHandler) ServeDNS(w dns.ResponseWriter, req *dns.Msg) {
 	if len(req.Question) == 1 {
-		h.stats.RecordDNS(req.Question[0].Name)
+		h.stats.RecordDNS(req.Question[0].Name, admin.ClientIPOf(w.RemoteAddr()))
 	}
 	h.Handler.ServeDNS(w, req)
 }
@@ -439,7 +439,8 @@ func (s *Server) handleTraffic(w http.ResponseWriter, r *http.Request) {
 	if !source.valid() {
 		source = SourceAll
 	}
-	writeJSON(w, http.StatusOK, s.opts.Stats.Snapshot(sort, source))
+	client := r.URL.Query().Get("client")
+	writeJSON(w, http.StatusOK, s.opts.Stats.Snapshot(sort, source, client))
 }

 func (s *Server) handleHistory(w http.ResponseWriter, r *http.Request) {
@@ -6,8 +6,8 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"os"
 	"net/http/httptest"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -386,7 +386,7 @@ func TestTrafficEndpoint(t *testing.T) {
 	t.Cleanup(ts.Close)

 	cookie := login(t, ts, "secret")
-	stats.RecordDNS("example.com.")
+	stats.RecordDNS("example.com.", "")
 	stats.WrapConn(&fakeConn{}, "socks5")

 	resp := authedRequest(t, ts, http.MethodGet, "/api/traffic", cookie, "")
@@ -519,7 +519,7 @@ func TestMetricsEndpointServesPrometheusFormat(t *testing.T) {
 	ts := httptest.NewServer(s.http.Handler)
 	defer ts.Close()

-	stats.RecordDNS("example.com")
+	stats.RecordDNS("example.com", "")
 	stats.RecordRoute("proxy")
 	conn := stats.WrapConn(&fakeConn{}, "https")
 	_ = conn.Close()
@@ -17,6 +17,7 @@ const (
 	maxDomainEntries = 10000
 	staleDomainAge   = time.Hour
 	snapshotTopN     = 100
+	clientsTopN      = 50

 	// historyCapacity keeps one hour of samples at the 5s poll cadence.
 	historyCapacity = 720
@@ -114,6 +115,17 @@ type TrafficSnapshot struct {
 	BytesUp   uint64       `json:"bytesUp"`
 	BytesDown uint64       `json:"bytesDown"`
 	Domains   []DomainStat `json:"domains"`
+	Clients   []ClientStat `json:"clients"`
+}
+
+// ClientStat aggregates traffic for a single client IP. Conns counts both
+// proxy connections and DNS queries originating from the IP.
+type ClientStat struct {
+	IP        string    `json:"ip"`
+	Conns     uint64    `json:"conns"`
+	BytesUp   uint64    `json:"bytesUp"`
+	BytesDown uint64    `json:"bytesDown"`
+	LastSeen  time.Time `json:"lastSeen"`
 }

 // HistorySample is one point in the in-process history ring. Byte, query,
@@ -190,6 +202,7 @@ type domainStat struct {
 	bytesDown uint64
 	lastSeen  time.Time
 	bySource  map[Source]*sourceStat
+	byClient  map[string]*clientStat // client IP -> per-domain stats
 }

 // sourceStat is the per-source breakdown of a domain's traffic. A domain can
@@ -202,6 +215,14 @@ type sourceStat struct {
 	lastSeen  time.Time
 }

+// clientStat is the per-client-IP breakdown of a domain's traffic.
+type clientStat struct {
+	conns     uint64
+	bytesUp   uint64
+	bytesDown uint64
+	lastSeen  time.Time
+}
+
 // NewStats creates the stats registry and its OpenTelemetry instruments.
 func NewStats() (*Stats, error) {
 	m, err := metrics.New()
@@ -218,17 +239,20 @@ func NewStats() (*Stats, error) {
 // Metrics exposes the OpenTelemetry export surface (Prometheus /metrics).
 func (s *Stats) Metrics() *metrics.Metrics { return s.metrics }

-// RecordDNS counts a DNS query for the given qname.
-func (s *Stats) RecordDNS(qname string) {
+// RecordDNS counts a DNS query for the given qname from the given client IP.
+func (s *Stats) RecordDNS(qname string, clientIP string) {
 	s.dnsQueries.Add(1)
 	s.metrics.RecordDNS()
 	domain := normalizeDomain(qname)
 	if domain == "" {
 		return
 	}
-	s.record(domain, SourceDNS, func(d *domainStat, src *sourceStat) {
+	s.record(domain, SourceDNS, clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
 		d.conns++
 		src.conns++
+		if dc != nil {
+			dc.conns++
+		}
 	})
 }

@@ -262,7 +286,13 @@ func (s *Stats) WrapConn(conn net.Conn, kind string) net.Conn {
 		s.activeSocks.Add(1)
 	}
 	s.metrics.ConnOpened(kind)
-	return &countingConn{Conn: conn, stats: s, kind: kind, created: time.Now()}
+	return &countingConn{
+		Conn:     conn,
+		stats:    s,
+		kind:     kind,
+		clientIP: ClientIPOf(conn.RemoteAddr()),
+		created:  time.Now(),
+	}
 }

 // BindConn attributes an already-wrapped connection to a domain and bumps the
@@ -279,17 +309,21 @@ func (s *Stats) BindConn(conn net.Conn, domain string) {
 	}
 	cc.bind(domain)
 	source := Source(cc.kind)
-	s.record(domain, source, func(d *domainStat, src *sourceStat) {
+	s.record(domain, source, cc.clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
 		d.conns++
 		src.conns++
+		if dc != nil {
+			dc.conns++
+		}
 	})
 }

 // Snapshot returns an immutable view of the current stats, advancing the
 // history ring and evicting stale domains along the way. sort selects the
-// domain ordering and source restricts domains to one entry point; invalid
-// values fall back to DomainSortBytes and SourceAll.
-func (s *Stats) Snapshot(sort DomainSort, source Source) TrafficSnapshot {
+// domain ordering, source restricts domains to one entry point, and client
+// restricts them to one client IP; invalid values fall back to
+// DomainSortBytes and SourceAll, and an empty client disables the filter.
+func (s *Stats) Snapshot(sort DomainSort, source Source, client string) TrafficSnapshot {
 	s.sample()

 	snap := TrafficSnapshot{
@@ -314,6 +348,7 @@ func (s *Stats) Snapshot(sort DomainSort, source Source) TrafficSnapshot {

 	s.mu.Lock()
 	defer s.mu.Unlock()
+	clientAgg := make(map[string]*ClientStat)
 	for domain, d := range s.domains {
 		if time.Since(d.lastSeen) > staleDomainAge {
 			delete(s.domains, domain)
@@ -336,15 +371,60 @@ func (s *Stats) Snapshot(sort DomainSort, source Source) TrafficSnapshot {
 			ds.BytesDown = src.bytesDown
 			ds.LastSeen = src.lastSeen
 		}
+		if client != "" {
+			dc, ok := d.byClient[client] // nil map read is safe
+			if !ok || (dc.conns == 0 && dc.bytesUp == 0 && dc.bytesDown == 0) {
+				continue
+			}
+			ds.Conns = dc.conns
+			ds.BytesUp = dc.bytesUp
+			ds.BytesDown = dc.bytesDown
+			ds.LastSeen = dc.lastSeen
+		}
 		snap.Domains = append(snap.Domains, ds)
+
+		for ip, c := range d.byClient {
+			agg := clientAgg[ip]
+			if agg == nil {
+				agg = &ClientStat{IP: ip}
+				clientAgg[ip] = agg
+			}
+			agg.Conns += c.conns
+			agg.BytesUp += c.bytesUp
+			agg.BytesDown += c.bytesDown
+			if c.lastSeen.After(agg.LastSeen) {
+				agg.LastSeen = c.lastSeen
+			}
+		}
 	}
 	sortDomains(snap.Domains, sort)
 	if len(snap.Domains) > snapshotTopN {
 		snap.Domains = snap.Domains[:snapshotTopN]
 	}
+	snap.Clients = topClients(clientAgg)
 	return snap
 }

+// topClients orders per-IP aggregates by total bytes and caps the list.
+func topClients(agg map[string]*ClientStat) []ClientStat {
+	out := make([]ClientStat, 0, len(agg))
+	for _, c := range agg {
+		out = append(out, *c)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		ti := out[i].BytesUp + out[i].BytesDown
+		tj := out[j].BytesUp + out[j].BytesDown
+		if ti == tj {
+			return out[i].IP < out[j].IP
+		}
+		return ti > tj
+	})
+	if len(out) > clientsTopN {
+		out = out[:clientsTopN]
+	}
+	return out
+}
+
 // sortDomains orders domains by the selected mode: total bytes (default),
 // most recently seen, or connection count. Ties break alphabetically.
 func sortDomains(ds []DomainStat, mode DomainSort) {
@@ -461,9 +541,10 @@ func (s *Stats) rates() (up, down, dns, conns float64) {
 	return float64(sumUp) / elapsed, float64(sumDown) / elapsed, float64(sumDNS) / elapsed, float64(sumConns) / elapsed
 }

-// record attributes one event to a domain and its source. It must be called
-// with the domain and source both normalized.
-func (s *Stats) record(domain string, source Source, update func(*domainStat, *sourceStat)) {
+// record attributes one event to a domain, its source, and optionally its
+// client IP. The client IP may be empty when it cannot be determined; the
+// per-client stat is then nil and callers must guard.
+func (s *Stats) record(domain string, source Source, clientIP string, update func(*domainStat, *sourceStat, *clientStat)) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	d := s.domains[domain]
@@ -479,9 +560,23 @@ func (s *Stats) record(domain string, source Source, update func(*domainStat, *s
 		src = &sourceStat{}
 		d.bySource[source] = src
 	}
-	update(d, src)
+	var dc *clientStat
+	if clientIP != "" {
+		if d.byClient == nil {
+			d.byClient = make(map[string]*clientStat)
+		}
+		dc = d.byClient[clientIP]
+		if dc == nil {
+			dc = &clientStat{}
+			d.byClient[clientIP] = dc
+		}
+	}
+	update(d, src, dc)
 	d.lastSeen = time.Now()
 	src.lastSeen = time.Now()
+	if dc != nil {
+		dc.lastSeen = time.Now()
+	}
 }

 // evictOldestLocked drops the single least-recently-seen entry to keep the
@@ -503,11 +598,12 @@ func (s *Stats) evictOldestLocked() {
 // countingConn wraps a client connection and feeds byte counts into Stats.
 type countingConn struct {
 	net.Conn
-	stats   *Stats
-	kind    string
-	created time.Time
-	domain  atomic.Value // string; empty until BindConn
-	closed  atomic.Bool
+	stats    *Stats
+	kind     string
+	clientIP string
+	created  time.Time
+	domain   atomic.Value // string; empty until BindConn
+	closed   atomic.Bool
 }

 func (c *countingConn) Read(p []byte) (int, error) {
@@ -559,13 +655,19 @@ func (c *countingConn) recordBytes(n int, up bool) {
 		return
 	}
 	source := Source(c.kind)
-	c.stats.record(domain, source, func(d *domainStat, src *sourceStat) {
+	c.stats.record(domain, source, c.clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
 		if up {
 			d.bytesUp += uint64(n)
 			src.bytesUp += uint64(n)
+			if dc != nil {
+				dc.bytesUp += uint64(n)
+			}
 		} else {
 			d.bytesDown += uint64(n)
 			src.bytesDown += uint64(n)
+			if dc != nil {
+				dc.bytesDown += uint64(n)
+			}
 		}
 	})
 }
@@ -577,6 +679,18 @@ func normalizeDomain(domain string) string {
 	return strings.ToLower(domain)
 }

+// clientIPOf extracts the host part of a network address, falling back to the
+// full address when it has no port.
+func ClientIPOf(addr net.Addr) string {
+	if addr == nil {
+		return ""
+	}
+	if host, _, err := net.SplitHostPort(addr.String()); err == nil {
+		return host
+	}
+	return addr.String()
+}
+
 // goroutineCount reports the current number of goroutines.
 func goroutineCount() uint64 {
 	var samples = []rtmetrics.Sample{{Name: "/sched/goroutines:goroutines"}}
@@ -26,6 +26,10 @@ func (c *fakeConn) Write(p []byte) (int, error) {

 func (c *fakeConn) Close() error { return nil }

+func (c *fakeConn) RemoteAddr() net.Addr {
+	return &net.TCPAddr{IP: net.ParseIP("192.0.2.1")}
+}
+
 func TestStatsWrapConnCountsBytesBeforeBind(t *testing.T) {
 	s := newTestStats(t)
 	conn := &fakeConn{readBuf: []byte("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")}
@@ -40,7 +44,7 @@ func TestStatsWrapConnCountsBytesBeforeBind(t *testing.T) {
 		t.Fatalf("write: %v", err)
 	}

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.Conns.HTTP != 1 {
 		t.Fatalf("expected 1 http conn, got %d", snap.Conns.HTTP)
 	}
@@ -73,7 +77,7 @@ func TestStatsBindConnAttributesBytesToDomain(t *testing.T) {
 		t.Fatalf("write: %v", err)
 	}

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if len(snap.Domains) != 1 {
 		t.Fatalf("expected 1 domain, got %v", snap.Domains)
 	}
@@ -98,17 +102,17 @@ func TestStatsBindConnAttributesBytesToDomain(t *testing.T) {
 func TestStatsBindConnIgnoresUnwrappedConn(t *testing.T) {
 	s := newTestStats(t)
 	s.BindConn(&fakeConn{}, "example.com")
-	if len(s.Snapshot(DomainSortBytes, SourceAll).Domains) != 0 {
+	if len(s.Snapshot(DomainSortBytes, SourceAll, "").Domains) != 0 {
 		t.Fatal("expected unwrapped conn to be ignored")
 	}
 }

 func TestStatsRecordDNS(t *testing.T) {
 	s := newTestStats(t)
-	s.RecordDNS("Example.COM.")
-	s.RecordDNS("example.com.")
+	s.RecordDNS("Example.COM.", "")
+	s.RecordDNS("example.com.", "")

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.DNSQueries != 2 {
 		t.Fatalf("expected 2 dns queries, got %d", snap.DNSQueries)
 	}
@@ -127,7 +131,7 @@ func TestStatsSnapshotEvictsStaleDomains(t *testing.T) {
 	s.domains["fresh.example"] = &domainStat{lastSeen: time.Now()}
 	s.mu.Unlock()

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if len(snap.Domains) != 1 || snap.Domains[0].Domain != "fresh.example" {
 		t.Fatalf("expected stale domain evicted, got %v", snap.Domains)
 	}
@@ -136,7 +140,7 @@ func TestStatsSnapshotEvictsStaleDomains(t *testing.T) {
 func TestStatsDomainCapIsEnforced(t *testing.T) {
 	s := newTestStats(t)
 	for i := 0; i < maxDomainEntries+50; i++ {
-		s.record(fmt.Sprintf("host%d.example", i), SourceHTTP, func(d *domainStat, src *sourceStat) {})
+		s.record(fmt.Sprintf("host%d.example", i), SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {})
 	}
 	s.mu.Lock()
 	n := len(s.domains)
@@ -165,11 +169,20 @@ func TestEvictOldestLocked(t *testing.T) {

 func TestStatsSnapshotSortsByTotalBytes(t *testing.T) {
 	s := newTestStats(t)
-	s.record("small.example", SourceHTTP, func(d *domainStat, src *sourceStat) { d.bytesUp, d.bytesDown = 1, 1; src.bytesUp, src.bytesDown = 1, 1 })
-	s.record("big.example", SourceHTTP, func(d *domainStat, src *sourceStat) { d.bytesUp, d.bytesDown = 100, 100; src.bytesUp, src.bytesDown = 100, 100 })
-	s.record("mid.example", SourceHTTP, func(d *domainStat, src *sourceStat) { d.bytesUp, d.bytesDown = 10, 10; src.bytesUp, src.bytesDown = 10, 10 })
+	s.record("small.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
+		d.bytesUp, d.bytesDown = 1, 1
+		src.bytesUp, src.bytesDown = 1, 1
+	})
+	s.record("big.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
+		d.bytesUp, d.bytesDown = 100, 100
+		src.bytesUp, src.bytesDown = 100, 100
+	})
+	s.record("mid.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
+		d.bytesUp, d.bytesDown = 10, 10
+		src.bytesUp, src.bytesDown = 10, 10
+	})

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if len(snap.Domains) != 3 {
 		t.Fatalf("unexpected domains: %v", snap.Domains)
 	}
@@ -189,7 +202,7 @@ func newTestStats(t *testing.T) *Stats {

 func TestStatsUptimeIsSeconds(t *testing.T) {
 	s := newTestStats(t)
-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	// A fresh stats object must report a small number of seconds, not
 	// nanoseconds (a time.Duration would serialize as ~1e9 per second).
 	if snap.Uptime < 0 || snap.Uptime > 60 {
@@ -201,7 +214,7 @@ func TestStatsActiveConnsTrackedAndIdempotentClose(t *testing.T) {
 	s := newTestStats(t)
 	wrapped := s.WrapConn(&fakeConn{}, "http")

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.Active.HTTP != 1 || snap.Conns.HTTP != 1 {
 		t.Fatalf("expected 1 active/1 total http conn, got %+v / %+v", snap.Active, snap.Conns)
 	}
@@ -212,7 +225,7 @@ func TestStatsActiveConnsTrackedAndIdempotentClose(t *testing.T) {
 	if err := wrapped.Close(); err != nil { // idempotent
 		t.Fatalf("second close: %v", err)
 	}
-	snap = s.Snapshot(DomainSortBytes, SourceAll)
+	snap = s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.Active.HTTP != 0 {
 		t.Fatalf("expected 0 active after close, got %d", snap.Active.HTTP)
 	}
@@ -227,14 +240,14 @@ func TestStatsActiveConnsPerProtocol(t *testing.T) {
 	https := s.WrapConn(&fakeConn{}, "https")
 	socks := s.WrapConn(&fakeConn{}, "socks5")

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.Active.HTTP != 1 || snap.Active.HTTPS != 1 || snap.Active.Socks5 != 1 {
 		t.Fatalf("unexpected active conns: %+v", snap.Active)
 	}
 	_ = http.Close()
 	_ = https.Close()
 	_ = socks.Close()
-	snap = s.Snapshot(DomainSortBytes, SourceAll)
+	snap = s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.Active.HTTP != 0 || snap.Active.HTTPS != 0 || snap.Active.Socks5 != 0 {
 		t.Fatalf("expected all active conns released: %+v", snap.Active)
 	}
@@ -247,7 +260,7 @@ func TestStatsRecordRoute(t *testing.T) {
 	s.RecordRoute("proxy")
 	s.RecordRoute("proxy")

-	snap := s.Snapshot(DomainSortBytes, SourceAll)
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if snap.RuleHits.Block != 1 || snap.RuleHits.Direct != 1 || snap.RuleHits.Proxy != 2 {
 		t.Fatalf("unexpected rule hits: %+v", snap.RuleHits)
 	}
@@ -297,7 +310,7 @@ func TestStatsHistoryRingBounded(t *testing.T) {

 func TestStatsSampleClampsLongGap(t *testing.T) {
 	s := newTestStats(t)
-	s.RecordDNS("example.com")
+	s.RecordDNS("example.com", "")
 	s.RecordRoute("proxy")
 	s.histMu.Lock()
 	s.lastSample = sampleCounters{at: time.Now().Add(-10 * time.Minute)}
@@ -322,22 +335,22 @@ func TestStatsSnapshotDomainSortModes(t *testing.T) {
 	s.domains["busy.example"] = &domainStat{conns: 20, bytesUp: 10, bytesDown: 10, lastSeen: now.Add(-2 * time.Minute)}
 	s.mu.Unlock()

-	byBytes := s.Snapshot(DomainSortBytes, SourceAll)
+	byBytes := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if byBytes.Domains[0].Domain != "big.example" {
 		t.Fatalf("expected big.example first by bytes, got %q", byBytes.Domains[0].Domain)
 	}

-	byRecent := s.Snapshot(DomainSortRecent, SourceAll)
+	byRecent := s.Snapshot(DomainSortRecent, SourceAll, "")
 	if byRecent.Domains[0].Domain != "hot.example" {
 		t.Fatalf("expected hot.example first by recency, got %q", byRecent.Domains[0].Domain)
 	}

-	byConns := s.Snapshot(DomainSortConns, SourceAll)
+	byConns := s.Snapshot(DomainSortConns, SourceAll, "")
 	if byConns.Domains[0].Domain != "busy.example" {
 		t.Fatalf("expected busy.example first by conns, got %q", byConns.Domains[0].Domain)
 	}

-	fallback := s.Snapshot(DomainSort("bogus"), SourceAll)
+	fallback := s.Snapshot(DomainSort("bogus"), SourceAll, "")
 	if fallback.Domains[0].Domain != "big.example" {
 		t.Fatalf("expected invalid sort to fall back to bytes, got %q", fallback.Domains[0].Domain)
 	}
@@ -363,12 +376,12 @@ func TestStatsSnapshotSourceFilter(t *testing.T) {
 	}
 	s.mu.Unlock()

-	all := s.Snapshot(DomainSortBytes, SourceAll)
+	all := s.Snapshot(DomainSortBytes, SourceAll, "")
 	if len(all.Domains) != 2 {
 		t.Fatalf("expected 2 domains in all view, got %d", len(all.Domains))
 	}

-	dns := s.Snapshot(DomainSortBytes, SourceDNS)
+	dns := s.Snapshot(DomainSortBytes, SourceDNS, "")
 	if len(dns.Domains) != 1 || dns.Domains[0].Domain != "queried.example" || dns.Domains[0].Conns != 5 {
 		t.Fatalf("unexpected dns view: %+v", dns.Domains)
 	}
@@ -376,12 +389,12 @@ func TestStatsSnapshotSourceFilter(t *testing.T) {
 		t.Fatalf("dns view should carry no bytes, got %+v", dns.Domains[0])
 	}

-	httpView := s.Snapshot(DomainSortBytes, SourceHTTP)
+	httpView := s.Snapshot(DomainSortBytes, SourceHTTP, "")
 	if len(httpView.Domains) != 1 || httpView.Domains[0].Domain != "proxied.example" || httpView.Domains[0].Conns != 2 {
 		t.Fatalf("unexpected http view: %+v", httpView.Domains)
 	}

-	socks := s.Snapshot(DomainSortBytes, SourceSocks5)
+	socks := s.Snapshot(DomainSortBytes, SourceSocks5, "")
 	if len(socks.Domains) != 0 {
 		t.Fatalf("expected no socks5 domains, got %+v", socks.Domains)
 	}
@@ -389,15 +402,86 @@ func TestStatsSnapshotSourceFilter(t *testing.T) {

 func TestRecordDNSAttributedToDNSSource(t *testing.T) {
 	s := newTestStats(t)
-	s.RecordDNS("example.com.")
+	s.RecordDNS("example.com.", "")

-	dns := s.Snapshot(DomainSortBytes, SourceDNS)
+	dns := s.Snapshot(DomainSortBytes, SourceDNS, "")
 	if len(dns.Domains) != 1 || dns.Domains[0].Domain != "example.com" || dns.Domains[0].Conns != 1 {
 		t.Fatalf("unexpected dns view: %+v", dns.Domains)
 	}

-	httpView := s.Snapshot(DomainSortBytes, SourceHTTP)
+	httpView := s.Snapshot(DomainSortBytes, SourceHTTP, "")
 	if len(httpView.Domains) != 0 {
 		t.Fatalf("dns query must not appear in http view: %+v", httpView.Domains)
 	}
 }
+
+func TestStatsSnapshotClientFilter(t *testing.T) {
+	s := newTestStats(t)
+	now := time.Now()
+
+	s.mu.Lock()
+	s.domains["a.example"] = &domainStat{
+		conns: 4, bytesUp: 400, bytesDown: 400, lastSeen: now,
+		byClient: map[string]*clientStat{
+			"192.168.1.10": {conns: 3, bytesUp: 300, bytesDown: 300, lastSeen: now},
+			"192.168.1.20": {conns: 1, bytesUp: 100, bytesDown: 100, lastSeen: now},
+		},
+	}
+	s.domains["b.example"] = &domainStat{
+		conns: 2, bytesUp: 200, bytesDown: 200, lastSeen: now,
+		byClient: map[string]*clientStat{
+			"192.168.1.10": {conns: 2, bytesUp: 200, bytesDown: 200, lastSeen: now},
+		},
+	}
+	s.mu.Unlock()
+
+	// filtered by client: only that client's domains, with that client's values
+	view := s.Snapshot(DomainSortBytes, SourceAll, "192.168.1.10")
+	if len(view.Domains) != 2 {
+		t.Fatalf("expected 2 domains for client, got %d", len(view.Domains))
+	}
+	if view.Domains[0].Domain != "a.example" || view.Domains[0].Conns != 3 {
+		t.Fatalf("a.example should show client 10's conns (3), got %+v", view.Domains[0])
+	}
+	if view.Domains[1].Domain != "b.example" || view.Domains[1].Conns != 2 {
+		t.Fatalf("b.example should show client 10's conns (2), got %+v", view.Domains[1])
+	}
+
+	// unknown client yields an empty list
+	empty := s.Snapshot(DomainSortBytes, SourceAll, "203.0.113.9")
+	if len(empty.Domains) != 0 {
+		t.Fatalf("expected no domains for unknown client, got %+v", empty.Domains)
+	}
+}
+
+func TestStatsSnapshotClientsAggregate(t *testing.T) {
+	s := newTestStats(t)
+	now := time.Now()
+
+	s.mu.Lock()
+	s.domains["a.example"] = &domainStat{
+		lastSeen: now,
+		byClient: map[string]*clientStat{
+			"192.168.1.10": {conns: 3, bytesUp: 300, bytesDown: 300, lastSeen: now},
+		},
+	}
+	s.domains["b.example"] = &domainStat{
+		lastSeen: now,
+		byClient: map[string]*clientStat{
+			"192.168.1.10": {conns: 2, bytesUp: 200, bytesDown: 200, lastSeen: now},
+			"192.168.1.20": {conns: 5, bytesUp: 50, bytesDown: 50, lastSeen: now},
+		},
+	}
+	s.mu.Unlock()
+
+	snap := s.Snapshot(DomainSortBytes, SourceAll, "")
+	if len(snap.Clients) != 2 {
+		t.Fatalf("expected 2 clients, got %+v", snap.Clients)
+	}
+	if snap.Clients[0].IP != "192.168.1.10" || snap.Clients[0].Conns != 5 || snap.Clients[0].BytesUp != 500 {
+		t.Fatalf("unexpected top client: %+v", snap.Clients[0])
+	}
+	if snap.Clients[1].IP != "192.168.1.20" {
+		t.Fatalf("unexpected second client: %+v", snap.Clients[1])
+	}
+}
@@ -5,8 +5,8 @@
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <meta name="color-scheme" content="light" />
    <title>Sower Admin</title>
-    <script type="module" crossorigin src="/assets/index-Dlfu0XwJ.js"></script>
-    <link rel="stylesheet" crossorigin href="/assets/index-BcqEQVll.css">
+    <script type="module" crossorigin src="/assets/index-WVE-D9IS.js"></script>
+    <link rel="stylesheet" crossorigin href="/assets/index-BmVtM7ex.css">
  </head>
  <body>
    <div id="app"></div>
@@ -19,6 +19,14 @@ export interface DomainStat {
 	lastSeen: string;
 }

+export interface ClientStat {
+	ip: string;
+	conns: number;
+	bytesUp: number;
+	bytesDown: number;
+	lastSeen: string;
+}
+
 export interface TrafficSnapshot {
 	uptime: number;
 	dnsQueries: number;
@@ -35,6 +43,7 @@ export interface TrafficSnapshot {
 	bytesUp: number;
 	bytesDown: number;
 	domains: DomainStat[];
+	clients: ClientStat[];
 }

 export interface HistorySample {
@@ -123,10 +132,11 @@ export const api = {
 			method: "DELETE",
 			body: JSON.stringify({ category, rules }),
 		}),
-	traffic: (sort?: DomainSort, source?: Source) => {
+	traffic: (sort?: DomainSort, source?: Source, client?: string) => {
 		const params: string[] = [];
 		if (sort) params.push(`sort=${sort}`);
 		if (source && source !== "all") params.push(`source=${source}`);
+		if (client) params.push(`client=${encodeURIComponent(client)}`);
 		return request<TrafficSnapshot>(params.length ? `/api/traffic?${params.join("&")}` : "/api/traffic");
 	},
 	history: () => request<History>("/api/history"),
@@ -26,6 +26,7 @@
  let paused = $state(false)
  let filter = $state('')
  let sort: SortMode = $state('bytes')
+  let client = $state('')

  const activeSortLabel = $derived(sortModes.find((m) => m.value === sort)?.label ?? '流量')
  const sourceLabels: Record<Source, string> = {
@@ -47,7 +48,7 @@

  async function reload() {
    try {
-      traffic = await api.traffic(sort, source)
+      traffic = await api.traffic(sort, source, client || undefined)
      error = ''
    } catch (e) {
      if (e instanceof ApiError && e.status === 401) {
@@ -65,9 +66,10 @@

  onMount(() => startPolling(refresh, 5000))

-  // Breadcrumb-driven source switch always reloads, even while paused.
+  // Breadcrumb-driven source or client switch always reloads, even while paused.
  $effect(() => {
    source
+    client
    traffic = null
    void reload()
  })
@@ -142,6 +144,34 @@
      {/if}
    </Badge>
  </div>
+  {#if (traffic.clients ?? []).length > 0}
+    <div class="mb-3 flex flex-wrap items-center gap-1.5" role="group" aria-label="客户端过滤">
+      <button
+        type="button"
+        class="rounded-full border px-3 py-1 text-xs font-medium transition-colors {!client
+          ? 'border-primary/40 bg-primary/10 text-foreground'
+          : 'text-muted-foreground hover:text-foreground'}"
+        aria-pressed={!client}
+        onclick={() => (client = '')}
+      >
+        全部客户端
+      </button>
+      {#each traffic.clients as c}
+        <button
+          type="button"
+          class="rounded-full border px-3 py-1 font-mono text-xs transition-colors {client === c.ip
+            ? 'border-primary/40 bg-primary/10 text-foreground'
+            : 'text-muted-foreground hover:text-foreground'}"
+          aria-pressed={client === c.ip}
+          title={`${formatCount(c.conns)} 次 · ↑${formatBytes(c.bytesUp)} ↓${formatBytes(c.bytesDown)}`}
+          onclick={() => (client = c.ip)}
+        >
+          {c.ip}
+          <span class="ms-1 text-muted-foreground">{formatCount(c.conns)}</span>
+        </button>
+      {/each}
+    </div>
+  {/if}
  <Card.Card class="max-h-[70vh] overflow-auto">
    <Table.Table>
      <Table.TableHeader>