Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 66 additions & 3 deletions peer/peer.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ import (
"math/rand/v2"
"net"
"runtime/debug"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
Expand Down Expand Up @@ -195,6 +197,12 @@ type Client struct {
// contend with Start/Stop holding c.mu.
connsMu sync.Mutex
connsByIP map[string]int
// verifierConns counts open verifier streams per "ip:port" source, so
// their closes, which carry no destination, are hidden like their
// accepts. A count rather than a set because samizdat multiplexes
// streams over one TCP conn, so several can share a source. Guarded by
// connsMu.
verifierConns map[string]int

// rejections tallies destinations refused by reject rules, across
// sessions, so allowlist gaps are visible. Flushed on each heartbeat and
Expand Down Expand Up @@ -509,12 +517,17 @@ func (c *Client) Start(ctx context.Context) (retErr error) {
// can flip the level. The listener-registration line below stays
// at Info — that's a once-per-session lifecycle event, not a
// per-connection breadcrumb.
slog.Debug("peer listener: forwarding connection event",
"state", state, "source", source)
c.trackConn(state, source)
if evt.Rejected {
c.rejections.record(evt.Destination)
}
if c.isVerifierConn(state, source, evt.Destination) {
slog.Debug("peer listener: hiding verifier connection",
"state", state, "source", source)
return
}
slog.Debug("peer listener: forwarding connection event",
"state", state, "source", source)
c.trackConn(state, source)
events.Emit(ConnectionEvent{State: state, Source: source, Timestamp: time.Now().UnixMilli()})
})
slog.Info("peer listener: registered with peerconn", "route_id", regResp.RouteID)
Expand Down Expand Up @@ -819,6 +832,56 @@ func (c *Client) resetConnTracking() {
c.connsMu.Lock()
defer c.connsMu.Unlock()
c.connsByIP = nil
c.verifierConns = nil
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

// verifierCallbackHosts are the hosts lantern-cloud's verifier fetches through
// a peer to prove it can proxy: "api." + the environment's public domain, per
// defaultPeerVerifyCallbackURL in lantern-cloud's cmd/api/peer_verify.go. They
// are lantern-cloud itself checking the route, not a person being helped.
var verifierCallbackHosts = []string{"api.iantem.io", "api.staging.iantem.io"}

// isVerifierConn reports whether a connection event belongs to lantern-cloud's
// verifier, which re-verifies serving peers too, so that it is neither shown
// as a person helped nor reported to the server as a client. An accept is
// matched by destination. A close carries no destination, so each hidden
// accept hides one later close from the same source. When ordinary and
// verifier streams share a source, which close is hidden may not be the
// verifier's own, but the per-IP totals still balance.
func (c *Client) isVerifierConn(state int, source, destination string) bool {
c.connsMu.Lock()
defer c.connsMu.Unlock()
// Same reason as trackConn: a late accept must not repopulate the set
// after resetConnTracking has cleared it.
if c.listenerDraining.Load() {
return false
}
if state < 0 {
n := c.verifierConns[source]
if n == 0 {
return false
}
if n == 1 {
delete(c.verifierConns, source)
} else {
c.verifierConns[source] = n - 1
}
return true
}
host, _, err := net.SplitHostPort(destination)
if err != nil {
host = destination
}
if !slices.ContainsFunc(verifierCallbackHosts, func(h string) bool {
return strings.EqualFold(h, host)
}) {
return false
}
if c.verifierConns == nil {
c.verifierConns = make(map[string]int)
}
c.verifierConns[source]++
return true
}

// heartbeatLoop closes done on exit so Stop can wait for the loop before
Expand Down
82 changes: 82 additions & 0 deletions peer/peer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"github.com/stretchr/testify/require"

box "github.com/getlantern/lantern-box"
"github.com/getlantern/lantern-box/tracker/peerconn"

"github.com/getlantern/radiance/common"
"github.com/getlantern/radiance/common/settings"
Expand Down Expand Up @@ -1229,6 +1230,87 @@ func TestTrackConn_IgnoresUnmatchedClose(t *testing.T) {
assert.Equal(t, 1, c.ActiveClients(), "and must not corrupt later counting")
}

// TestIsVerifierConn_PairsCloseWithAccept pins that a verifier close, which
// carries no destination, is matched by the source its accept recorded, and
// that an ordinary connection from the same IP is not swept up with it.
func TestIsVerifierConn_PairsCloseWithAccept(t *testing.T) {
c := &Client{}

assert.True(t, c.isVerifierConn(1, "35.233.177.182:33792", "api.iantem.io:443"))
assert.True(t, c.isVerifierConn(1, "35.233.177.183:1000", "API.staging.iantem.io:443"))
assert.False(t, c.isVerifierConn(1, "35.233.177.182:33793", "google.com:443"))

assert.True(t, c.isVerifierConn(-1, "35.233.177.182:33792", ""))
assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33792", ""),
"a second close for the same source is not the verifier's")
assert.False(t, c.isVerifierConn(-1, "35.233.177.182:33793", ""))

c.resetConnTracking()
assert.False(t, c.isVerifierConn(-1, "35.233.177.183:1000", ""),
"a reset forgets verifier connections from the old session")
}

// TestClient_VerifierStreamsSharingASource covers samizdat multiplexing: a
// client's ordinary stream and two api.iantem.io streams share one source.
// Both hidden accepts must hide a close each, so the ordinary stream stays
// counted until its own close.
func TestClient_VerifierStreamsSharingASource(t *testing.T) {
fwd := &fakeForwarder{externalIP: "203.0.113.42"}
c := newTestClient(t, fwd, &fakeBoxService{}, newStubServer(t))
ctx := context.Background()
require.NoError(t, c.Start(ctx))
defer func() { require.NoError(t, c.Stop(ctx)) }()

const src = "5.119.14.105:10304"
peerconn.Notify(peerconn.Event{State: +1, Source: src, Destination: "google.com:443"})
peerconn.Notify(peerconn.Event{State: +1, Source: src, Destination: "api.iantem.io:443"})
peerconn.Notify(peerconn.Event{State: +1, Source: src, Destination: "api.iantem.io:443"})
require.Equal(t, 1, c.ActiveClients())

peerconn.Notify(peerconn.Event{State: -1, Source: src})
peerconn.Notify(peerconn.Event{State: -1, Source: src})
assert.Equal(t, 1, c.ActiveClients(), "two hidden accepts absorb two closes")
peerconn.Notify(peerconn.Event{State: -1, Source: src})
assert.Equal(t, 0, c.ActiveClients())
}

// TestClient_HidesVerifierConnections covers lantern-cloud's verify callback,
// which reaches api.iantem.io through the peer: it must not reach the UI as a
// person helped nor be reported to the server as a client.
func TestClient_HidesVerifierConnections(t *testing.T) {
fwd := &fakeForwarder{externalIP: "203.0.113.42"}
c := newTestClient(t, fwd, &fakeBoxService{}, newStubServer(t))
ctx := context.Background()

got := make(chan ConnectionEvent, 8)
sub := events.Subscribe(func(evt ConnectionEvent) { got <- evt })
defer sub.Unsubscribe()

require.NoError(t, c.Start(ctx))
peerconn.Notify(peerconn.Event{State: +1, Source: "35.233.177.182:33792", Destination: "api.iantem.io:443"})
peerconn.Notify(peerconn.Event{State: +1, Source: "5.119.14.105:10304", Destination: "google.com:443"})
assert.Equal(t, 1, c.ActiveClients(), "the verifier is not a client")
peerconn.Notify(peerconn.Event{State: -1, Source: "35.233.177.182:33792"})
peerconn.Notify(peerconn.Event{State: -1, Source: "5.119.14.105:10304"})

var sources []string
for len(sources) < 2 {
select {
case evt := <-got:
sources = append(sources, evt.Source)
case <-time.After(2 * time.Second):
t.Fatalf("got %d connection events, want 2", len(sources))
}
}
assert.Equal(t, []string{"5.119.14.105:10304", "5.119.14.105:10304"}, sources)
select {
case evt := <-got:
t.Fatalf("unexpected connection event for %s", evt.Source)
case <-time.After(100 * time.Millisecond):
}
require.NoError(t, c.Stop(ctx))
}

// TestResetConnTracking_ClearsAcrossSessions covers Stop → Start: the new
// session's box has its own connections, so inheriting the old tally would
// report load that no longer exists.
Expand Down
Loading