From f6e2ec02ce63e5f50c6e647ab7b0253da8dc0e7e Mon Sep 17 00:00:00 2001 From: Chase Adams Date: Tue, 27 Oct 2026 19:56:34 -0700 Subject: [PATCH 1/2] feat: upgrade retained native connections to direct UDP --- docs/maintainers/transport/peer-packets.md | 21 +- internal/connector/stream/connect.go | 4 +- internal/connector/stream/peer_connection.go | 15 +- .../daemon/peer_packets_integration_test.go | 95 ++++- .../daemon/service_separated_fixture_test.go | 2 + .../workload_transport_integration_test.go | 11 +- internal/publisher/service/lifecycle.go | 14 +- internal/publisher/service/peer_packets.go | 17 +- internal/publisher/service/run.go | 13 +- internal/transport/peer/connectivity.go | 250 +++++++++++ internal/transport/peer/connectivity_test.go | 390 ++++++++++++++++++ .../tunnelv1/native_peer_connectivity.go | 74 ++++ .../tunnelv1/native_peer_connectivity_test.go | 108 +++++ .../native-peer-connectivity.frame.hex | 1 + .../testdata/native-peer-connectivity.json | 10 + 15 files changed, 1000 insertions(+), 25 deletions(-) create mode 100644 internal/transport/peer/connectivity.go create mode 100644 internal/transport/peer/connectivity_test.go create mode 100644 pkg/protocol/tunnelv1/native_peer_connectivity.go create mode 100644 pkg/protocol/tunnelv1/native_peer_connectivity_test.go create mode 100644 pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex create mode 100644 pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json diff --git a/docs/maintainers/transport/peer-packets.md b/docs/maintainers/transport/peer-packets.md index cbcb936f4..1347267b0 100644 --- a/docs/maintainers/transport/peer-packets.md +++ b/docs/maintainers/transport/peer-packets.md @@ -152,6 +152,22 @@ and eligible local host candidates come from that mux. closing an ICE agent leaves sibling agents and relay QUIC alive. library logging is disabled; role owners report bounded outcomes through the existing observability boundary. +an opt-in native publisher run shares one socket across its two relay connections +and peer agents. the connector shares its socket across both relay attempts and +its peer agent. after the inner QUIC connection starts over relay delivery, +[connectivity preparation](../../../internal/transport/peer/connectivity.go) +exchanges one bounded message on a dedicated unidirectional stream each way. +the exact pair and delivery bind at most 32 IP-only UDP candidates and short-lived +ICE credentials. neither endpoint accepts source authority, application permission, +or a serving deadline from that message. + +gathering and consent run outside application admission. successful ICE consent +installs a direct carrier; fresh packet MAC probes must succeed before selection. +consent loss removes that carrier while the inner QUIC connection retains its +streams over relay delivery. cancellation interrupts partial signaling and joins +the agent before releasing the shared socket. this initial negotiation does not +yet restart ICE after network changes or a failed setup. + relay and standalone answer STUN Binding requests on their existing QUIC socket. the server bounds request size and response rate and ignores ICE-authenticated requests intended for a peer. its operational counter has fixed result labels. @@ -326,6 +342,9 @@ and its monotonic application counter intact. the reconnect case replaces each TCP carrier, then closes both replacements together and requires recovery on the same target. this fixture does not qualify load-balancer source handling or assignment replacement. +relay-fault cases explicitly disable direct delivery. a separate direct case +holds ICE until the first target reply, then upgrades and serves through a loss +of both relay UDP paths. it includes the default publisher socket-sharing dialer. the renewal case also retains that target past the original pair deadline, checks that both durable leg claims advanced, then applies the carrier faults. @@ -333,5 +352,5 @@ production publishers and connectors do not enable `NativePeers` yet, and production relays leave native admission disabled. a relay hello receipt proves its own checked claim, not publisher installation or application permission. pair renewal does not bypass the separate checked relay renewal. assignment/source -rebinding, deployed listener source handling, and ICE integration remain required +rebinding, deployed listener source handling, and ICE restart remain required before enabling the native path. diff --git a/internal/connector/stream/connect.go b/internal/connector/stream/connect.go index ce8a1ae5f..e9771d2c4 100644 --- a/internal/connector/stream/connect.go +++ b/internal/connector/stream/connect.go @@ -17,6 +17,7 @@ import ( "github.com/tnldotdev/tnl/internal/naming/opaqueid" "github.com/tnldotdev/tnl/internal/observability" "github.com/tnldotdev/tnl/internal/transport/muxsession" + "github.com/tnldotdev/tnl/internal/transport/peer" "github.com/tnldotdev/tnl/internal/transport/streamcopy" "github.com/tnldotdev/tnl/internal/transport/targettransport" "github.com/tnldotdev/tnl/internal/transport/tunnel" @@ -90,7 +91,8 @@ type Config struct { CleanupContext func() context.Context `json:"-"` // NativePeers requires prepared relay packet delivery. it never falls back // to ingress; production owners leave it off until rollout qualification. - NativePeers bool + NativePeers bool + PeerConnectivity peer.ConnectivityOptions } func (Config) String() string { return "private connector configuration" } diff --git a/internal/connector/stream/peer_connection.go b/internal/connector/stream/peer_connection.go index b1ea55778..7c6ceca26 100644 --- a/internal/connector/stream/peer_connection.go +++ b/internal/connector/stream/peer_connection.go @@ -27,6 +27,7 @@ type peerConnection struct { socket *peer.Socket legs [2]*peerRelayLeg paths [2]*peer.CarrierSlot + direct *peer.CarrierSlot packets *peer.Endpoint transport *quic.Transport session muxsession.Session @@ -130,7 +131,8 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare var packetKey [32]byte copy(packetKey[:], key) clear(key) - paths := [3]peer.Carrier{p.paths[0], p.paths[1]} + p.direct = peer.NewCarrierSlot() + paths := [3]peer.Carrier{p.paths[0], p.paths[1], p.direct} p.packets, err = peer.New(life, peer.Config{Kind: peer.NativePeer, Role: 0, Key: packetKey, Paths: paths}) clear(packetKey[:]) if err != nil { @@ -142,7 +144,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare connection, err := p.transport.Dial(startup, p.packets.RemoteAddr(), verification, &quic.Config{ InitialPacketSize: tunnelv1.MaxPeerPacketBytes, DisablePathMTUDiscovery: true, MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second, - MaxIncomingStreams: -1, MaxIncomingUniStreams: -1, + MaxIncomingStreams: -1, MaxIncomingUniStreams: 1, }) if err != nil { return nil, err @@ -151,6 +153,12 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare if err != nil { return nil, err } + p.renewals.Go(func() { + _ = peer.ServeConnectivity(life, p.socket, connection, p.packets, p.direct, peer.ConnectivityBinding{ + PairID: initial.pair.PairId, DeliveryID: initial.pair.DeliveryId, Generation: initial.pair.DeliveryGeneration, + RelayAddresses: [2]string{initial.pair.Relays[0].RelayAddress, initial.pair.Relays[1].RelayAddress}, + }, true, config.PeerConnectivity) + }) go p.watch() return p, nil } @@ -204,6 +212,9 @@ func (p *peerConnection) closeResources() { } } if p.socket != nil { + if p.direct != nil { + _ = p.direct.Close() + } _ = p.socket.Close() } } diff --git a/internal/daemon/peer_packets_integration_test.go b/internal/daemon/peer_packets_integration_test.go index 8716a8e28..2ed7c2e33 100644 --- a/internal/daemon/peer_packets_integration_test.go +++ b/internal/daemon/peer_packets_integration_test.go @@ -15,6 +15,7 @@ import ( "github.com/tnldotdev/tnl/internal/connector/workload" "github.com/tnldotdev/tnl/internal/testutil" "github.com/tnldotdev/tnl/internal/testutil/githubtest" + "github.com/tnldotdev/tnl/internal/transport/peer" controlv1 "github.com/tnldotdev/tnl/pkg/api/control/public/v1" ) @@ -38,7 +39,16 @@ func startNativePacketFaultSocket(t *testing.T, address, target string) *atomic. done := make(chan struct{}) go func() { defer close(done) - var client *net.UDPAddr + // preserve each sender's UDP mapping. the default publisher and connector + // both use this advertised address; a last-sender proxy mixes their QUIC. + upstreams := make(map[string]*net.UDPConn) + var readers sync.WaitGroup + defer func() { + for _, upstream := range upstreams { + _ = upstream.Close() + } + readers.Wait() + }() packet := make([]byte, 64*1024) for { n, source, err := socket.ReadFromUDP(packet) @@ -48,14 +58,30 @@ func startNativePacketFaultSocket(t *testing.T, address, target string) *atomic. if drop.Load() { continue } - if source.String() == remote.String() { - if client != nil { - _, _ = socket.WriteToUDP(packet[:n], client) + upstream := upstreams[source.String()] + if upstream == nil { + if len(upstreams) == 64 { + continue + } + upstream, err = net.DialUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}, remote) + if err != nil { + return } - } else { - client = source - _, _ = socket.WriteToUDP(packet[:n], remote) + upstreams[source.String()] = upstream + readers.Go(func() { + response := make([]byte, 64*1024) + for { + n, err := upstream.Read(response) + if err != nil { + return + } + if !drop.Load() { + _, _ = socket.WriteToUDP(response[:n], source) + } + } + }) } + _, _ = upstream.Write(packet[:n]) } }() t.Cleanup(func() { _ = socket.Close(); <-done }) @@ -131,7 +157,7 @@ func TestIntegrationPeerPacketsKeepNativeTargetAcrossRelayPacketLoss(t *testing. {"tcp-tcp-lose-b", [2]bool{true, true}, 1}, } { t.Run(carriers.name, func(t *testing.T) { - exerciseNativePacketContinuity(t, publisher, carriers.tcp, carriers.loseTCP, false) + exerciseNativePacketContinuity(t, publisher, carriers.tcp, carriers.loseTCP, false, false) }) } }) @@ -140,21 +166,44 @@ func TestIntegrationPeerPacketsKeepNativeTargetAcrossRelayPacketLoss(t *testing. func TestIntegrationPeerPacketsRenewNativeClaimsWithoutReopeningTarget(t *testing.T) { testutil.RequireTestTier(t, testutil.TestTierIntegration) - exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true) + exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true, false) } func TestIntegrationPeerPacketsReconnectBothNativeCarriersWithoutReopeningTarget(t *testing.T) { testutil.RequireTestTier(t, testutil.TestTierIntegration) for _, publisher := range []string{"quic", "tls-tcp"} { t.Run(publisher, func(t *testing.T) { - exerciseNativePacketContinuity(t, publisher, [2]bool{true, true}, -2, false) + exerciseNativePacketContinuity(t, publisher, [2]bool{true, true}, -2, false, false) + }) + } +} + +func TestIntegrationPeerPacketsUpgradeNativeTargetToDirectUDP(t *testing.T) { + testutil.RequireTestTier(t, testutil.TestTierIntegration) + for _, publisher := range []string{"default", "tls-tcp"} { + t.Run(publisher, func(t *testing.T) { + exerciseNativePacketContinuity(t, publisher, [2]bool{}, -1, false, true) }) } } -func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew bool) { +func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew, direct bool) { t.Helper() f, issuer := newWorkloadTransportFixtureTopology(t, nil, serviceRuntimeTopologyOptions{nativePeers: true}) + var directReady [2]atomic.Bool + iceGate := make(chan struct{}) + var release sync.Once + releaseICE := func() { release.Do(func() { close(iceGate) }) } + defer releaseICE() + connectivity := func(side int) peer.ConnectivityOptions { + return peer.ConnectivityOptions{DisableDirect: !direct, IncludeLoopback: true, Observe: func(state peer.ConnectivityState) { + if state == peer.ConnectivityChecking { + <-iceGate + } + directReady[side].Store(state == peer.ConnectivityDirect) + }} + } + f.peerConnectivity = connectivity(1) // health probes may connect and close, but only a socket that receives // application bytes gets an ID. a reopened target cannot restart its counter. target := serviceRuntimeListener(t) @@ -243,10 +292,11 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, defer cancel() ready, done := make(chan struct{}), make(chan error, 1) go func() { - done <- connectorstream.Serve(ctx, listener, connectorstream.Config{Control: control, NativePeers: true, ServiceID: run.ServiceID, Protocol: run.Protocol, + done <- connectorstream.Serve(ctx, listener, connectorstream.Config{Control: control, NativePeers: true, PeerConnectivity: connectivity(0), ServiceID: run.ServiceID, Protocol: run.Protocol, TrustRevision: run.TrustRevision, Endpoint: run.Endpoint, TransportTLS: f.roots, ServiceTLS: f.visitorRoots, Ready: func() error { close(ready); return nil }}) }() defer func() { + releaseICE() cancel() select { case err := <-done: @@ -300,6 +350,25 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, } check(1) counter := uint64(1) + if direct { + if directReady[0].Load() || directReady[1].Load() { + t.Fatal("ICE gate did not keep startup relay-first") + } + releaseICE() + privateWait(t, 10*time.Second, func(context.Context) (bool, error) { + return directReady[0].Load() && directReady[1].Load(), nil + }) + // both relay UDP deliveries are unavailable: only the selected ICE path + // can carry the retained inner connection's application bytes. + for _, fault := range faults { + fault.Store(true) + } + counter++ + check(counter) + for _, fault := range faults { + fault.Store(false) + } + } if renew { var original, firstClaim time.Time if err := f.inspect.QueryRowContext(t.Context(), `SELECT expires_at FROM control.native_peer_pairs WHERE service_id=$1`, run.ServiceID).Scan(&original); err != nil { @@ -342,7 +411,7 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, t.Logf("one target survived the original pair deadline; both relay claims advanced by at least %s", lastClaim.Sub(firstClaim).Round(time.Millisecond)) } for slot := range 2 { - if tcp[slot] { + if tcp[slot] || direct { continue } faults[slot].Store(true) diff --git a/internal/daemon/service_separated_fixture_test.go b/internal/daemon/service_separated_fixture_test.go index cb4fe192e..414a8539c 100644 --- a/internal/daemon/service_separated_fixture_test.go +++ b/internal/daemon/service_separated_fixture_test.go @@ -30,6 +30,7 @@ import ( "github.com/tnldotdev/tnl/internal/serviceapi" "github.com/tnldotdev/tnl/internal/testutil" "github.com/tnldotdev/tnl/internal/transport/muxsession" + "github.com/tnldotdev/tnl/internal/transport/peer" ) const serviceRuntimeClusterSecret = "0123456789abcdef0123456789abcdef" @@ -62,6 +63,7 @@ type privateSeparatedFixture struct { visitorRootPEM []byte processContext func(string) context.Context nativePeers bool + peerConnectivity peer.ConnectivityOptions relays []*daemon nativePacketDrops map[string]*atomic.Bool nativeTCPListeners map[string]*nativeTrackedListener diff --git a/internal/daemon/workload_transport_integration_test.go b/internal/daemon/workload_transport_integration_test.go index 14c9a1424..92d8e8da6 100644 --- a/internal/daemon/workload_transport_integration_test.go +++ b/internal/daemon/workload_transport_integration_test.go @@ -336,9 +336,14 @@ func startWorkloadPublishers(t *testing.T, f *privateSeparatedFixture, group, tr } ready, done := make(chan struct{}), make(chan error, 1) go func() { - done <- publisherservice.Publish(ctx, publisherservice.RunConfig{Control: control, State: f.identity.state, Access: native.Access, Identity: identity, ServiceID: service.Id, Target: assignment.Configuration.Target, Metadata: publisheridentity.ServiceMetadata(service), HeartbeatInterval: time.Second, Ready: func() error { close(ready); return nil }, Connect: func(ctx context.Context, bound *publisherservice.Bound, slot uint8) (publisherservice.SlotSession, error) { - return bound.ConnectSlotWithTransports(ctx, slot, false, quic, tcp) - }, NativePeers: f.nativePeers}) + config := publisherservice.RunConfig{Control: control, State: f.identity.state, Access: native.Access, Identity: identity, ServiceID: service.Id, Target: assignment.Configuration.Target, Metadata: publisheridentity.ServiceMetadata(service), HeartbeatInterval: time.Second, Ready: func() error { close(ready); return nil }, + NativePeers: f.nativePeers, PeerConnectivity: f.peerConnectivity, TransportTLS: f.roots} + if transport != "default" { + config.Connect = func(ctx context.Context, bound *publisherservice.Bound, slot uint8) (publisherservice.SlotSession, error) { + return bound.ConnectSlotWithTransports(ctx, slot, false, quic, tcp) + } + } + done <- publisherservice.Publish(ctx, config) }() t.Cleanup(func() { cancel() diff --git a/internal/publisher/service/lifecycle.go b/internal/publisher/service/lifecycle.go index 05b757dfe..e12a6b595 100644 --- a/internal/publisher/service/lifecycle.go +++ b/internal/publisher/service/lifecycle.go @@ -21,6 +21,7 @@ import ( "github.com/tnldotdev/tnl/internal/region" "github.com/tnldotdev/tnl/internal/security/credentials" "github.com/tnldotdev/tnl/internal/security/targettrust" + "github.com/tnldotdev/tnl/internal/transport/peer" "github.com/tnldotdev/tnl/internal/transport/targettransport" "github.com/tnldotdev/tnl/internal/transport/tunnel" authorityv1 "github.com/tnldotdev/tnl/pkg/api/control/authority/v1" @@ -65,7 +66,8 @@ type RunConfig struct { ObserveReadiness func(RunStatus) // NativePeers opts this run into background pair discovery and guarded // packet serving. server-side admission must be prepared independently. - NativePeers bool + NativePeers bool + PeerConnectivity peer.ConnectivityOptions } // RunStatus separates current readiness from the one-time Ready notification. @@ -259,9 +261,19 @@ func Publish(ctx context.Context, config RunConfig) (result error) { if err != nil { return err } + // all native pairs and both relay connections borrow this run-owned socket. + // invalidate joins their packet work before the physical socket is closed. + if config.NativePeers { + bound.peerSocket, err = peer.ListenSocket(ctx, "udp", ":0") + if err != nil { + return err + } + defer bound.peerSocket.Close() + } defer bound.Invalidate() bound.streamCapacity = int(config.StreamCapacity) bound.nativePeers = config.NativePeers + bound.peerConnectivity = config.PeerConnectivity policy, err := visitorpolicy.Prepare(ctx, managedControl, visitorscope.Scope{Kind: visitorscope.Service, ID: setup.ServiceId}, setup.ServiceRunId, setup.RunNumber, setup.RunToken, nil) if err != nil { return err diff --git a/internal/publisher/service/peer_packets.go b/internal/publisher/service/peer_packets.go index 396d38b74..33cff0735 100644 --- a/internal/publisher/service/peer_packets.go +++ b/internal/publisher/service/peer_packets.go @@ -3,6 +3,7 @@ package service import ( "context" "crypto/tls" + "encoding/base64" "errors" "sync" "time" @@ -48,6 +49,12 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding return err } defer closePeerCarriers(paths) + var direct *peer.CarrierSlot + if p.bound.peerSocket != nil { + direct = peer.NewCarrierSlot() + paths[peer.Direct] = direct + defer direct.Close() + } packets, err := peer.New(life, peer.Config{Kind: peer.NativePeer, Key: binding.key, Role: 1, Paths: paths}) if err != nil { return err @@ -62,7 +69,7 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding } return p.bound.prepared.currentCertificate() }}, &quic.Config{InitialPacketSize: tunnelv1.MaxPeerPacketBytes, DisablePathMTUDiscovery: true, - MaxIncomingStreams: 64, MaxIncomingUniStreams: -1, MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second}) + MaxIncomingStreams: 64, MaxIncomingUniStreams: 1, MaxIdleTimeout: 30 * time.Second, KeepAlivePeriod: 10 * time.Second}) if err != nil { return err } @@ -85,6 +92,14 @@ func (p *PeerSession) servePackets(ctx context.Context, prepare func(PeerBinding closed := make(chan struct{}) stopSession := context.AfterFunc(life, func() { _ = session.Close(); close(closed) }) var workers sync.WaitGroup + if direct != nil { + workers.Go(func() { + _ = peer.ServeConnectivity(life, p.bound.peerSocket, connection, packets, direct, peer.ConnectivityBinding{ + PairID: binding.pairID, DeliveryID: base64.RawURLEncoding.EncodeToString(binding.deliveryID[:]), Generation: binding.generation, + RelayAddresses: [2]string{binding.relays[0].RelayAddress, binding.relays[1].RelayAddress}, + }, false, p.bound.peerConnectivity) + }) + } slots := make(chan struct{}, 64) defer func() { cancel() diff --git a/internal/publisher/service/run.go b/internal/publisher/service/run.go index b1170e8a5..7c0ea1d1b 100644 --- a/internal/publisher/service/run.go +++ b/internal/publisher/service/run.go @@ -20,6 +20,7 @@ import ( "github.com/tnldotdev/tnl/internal/policy/visitorlease" "github.com/tnldotdev/tnl/internal/publisher" "github.com/tnldotdev/tnl/internal/transport/muxsession" + "github.com/tnldotdev/tnl/internal/transport/peer" "github.com/tnldotdev/tnl/internal/transport/proxyproto" "github.com/tnldotdev/tnl/internal/transport/tunnel" controlv1 "github.com/tnldotdev/tnl/pkg/api/control/public/v1" @@ -82,8 +83,10 @@ func (p *Prepared) currentCertificate() (*tls.Certificate, error) { type Bound struct { prepared *Prepared *boundState - streamCapacity int - nativePeers bool + streamCapacity int + nativePeers bool + peerSocket *peer.Socket + peerConnectivity peer.ConnectivityOptions } type boundState struct { @@ -274,7 +277,11 @@ func (b *Bound) ConnectSlotWithPreference(ctx context.Context, slot uint8, prefe if b.nativePeers { config.Config = &quic.Config{EnableDatagrams: true} } - return b.ConnectSlotWithTransports(ctx, slot, preferTCP, muxsession.QUICConnector{TLSConfig: roots, Config: config}, muxsession.TLSYamuxConnector{TLSConfig: roots}) + var dialer muxsession.QUICDialer + if b.peerSocket != nil { + dialer = b.peerSocket + } + return b.ConnectSlotWithTransports(ctx, slot, preferTCP, muxsession.QUICConnector{TLSConfig: roots, Config: config, Dialer: dialer}, muxsession.TLSYamuxConnector{TLSConfig: roots}) } // ConnectSlotWithTransports retains the production claim and stale-assignment diff --git a/internal/transport/peer/connectivity.go b/internal/transport/peer/connectivity.go new file mode 100644 index 000000000..f0b74a34a --- /dev/null +++ b/internal/transport/peer/connectivity.go @@ -0,0 +1,250 @@ +package peer + +import ( + "context" + "errors" + "io" + "net" + "net/netip" + "strconv" + "sync" + "time" + + "github.com/pion/ice/v4" + "github.com/pion/stun/v4" + quic "github.com/quic-go/quic-go" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +type ConnectivityState string + +var ErrConnectivityConsent = errors.New("native direct connectivity consent expired") + +const ( + ConnectivityChecking ConnectivityState = "checking" + ConnectivityDirect ConnectivityState = "direct" + ConnectivityRelay ConnectivityState = "relay" +) + +// ConnectivityOptions are supplied by native owners, never by application +// requests. qualification can keep delivery relay-only or include loopback. +type ConnectivityOptions struct { + DisableDirect bool + IncludeLoopback bool + Observe func(ConnectivityState) +} + +type ConnectivityBinding struct { + PairID, DeliveryID string + Generation int64 + RelayAddresses [2]string +} + +type ConnectivityError struct { + Code tunnelv1.ErrorCode + Cause error +} + +func (e *ConnectivityError) Error() string { return "native direct packet path could not be prepared" } +func (e *ConnectivityError) Unwrap() error { + if e == nil { + return nil + } + return e.Cause +} + +// ServeConnectivity is background work on an already-established inner QUIC +// connection. one bounded signaling stream in each direction is independent of +// application admission. the owner joins this function before closing its socket. +// ICE consent installs delivery only; endpoint MAC probes choose the new path. +func ServeConnectivity(ctx context.Context, socket *Socket, connection *quic.Conn, packets *Endpoint, slot *CarrierSlot, binding ConnectivityBinding, controlling bool, options ConnectivityOptions) (retErr error) { + if options.DisableDirect { + return nil + } + if socket == nil || connection == nil || packets == nil || slot == nil { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: ErrConfiguration} + } + if options.Observe != nil { + defer options.Observe(ConnectivityRelay) + } + defer func() { + if ctx.Err() != nil { + retErr = errors.Join(context.Cause(ctx), retErr) + return + } + if retErr != nil { + var classified *ConnectivityError + if !errors.As(retErr, &classified) || classified == nil { + retErr = &ConnectivityError{Code: tunnelv1.Unavailable, Cause: retErr} + } + } + }() + urls := make([]*stun.URI, 0, 2) + for _, address := range binding.RelayAddresses { + host, port, err := net.SplitHostPort(address) + if err != nil { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: err} + } + number, err := strconv.Atoi(port) + if err != nil || number < 1 || number > 65535 { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(ErrConfiguration, err)} + } + urls = append(urls, &stun.URI{Scheme: stun.SchemeTypeSTUN, Proto: stun.ProtoTypeUDP, Host: host, Port: number}) + } + agent, err := socket.NewICE(urls, options.IncludeLoopback) + if err != nil { + return err + } + defer func() { retErr = errors.Join(retErr, agent.Close()) }() + setup, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + message := tunnelv1.NativePeerConnectivity{ProtocolVersion: tunnelv1.Version, Kind: "native_peer_connectivity", PairID: binding.PairID, DeliveryID: binding.DeliveryID, DeliveryGeneration: binding.Generation} + message.UsernameFragment, message.Password, err = agent.GetLocalUserCredentials() + if err != nil { + return err + } + var mu sync.Mutex + var completed sync.Once + complete := make(chan struct{}) + if err := agent.OnCandidate(func(candidate ice.Candidate) { + if candidate == nil { + completed.Do(func() { close(complete) }) + return + } + mu.Lock() + if len(message.Candidates) < tunnelv1.MaxNativePeerCandidates && validConnectivityCandidate(candidate) { + message.Candidates = append(message.Candidates, candidate.Marshal()) + } + mu.Unlock() + }); err != nil { + return err + } + if err := agent.GatherCandidates(); err != nil { + return err + } + select { + case <-setup.Done(): + return context.Cause(setup) + case <-complete: + } + mu.Lock() + local := message + local.Candidates = append([]string(nil), message.Candidates...) + mu.Unlock() + if err := local.Validate(); err != nil { + return &ConnectivityError{Code: tunnelv1.Unavailable, Cause: err} + } + remote, err := exchangeConnectivity(setup, connection, local) + if err != nil { + return err + } + if remote.PairID != binding.PairID || remote.DeliveryID != binding.DeliveryID || remote.DeliveryGeneration != binding.Generation { + return &ConnectivityError{Code: tunnelv1.Unauthenticated, Cause: tunnelv1.ErrNativePeerConnectivity} + } + for _, encoded := range remote.Candidates { + candidate, err := ice.UnmarshalCandidate(encoded) + if err != nil || !validConnectivityCandidate(candidate) { + return &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} + } + if err := agent.AddRemoteCandidate(candidate); err != nil { + return err + } + } + if options.Observe != nil { + options.Observe(ConnectivityChecking) + } + lostConsent := make(chan struct{}) + var lost sync.Once + if err := agent.OnConnectionStateChange(func(state ice.ConnectionState) { + if state == ice.ConnectionStateDisconnected || state == ice.ConnectionStateFailed || state == ice.ConnectionStateClosed { + lost.Do(func() { close(lostConsent) }) + } + }); err != nil { + return err + } + var direct *ice.Conn + if controlling { + direct, err = agent.Dial(setup, remote.UsernameFragment, remote.Password) + } else { + direct, err = agent.Accept(setup, remote.UsernameFragment, remote.Password) + } + if err != nil { + return err + } + if ctx.Err() != nil { + _ = direct.Close() + return context.Cause(ctx) + } + if err := slot.Install(ICECarrier{Conn: direct}); err != nil { + return err + } + defer slot.Install(nil) + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + last := ConnectivityRelay + for { + select { + case <-ctx.Done(): + return nil + case <-connection.Context().Done(): + return nil + case <-lostConsent: + return &ConnectivityError{Code: tunnelv1.Unavailable, Cause: ErrConnectivityConsent} + case <-ticker.C: + next := ConnectivityRelay + if packets.ActivePath() == Direct { + next = ConnectivityDirect + } + if next != last && options.Observe != nil { + options.Observe(next) + } + last = next + } + } +} + +func exchangeConnectivity(ctx context.Context, connection *quic.Conn, local tunnelv1.NativePeerConnectivity) (remote tunnelv1.NativePeerConnectivity, retErr error) { + out, err := connection.OpenUniStreamSync(ctx) + if err != nil { + return remote, err + } + stopWrite := interruptOnCancel(ctx, func() error { out.CancelWrite(0); return nil }) + defer stopWrite() + defer func() { + if retErr != nil { + out.CancelWrite(0) + } + }() + if err := tunnelv1.WriteNativePeerConnectivity(out, local); err != nil { + return remote, err + } + if err := out.Close(); err != nil { + return remote, err + } + in, err := connection.AcceptUniStream(ctx) + if err != nil { + return remote, err + } + stopRead := interruptOnCancel(ctx, func() error { in.CancelRead(0); return nil }) + defer stopRead() + defer in.CancelRead(0) + remote, err = tunnelv1.ReadNativePeerConnectivity(in) + if err != nil { + return remote, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: err} + } + // a signaling stream has exactly one frame. reject trailing data rather than + // leaving an unbounded body or a second credential set attached to the pair. + var extra [1]byte + if n, err := in.Read(extra[:]); n != 0 || !errors.Is(err, io.EOF) { + return remote, &ConnectivityError{Code: tunnelv1.InvalidMessage, Cause: errors.Join(tunnelv1.ErrNativePeerConnectivity, err)} + } + return remote, nil +} + +func validConnectivityCandidate(candidate ice.Candidate) bool { + if candidate == nil || candidate.Component() != 1 || candidate.Port() <= 0 || candidate.Port() > 65535 || !candidate.NetworkType().IsUDP() || (candidate.Type() != ice.CandidateTypeHost && candidate.Type() != ice.CandidateTypeServerReflexive) { + return false + } + address, err := netip.ParseAddr(candidate.Address()) + return err == nil && address.IsValid() && !address.IsUnspecified() && !address.IsMulticast() && address.Zone() == "" +} diff --git a/internal/transport/peer/connectivity_test.go b/internal/transport/peer/connectivity_test.go new file mode 100644 index 000000000..7ab490790 --- /dev/null +++ b/internal/transport/peer/connectivity_test.go @@ -0,0 +1,390 @@ +package peer + +import ( + "context" + "encoding/binary" + "errors" + "io" + "net" + "sync" + "testing" + "time" + + "github.com/pion/ice/v4" + quic "github.com/quic-go/quic-go" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +type connectivityFixture struct { + ctx context.Context + sockets [2]*Socket + packets [2]*Endpoint + slots [2]*CarrierSlot + connections [2]*quic.Conn + binding ConnectivityBinding +} + +func newConnectivityFixture(t *testing.T) *connectivityFixture { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + t.Cleanup(cancel) + f := &connectivityFixture{ctx: ctx, binding: ConnectivityBinding{ + PairID: "npp_1234567890123456789012", DeliveryID: "AQEBAQEBAQEBAQEBAQEBAQ", Generation: 1, + RelayAddresses: [2]string{"127.0.0.1:9", "127.0.0.1:9"}, + }} + a, b := memoryPair() + c, d := memoryPair() + for side, relays := range [2][2]Carrier{{a, c}, {b, d}} { + socket, err := ListenSocket(ctx, "udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + f.sockets[side] = socket + t.Cleanup(func() { _ = socket.Close() }) + f.slots[side] = NewCarrierSlot() + packets, err := New(ctx, Config{Kind: NativePeer, Key: [32]byte{1}, Role: uint8(side), + Paths: [3]Carrier{relays[0], relays[1], f.slots[side]}, ProbeInterval: 20 * time.Millisecond, PathTimeout: 120 * time.Millisecond}) + if err != nil { + t.Fatal(err) + } + f.packets[side] = packets + t.Cleanup(func() { _ = packets.Close() }) + } + serverTLS, clientTLS := peerTLS(t) + clientTransport, serverTransport := &quic.Transport{Conn: f.packets[0]}, &quic.Transport{Conn: f.packets[1]} + t.Cleanup(func() { _ = clientTransport.Close(); _ = serverTransport.Close() }) + config := &quic.Config{InitialPacketSize: 1200, DisablePathMTUDiscovery: true, MaxIncomingUniStreams: 1, + MaxIdleTimeout: 5 * time.Second, KeepAlivePeriod: 100 * time.Millisecond} + listener, err := serverTransport.Listen(serverTLS, config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + f.connections[0], err = clientTransport.Dial(ctx, f.packets[0].RemoteAddr(), clientTLS, config) + if err != nil { + t.Fatal(err) + } + f.connections[1], err = listener.Accept(ctx) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + for _, c := range f.connections { + _ = c.CloseWithError(0, "") + } + }) + return f +} + +func TestNativePeerConnectivityUpgradesAndFallsBackOnSameStream(t *testing.T) { + f := newConnectivityFixture(t) + relayExchange := sharedSocketQUICExchange(t, f.ctx, f.sockets[1]) + relayExchange() + stream, err := f.connections[0].OpenStreamSync(f.ctx) + if err != nil { + t.Fatal(err) + } + if _, err := stream.Write([]byte{0}); err != nil { + t.Fatal(err) + } + in, err := f.connections[1].AcceptStream(f.ctx) + if err != nil { + t.Fatal(err) + } + var first [1]byte + if _, err := io.ReadFull(in, first[:]); err != nil { + t.Fatal(err) + } + // one actual target socket retains a counter through direct upgrade and loss. + target, targetSide := tcpPair(t) + var workers sync.WaitGroup + workers.Go(func() { + defer target.Close() + var counter uint64 + for { + if _, err := io.ReadFull(target, first[:]); err != nil { + return + } + counter++ + if _, err := target.Write(binary.BigEndian.AppendUint64(nil, counter)); err != nil { + return + } + } + }) + workers.Go(func() { + defer targetSide.Close() + for { + var request [1]byte + if _, err := io.ReadFull(in, request[:]); err != nil { + return + } + if _, err := targetSide.Write(request[:]); err != nil { + return + } + var reply [8]byte + if _, err := io.ReadFull(targetSide, reply[:]); err != nil { + return + } + if _, err := in.Write(reply[:]); err != nil { + return + } + } + }) + t.Cleanup(func() { + in.CancelRead(0) + in.CancelWrite(0) + _ = target.Close() + _ = targetSide.Close() + workers.Wait() + }) + var counter uint64 + exchange := func() { + t.Helper() + if err := stream.SetDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + if _, err := stream.Write([]byte{1}); err != nil { + t.Fatal(err) + } + var reply [8]byte + if _, err := io.ReadFull(stream, reply[:]); err != nil { + t.Fatal(err) + } + counter++ + if binary.BigEndian.Uint64(reply[:]) != counter { + t.Fatal("target socket was replaced or bytes replayed") + } + } + exchange() + if f.packets[0].ActivePath() == Direct { + t.Fatal("application did not start over a relay") + } + ctx, stop := context.WithCancel(f.ctx) + results := make(chan error, 2) + for side := range 2 { + go func() { + results <- ServeConnectivity(ctx, f.sockets[side], f.connections[side], f.packets[side], f.slots[side], f.binding, side == 0, ConnectivityOptions{IncludeLoopback: true}) + }() + } + t.Cleanup(func() { + stop() + for range 2 { + select { + case err := <-results: + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, ErrConnectivityConsent) { + t.Error(err) + } + case <-time.After(time.Second): + t.Error("connectivity shutdown did not join") + } + } + }) + // signaling is still waiting on STUN, while the application remains usable. + exchange() + awaitPath(t, f.ctx, f.packets[0], f.packets[1], Direct) + exchange() + relayExchange() + stop() + deadline, finish := context.WithTimeout(f.ctx, 2*time.Second) + defer finish() + for f.packets[0].ActivePath() == Direct || f.packets[1].ActivePath() == Direct { + select { + case <-deadline.Done(): + t.Fatal("direct loss did not fall back") + case <-time.After(10 * time.Millisecond): + } + } + exchange() + // canceling a peer must leave an actual relay QUIC connection on the same + // physical socket usable, rather than merely allowing another agent object. + relayExchange() +} + +func sharedSocketQUICExchange(t *testing.T, ctx context.Context, socket *Socket) func() { + t.Helper() + serverTLS, clientTLS := peerTLS(t) + listener, err := quic.ListenAddr("127.0.0.1:0", serverTLS, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + client, err := socket.DialQUIC(ctx, listener.Addr(), clientTLS, nil) + if err != nil { + t.Fatal(err) + } + server, err := listener.Accept(ctx) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = client.CloseWithError(0, ""); _ = server.CloseWithError(0, "") }) + if server.RemoteAddr().String() != socket.LocalAddr().String() { + t.Fatal("relay QUIC did not share physical socket") + } + stream, err := client.OpenStreamSync(ctx) + if err != nil { + t.Fatal(err) + } + if _, err := stream.Write([]byte{0}); err != nil { + t.Fatal(err) + } + incoming, err := server.AcceptStream(ctx) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + go func() { defer close(done); _, _ = io.Copy(incoming, incoming) }() + t.Cleanup(func() { incoming.CancelRead(0); incoming.CancelWrite(0); <-done }) + var ready [1]byte + if _, err := io.ReadFull(stream, ready[:]); err != nil { + t.Fatal(err) + } + return func() { + t.Helper() + if err := stream.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := stream.Write([]byte{42}); err != nil { + t.Fatal("shared relay socket write", err) + } + var reply [1]byte + if _, err := io.ReadFull(stream, reply[:]); err != nil || reply[0] != 42 { + t.Fatal("shared relay socket read", err) + } + } +} + +func TestNativePeerConnectivityRejectsCrossBindingAndBadCandidates(t *testing.T) { + for name, test := range map[string]struct { + change func(*tunnelv1.NativePeerConnectivity) + code tunnelv1.ErrorCode + trailing bool + }{ + "pair": {change: func(m *tunnelv1.NativePeerConnectivity) { m.PairID = "npp_2234567890123456789012" }, code: tunnelv1.Unauthenticated}, + "delivery": {change: func(m *tunnelv1.NativePeerConnectivity) { m.DeliveryID = "AgICAgICAgICAgICAgICAg" }, code: tunnelv1.Unauthenticated}, + "generation": {change: func(m *tunnelv1.NativePeerConnectivity) { m.DeliveryGeneration++ }, code: tunnelv1.Unauthenticated}, + "hostname": {change: func(m *tunnelv1.NativePeerConnectivity) { + m.Candidates = []string{"1 1 udp 2130706431 example.local 4443 typ host"} + }, code: tunnelv1.InvalidMessage}, + "component": {change: func(m *tunnelv1.NativePeerConnectivity) { + m.Candidates = []string{"1 2 udp 2130706431 127.0.0.1 4443 typ host"} + }, code: tunnelv1.InvalidMessage}, + "malformed": {change: func(m *tunnelv1.NativePeerConnectivity) { m.Candidates = []string{"invalid"} }, code: tunnelv1.InvalidMessage}, + "trailing": {change: func(*tunnelv1.NativePeerConnectivity) {}, code: tunnelv1.InvalidMessage, trailing: true}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + f := newConnectivityFixture(t) + ctx, cancel := context.WithCancel(f.ctx) + result := make(chan error, 1) + go func() { + result <- ServeConnectivity(ctx, f.sockets[0], f.connections[0], f.packets[0], f.slots[0], f.binding, true, ConnectivityOptions{IncludeLoopback: true}) + }() + joined := false + t.Cleanup(func() { + cancel() + if !joined { + <-result + } + }) + in, err := f.connections[1].AcceptUniStream(f.ctx) + if err != nil { + t.Fatal(err) + } + message, err := tunnelv1.ReadNativePeerConnectivity(in) + if err != nil { + t.Fatal(err) + } + test.change(&message) + out, err := f.connections[1].OpenUniStreamSync(f.ctx) + if err != nil { + t.Fatal(err) + } + if err := tunnelv1.WriteNativePeerConnectivity(out, message); err != nil { + t.Fatal(err) + } + if test.trailing { + if _, err := out.Write([]byte{0}); err != nil { + t.Fatal(err) + } + } + if err := out.Close(); err != nil { + t.Fatal(err) + } + select { + case err := <-result: + joined = true + var classified *ConnectivityError + if !errors.As(err, &classified) || classified == nil || classified.Code != test.code { + t.Fatal("wrong connectivity failure", err) + } + case <-time.After(time.Second): + t.Fatal("invalid signal did not finish") + } + if f.slots[0].Generation() != 0 { + t.Fatal("invalid signal installed packet authority") + } + }) + } +} + +func TestNativePeerConnectivityCancellationInterruptsPartialSignal(t *testing.T) { + f := newConnectivityFixture(t) + ctx, cancel := context.WithCancel(f.ctx) + result := make(chan error, 1) + go func() { + result <- ServeConnectivity(ctx, f.sockets[0], f.connections[0], f.packets[0], f.slots[0], f.binding, true, ConnectivityOptions{IncludeLoopback: true}) + }() + t.Cleanup(cancel) + in, err := f.connections[1].AcceptUniStream(f.ctx) + if err != nil { + t.Fatal(err) + } + if _, err := tunnelv1.ReadNativePeerConnectivity(in); err != nil { + t.Fatal(err) + } + out, err := f.connections[1].OpenUniStreamSync(f.ctx) + if err != nil { + t.Fatal(err) + } + // announce a body but withhold it; only cancellation can release the reader. + if _, err := out.Write(binary.BigEndian.AppendUint32(nil, 1024)); err != nil { + t.Fatal(err) + } + cancel() + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("partial signaling frame blocked shutdown") + } + if f.packets[0].ActivePath() == Direct { + t.Fatal("partial signal installed direct carrier") + } +} + +func TestNativePeerConnectivityCandidateValidation(t *testing.T) { + for _, address := range []string{"0.0.0.0", "::", "224.0.0.1", "ff02::1", "host.local", "fe80::1%en0"} { + candidate, err := ice.NewCandidateHost(&ice.CandidateHostConfig{Network: "udp", Address: address, Port: 1234, Component: 1}) + if err == nil && validConnectivityCandidate(candidate) { + t.Fatalf("accepted candidate address %q", address) + } + } + for _, address := range []string{"127.0.0.1", "192.0.2.1", "2001:db8::1"} { + candidate, err := ice.NewCandidateHost(&ice.CandidateHostConfig{Network: "udp", Address: address, Port: 1234, Component: 1}) + if err != nil || !validConnectivityCandidate(candidate) { + t.Fatalf("rejected IP candidate %q: %v", address, err) + } + } + if validConnectivityCandidate(nil) { + t.Fatal("accepted nil candidate") + } + var cause *ConnectivityError + if errors.Unwrap(cause) != nil { + t.Fatal("typed nil cause") + } + if !errors.Is(&ConnectivityError{Cause: net.ErrClosed}, net.ErrClosed) { + t.Fatal("lost original cause") + } +} diff --git a/pkg/protocol/tunnelv1/native_peer_connectivity.go b/pkg/protocol/tunnelv1/native_peer_connectivity.go new file mode 100644 index 000000000..62817282c --- /dev/null +++ b/pkg/protocol/tunnelv1/native_peer_connectivity.go @@ -0,0 +1,74 @@ +package tunnelv1 + +import ( + "errors" + "io" + "strings" + + "github.com/tnldotdev/tnl/internal/naming/opaqueid" +) + +const ( + MaxNativePeerCandidates = 32 + MaxNativePeerConnectivityBytes = 32 << 10 +) + +var ErrNativePeerConnectivity = errors.New("native peer connectivity message is invalid") + +// NativePeerConnectivity travels on the one dedicated unidirectional stream in +// each direction of the existing inner QUIC connection. it grants no application +// permission and carries no source-IP authority or renewable deadline. +type NativePeerConnectivity struct { + ProtocolVersion uint32 `json:"protocol_version"` + Kind string `json:"kind"` + PairID string `json:"pair_id"` + DeliveryID string `json:"delivery_id"` + DeliveryGeneration int64 `json:"delivery_generation"` + UsernameFragment string `json:"username_fragment"` + Password string `json:"password"` + Candidates []string `json:"candidates"` +} + +func (NativePeerConnectivity) String() string { return "native peer connectivity" } +func (NativePeerConnectivity) GoString() string { return "native peer connectivity" } + +func (m NativePeerConnectivity) Validate() error { + if m.ProtocolVersion != Version || m.Kind != "native_peer_connectivity" || !opaqueid.Valid(m.PairID, "npp_") || m.DeliveryGeneration < 1 { + return ErrNativePeerConnectivity + } + if _, err := ParsePeerDeliveryIdentifier(m.DeliveryID); err != nil { + return ErrNativePeerConnectivity + } + credential := func(value string, minimum int) bool { + return len(value) >= minimum && len(value) <= 256 && strings.IndexFunc(value, func(r rune) bool { + return !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '+' || r == '/') + }) == -1 + } + if !credential(m.UsernameFragment, 4) || !credential(m.Password, 22) || len(m.Candidates) == 0 || len(m.Candidates) > MaxNativePeerCandidates { + return ErrNativePeerConnectivity + } + for _, candidate := range m.Candidates { + if len(candidate) == 0 || len(candidate) > 768 || strings.IndexFunc(candidate, func(r rune) bool { return r < 32 || r > 126 }) != -1 { + return ErrNativePeerConnectivity + } + } + return nil +} + +func WriteNativePeerConnectivity(w io.Writer, m NativePeerConnectivity) error { + if err := m.Validate(); err != nil { + return err + } + return writeJSONFrame(w, m, MaxNativePeerConnectivityBytes) +} + +func ReadNativePeerConnectivity(r io.Reader) (NativePeerConnectivity, error) { + var m NativePeerConnectivity + if err := readJSONFrame(r, &m, MaxNativePeerConnectivityBytes); err != nil { + return NativePeerConnectivity{}, errors.Join(ErrNativePeerConnectivity, err) + } + if err := m.Validate(); err != nil { + return NativePeerConnectivity{}, err + } + return m, nil +} diff --git a/pkg/protocol/tunnelv1/native_peer_connectivity_test.go b/pkg/protocol/tunnelv1/native_peer_connectivity_test.go new file mode 100644 index 000000000..46abf536e --- /dev/null +++ b/pkg/protocol/tunnelv1/native_peer_connectivity_test.go @@ -0,0 +1,108 @@ +package tunnelv1 + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" +) + +func connectivityFixture() NativePeerConnectivity { + return NativePeerConnectivity{ProtocolVersion: Version, Kind: "native_peer_connectivity", + PairID: "npp_1234567890123456789012", DeliveryID: "AQEBAQEBAQEBAQEBAQEBAQ", DeliveryGeneration: 1, + UsernameFragment: "abcd", Password: "abcdefghijklmnopqrstuv", Candidates: []string{"1 1 udp 2130706431 192.0.2.1 4443 typ host"}} +} + +func TestNativePeerConnectivityGoldenAndTruncation(t *testing.T) { + m := connectivityFixture() + assertGoldenFrame(t, "native-peer-connectivity", m, WriteNativePeerConnectivity, ReadNativePeerConnectivity) + wire := goldenWire(t, "native-peer-connectivity") + for length := range len(wire) { + if _, err := ReadNativePeerConnectivity(bytes.NewReader(wire[:length])); !errors.Is(err, ErrNativePeerConnectivity) { + t.Fatalf("accepted partial signaling at byte %d: %v", length, err) + } + } + if strings.Contains(fmt.Sprintf("%v %#v", m, m), m.Password) { + t.Fatal("formatted connectivity exposed credentials") + } +} + +func TestNativePeerConnectivityRejectsInvalidAndBounds(t *testing.T) { + for name, change := range map[string]func(*NativePeerConnectivity){ + "version": func(m *NativePeerConnectivity) { m.ProtocolVersion++ }, + "kind": func(m *NativePeerConnectivity) { m.Kind = "other" }, + "pair": func(m *NativePeerConnectivity) { m.PairID = "npp_invalid" }, + "delivery": func(m *NativePeerConnectivity) { m.DeliveryID = strings.Repeat("A", 22) }, + "generation": func(m *NativePeerConnectivity) { m.DeliveryGeneration = 0 }, + "ufrag-short": func(m *NativePeerConnectivity) { m.UsernameFragment = "abc" }, + "ufrag-oversize": func(m *NativePeerConnectivity) { m.UsernameFragment = strings.Repeat("a", 257) }, + "password-short": func(m *NativePeerConnectivity) { m.Password = "short" }, + "password-oversize": func(m *NativePeerConnectivity) { m.Password = strings.Repeat("a", 257) }, + "credential-character": func(m *NativePeerConnectivity) { m.Password += "\n" }, + "no-candidates": func(m *NativePeerConnectivity) { m.Candidates = nil }, + "many-candidates": func(m *NativePeerConnectivity) { m.Candidates = make([]string, MaxNativePeerCandidates+1) }, + "large-candidate": func(m *NativePeerConnectivity) { m.Candidates[0] = strings.Repeat("a", 769) }, + "candidate-control": func(m *NativePeerConnectivity) { m.Candidates[0] += "\n" }, + } { + t.Run(name, func(t *testing.T) { + m := connectivityFixture() + change(&m) + var output bytes.Buffer + if err := WriteNativePeerConnectivity(&output, m); !errors.Is(err, ErrNativePeerConnectivity) || output.Len() != 0 { + t.Fatal("wrote invalid signal", err) + } + body, err := json.Marshal(m) + if err != nil { + t.Fatal(err) + } + if _, err := ReadNativePeerConnectivity(frame(string(body))); !errors.Is(err, ErrNativePeerConnectivity) { + t.Fatal("accepted invalid signal", err) + } + }) + } + body, _ := json.Marshal(connectivityFixture()) + for _, invalid := range []string{strings.Replace(string(body), `"pair_id"`, `"unknown"`, 1), strings.TrimSuffix(string(body), "}") + `,"source_ip":"127.0.0.1"}`} { + if _, err := ReadNativePeerConnectivity(frame(invalid)); !errors.Is(err, ErrNativePeerConnectivity) { + t.Fatal("accepted unknown or missing field", err) + } + } + for _, size := range []uint32{0, MaxNativePeerConnectivityBytes + 1} { + if _, err := ReadNativePeerConnectivity(bytes.NewReader(binary.BigEndian.AppendUint32(nil, size))); !errors.Is(err, ErrNativePeerConnectivity) { + t.Fatal("accepted frame size", size, err) + } + } + m := connectivityFixture() + m.UsernameFragment, m.Password = strings.Repeat("a", 256), strings.Repeat("b", 256) + m.Candidates = make([]string, MaxNativePeerCandidates) + for i := range m.Candidates { + m.Candidates[i] = strings.Repeat("c", 768) + } + var output bytes.Buffer + if err := WriteNativePeerConnectivity(&output, m); err != nil { + t.Fatal("rejected maximum bounded fields", err) + } + if _, err := ReadNativePeerConnectivity(&output); err != nil { + t.Fatal(err) + } +} + +func FuzzNativePeerConnectivity(f *testing.F) { + f.Add(goldenWire(f, "native-peer-connectivity")) + f.Add([]byte{0, 0, 0, 1, '{'}) + f.Fuzz(func(t *testing.T, data []byte) { + message, err := ReadNativePeerConnectivity(bytes.NewReader(data)) + if err != nil { + return + } + var output bytes.Buffer + if err := WriteNativePeerConnectivity(&output, message); err != nil { + t.Fatal("accepted signal could not be encoded", err) + } + if output.Len() > MaxNativePeerConnectivityBytes+4 { + t.Fatal("unbounded connectivity frame") + } + }) +} diff --git a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex new file mode 100644 index 000000000..8403a7334 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.frame.hex @@ -0,0 +1 @@ +000001197b2270726f746f636f6c5f76657273696f6e223a312c226b696e64223a226e61746976655f706565725f636f6e6e6563746976697479222c22706169725f6964223a226e70705f31323334353637383930313233343536373839303132222c2264656c69766572795f6964223a2241514542415145424151454241514542415145424151222c2264656c69766572795f67656e65726174696f6e223a312c22757365726e616d655f667261676d656e74223a2261626364222c2270617373776f7264223a226162636465666768696a6b6c6d6e6f70717273747576222c2263616e64696461746573223a5b22312031207564702032313330373036343331203139322e302e322e3120343434332074797020686f7374225d7d diff --git a/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json new file mode 100644 index 000000000..254cfe273 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/native-peer-connectivity.json @@ -0,0 +1,10 @@ +{ + "protocol_version": 1, + "kind": "native_peer_connectivity", + "pair_id": "npp_1234567890123456789012", + "delivery_id": "AQEBAQEBAQEBAQEBAQEBAQ", + "delivery_generation": 1, + "username_fragment": "abcd", + "password": "abcdefghijklmnopqrstuv", + "candidates": ["1 1 udp 2130706431 192.0.2.1 4443 typ host"] +} From 3473c05b542ed4d1a7cd94f101d581cd84cc89af Mon Sep 17 00:00:00 2001 From: Chase Adams Date: Tue, 27 Oct 2026 20:27:54 -0700 Subject: [PATCH 2/2] test: retain safe Drizzle subprocess failure details --- .../fixtures/connections/installed-drizzle.ts | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/packages/ts/test/fixtures/connections/installed-drizzle.ts b/packages/ts/test/fixtures/connections/installed-drizzle.ts index 80ba97abb..258694f81 100644 --- a/packages/ts/test/fixtures/connections/installed-drizzle.ts +++ b/packages/ts/test/fixtures/connections/installed-drizzle.ts @@ -178,9 +178,11 @@ try { } `, ); + stage = "sdk synchronous options"; await run(process.execPath, [verification], { cwd: consumer, timeout: 60_000 }); // drizzle kit loads this ordinary typescript config through its commonjs // loader. there is no module cache injection, custom loader or top-level await. + stage = "drizzle migration"; await run("pnpm", ["exec", "drizzle-kit", "migrate"], { cwd: consumer, timeout: 60_000, @@ -209,12 +211,25 @@ try { } `, ); + stage = "drizzle migration query"; await run(process.execPath, [verification], { cwd: consumer, timeout: 60_000 }); } process.stdout.write("installed package connection checks passed\n"); -} catch { +} catch (cause) { // driver and subprocess output may contain application account details. - throw new Error(`installed pnpm package verification failed during ${stage}`); + const status = z + .object({ + code: z.union([z.number().int(), z.string().regex(/^[A-Z0-9_]{1,64}$/)]).optional(), + signal: z.enum(["SIGTERM", "SIGKILL", "SIGABRT", "SIGSEGV", "SIGINT"]).nullable().optional(), + killed: z.boolean().optional(), + }) + .safeParse(cause); + const failure = new Error( + `installed pnpm package verification failed during ${stage}: ${JSON.stringify(status.success ? status.data : {})}`, + { cause }, + ); + process.stderr.write(failure.message + "\n"); + process.exitCode = 1; } finally { await rm(directory, { recursive: true, force: true }); }