From 41906d4a1ac6202e1bbd8746b48b1cd7bcc4bbc4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Tue, 25 Aug 2026 18:51:21 +0200 Subject: [PATCH] [client] Pin the file drop port over the tunnel The file drop server listened on 41421, which falls inside the ephemeral port range on Linux (32768-60999) and Windows (49152-65535), so any outbound connection could take it after boot. The receiver then bound a dynamic port and advertised it over signaling, and the sender waited for that advertisement before retrying. That coupled a data plane feature to signaling traffic: once a peer connection is established there is no reason for another offer or answer to go out, so a sender could wait out the grace period for an advertisement that never came. Move the port to 22042, next to the SSH (22022) and DNS forwarder (22054) ports and clear of both ephemeral ranges, and keep the tunnel side fixed the way SSH does. A receiver that cannot bind it falls back to a dynamic port and redirects 22042 to it with an inbound DNAT rule, so senders always dial the well known port and never negotiate. NB_FILEDROP_PORT overrides the local bind only. The DNAT runs ahead of the filter on every backend (nftables prerouting at NAT dest priority, iptables nat/PREROUTING, and the userspace filter's translate-then-redecode path), so the netstack service registry keeps taking the bound port. This drops the port registry, the retry that waited on it, and the signaling plumbing that fed it. --- client/internal/engine.go | 2 - client/internal/engine_filedrop.go | 77 +++++++++++++++++---- client/internal/filedrop/filedrop_test.go | 64 ++---------------- client/internal/filedrop/history.go | 2 +- client/internal/filedrop/manager.go | 73 ++------------------ client/internal/filedrop/ports.go | 81 ----------------------- client/internal/filedrop/protocol.go | 8 ++- client/internal/peer/signaler.go | 10 --- shared/signal/client/client.go | 2 - 9 files changed, 83 insertions(+), 236 deletions(-) delete mode 100644 client/internal/filedrop/ports.go diff --git a/client/internal/engine.go b/client/internal/engine.go index 86331ea3c..17a4cf5cf 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -1971,8 +1971,6 @@ func (e *Engine) receiveSignalEvents() error { return err } - e.recordFiledropPort(msg.Key, msg.GetBody().GetFiledropPort()) - log.Debugf("receiveMSG: took %s to get lock for peer %s with session id %s", gotLock, msg.Key, offerAnswer.SessionID) if msg.Body.Type == sProto.Body_OFFER { diff --git a/client/internal/engine_filedrop.go b/client/internal/engine_filedrop.go index 734fdf52f..15aee222f 100644 --- a/client/internal/engine_filedrop.go +++ b/client/internal/engine_filedrop.go @@ -4,9 +4,12 @@ import ( "context" "net" "net/netip" + "os" + "strconv" log "github.com/sirupsen/logrus" + firewallManager "github.com/netbirdio/netbird/client/firewall/manager" "github.com/netbirdio/netbird/client/internal/filedrop" nftypes "github.com/netbirdio/netbird/client/internal/netflow/types" "github.com/netbirdio/netbird/client/internal/peer" @@ -36,7 +39,7 @@ func (e *Engine) startFileDrop() { } wgAddr := e.wgInterface.Address() - addr := netip.AddrPortFrom(wgAddr.IP, filedrop.Port) + addr := netip.AddrPortFrom(wgAddr.IP, fileDropListenPort()) resolver := filedropResolver{status: e.statusRecorder} netstackNet := e.wgInterface.GetNet() @@ -47,7 +50,7 @@ func (e *Engine) startFileDrop() { bound := e.fileDrop.ReceiverPort() if bound == 0 { - bound = filedrop.Port + bound = addr.Port() } e.fileDropPort = bound @@ -65,24 +68,56 @@ func (e *Engine) startFileDrop() { } } - if bound != filedrop.Port { - e.signaler.SetFiledropPort(bound) - } + e.setupFileDropPortRedirection(bound) e.setFileDropTunnel() e.fileDropRunning = true } -// recordFiledropPort stores the file drop port a peer advertised over signaling; -// a value that does not fit a port is treated as the default. -func (e *Engine) recordFiledropPort(peerKey string, port uint32) { - if e.fileDrop == nil { +// setupFileDropPortRedirection keeps the tunnel-side port fixed when the receiver +// could not bind it, so senders always reach the well-known port. +func (e *Engine) setupFileDropPortRedirection(bound uint16) { + if e.firewall == nil || bound == filedrop.Port { return } - if port > 65535 { - port = 0 + + for _, addr := range e.fileDropLocalAddrs() { + if err := e.firewall.AddInboundDNAT(addr, firewallManager.ProtocolTCP, filedrop.Port, bound); err != nil { + log.Warnf("failed to add file drop port redirection on %s: %v", addr, err) + continue + } + log.Infof("file drop port redirection enabled: %s:%d -> %s:%d", addr, filedrop.Port, addr, bound) } - e.fileDrop.Ports().Set(filedrop.PeerKey(peerKey), uint16(port)) +} + +func (e *Engine) removeFileDropPortRedirection(bound uint16) { + if e.firewall == nil || bound == 0 || bound == filedrop.Port { + return + } + + for _, addr := range e.fileDropLocalAddrs() { + if err := e.firewall.RemoveInboundDNAT(addr, firewallManager.ProtocolTCP, filedrop.Port, bound); err != nil { + log.Warnf("failed to remove file drop port redirection on %s: %v", addr, err) + continue + } + log.Debugf("file drop port redirection removed: %s:%d -> %s:%d", addr, filedrop.Port, addr, bound) + } +} + +func (e *Engine) fileDropLocalAddrs() []netip.Addr { + if e.wgInterface == nil { + return nil + } + + wgAddr := e.wgInterface.Address() + var addrs []netip.Addr + if wgAddr.IP.IsValid() { + addrs = append(addrs, wgAddr.IP) + } + if wgAddr.IPv6.IsValid() { + addrs = append(addrs, wgAddr.IPv6) + } + return addrs } func (e *Engine) setFileDropTunnel() { @@ -132,7 +167,7 @@ func (e *Engine) stopFileDrop() { registrar.UnregisterNetstackService(nftypes.TCP, e.fileDropPort) } } - e.signaler.SetFiledropPort(0) + e.removeFileDropPortRedirection(e.fileDropPort) } if err := e.fileDrop.StopReceiver(); err != nil { @@ -141,3 +176,19 @@ func (e *Engine) stopFileDrop() { e.fileDropRunning = false e.fileDropPort = 0 } + +// fileDropListenPort is the port the receiver tries to bind locally; the +// tunnel-side port stays filedrop.Port whatever this resolves to. +func fileDropListenPort() uint16 { + raw := os.Getenv(filedrop.EnvPort) + if raw == "" { + return filedrop.Port + } + + port, err := strconv.ParseUint(raw, 10, 16) + if err != nil { + log.Warnf("invalid %s value %q, using %d", filedrop.EnvPort, raw, filedrop.Port) + return filedrop.Port + } + return uint16(port) +} diff --git a/client/internal/filedrop/filedrop_test.go b/client/internal/filedrop/filedrop_test.go index 3419d1724..c8da0d8e1 100644 --- a/client/internal/filedrop/filedrop_test.go +++ b/client/internal/filedrop/filedrop_test.go @@ -744,30 +744,9 @@ func TestServerFallsBackWhenPortBusy(t *testing.T) { assert.NotEqual(t, busyPort, bound, "fallback must pick a different port") } -func TestPortRegistryAwait(t *testing.T) { - reg := NewPortRegistry() - - reg.Set(testPeer, 5000) - port, changed := reg.Await(context.Background(), testPeer, 0) - assert.True(t, changed, "known differing port must return immediately") - assert.Equal(t, uint16(5000), port) - - go func() { - time.Sleep(50 * time.Millisecond) - reg.Set(testPeer, 5000) - }() - _, changed = reg.Await(context.Background(), testPeer, 5000) - assert.False(t, changed, "an advertisement equal to the used port must release the waiter as unchanged") - - ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) - defer cancel() - _, changed = reg.Await(ctx, testPeer, 5000) - assert.False(t, changed, "timeout without advertisement must report unchanged") -} - // senderManager builds a send-only manager whose dialer reaches the test server only -// on realPort; other ports behave per defaultPortBehavior ("refuse" or "hang"). -func senderManager(t *testing.T, serverAddr string, realPort uint16, defaultPortBehavior string) *Manager { +// on wantPort; any other port is refused. +func senderManager(t *testing.T, serverAddr string, wantPort uint16) *Manager { t.Helper() mgr, err := NewManager(ManagerConfig{Profile: testProfile, DataDir: t.TempDir()}) @@ -779,14 +758,10 @@ func senderManager(t *testing.T, serverAddr string, realPort uint16, defaultPort mgr.SetTunnel(func(ctx context.Context, network, addr string) (net.Conn, error) { ap, err := netip.ParseAddrPort(addr) require.NoError(t, err, "dialer must receive a valid addr") - if ap.Port() == realPort { + if ap.Port() == wantPort { var d net.Dialer return d.DialContext(ctx, network, serverAddr) } - if defaultPortBehavior == "hang" { - <-ctx.Done() - return nil, ctx.Err() - } return nil, &net.OpError{Op: "dial", Net: network, Err: errors.New("connection refused")} }, "sender") @@ -801,50 +776,25 @@ func waitForState(t *testing.T, mgr *Manager, id OfferID, want State) { }, 10*time.Second, 20*time.Millisecond, "transfer must reach state %s", want) } -func TestSendRetriesOnAdvertisedPort(t *testing.T) { +func TestSendUsesTheWellKnownPort(t *testing.T) { srv, _, _ := startTestServer(t, ModeAutoAccept, staticResolver{key: PeerKey("sender-key")}) srv.mu.RLock() serverAddr := srv.listener.Addr().String() srv.mu.RUnlock() - realPort := srv.BoundPort() - mgr := senderManager(t, serverAddr, realPort, "refuse") + mgr := senderManager(t, serverAddr, Port) id, err := mgr.Send(testPeer, "receiver", netip.AddrFrom4([4]byte{100, 64, 0, 9}), []Payload{TextPayload("t", "hello")}) require.NoError(t, err) - time.Sleep(100 * time.Millisecond) - mgr.Ports().Set(testPeer, realPort) - waitForState(t, mgr, id, StateCompleted) } -func TestSendAbortsHangingAttemptOnAdvertisedPort(t *testing.T) { - srv, _, _ := startTestServer(t, ModeAutoAccept, staticResolver{key: PeerKey("sender-key")}) - srv.mu.RLock() - serverAddr := srv.listener.Addr().String() - srv.mu.RUnlock() - realPort := srv.BoundPort() - - mgr := senderManager(t, serverAddr, realPort, "hang") +func TestSendFailsWhenNothingListensOnTheWellKnownPort(t *testing.T) { + mgr := senderManager(t, "127.0.0.1:1", 1) id, err := mgr.Send(testPeer, "receiver", netip.AddrFrom4([4]byte{100, 64, 0, 9}), []Payload{TextPayload("t", "hello")}) require.NoError(t, err) - time.Sleep(100 * time.Millisecond) - mgr.Ports().Set(testPeer, realPort) - - waitForState(t, mgr, id, StateCompleted) -} - -func TestSendFailsWhenSignalConfirmsUsedPort(t *testing.T) { - mgr := senderManager(t, "127.0.0.1:1", 1, "refuse") - - id, err := mgr.Send(testPeer, "receiver", netip.AddrFrom4([4]byte{100, 64, 0, 9}), []Payload{TextPayload("t", "hello")}) - require.NoError(t, err) - - time.Sleep(100 * time.Millisecond) - mgr.Ports().Set(testPeer, 0) - waitForState(t, mgr, id, StateFailed) } diff --git a/client/internal/filedrop/history.go b/client/internal/filedrop/history.go index 7a0c24ce3..eed2f13e7 100644 --- a/client/internal/filedrop/history.go +++ b/client/internal/filedrop/history.go @@ -19,7 +19,7 @@ const ( const ( ReasonNone FailureReason = iota // ReasonUnreachable marks a transport-level failure: nothing listens on the - // peer's file drop port, so the client is old or receiving is off. + // peer's file drop port, so the client is old or the tunnel is down. ReasonUnreachable // ReasonInterrupted marks a transfer that was still moving when the process // stopped; nothing survived to finish or resume it. diff --git a/client/internal/filedrop/manager.go b/client/internal/filedrop/manager.go index b3bcef9d2..dae7dc79a 100644 --- a/client/internal/filedrop/manager.go +++ b/client/internal/filedrop/manager.go @@ -28,10 +28,6 @@ const ( EventProgress ) -// portSignalGrace bounds how long a failed attempt waits for one signal message -// that may advertise the receiver's actual port before giving up. -const portSignalGrace = 3 * time.Second - // ErrNotConnected indicates the operation needs a running tunnel. var ErrNotConnected = errors.New("not connected") @@ -73,7 +69,6 @@ type Manager struct { sink Sink server *Server - ports *PortRegistry dial DialFunc senderName string sends map[OfferID]*sendHandle @@ -98,7 +93,6 @@ func NewManager(cfg ManagerConfig) (*Manager, error) { events: cfg.Events, offerTTL: cfg.OfferTTL, sink: cfg.Sink, - ports: NewPortRegistry(), sends: make(map[OfferID]*sendHandle), } return m, nil @@ -114,12 +108,6 @@ func (m *Manager) Policy() *PolicyStore { return m.policy } -// Ports returns the registry of peer-advertised listen ports; the engine feeds it -// from incoming signal messages. -func (m *Manager) Ports() *PortRegistry { - return m.ports -} - // ReceiverPort returns the port the receiver is actually bound to, 0 when stopped. func (m *Manager) ReceiverPort() uint16 { m.mu.Lock() @@ -394,11 +382,13 @@ func (m *Manager) SetSenderRule(peer PeerKey, rule SenderRule) error { } func (m *Manager) runSend(ctx context.Context, client *Client, handle *sendHandle, transfer Transfer, payloads []Payload) { - addr, remoteID, decision, err := m.offerWithPortRetry(ctx, client, handle, transfer.PeerKey, payloads) + addr := netip.AddrPortFrom(handle.ip, Port) + remoteID, decision, err := client.Offer(ctx, addr, payloads) if err != nil { m.failSend(ctx, transfer.ID, err) return } + m.storeRemote(handle, addr, remoteID) decision, err = client.AwaitDecision(ctx, addr, remoteID, decision) if err != nil { @@ -433,54 +423,6 @@ func (m *Manager) runSend(ctx context.Context, client *Client, handle *sendHandl m.emit(EventCompleted, m.transferOf(transfer.ID)) } -// offerWithPortRetry places the offer on the last advertised port, falling back to -// the default. When the attempt fails on the transport, it waits out one signal -// message that may carry the receiver's actual port and retries there once. A port -// learned mid-attempt aborts the attempt immediately instead of letting it hang. -func (m *Manager) offerWithPortRetry(ctx context.Context, client *Client, handle *sendHandle, key PeerKey, payloads []Payload) (netip.AddrPort, OfferID, Decision, error) { - used := m.ports.Port(key) - addr := netip.AddrPortFrom(handle.ip, effectivePort(used)) - - remoteID, decision, err := m.offerWatchingPorts(ctx, client, key, used, addr, payloads) - if err == nil { - m.storeRemote(handle, addr, remoteID) - return addr, remoteID, decision, nil - } - if ctx.Err() != nil || !transportFailure(err) { - return addr, remoteID, decision, err - } - - graceCtx, cancel := context.WithTimeout(ctx, portSignalGrace) - port, changed := m.ports.Await(graceCtx, key, used) - cancel() - if !changed { - return addr, remoteID, decision, err - } - - addr = netip.AddrPortFrom(handle.ip, effectivePort(port)) - remoteID, decision, err = client.Offer(ctx, addr, payloads) - if err != nil { - return addr, remoteID, decision, err - } - m.storeRemote(handle, addr, remoteID) - return addr, remoteID, decision, nil -} - -// offerWatchingPorts runs the offer while watching for a port advertisement that -// differs from the one in use; such an advertisement aborts the in-flight attempt. -func (m *Manager) offerWatchingPorts(ctx context.Context, client *Client, key PeerKey, used uint16, addr netip.AddrPort, payloads []Payload) (OfferID, Decision, error) { - watchCtx, cancel := context.WithCancel(ctx) - defer cancel() - - go func() { - if _, changed := m.ports.Await(watchCtx, key, used); changed { - cancel() - } - }() - - return client.Offer(watchCtx, addr, payloads) -} - func (m *Manager) storeRemote(handle *sendHandle, addr netip.AddrPort, remoteID OfferID) { m.mu.Lock() defer m.mu.Unlock() @@ -664,15 +606,8 @@ func payloadTotal(payloads []Payload) int64 { return total } -func effectivePort(advertised uint16) uint16 { - if advertised == 0 { - return Port - } - return advertised -} - // transportFailure reports whether the offer never reached the receiver; any HTTP -// response, refusal included, proves the port right and is not retried elsewhere. +// response, refusal included, means the receiver answered. func transportFailure(err error) bool { var urlErr *url.Error return errors.As(err, &urlErr) diff --git a/client/internal/filedrop/ports.go b/client/internal/filedrop/ports.go deleted file mode 100644 index 912333fe7..000000000 --- a/client/internal/filedrop/ports.go +++ /dev/null @@ -1,81 +0,0 @@ -package filedrop - -import ( - "context" - "sync" -) - -// PortRegistry tracks the file drop listen port each peer advertised over -// signaling; 0 means the well-known default. Senders can wait on it to learn a -// better port after a failed attempt. -type PortRegistry struct { - mu sync.Mutex - ports map[PeerKey]uint16 - waits map[PeerKey][]chan uint16 -} - -// NewPortRegistry returns an empty registry. -func NewPortRegistry() *PortRegistry { - return &PortRegistry{ - ports: make(map[PeerKey]uint16), - waits: make(map[PeerKey][]chan uint16), - } -} - -// Set records the port a peer advertised and releases every waiter for it. -func (r *PortRegistry) Set(key PeerKey, port uint16) { - r.mu.Lock() - r.ports[key] = port - waiters := r.waits[key] - delete(r.waits, key) - r.mu.Unlock() - - for _, ch := range waiters { - ch <- port - } -} - -// Port returns the last advertised port for a peer; 0 means default or unknown. -func (r *PortRegistry) Port(key PeerKey) uint16 { - r.mu.Lock() - defer r.mu.Unlock() - return r.ports[key] -} - -// Await returns the peer's port as soon as it differs from used, or after the next -// advertisement even when it does not, reporting whether it differs. It returns -// immediately when the currently known port already differs. -func (r *PortRegistry) Await(ctx context.Context, key PeerKey, used uint16) (uint16, bool) { - r.mu.Lock() - if port, ok := r.ports[key]; ok && port != used { - r.mu.Unlock() - return port, true - } - ch := make(chan uint16, 1) - r.waits[key] = append(r.waits[key], ch) - r.mu.Unlock() - - select { - case port := <-ch: - return port, port != used - case <-ctx.Done(): - r.drop(key, ch) - return 0, false - } -} - -func (r *PortRegistry) drop(key PeerKey, ch chan uint16) { - r.mu.Lock() - defer r.mu.Unlock() - - waiters := r.waits[key] - for i, w := range waiters { - if w == ch { - r.waits[key] = append(waiters[:i], waiters[i+1:]...) - break - } - } - if len(r.waits[key]) == 0 { - delete(r.waits, key) - } -} diff --git a/client/internal/filedrop/protocol.go b/client/internal/filedrop/protocol.go index bed9c8248..d0a65ff6e 100644 --- a/client/internal/filedrop/protocol.go +++ b/client/internal/filedrop/protocol.go @@ -6,7 +6,13 @@ import ( "time" ) -const Port uint16 = 41421 +// Port is the file drop port over the tunnel. It stays fixed whatever the +// receiver ends up binding locally: a receiver that cannot take this port binds +// another one and redirects this port to it, so senders never negotiate. +const Port uint16 = 22042 + +// EnvPort overrides the local listen port; the tunnel-side port stays Port. +const EnvPort = "NB_FILEDROP_PORT" // HeaderReceivedBytes carries the receiver's confirmed byte count in a HEAD response. const HeaderReceivedBytes = "Netbird-Received-Bytes" diff --git a/client/internal/peer/signaler.go b/client/internal/peer/signaler.go index abc3ec4b3..5e437d96b 100644 --- a/client/internal/peer/signaler.go +++ b/client/internal/peer/signaler.go @@ -1,8 +1,6 @@ package peer import ( - "sync/atomic" - "github.com/pion/ice/v4" log "github.com/sirupsen/logrus" "golang.zx2c4.com/wireguard/wgctrl/wgtypes" @@ -14,7 +12,6 @@ import ( type Signaler struct { signal signal.Client wgPrivateKey wgtypes.Key - filedropPort atomic.Uint32 } func NewSignaler(signal signal.Client, wgPrivateKey wgtypes.Key) *Signaler { @@ -47,12 +44,6 @@ func (s *Signaler) Ready() bool { return s.signal.Ready() } -// SetFiledropPort sets the file drop listen port advertised in offers and answers; -// 0 means the well-known default and is not put on the wire. -func (s *Signaler) SetFiledropPort(port uint16) { - s.filedropPort.Store(uint32(port)) -} - // SignalOfferAnswer signals either an offer or an answer to remote peer func (s *Signaler) signalOfferAnswer(offerAnswer OfferAnswer, remoteKey string, bodyType sProto.Body_Type) error { var sessionIDBytes []byte @@ -66,7 +57,6 @@ func (s *Signaler) signalOfferAnswer(offerAnswer OfferAnswer, remoteKey string, msg, err := signal.MarshalCredential(s.wgPrivateKey, remoteKey, signal.CredentialPayload{ Type: bodyType, WgListenPort: offerAnswer.WgListenPort, - FiledropPort: uint16(s.filedropPort.Load()), Credential: &signal.Credential{ UFrag: offerAnswer.IceCredentials.UFrag, Pwd: offerAnswer.IceCredentials.Pwd, diff --git a/shared/signal/client/client.go b/shared/signal/client/client.go index efd2e7e27..fb77cb90f 100644 --- a/shared/signal/client/client.go +++ b/shared/signal/client/client.go @@ -49,7 +49,6 @@ type Credential struct { type CredentialPayload struct { Type proto.Body_Type WgListenPort int - FiledropPort uint16 Credential *Credential RosenpassPubKey []byte RosenpassAddr string @@ -77,7 +76,6 @@ func MarshalCredential(myKey wgtypes.Key, remoteKey string, p CredentialPayload) Type: p.Type, Payload: fmt.Sprintf("%s:%s", p.Credential.UFrag, p.Credential.Pwd), WgListenPort: uint32(p.WgListenPort), - FiledropPort: uint32(p.FiledropPort), NetBirdVersion: version.NetbirdVersion(), RosenpassConfig: &proto.RosenpassConfig{ RosenpassPubKey: p.RosenpassPubKey,