peerconn: carry destination in accept events for abuse-detection

protocol/samizdat/inbound.go

@@ -168,10 +168,12 @@ func (i *Inbound) handleConnection(ctx context.Context, conn net.Conn, destinati
 
 	// Surface accept/close to the peer-connection listener (registered by
 	// the radiance peer client when Share My Connection is active). Cheap
-	// no-op when no listener is set.
+	// no-op when no listener is set. Destination is carried on the accept
+	// event so abuse aggregators can bucket per-(source, destination)
+	// without re-parsing protocol metadata.
 	source := metadata.Source.String()
-	peerconn.Notify(+1, source)
-	defer peerconn.Notify(-1, source)
+	peerconn.NotifyAccept(source, destination)
+	defer peerconn.NotifyClose(source)
 
 	i.logger.InfoContext(ctx, "inbound connection to ", destination)
 	done := make(chan struct{})

tracker/peerconn/listener.go

@@ -19,16 +19,37 @@ package peerconn
 
 import "sync"
 
-// Listener receives connection lifecycle notifications from lantern-box
-// inbound protocols.
+// Event describes a single connection lifecycle transition surfaced from
+// a lantern-box inbound to the peer-share consumer (radiance peer client).
+//
+//   State        +1 on accept, -1 on close
+//   Source       remote peer's "ip:port" string (empty if unavailable)
+//   Destination  the host:port the remote peer requested through the
+//                inbound (only meaningful on +1; close events leave it
+//                empty since the abuse aggregator already pairs each
+//                close with the prior accept by source identity)
 //
-//   state  +1 on accept, -1 on close
-//   source remote peer's "ip:port" string (empty if unavailable)
+// Destination carries the load-bearing abuse-detection signal: source
+// IP alone is insufficient (mobile clients change IPs, NAT pools
+// collapse to one address) but the destination distribution per source
+// bucket is highly diagnostic — one source making thousands of port-25
+// connections is a spam relay; one source making millions of HTTPS
+// connections to a fixed e-commerce list is credential stuffing.
 //
-// The listener is invoked synchronously on the inbound's accept/close path,
-// so implementations should not block on heavy work — fan out to a buffered
-// channel or goroutine if more than a few microseconds is expected.
-type Listener func(state int, source string)
+// Bucket aggregation lives outside this package (see radiance/peer);
+// the registry just hands raw events off to whoever's listening.
+type Event struct {
+	State       int
+	Source      string
+	Destination string
+}
+
+// Listener receives connection lifecycle notifications from lantern-box
+// inbound protocols. Invoked synchronously on the inbound's accept/close
+// path, so implementations should not block on heavy work — fan out to
+// a buffered channel or goroutine if more than a few microseconds is
+// expected.
+type Listener func(evt Event)
 
 var (
 	listenerMu sync.RWMutex
@@ -44,15 +65,27 @@ func SetListener(l Listener) {
 	listener = l
 }
 
-// Notify dispatches a state transition to the registered listener if one
-// is set. No-op when nothing is registered, so lantern-box uses that don't
-// care about peer-share connection events (cmd_run, the standalone CLI,
-// the radiance VPN client) pay zero cost beyond a mutex read.
-func Notify(state int, source string) {
+// Notify dispatches an event to the registered listener if one is set.
+// No-op when nothing is registered — non-peer-share consumers (cmd_run,
+// the standalone CLI, the radiance VPN client) pay zero cost beyond a
+// mutex read. Lower-level entry exposed for future hooks (byte-counting
+// conn wrappers, periodic flush emitters); accept/close call sites
+// should prefer NotifyAccept / NotifyClose.
+func Notify(evt Event) {
 	listenerMu.RLock()
 	l := listener
 	listenerMu.RUnlock()
 	if l != nil {
-		l(state, source)
+		l(evt)
 	}
 }
+
+// NotifyAccept is the convenience caller for inbound accept paths.
+func NotifyAccept(source, destination string) {
+	Notify(Event{State: +1, Source: source, Destination: destination})
+}
+
+// NotifyClose is the convenience caller for inbound close paths.
+func NotifyClose(source string) {
+	Notify(Event{State: -1, Source: source})
+}

tracker/peerconn/listener_test.go

@@ -13,49 +13,71 @@ func TestNotify_NoListener_NoOp(t *testing.T) {
 	SetListener(nil)
 	// Just must not panic. Cheap path the standalone CLI exercises on every
 	// connection.
-	Notify(+1, "1.2.3.4:5555")
-	Notify(-1, "1.2.3.4:5555")
+	NotifyAccept("1.2.3.4:5555", "example.com:443")
+	NotifyClose("1.2.3.4:5555")
 }
 
 func TestSetListener_FiresOnNotify(t *testing.T) {
 	t.Cleanup(func() { SetListener(nil) })
 
-	type call struct {
-		state  int
-		source string
-	}
 	var (
-		mu    sync.Mutex
-		calls []call
+		mu     sync.Mutex
+		events []Event
 	)
-	SetListener(func(state int, source string) {
+	SetListener(func(evt Event) {
 		mu.Lock()
 		defer mu.Unlock()
-		calls = append(calls, call{state, source})
+		events = append(events, evt)
 	})
 
-	Notify(+1, "10.0.0.1:443")
-	Notify(-1, "10.0.0.1:443")
-	Notify(+1, "10.0.0.2:443")
+	NotifyAccept("10.0.0.1:443", "example.com:443")
+	NotifyClose("10.0.0.1:443")
+	NotifyAccept("10.0.0.2:443", "spamhaus.example.org:25")
 
 	mu.Lock()
 	defer mu.Unlock()
-	assert.Equal(t, []call{
-		{+1, "10.0.0.1:443"},
-		{-1, "10.0.0.1:443"},
-		{+1, "10.0.0.2:443"},
-	}, calls)
+	assert.Equal(t, []Event{
+		{State: +1, Source: "10.0.0.1:443", Destination: "example.com:443"},
+		{State: -1, Source: "10.0.0.1:443"},
+		{State: +1, Source: "10.0.0.2:443", Destination: "spamhaus.example.org:25"},
+	}, events)
+}
+
+func TestNotifyAccept_CarriesDestination(t *testing.T) {
+	t.Cleanup(func() { SetListener(nil) })
+
+	var got Event
+	SetListener(func(evt Event) { got = evt })
+	NotifyAccept("203.0.113.1:5698", "smtp.botnet.example:25")
+
+	assert.Equal(t, +1, got.State)
+	assert.Equal(t, "203.0.113.1:5698", got.Source)
+	assert.Equal(t, "smtp.botnet.example:25", got.Destination,
+		"destination is the abuse-detection load-bearing field — must round-trip")
+}
+
+func TestNotifyClose_OmitsDestination(t *testing.T) {
+	t.Cleanup(func() { SetListener(nil) })
+
+	var got Event
+	SetListener(func(evt Event) { got = evt })
+	NotifyClose("203.0.113.1:5698")
+
+	assert.Equal(t, -1, got.State)
+	assert.Equal(t, "203.0.113.1:5698", got.Source)
+	assert.Empty(t, got.Destination,
+		"close events must NOT carry destination — abuse aggregator pairs by source identity at +1 time")
 }
 
 func TestSetListener_LastWriterWins(t *testing.T) {
 	t.Cleanup(func() { SetListener(nil) })
 
 	var firstHits, secondHits atomic.Int32
-	SetListener(func(_ int, _ string) { firstHits.Add(1) })
-	Notify(+1, "")
-	SetListener(func(_ int, _ string) { secondHits.Add(1) })
-	Notify(+1, "")
-	Notify(-1, "")
+	SetListener(func(_ Event) { firstHits.Add(1) })
+	NotifyAccept("", "")
+	SetListener(func(_ Event) { secondHits.Add(1) })
+	NotifyAccept("", "")
+	NotifyClose("")
 
 	assert.Equal(t, int32(1), firstHits.Load(),
 		"first listener should see only the pre-replace Notify")
@@ -67,11 +89,11 @@ func TestSetListener_NilUnregisters(t *testing.T) {
 	t.Cleanup(func() { SetListener(nil) })
 
 	var hits atomic.Int32
-	SetListener(func(_ int, _ string) { hits.Add(1) })
-	Notify(+1, "")
+	SetListener(func(_ Event) { hits.Add(1) })
+	NotifyAccept("", "")
 	SetListener(nil)
-	Notify(+1, "")
-	Notify(-1, "")
+	NotifyAccept("", "")
+	NotifyClose("")
 	assert.Equal(t, int32(1), hits.Load(),
 		"after SetListener(nil) further Notifys must not fire")
 }