diff --git a/docs/maintainers/transport/architecture.md b/docs/maintainers/transport/architecture.md index 19037d88c..adebd6b0e 100644 --- a/docs/maintainers/transport/architecture.md +++ b/docs/maintainers/transport/architecture.md @@ -133,9 +133,13 @@ control request, authority time, or application admission. the relay rechecks it current process lease, publisher assignment, and claim deadline before installing packet forwarding. closing a passive loser cannot retire the winner's publisher registration. a duplicate activation cannot replace a live registration or reopen -a retired delivery generation. this is startup selection, not reconnect handoff: -closing the active outer session still retires that registration. production -native admission remains disabled until its source authority is qualified. +a retired delivery generation. startup selection and reconnect handoff have +separate control messages. `native_leg_handoff` binds the new signed hello to +the previous selected nonce while retaining the publisher registration and its +replay state. connector loss detaches the carrier; the bridge retains only its +last confirmed deadline. rejected or ambiguous handoffs cannot retry selection, +and expiry still retires the registration. production native admission remains +disabled until its source authority is qualified. `httpstream` enables HTTP/1 full duplex before target responses and bounds the upload, including trickles, to 30 seconds. Body completion ends the read deadline; diff --git a/docs/maintainers/transport/peer-packets.md b/docs/maintainers/transport/peer-packets.md index f74862a92..cbcb936f4 100644 --- a/docs/maintainers/transport/peer-packets.md +++ b/docs/maintainers/transport/peer-packets.md @@ -186,8 +186,22 @@ deadline from request start plus the observed remaining duration, at most five minutes, rejecting a response already expired in transit. it cannot restart this deadline when delivery begins. publisher replacement, drain, close, cancellation, or claim expiry interrupt and join the packet workers and partial introductions. -closing one pair removes its live registration while retaining the generation -tombstone; it leaves sibling pairs and the borrowed publisher session alive. +connector socket loss detaches its carrier while the relay bridge retains the +publisher registration, replay history, and last confirmed deadline. bridge +expiry removes the registration and keeps its generation tombstone; sibling pairs +and the borrowed publisher session stay alive. + +`native_leg_handoff` selects a new signed hello nonce and names the exact previous +selected nonce. the relay requires the same pair, delivery, source, assignment, +and process lease, consumes the predecessor once, and clips selection to both +existing deadlines. it retains at most 64 selected candidate nonces per bridge; +a consumed nonce cannot return as a later candidate. an ordinary +`native_leg_activate` cannot perform this handoff or reopen a retired generation. +the connector joins the failed leg and its renewal work before preparing a +replacement. setup failures may retry before selection; an ambiguous handoff +response stops recovery of that slot. replacing either carrier preserves the +inner QUIC connection and target sockets. assignment, source, and endpoint +replacement still fence the existing pair rather than inherit its authority. pair renewal alone never advances an existing relay claim. connector background work sends a fresh `native_leg_renew` proof on the existing native control @@ -308,7 +322,9 @@ the control-backed continuity proof runs with it uses disposable PostgreSQL, two real direct-socket relays, and no ingress. QUIC and TLS/TCP publishers each serve all connector carrier combinations. UDP packet loss and closing a native outer TCP session leave one target socket -and its monotonic application counter intact. this fixture does not qualify +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. the renewal case also retains that target past the original pair deadline, checks that both durable leg claims advanced, then applies the carrier faults. diff --git a/internal/connector/stream/peer_connection.go b/internal/connector/stream/peer_connection.go index fb7fc949e..b1ea55778 100644 --- a/internal/connector/stream/peer_connection.go +++ b/internal/connector/stream/peer_connection.go @@ -26,6 +26,7 @@ type peerConnection struct { source string socket *peer.Socket legs [2]*peerRelayLeg + paths [2]*peer.CarrierSlot packets *peer.Endpoint transport *quic.Transport session muxsession.Session @@ -47,6 +48,7 @@ type peerRelayLeg struct { until time.Time changed chan struct{} retired bool + nonce [32]byte } func (l *peerRelayLeg) Close() { @@ -81,7 +83,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare results := make(chan result, 2) for slot := range 2 { go func() { - leg, err := racePeerRelayLeg(startup, life, config, p.socket, initial, signer, slot) + leg, err := racePeerRelayLeg(startup, life, config, p.socket, initial, signer, slot, nil) results <- result{slot, leg, err} }() } @@ -102,7 +104,9 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare } p.source = p.legs[0].receipt.SourceIP for slot, leg := range p.legs { - p.renewals.Go(func() { leg.renewPrepared(prepared, initial, signer, slot) }) + p.paths[slot] = peer.NewCarrierSlot() + _ = p.paths[slot].Install(leg.carrier) + p.renewals.Go(func() { p.maintainRelayLeg(config, slot) }) } // both relay claims can now be discovered by the publisher. do not start // even packet probes until it has installed its independent registrations. @@ -126,7 +130,7 @@ func preparePeerConnection(ctx context.Context, config Config, prepared *prepare var packetKey [32]byte copy(packetKey[:], key) clear(key) - paths := [3]peer.Carrier{p.legs[0].carrier, p.legs[1].carrier} + paths := [3]peer.Carrier{p.paths[0], p.paths[1]} p.packets, err = peer.New(life, peer.Config{Kind: peer.NativePeer, Role: 0, Key: packetKey, Paths: paths}) clear(packetKey[:]) if err != nil { @@ -169,7 +173,7 @@ func (p *peerConnection) watch() { return case <-ticker.C: _, err := p.prepared.peerRunForRenewal(p.initial) - if err != nil || p.legs[0].ctx.Err() != nil && p.legs[1].ctx.Err() != nil { + if err != nil { p.cancel(ErrSelectionExpired) return } @@ -188,12 +192,17 @@ func (p *peerConnection) closeResources() { if p.packets != nil { _ = p.packets.Close() } + p.renewals.Wait() for _, leg := range p.legs { if leg != nil { leg.Close() } } - p.renewals.Wait() + for _, path := range p.paths { + if path != nil { + _ = path.Close() + } + } if p.socket != nil { _ = p.socket.Close() } @@ -205,13 +214,13 @@ func (p *peerConnection) Close() { // the race joins both attempts, including a late authenticated loser. each // attempt gets its own signed nonce so it cannot replay a consumed leg proof. -func racePeerRelayLeg(ctx, life context.Context, config Config, socket *peer.Socket, binding preparedPeer, signer Signer, slot int) (*peerRelayLeg, error) { +func racePeerRelayLeg(ctx, life context.Context, config Config, socket *peer.Socket, binding preparedPeer, signer Signer, slot int, previous *peerLegPredecessor) (*peerRelayLeg, error) { quicConnector := muxsession.QUICConnector{TLSConfig: config.TransportTLS, Dialer: socket, Config: muxsession.QUICConfig{Config: &quic.Config{EnableDatagrams: true}}} tcpConnector := muxsession.TLSYamuxConnector{TLSConfig: config.TransportTLS} - return racePeerRelayCandidates(ctx, life, [2]muxsession.Connector{quicConnector, tcpConnector}, binding, signer, slot) + return racePeerRelayCandidates(ctx, life, [2]muxsession.Connector{quicConnector, tcpConnector}, binding, signer, slot, previous) } -func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession.Connector, binding preparedPeer, signer Signer, slot int) (*peerRelayLeg, error) { +func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession.Connector, binding preparedPeer, signer Signer, slot int, previous *peerLegPredecessor) (*peerRelayLeg, error) { call, cancel := context.WithCancel(ctx) defer cancel() type result struct { @@ -264,21 +273,41 @@ func racePeerRelayCandidates(ctx, life context.Context, connectors [2]muxsession winner.mu.Lock() until, retired := winner.until, winner.retired winner.mu.Unlock() + if previous != nil { + until = minSelectionTime(until, previous.until) + if winner.receipt.SourceIP != previous.source { + winner.Close() + return nil, ErrServiceConnection + } + } if retired || ctx.Err() != nil || winner.ctx.Err() != nil || !until.After(time.Now()) { winner.Close() return nil, errors.Join(context.Cause(ctx), ErrSelectionExpired) } activation, finish := context.WithDeadline(ctx, until) - err := winner.session.ActivateNativeLeg(activation, "activate") + var err error + if previous == nil { + err = winner.session.ActivateNativeLeg(activation, "activate") + } else { + err = winner.session.HandoffNativeLeg(activation, "handoff", previous.nonce[:]) + } finish() if err != nil { winner.Close() + if previous != nil { + err = errors.Join(errPeerHandoffConsumed, err) + } return nil, err } if ctx.Err() != nil || winner.ctx.Err() != nil || !until.After(time.Now()) { winner.Close() - return nil, errors.Join(context.Cause(ctx), ErrSelectionExpired) + return nil, errors.Join(errPeerHandoffConsumed, context.Cause(ctx), ErrSelectionExpired) } + winner.mu.Lock() + winner.until = until + close(winner.changed) + winner.changed = make(chan struct{}) + winner.mu.Unlock() return winner, nil } return nil, errors.Join(ErrServiceConnection, context.Cause(ctx), errors.Join(failures...)) @@ -352,6 +381,8 @@ func dialPeerRelayLeg(ctx, lifetime context.Context, connector muxsession.Connec return nil, err } leg := &peerRelayLeg{session: session, carrier: carrier, receipt: receipt, ctx: life, cancel: cancel, done: make(chan struct{}), until: until, changed: make(chan struct{})} + leg.carrier = &peerRelayCarrier{Carrier: carrier, stop: cancel} + copy(leg.nonce[:], proof.Nonce) go func() { defer close(leg.done) leg.watchAuthority() diff --git a/internal/connector/stream/peer_connection_test.go b/internal/connector/stream/peer_connection_test.go index e3c7c248d..e86ac397c 100644 --- a/internal/connector/stream/peer_connection_test.go +++ b/internal/connector/stream/peer_connection_test.go @@ -40,6 +40,7 @@ type peerTestSession struct { packets atomic.Int32 activations atomic.Int32 activate func() error + selection func(tunnelv1.NativeLegBinding) } func newPeerTestSession(t *testing.T, datagrams bool, open func(muxsession.Stream)) *peerTestSession { @@ -153,8 +154,11 @@ func peerHelloSession(t *testing.T, binding preparedPeer, signer Signer, change if err != nil { return } - if request.Type == tunnelv1.NativeLegActivate && request.NativeLegBinding != nil && request.NativeLegBinding.Matches(proof) { + if (request.Type == tunnelv1.NativeLegActivate || request.Type == tunnelv1.NativeLegHandoff) && request.NativeLegBinding != nil && request.NativeLegBinding.Matches(proof) { session.activations.Add(1) + if session.selection != nil { + session.selection(*request.NativeLegBinding) + } if session.activate != nil { if err := session.activate(); err != nil { _ = tunnelv1.WriteControl(stream, tunnelv1.Message{Type: tunnelv1.Error, ProtocolVersion: tunnelv1.Version, @@ -162,7 +166,11 @@ func peerHelloSession(t *testing.T, binding preparedPeer, signer Signer, change return } } - if err := tunnelv1.WriteControl(stream, tunnelv1.Message{Type: tunnelv1.NativeLegActivated, ProtocolVersion: tunnelv1.Version, + accepted := tunnelv1.NativeLegActivated + if request.Type == tunnelv1.NativeLegHandoff { + accepted = tunnelv1.NativeLegHandedOff + } + if err := tunnelv1.WriteControl(stream, tunnelv1.Message{Type: accepted, ProtocolVersion: tunnelv1.Version, RequestID: request.RequestID, NativeLegBinding: request.NativeLegBinding}); err != nil { return } @@ -260,7 +268,7 @@ func TestNativePeerRelayRaceJoinsCanceledLateConnect(t *testing.T) { } done := make(chan *peerRelayLeg, 1) go func() { - leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0) + leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0, nil) if err != nil { t.Error(err) } @@ -323,7 +331,7 @@ func TestNativePeerRelayActivationFailureClosesWinnerWithoutRetry(t *testing.T) } ctx, cancel := context.WithTimeout(t.Context(), time.Second) defer cancel() - leg, err := racePeerRelayCandidates(ctx, t.Context(), connectors, binding, signer, 0) + leg, err := racePeerRelayCandidates(ctx, t.Context(), connectors, binding, signer, 0, nil) if leg != nil || err == nil || primary.activations.Load() != 1 || connects.Load() != 1 || signatures.Load() != 1 { t.Fatal("activation failure retried setup or admitted a carrier", err) } @@ -360,7 +368,7 @@ func TestNativePeerRelayOverlappingHellosActivateOnlySelectedCandidate(t *testin muxsession.ConnectorFunc(func(context.Context, muxsession.Endpoint) (muxsession.Session, error) { return primary, nil }), muxsession.ConnectorFunc(func(context.Context, muxsession.Endpoint) (muxsession.Session, error) { return fallback, nil }), } - leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, signer, 0) + leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, signer, 0, nil) if err != nil || leg == nil { t.Fatal("overlapping native hellos did not select a winner", err) } @@ -407,7 +415,7 @@ func TestNativePeerRelayWinnerExpiryWhileJoiningDoesNotActivate(t *testing.T) { } done := make(chan error, 1) go func() { - leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0) + leg, err := racePeerRelayCandidates(t.Context(), t.Context(), connectors, binding, owner.signer, 0, nil) if leg != nil { leg.Close() t.Error("expired winner was returned") @@ -558,7 +566,7 @@ func TestNativePeerPacketStartupRequiresMatchingSourcesAndPublisherConfirmation( accepted <- struct{}{} controlDone := make(chan error, 1) go func() { - controlDone <- session.HandleNativeLegControl(ctx, func(context.Context) error { return nil }, + controlDone <- session.HandleNativeLegControl(ctx, func(context.Context, tunnelv1.NativeLegBinding) error { return nil }, func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) { t.Error("startup unexpectedly renewed authority") return tunnelv1.NativePairLegAccepted{}, &tunnel.ProtocolError{Code: tunnelv1.Unauthenticated} diff --git a/internal/connector/stream/peer_handoff.go b/internal/connector/stream/peer_handoff.go new file mode 100644 index 000000000..08c1209b2 --- /dev/null +++ b/internal/connector/stream/peer_handoff.go @@ -0,0 +1,107 @@ +package stream + +import ( + "context" + "errors" + "time" + + "github.com/tnldotdev/tnl/internal/transport/peer" +) + +var errPeerHandoffConsumed = errors.New("native carrier handoff attempt was consumed") + +type peerLegPredecessor struct { + nonce [32]byte + source string + until time.Time +} + +// a failed packet carrier retires the associated control owner as well. otherwise +// a live control stream could keep renewing an unusable packet registration. +type peerRelayCarrier struct { + peer.Carrier + stop context.CancelFunc +} + +var _ peer.Carrier = (*peerRelayCarrier)(nil) + +func (c *peerRelayCarrier) Send(ctx context.Context, frame []byte) error { + err := c.Carrier.Send(ctx, frame) + if err != nil { + c.stop() + } + return err +} + +func (c *peerRelayCarrier) Receive(ctx context.Context) ([]byte, error) { + frame, err := c.Carrier.Receive(ctx) + if err != nil { + c.stop() + } + return frame, err +} + +func (c *peerRelayCarrier) Close() error { + c.stop() + return c.Carrier.Close() +} + +// maintainRelayLeg owns this slot's renewal and reconnect work. replacement +// touches neither the inner QUIC connection nor any application stream. only +// failures before handoff may retry, within the predecessor's fixed deadline. +func (p *peerConnection) maintainRelayLeg(config Config, slot int) { + leg := p.legs[slot] + defer func() { leg.Close() }() + for { + renewed := make(chan struct{}) + go func() { + defer close(renewed) + leg.renewPrepared(p.prepared, p.initial, p.signer, slot) + }() + select { + case <-p.ctx.Done(): + case <-leg.ctx.Done(): + } + leg.Close() + <-renewed + if p.ctx.Err() != nil { + return + } + leg.mu.Lock() + previous := peerLegPredecessor{nonce: leg.nonce, source: leg.receipt.SourceIP, until: leg.until} + leg.mu.Unlock() + _ = p.paths[slot].Install(nil) + for { + if p.ctx.Err() != nil || !previous.until.After(time.Now()) { + return + } + binding, err := p.prepared.peerCurrent() + if err == nil && !samePeerDelivery(p.initial, binding) { + return + } + if err == nil { + call, cancel := context.WithTimeout(p.ctx, min(5*time.Second, time.Until(previous.until))) + next, err := racePeerRelayLeg(call, p.ctx, config, p.socket, binding, p.signer, slot, &previous) + cancel() + if err == nil { + leg = next + p.legs[slot] = next + if p.paths[slot].Install(next.carrier) != nil { + return + } + break + } + if errors.Is(err, errPeerHandoffConsumed) { + return + } + } + timer := time.NewTimer(min(250*time.Millisecond, time.Until(previous.until))) + select { + case <-p.ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + } +} diff --git a/internal/connector/stream/peer_handoff_test.go b/internal/connector/stream/peer_handoff_test.go new file mode 100644 index 000000000..641d9ee07 --- /dev/null +++ b/internal/connector/stream/peer_handoff_test.go @@ -0,0 +1,68 @@ +package stream + +import ( + "bytes" + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/tnldotdev/tnl/internal/transport/muxsession" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +func TestNativePeerHandoffConsumerKeepsSourceDeadlineAndOneShotSelection(t *testing.T) { + for _, outcome := range []string{"accepted", "changed source", "expired", "rejected", "ambiguous"} { + t.Run(outcome, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + _, prepared, owner := preparedPeerFixture(t) + binding := *prepared.peer + previous := peerLegPredecessor{nonce: [32]byte{1}, source: "127.0.0.1", until: time.Now().Add(3 * time.Second)} + if outcome == "expired" { + previous.until = time.Now() + } + session := peerHelloSession(t, binding, owner.signer, func(r *tunnelv1.NativePairLegAccepted) { + if outcome == "changed source" { + r.SourceIP = "127.0.0.2" + } + }, nil, true) + session.selection = func(selection tunnelv1.NativeLegBinding) { + if !bytes.Equal(selection.PreviousNonce, previous.nonce[:]) { + t.Error("handoff lost the exact predecessor") + } + } + session.activate = func() error { + switch outcome { + case "rejected": + return ErrServiceConnection + case "ambiguous": + time.Sleep(4 * time.Second) + } + return nil + } + connector := muxsession.ConnectorFunc(func(context.Context, muxsession.Endpoint) (muxsession.Session, error) { return session, nil }) + next, err := racePeerRelayCandidates(t.Context(), t.Context(), [2]muxsession.Connector{connector, connector}, binding, owner.signer, 0, &previous) + if (err == nil) != (outcome == "accepted") { + t.Fatal("unexpected handoff result", err) + } + if next != nil { + defer next.Close() + next.mu.Lock() + until := next.until + next.mu.Unlock() + if !until.Equal(previous.until) { + t.Fatal("handoff extended the predecessor deadline") + } + } + consumed := outcome == "rejected" || outcome == "ambiguous" + if errors.Is(err, errPeerHandoffConsumed) != consumed { + t.Fatal("handoff lost its retry boundary", err) + } + if (session.activations.Load() == 0) != (outcome == "changed source" || outcome == "expired") { + t.Fatal("handoff selected before checking source and authority") + } + }) + }) + } +} diff --git a/internal/daemon/peer_packets_integration_test.go b/internal/daemon/peer_packets_integration_test.go index 497b781c8..8716a8e28 100644 --- a/internal/daemon/peer_packets_integration_test.go +++ b/internal/daemon/peer_packets_integration_test.go @@ -143,6 +143,15 @@ func TestIntegrationPeerPacketsRenewNativeClaimsWithoutReopeningTarget(t *testin exerciseNativePacketContinuity(t, "quic", [2]bool{false, true}, 1, true) } +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) + }) + } +} + func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, loseTCP int, renew bool) { t.Helper() f, issuer := newWorkloadTransportFixtureTopology(t, nil, serviceRuntimeTopologyOptions{nativePeers: true}) @@ -355,6 +364,31 @@ func exerciseNativePacketContinuity(t *testing.T, publisher string, tcp [2]bool, counter++ check(counter) } + if loseTCP == -2 { + // require each recovered socket before closing both successors together. + // no surviving old carrier can hide a failed handoff on either slot. + for round := range 2 { + for slot := range 2 { + connections := listeners[slot].after(accepted[slot]) + if len(connections) != 1 { + t.Fatal("reconnect fault did not select one exact native socket", slot, len(connections)) + } + accepted[slot] = listeners[slot].snapshot() + if err := connections[0].Close(); err != nil { + t.Fatal(err) + } + if round == 0 { + privateWait(t, 5*time.Second, func(context.Context) (bool, error) { + return len(listeners[slot].after(accepted[slot])) == 1, nil + }) + counter++ + check(counter) + } + } + counter++ + check(counter) + } + } counter++ check(counter) var pairs, claims, ingresses int diff --git a/internal/relay/native_peer_delivery.go b/internal/relay/native_peer_delivery.go index 6919033ab..4ccdca7e2 100644 --- a/internal/relay/native_peer_delivery.go +++ b/internal/relay/native_peer_delivery.go @@ -25,12 +25,13 @@ type nativePacketSession struct { mu sync.Mutex closed bool jobs map[*nativePairDelivery]struct{} + bridges map[[16]byte]*nativePairBridge once sync.Once } func newNativePacketSession(parent context.Context, transport muxsession.Session) (*nativePacketSession, error) { ctx, cancel := context.WithCancel(parent) - p := &nativePacketSession{ctx: ctx, cancel: cancel, jobs: make(map[*nativePairDelivery]struct{})} + p := &nativePacketSession{ctx: ctx, cancel: cancel, jobs: make(map[*nativePairDelivery]struct{}), bridges: make(map[[16]byte]*nativePairBridge)} var err error if datagrams, ok := transport.(muxsession.DatagramSession); ok { // a QUIC session must have negotiated datagrams on both sides; do not @@ -68,6 +69,10 @@ func (p *nativePacketSession) Close() { for job := range p.jobs { jobs = append(jobs, job) } + bridges := make([]*nativePairBridge, 0, len(p.bridges)) + for _, bridge := range p.bridges { + bridges = append(bridges, bridge) + } p.mu.Unlock() p.cancel() if p.datagrams != nil { @@ -79,6 +84,9 @@ func (p *nativePacketSession) Close() { for _, job := range jobs { <-job.done } + for _, bridge := range bridges { + <-bridge.done + } }) } @@ -100,6 +108,7 @@ type nativePairDelivery struct { deadline time.Time changed chan struct{} authorityDone chan struct{} + bridge *nativePairBridge } func (*nativePairDelivery) String() string { return "native pair packet delivery" } @@ -174,18 +183,30 @@ func (a *NativeConnectorAcceptor) prepareDelivery(setup, lifetime context.Contex return d, nil } -func (d *nativePairDelivery) serve(a *NativeConnectorAcceptor, session *tunnel.Session, proof tunnelv1.NativePairLegProof, claim nativeConnectorClaim) error { - defer d.Close() +func (d *nativePairDelivery) serve(a *NativeConnectorAcceptor, session *tunnel.Session, proof tunnelv1.NativePairLegProof, claim nativeConnectorClaim) (retErr error) { + finished := make(chan error, 1) + defer func() { + d.Close() + // session shutdown may win the select after the peer reads a denial. + // joining the control owner must retain that completed protocol result. + select { + case err := <-finished: + var protocol *tunnel.ProtocolError + if errors.As(err, &protocol) { + retErr = errors.Join(retErr, err) + } + default: + } + }() d.mu.Lock() if d.closed || d.serving || d.ctx.Err() != nil { d.mu.Unlock() return net.ErrClosed } d.serving = true - finished := make(chan error, 3) d.workers.Go(func() { - finished <- session.HandleNativeLegControl(d.ctx, func(ctx context.Context) error { - return d.activate(ctx, a, proof, claim, finished) + finished <- session.HandleNativeLegControl(d.ctx, func(ctx context.Context, selection tunnelv1.NativeLegBinding) error { + return d.activate(ctx, a, proof, claim, selection) }, func(ctx context.Context, proof tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) { return d.renewClaim(ctx, a, d.transport, proof, claim) }) @@ -200,6 +221,10 @@ func (d *nativePairDelivery) serve(a *NativeConnectorAcceptor, session *tunnel.S case <-session.Done(): return nil case err := <-finished: + var protocol *tunnel.ProtocolError + if errors.As(err, &protocol) { + return err + } if errors.Is(err, net.ErrClosed) || errors.Is(err, context.Canceled) || errors.Is(err, io.EOF) { return nil } @@ -218,8 +243,8 @@ func (d *nativePairDelivery) serve(a *NativeConnectorAcceptor, session *tunnel.S // only the explicitly selected carrier may install publisher registration state. // closing an authenticated but unselected candidate cannot spend its tombstone -// or displace a sibling's live replay state. reconnect handoff is separate. -func (d *nativePairDelivery) activate(ctx context.Context, a *NativeConnectorAcceptor, proof tunnelv1.NativePairLegProof, claim nativeConnectorClaim, finished chan<- error) error { +// or displace a sibling's live replay state. handoff selects an exact predecessor. +func (d *nativePairDelivery) activate(ctx context.Context, a *NativeConnectorAcceptor, proof tunnelv1.NativePairLegProof, claim nativeConnectorClaim, selection tunnelv1.NativeLegBinding) error { d.mu.Lock() defer d.mu.Unlock() if d.closed || d.activated || d.ctx.Err() != nil || ctx.Err() != nil || !d.deadline.After(a.now()) { @@ -233,28 +258,12 @@ func (d *nativePairDelivery) activate(ctx context.Context, a *NativeConnectorAcc if err != nil { return errors.Join(&tunnel.ProtocolError{Code: tunnelv1.Unauthenticated}, err) } - binding := peer.PairBinding{Delivery: tunnelv1.PeerDelivery{Pair: id, Generation: uint64(claim.receipt.DeliveryGeneration)}, - Key: claim.key, Kind: peer.NativePeer, Path: uint8(claim.receipt.ConnectionSlot), Role: 0} - d.towardPublisher, err = d.publisher.register(ctx, binding, true) + d.bridge, err = d.publisher.selectNativeBridge(ctx, a, d, proof, claim, id, selection) if err != nil { - return errors.Join(&tunnel.ProtocolError{Code: tunnelv1.Unavailable}, err) + return err } + d.towardPublisher = d.bridge.publisher d.activated = true - connectorReady := make(chan struct{}) - d.workers.Go(func() { - finished <- forwardNativePackets(d.ctx, d.towardConnector, d.towardPublisher, connectorReady) - }) - d.workers.Go(func() { - // a publisher probe must not spend a send timeout waiting for the - // connector's first tagged TLS/yamux introduction. the connector can - // still be preparing its other leg after reading hello_accepted. - select { - case <-connectorReady: - finished <- forwardNativePackets(d.ctx, d.towardPublisher, d.towardConnector, nil) - case <-d.ctx.Done(): - finished <- d.ctx.Err() - } - }) return nil } @@ -285,8 +294,8 @@ func (d *nativePairDelivery) Close() { d.cancel() d.stopPublisher() <-d.authorityDone - if d.towardPublisher != nil { - _ = d.towardPublisher.Close() + if d.bridge != nil { + d.bridge.detach(d) } if d.towardConnector != nil { _ = d.towardConnector.Close() diff --git a/internal/relay/native_peer_delivery_shutdown_test.go b/internal/relay/native_peer_delivery_shutdown_test.go new file mode 100644 index 000000000..0860c7b1d --- /dev/null +++ b/internal/relay/native_peer_delivery_shutdown_test.go @@ -0,0 +1,109 @@ +package relay + +import ( + "context" + "errors" + "io" + "net" + "net/netip" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/tnldotdev/tnl/internal/transport/muxsession" + "github.com/tnldotdev/tnl/internal/transport/tunnel" + relayv1 "github.com/tnldotdev/tnl/pkg/api/control/private/relay/v1" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +// hold the denial write until the serving loop has selected session shutdown. +// returning EOF then proves that both the joined cause and the denial survive. +type nativeDenialWriteGate struct { + muxsession.Stream + armed atomic.Bool + started chan struct{} + release chan struct{} +} + +func (s *nativeDenialWriteGate) Write(data []byte) (int, error) { + if s.armed.Load() { + close(s.started) + <-s.release + return 0, io.EOF + } + return s.Stream.Write(data) +} + +func TestNativeDeliveryRetainsDenialAcrossControlWriteFailure(t *testing.T) { + for _, shutdown := range []bool{false, true} { + name := "control result" + if shutdown { + name = "session shutdown" + } + t.Run(name, func(t *testing.T) { + claimClient := nativeClaimFunc(func(_ context.Context, _ relayv1.RelayLease, source netip.Addr, proof tunnelv1.NativePairLegProof, _ []byte) (relayv1.NativePeerPairLegClaim, error) { + return nativeTestClaim(source, proof), nil + }) + _, owner, hello := nativeAdmissionFixture(t, 2, claimClient) + proof := *hello.NativePairLegProof + claim, err := owner.claim(t.Context(), socketIdentityOnly{}, proof) + if err != nil { + t.Fatal(err) + } + server, client := net.Pipe() + t.Cleanup(func() { _ = client.Close(); _ = server.Close() }) + if err := client.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatal(err) + } + gate := &nativeDenialWriteGate{Stream: &forwardingTestStream{Conn: server}, started: make(chan struct{}), release: make(chan struct{})} + release := sync.OnceFunc(func() { close(gate.release) }) + transport := newForwardingTestSession(gate) + ctx, cancel := context.WithCancel(t.Context()) + // a second activation is rejected before any publisher installation. + delivery := &nativePairDelivery{ctx: ctx, cancel: cancel, activated: true, + stopPublisher: func() bool { return true }, publisher: &nativePacketSession{}, + deadline: time.Now().Add(time.Minute), changed: make(chan struct{}), + done: make(chan struct{}), authorityDone: make(chan struct{})} + close(delivery.authorityDone) + ready := make(chan struct{}) + result := relayWorker(t, func() { release(); cancel(); _ = transport.Close() }, func() error { + session, _, err := tunnel.AcceptWithResult(ctx, transport, func(context.Context, tunnelv1.Message) (*tunnelv1.NativePairLegAccepted, error) { + return &claim.receipt, nil + }) + if err != nil { + return err + } + defer session.Close() + close(ready) + return delivery.serve(owner, session, proof, claim) + }) + if err := tunnelv1.WriteControl(client, hello); err != nil { + t.Fatal(err) + } + if message, err := tunnelv1.ReadControl(client); err != nil || message.Type != tunnelv1.HelloAccepted { + t.Fatal("native handshake failed", err) + } + relayAwait(t, ready) + gate.armed.Store(true) + if err := tunnelv1.WriteControl(client, tunnelv1.Message{Type: tunnelv1.NativeLegActivate, ProtocolVersion: tunnelv1.Version, + RequestID: "repeated", Role: tunnelv1.NativeConnector, NativeLegBinding: &tunnelv1.NativeLegBinding{Nonce: proof.Nonce}}); err != nil { + t.Fatal(err) + } + relayAwait(t, gate.started) + if shutdown { + _ = transport.Close() + // only delivery.Close cancels this context. wait until the outer + // loop is joining the still-blocked control worker before releasing it. + relayAwait(t, ctx.Done()) + } + release() + err = relayAwait(t, result) + var denied *tunnel.ProtocolError + if !errors.As(err, &denied) || denied == nil || denied.Code != tunnelv1.Unauthenticated || !errors.Is(err, io.EOF) { + t.Fatal("control denial or failed write was lost", err) + } + relayAwait(t, delivery.done) + }) + } +} diff --git a/internal/relay/native_peer_handoff.go b/internal/relay/native_peer_handoff.go new file mode 100644 index 000000000..adb02cd49 --- /dev/null +++ b/internal/relay/native_peer_handoff.go @@ -0,0 +1,231 @@ +package relay + +import ( + "bytes" + "context" + "errors" + "net" + "sync" + "sync/atomic" + "time" + + "github.com/tnldotdev/tnl/internal/transport/peer" + "github.com/tnldotdev/tnl/internal/transport/tunnel" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +const maxNativeCarrierCandidates = 64 + +// nativePairBridge owns the publisher registration across connector socket loss. +// its deadline and replay state survive detachment; only a current signed claim +// and an exact predecessor handoff can install another connector carrier. +type nativePairBridge struct { + owner *nativePacketSession + ctx context.Context + cancel context.CancelFunc + done chan struct{} + publisher peer.Carrier + connector *peer.CarrierSlot + claim nativeConnectorClaim + proof tunnelv1.NativePairLegProof + mu sync.Mutex + active *nativePairDelivery + nonce []byte + used map[[32]byte]struct{} + deadline time.Time +} + +func (*nativePairBridge) String() string { return "native pair packet bridge" } +func (*nativePairBridge) GoString() string { return "native pair packet bridge" } + +func (p *nativePacketSession) selectNativeBridge(ctx context.Context, a *NativeConnectorAcceptor, d *nativePairDelivery, proof tunnelv1.NativePairLegProof, claim nativeConnectorClaim, id [16]byte, selection tunnelv1.NativeLegBinding) (*nativePairBridge, error) { + denied := &tunnel.ProtocolError{Code: tunnelv1.Unauthenticated} + if !selection.Matches(proof) { + return nil, denied + } + p.mu.Lock() + defer p.mu.Unlock() + if p.closed || p.ctx.Err() != nil || ctx.Err() != nil { + return nil, errors.Join(denied, context.Cause(ctx), net.ErrClosed) + } + b := p.bridges[id] + nonce := [32]byte(selection.Nonce) + if len(selection.PreviousNonce) != 0 { + if b == nil { + return nil, denied + } + b.mu.Lock() + defer b.mu.Unlock() + if _, spent := b.used[nonce]; spent { + return nil, denied + } + if len(b.used) >= maxNativeCarrierCandidates { + return nil, &tunnel.ProtocolError{Code: tunnelv1.CapacityExceeded} + } + if b.ctx.Err() != nil || !b.deadline.After(a.now()) || !bytes.Equal(b.nonce, selection.PreviousNonce) || + !b.proof.SameLeg(proof) || b.claim.key != claim.key || b.claim.ref != claim.ref || b.claim.receipt.SourceIP != claim.receipt.SourceIP { + return nil, denied + } + // selection transfers delivery, not serving time. renewal on the selected + // control stream can later install freshly checked authority. + until := b.deadline + if d.deadline.Before(until) { + until = d.deadline + } + if !until.After(a.now()) || ctx.Err() != nil { + return nil, denied + } + old := b.active + b.nonce, b.active, b.deadline = bytes.Clone(selection.Nonce), d, until + b.used[nonce] = struct{}{} + d.deadline = until + close(d.changed) + d.changed = make(chan struct{}) + if err := b.connector.Install(&nativeReadyCarrier{Carrier: d.towardConnector, stop: d.cancel}); err != nil { + b.cancel() + return nil, errors.Join(denied, err) + } + if old != nil { + old.cancel() + } + return b, nil + } + if b != nil { + return nil, denied + } + if len(p.bridges) >= defaultNativeConnectorSessions { + return nil, &tunnel.ProtocolError{Code: tunnelv1.CapacityExceeded} + } + binding := peer.PairBinding{Delivery: tunnelv1.PeerDelivery{Pair: id, Generation: uint64(claim.receipt.DeliveryGeneration)}, + Key: claim.key, Kind: peer.NativePeer, Path: uint8(claim.receipt.ConnectionSlot), Role: 0} + published, err := p.register(ctx, binding, true) + if err != nil { + return nil, errors.Join(&tunnel.ProtocolError{Code: tunnelv1.Unavailable}, err) + } + life, cancel := context.WithCancel(p.ctx) + b = &nativePairBridge{owner: p, ctx: life, cancel: cancel, done: make(chan struct{}), publisher: published, + connector: peer.NewCarrierSlot(), claim: claim, proof: proof, active: d, nonce: bytes.Clone(selection.Nonce), + used: map[[32]byte]struct{}{nonce: {}}, deadline: d.deadline} + _ = b.connector.Install(&nativeReadyCarrier{Carrier: d.towardConnector, stop: d.cancel}) + p.bridges[id] = b + go b.serve(a, d.connection, id) + return b, nil +} + +func (b *nativePairBridge) detach(d *nativePairDelivery) { + b.mu.Lock() + defer b.mu.Unlock() + if b.active == d { + b.active = nil + _ = b.connector.Install(nil) + } +} + +func (b *nativePairBridge) renew(d *nativePairDelivery, until, now time.Time) error { + b.mu.Lock() + defer b.mu.Unlock() + if b.active != d || b.ctx.Err() != nil || !b.deadline.After(now) || !until.After(now) { + return &tunnel.ProtocolError{Code: tunnelv1.Unauthenticated} + } + b.deadline = until + return nil +} + +func (b *nativePairBridge) serve(a *NativeConnectorAcceptor, connection *serviceConnection, id [16]byte) { + defer close(b.done) + var workers sync.WaitGroup + finished := make(chan error, 2) + ready := make(chan struct{}) + workers.Go(func() { + finished <- forwardNativePackets(b.ctx, peer.RetainCarrierReplay(b.connector), b.publisher, ready) + }) + workers.Go(func() { + select { + case <-ready: + finished <- forwardNativePackets(b.ctx, b.publisher, b.connector, nil) + case <-b.ctx.Done(): + finished <- b.ctx.Err() + } + }) + defer func() { + b.cancel() + b.mu.Lock() + if b.active != nil { + b.active.cancel() + } + b.mu.Unlock() + _ = b.connector.Close() + _ = b.publisher.Close() + workers.Wait() + b.owner.mu.Lock() + delete(b.owner.bridges, id) + b.owner.mu.Unlock() + }() + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + b.mu.Lock() + until := b.deadline + expired := !until.After(a.now()) + if expired { + b.cancel() + } + b.mu.Unlock() + if expired { + return + } + timer := time.NewTimer(until.Sub(a.now())) + select { + case <-b.ctx.Done(): + timer.Stop() + return + case <-finished: + timer.Stop() + return + case <-timer.C: + case <-ticker.C: + timer.Stop() + lease, valid := a.currentLease(b.proof, a.now()) + if current, ok := a.Publishers.nativeCandidate(b.claim.ref, lease, a.now()); !valid || !ok || current != connection { + return + } + } + } +} + +// publisher probes may arrive before a replacement TLS carrier introduces its +// framed stream. drop those packets until the connector sends its first frame; +// inner QUIC retransmits without repeating an application admission. +type nativeReadyCarrier struct { + peer.Carrier + ready atomic.Bool + stop context.CancelFunc +} + +var _ peer.Carrier = (*nativeReadyCarrier)(nil) + +func (c *nativeReadyCarrier) Send(ctx context.Context, frame []byte) error { + if !c.ready.Load() { + return ctx.Err() + } + err := c.Carrier.Send(ctx, frame) + if err != nil { + c.stop() + } + return err +} + +func (c *nativeReadyCarrier) Receive(ctx context.Context) ([]byte, error) { + frame, err := c.Carrier.Receive(ctx) + if err == nil { + c.ready.Store(true) + } else { + c.stop() + } + return frame, err +} + +func (c *nativeReadyCarrier) Close() error { + c.stop() + return c.Carrier.Close() +} diff --git a/internal/relay/native_peer_handoff_test.go b/internal/relay/native_peer_handoff_test.go new file mode 100644 index 000000000..a24eea828 --- /dev/null +++ b/internal/relay/native_peer_handoff_test.go @@ -0,0 +1,229 @@ +package relay + +import ( + "bytes" + "context" + "errors" + "net/netip" + "testing" + "time" + + "github.com/tnldotdev/tnl/internal/transport/peer" + "github.com/tnldotdev/tnl/internal/transport/tunnel" + relayv1 "github.com/tnldotdev/tnl/pkg/api/control/private/relay/v1" + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +func TestNativePeerHandoffKeepsPublisherRegistrationAndReplay(t *testing.T) { + for _, publisherKind := range []string{"quic", "tls_yamux"} { + for _, firstKind := range []string{"quic", "tls_yamux"} { + t.Run(publisherKind+"/"+firstKind, func(t *testing.T) { + claim := nativeClaimFunc(func(_ context.Context, _ relayv1.RelayLease, source netip.Addr, proof tunnelv1.NativePairLegProof, _ []byte) (relayv1.NativePeerPairLegClaim, error) { + return nativeTestClaim(source, proof), nil + }) + acceptor, owner, hello := nativeAdmissionFixture(t, 4, claim) + publisherTransport, _, _ := nativeDeliveryPublisher(t, publisherKind, acceptor, owner, hello) + id, _ := tunnelv1.ParsePeerDeliveryIdentifier(nativeTestDeliveryID) + binding := peer.PairBinding{Delivery: tunnelv1.PeerDelivery{Pair: id, Generation: 1}, Key: [32]byte{1}, Kind: peer.NativePeer, Role: 1} + published := nativeTestPacketOwner(t, publisherTransport).register(t, binding, false) + binding.Role = 0 + kind := firstKind + previous := bytes.Clone(hello.NativePairLegProof.Nonce) + var original *nativePairBridge + var deadline time.Time + var retained *tunnel.Session + var retainedDone <-chan error + for attempt := range 3 { + transport, exited := nativePacketTransport(t, kind, acceptor) + proof := *hello.NativePairLegProof + proof.Nonce = bytes.Repeat([]byte{byte(attempt)}, 32) + message := hello + message.NativePairLegProof = &proof + session, _, err := tunnel.DialNative(t.Context(), transport, message, nativeTestDeliveryID) + if err != nil { + t.Fatal(err) + } + if attempt == 0 { + err = session.ActivateNativeLeg(t.Context(), "initial") + } else { + err = session.HandoffNativeLeg(t.Context(), "handoff", previous) + } + if err != nil { + t.Fatal("selected carrier could not install", err) + } + if retained != nil { + _ = relayAwait(t, retainedDone) + select { + case <-retained.Done(): + case <-time.After(time.Second): + t.Fatal("handoff did not retire the old live candidate") + } + retained = nil + } + owner.Publishers.mu.Lock() + for _, connection := range owner.Publishers.connections { + connection.packets.mu.Lock() + current := connection.packets.bridges[id] + if current == nil { + t.Fatal("selected carrier has no bridge") + } + current.mu.Lock() + if attempt == 0 { + original, deadline = current, current.deadline + } else if current != original || current.deadline.After(deadline) { + t.Error("handoff replaced registration or extended authority") + } + current.mu.Unlock() + connection.packets.mu.Unlock() + } + owner.Publishers.mu.Unlock() + connected := nativeTestPacketOwner(t, transport).register(t, binding, true) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + // replay a previous frame through a new outer session. the retained + // publisher registration must discard it before delivering the next. + if attempt > 0 { + if err := connected.carrier(t, ctx).Send(ctx, nativeDeliveryFrame(t, binding, uint64(attempt), byte(attempt))); err != nil { + t.Fatal(err) + } + } + frame := nativeDeliveryFrame(t, binding, uint64(attempt+1), byte(attempt+1)) + if err := connected.carrier(t, ctx).Send(ctx, frame); err != nil { + t.Fatal(err) + } + publishedCarrier := published.carrier(t, ctx) + if got, err := publishedCarrier.Receive(ctx); err != nil || !bytes.Equal(got, frame) { + t.Fatal("handoff lost delivery or publisher replay state", err) + } + reverse := binding + reverse.Role = 1 + frame = nativeDeliveryFrame(t, reverse, uint64(attempt+1), byte(attempt+2)) + if err := publishedCarrier.Send(ctx, frame); err != nil { + t.Fatal(err) + } + if got, err := connected.carrier(t, ctx).Receive(ctx); err != nil || !bytes.Equal(got, frame) { + t.Fatal("handoff lost reverse delivery", err) + } + cancel() + if attempt == 0 { + retained, retainedDone = session, exited + } else { + if attempt == 1 && kind == "tls_yamux" { + // a protocol failure on the packet stream must also retire + // its control owner so the next handoff can recover it. + ctx, finish := context.WithTimeout(t.Context(), time.Second) + if err := connected.carrier(t, ctx).Send(ctx, nativeDeliveryFrame(t, binding, uint64(attempt+1), byte(attempt+1))); err != nil { + t.Fatal(err) + } + select { + case <-session.Done(): + case <-ctx.Done(): + t.Fatal("failed packet carrier left its control owner alive") + } + finish() + } + _ = session.Close() + _ = relayAwait(t, exited) + } + previous = bytes.Clone(proof.Nonce) + kind = map[string]string{"quic": "tls_yamux", "tls_yamux": "quic"}[kind] + } + for _, failure := range []string{"consumed predecessor", "reused candidate", "expired bridge"} { + transport, exited := nativePacketTransport(t, firstKind, acceptor) + proof := *hello.NativePairLegProof + proof.Nonce = bytes.Repeat([]byte{9}, 32) + predecessor := previous + switch failure { + case "consumed predecessor": + predecessor = make([]byte, 32) + case "reused candidate": + proof.Nonce = make([]byte, 32) + case "expired bridge": + original.mu.Lock() + original.deadline = time.Now() + original.mu.Unlock() + } + message := hello + message.NativePairLegProof = &proof + session, _, err := tunnel.DialNative(t.Context(), transport, message, nativeTestDeliveryID) + if err != nil { + t.Fatal(err) + } + var rejected *tunnel.ProtocolError + if err := session.HandoffNativeLeg(t.Context(), "stale", predecessor); !errors.As(err, &rejected) || (rejected.Code != tunnelv1.Unauthenticated && rejected.Code != tunnelv1.Unavailable) { + t.Fatal("invalid handoff displaced current registration", failure, err) + } + _ = session.Close() + // outer close can overtake the queued error frame. the server + // must retain the exact denial even when the client sees ambiguity. + if err := relayAwait(t, exited); !errors.As(err, &rejected) || rejected.Code != tunnelv1.Unauthenticated { + t.Fatal("handoff lost its server-side rejection", failure, err) + } + original.mu.Lock() + unchanged := bytes.Equal(original.nonce, previous) && !original.deadline.After(deadline) + original.mu.Unlock() + if !unchanged { + t.Fatal("denied handoff changed the selected binding or deadline") + } + } + }) + } + } +} + +func TestNativePeerHandoffBoundsAndFencesRetainedAuthority(t *testing.T) { + for _, changed := range []string{"source", "key", "assignment", "expired", "canceled", "request canceled", "closed publisher", "candidate limit"} { + t.Run(changed, func(t *testing.T) { + claimClient := nativeClaimFunc(func(_ context.Context, _ relayv1.RelayLease, source netip.Addr, proof tunnelv1.NativePairLegProof, _ []byte) (relayv1.NativePeerPairLegClaim, error) { + return nativeTestClaim(source, proof), nil + }) + _, owner, hello := nativeAdmissionFixture(t, 2, claimClient) + proof := *hello.NativePairLegProof + claim, err := owner.claim(t.Context(), socketIdentityOnly{}, proof) + if err != nil { + t.Fatal(err) + } + life, stop := context.WithCancel(t.Context()) + defer stop() + id, _ := tunnelv1.ParsePeerDeliveryIdentifier(nativeTestDeliveryID) + oldNonce := bytes.Repeat([]byte{7}, 32) + deadline := time.Now().Add(time.Minute) + bridge := &nativePairBridge{ctx: life, cancel: stop, proof: proof, claim: claim, nonce: oldNonce, + deadline: deadline, used: map[[32]byte]struct{}{[32]byte(oldNonce): {}}} + packets := &nativePacketSession{ctx: t.Context(), bridges: map[[16]byte]*nativePairBridge{id: bridge}} + delivery := &nativePairDelivery{deadline: deadline, changed: make(chan struct{})} + request, cancel := context.WithCancel(t.Context()) + defer cancel() + want := tunnelv1.Unauthenticated + switch changed { + case "source": + claim.receipt.SourceIP = "127.0.0.2" + case "key": + claim.key[0]++ + case "assignment": + proof.AssignmentRevision++ + case "expired": + bridge.deadline = time.Now() + case "canceled": + stop() + case "request canceled": + cancel() + case "closed publisher": + packets.closed = true + case "candidate limit": + for n := byte(1); len(bridge.used) < maxNativeCarrierCandidates; n++ { + bridge.used[[32]byte{n}] = struct{}{} + } + want = tunnelv1.CapacityExceeded + } + selection := tunnelv1.NativeLegBinding{Nonce: proof.Nonce, PreviousNonce: oldNonce} + selected, err := packets.selectNativeBridge(request, owner, delivery, proof, claim, id, selection) + var rejected *tunnel.ProtocolError + if selected != nil || !errors.As(err, &rejected) || rejected.Code != want { + t.Fatal("invalid handoff reached carrier installation", changed, err) + } + if bridge.active != nil || !bytes.Equal(bridge.nonce, oldNonce) || bridge.deadline.After(deadline) { + t.Fatal("rejected handoff changed retained authority") + } + }) + } +} diff --git a/internal/relay/native_peer_renewal.go b/internal/relay/native_peer_renewal.go index c05f43d29..822ddca1a 100644 --- a/internal/relay/native_peer_renewal.go +++ b/internal/relay/native_peer_renewal.go @@ -69,6 +69,11 @@ func (d *nativePairDelivery) renewClaim(ctx context.Context, a *NativeConnectorA if d.closed || d.ctx.Err() != nil || !d.deadline.After(a.now()) || !next.receipt.ExpiresAt.After(a.now()) { return tunnelv1.NativePairLegAccepted{}, denied } + if d.bridge != nil { + if err := d.bridge.renew(d, next.receipt.ExpiresAt, a.now()); err != nil { + return tunnelv1.NativePairLegAccepted{}, err + } + } d.deadline = next.receipt.ExpiresAt close(d.changed) d.changed = make(chan struct{}) diff --git a/internal/transport/peer/replay_carrier.go b/internal/transport/peer/replay_carrier.go new file mode 100644 index 000000000..3dd6876b5 --- /dev/null +++ b/internal/transport/peer/replay_carrier.go @@ -0,0 +1,36 @@ +package peer + +import ( + "context" + "sync" +) + +// RetainCarrierReplay keeps one bounded receive history across replacements of +// the underlying carrier. that carrier must authenticate frames first. this +// filter prevents a replay on a new outer session from reaching a retained +// TLS/yamux stream, whose per-registration history would reject the stream. +func RetainCarrierReplay(carrier Carrier) Carrier { + return &replayCarrier{Carrier: carrier} +} + +type replayCarrier struct { + Carrier + mu sync.Mutex + replay replayWindow +} + +var _ Carrier = (*replayCarrier)(nil) + +func (c *replayCarrier) Receive(ctx context.Context) ([]byte, error) { + c.mu.Lock() + defer c.mu.Unlock() + for { + frame, err := c.Carrier.Receive(ctx) + if err != nil { + return nil, err + } + if c.replay.accept(frame) { + return frame, nil + } + } +} diff --git a/internal/transport/tunnel/native_activation.go b/internal/transport/tunnel/native_activation.go index f6e8abe02..a4a579ed9 100644 --- a/internal/transport/tunnel/native_activation.go +++ b/internal/transport/tunnel/native_activation.go @@ -13,6 +13,19 @@ import ( // candidates remain passive until their owner joins the losers and chooses one. // activation does not sign another proof, contact control, or extend a deadline. func (s *Session) ActivateNativeLeg(ctx context.Context, requestID string) (err error) { + return s.selectNativeLeg(ctx, requestID, nil) +} + +// HandoffNativeLeg replaces only the exact previously selected candidate. a +// lost acknowledgment consumes this attempt just like initial activation. +func (s *Session) HandoffNativeLeg(ctx context.Context, requestID string, previousNonce []byte) error { + if len(previousNonce) != 32 { + return &ProtocolError{Code: tunnelv1.InvalidMessage} + } + return s.selectNativeLeg(ctx, requestID, bytes.Clone(previousNonce)) +} + +func (s *Session) selectNativeLeg(ctx context.Context, requestID string, previousNonce []byte) (err error) { if s == nil || s.role != tunnelv1.NativeConnector || s.nativeLeg == nil { return &ProtocolError{Code: tunnelv1.Unauthenticated} } @@ -42,8 +55,12 @@ func (s *Session) ActivateNativeLeg(ctx context.Context, requestID string) (err // an ambiguous response cannot be retried on this control stream. the role // owner closes this candidate rather than risking a second installation. s.nativeActivated = true - binding := tunnelv1.NativeLegBinding{Nonce: bytes.Clone(s.nativeLeg.Nonce)} - if err := tunnelv1.WriteControl(s.control, tunnelv1.Message{Type: tunnelv1.NativeLegActivate, ProtocolVersion: tunnelv1.Version, + binding := tunnelv1.NativeLegBinding{Nonce: bytes.Clone(s.nativeLeg.Nonce), PreviousNonce: previousNonce} + kind, accepted := tunnelv1.NativeLegActivate, tunnelv1.NativeLegActivated + if len(previousNonce) != 0 { + kind, accepted = tunnelv1.NativeLegHandoff, tunnelv1.NativeLegHandedOff + } + if err := tunnelv1.WriteControl(s.control, tunnelv1.Message{Type: kind, ProtocolVersion: tunnelv1.Version, RequestID: requestID, Role: tunnelv1.NativeConnector, NativeLegBinding: &binding}); err != nil { return err } @@ -57,7 +74,7 @@ func (s *Session) ActivateNativeLeg(ctx context.Context, requestID string) (err if response.Type == tunnelv1.Error { return &ProtocolError{Code: response.Code} } - if response.Type != tunnelv1.NativeLegActivated || response.NativeLegBinding == nil || !response.NativeLegBinding.Matches(*s.nativeLeg) { + if response.Type != accepted || response.NativeLegBinding == nil || !response.NativeLegBinding.Matches(*s.nativeLeg) || !bytes.Equal(response.NativeLegBinding.PreviousNonce, previousNonce) { return &ProtocolError{Code: tunnelv1.InvalidMessage} } if cause := context.Cause(ctx); cause != nil { diff --git a/internal/transport/tunnel/native_activation_test.go b/internal/transport/tunnel/native_activation_test.go index cd8a5d6b8..da987c554 100644 --- a/internal/transport/tunnel/native_activation_test.go +++ b/internal/transport/tunnel/native_activation_test.go @@ -19,7 +19,7 @@ func TestNativeLegActivationSelectsOriginalHelloAfterPassiveRenewal(t *testing.T defer cancel() var activations, renewals atomic.Int32 done := tunnelWorker(t, func() { cancel(); _ = server.Close() }, func() error { - return server.HandleNativeLegControl(ctx, func(context.Context) error { + return server.HandleNativeLegControl(ctx, func(context.Context, tunnelv1.NativeLegBinding) error { activations.Add(1) return nil }, func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) { @@ -62,7 +62,7 @@ func TestNativeLegActivationRejectsForeignCandidate(t *testing.T) { defer cancel() var calls atomic.Int32 done := tunnelWorker(t, func() { cancel(); _ = server.Close() }, func() error { - return server.HandleNativeLegControl(ctx, func(context.Context) error { calls.Add(1); return nil }, + return server.HandleNativeLegControl(ctx, func(context.Context, tunnelv1.NativeLegBinding) error { calls.Add(1); return nil }, func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) { t.Error("selection renewed authority") return tunnelv1.NativePairLegAccepted{}, nil @@ -92,7 +92,7 @@ func TestNativeLegActivationKeepsCallbackClassificationCauseAndSafeWire(t *testi ctx, cancel := context.WithCancel(t.Context()) defer cancel() done := tunnelWorker(t, func() { cancel(); _ = server.Close() }, func() error { - return server.HandleNativeLegControl(ctx, func(context.Context) error { + return server.HandleNativeLegControl(ctx, func(context.Context, tunnelv1.NativeLegBinding) error { return errors.Join(classification, cause) }, func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error) { t.Error("selection renewed authority") diff --git a/internal/transport/tunnel/native_handoff_test.go b/internal/transport/tunnel/native_handoff_test.go new file mode 100644 index 000000000..1bb214116 --- /dev/null +++ b/internal/transport/tunnel/native_handoff_test.go @@ -0,0 +1,58 @@ +package tunnel + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/tnldotdev/tnl/pkg/protocol/tunnelv1" +) + +func TestNativeLegHandoffBindsBothCandidates(t *testing.T) { + for _, outcome := range []string{"accepted", "wrong predecessor", "wrong response kind"} { + t.Run(outcome, func(t *testing.T) { + client, server, proof, _ := nativeRenewalPair(t) + previous := bytes.Repeat([]byte{1}, 32) + done := tunnelWorker(t, func() { _ = server.Close() }, func() error { + request, err := tunnelv1.ReadControl(server.control) + if err != nil { + return err + } + if request.Type != tunnelv1.NativeLegHandoff || !request.NativeLegBinding.Matches(proof) || !bytes.Equal(request.NativeLegBinding.PreviousNonce, previous) { + t.Error("handoff lost its exact old and new candidate bindings") + } + response := tunnelv1.Message{Type: tunnelv1.NativeLegHandedOff, ProtocolVersion: tunnelv1.Version, + RequestID: request.RequestID, NativeLegBinding: request.NativeLegBinding} + if outcome == "wrong predecessor" { + response.NativeLegBinding.PreviousNonce = bytes.Repeat([]byte{2}, 32) + } + if outcome == "wrong response kind" { + response.Type = tunnelv1.NativeLegActivated + response.NativeLegBinding.PreviousNonce = nil + } + return tunnelv1.WriteControl(server.control, response) + }) + err := client.HandoffNativeLeg(t.Context(), "handoff", previous) + if outcome == "accepted" { + if err != nil { + t.Fatal(err) + } + } else { + var rejected *ProtocolError + if !errors.As(err, &rejected) || rejected.Code != tunnelv1.InvalidMessage { + t.Fatal("handoff trusted a mismatched acknowledgment", err) + } + } + if err := client.HandoffNativeLeg(context.Background(), "again", previous); err == nil { + t.Fatal("handoff was repeated") + } + if err := client.ActivateNativeLeg(context.Background(), "startup"); err == nil { + t.Fatal("handoff retried as startup activation") + } + if err := await(t, done); err != nil { + t.Fatal(err) + } + }) + } +} diff --git a/internal/transport/tunnel/native_renewal.go b/internal/transport/tunnel/native_renewal.go index e8bce7c4c..94ab627fe 100644 --- a/internal/transport/tunnel/native_renewal.go +++ b/internal/transport/tunnel/native_renewal.go @@ -52,7 +52,7 @@ func (s *Session) RenewNativeLeg(ctx context.Context, requestID string, proof tu // HandleNativeLegControl owns the accepted native control stream until it // stops. the callback must check current authority without reviving an expired // owner. packet stream/datagram readers stay with their existing role owner. -func (s *Session) HandleNativeLegControl(ctx context.Context, activate func(context.Context) error, authorize func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error)) (err error) { +func (s *Session) HandleNativeLegControl(ctx context.Context, activate func(context.Context, tunnelv1.NativeLegBinding) error, authorize func(context.Context, tunnelv1.NativePairLegProof) (tunnelv1.NativePairLegAccepted, error)) (err error) { if s == nil || s.role != tunnelv1.NativeConnector || s.nativeLeg == nil || authorize == nil { return &ProtocolError{Code: tunnelv1.Unauthenticated} } @@ -68,12 +68,12 @@ func (s *Session) HandleNativeLegControl(ctx context.Context, activate func(cont if err != nil { return err } - if request.Type == tunnelv1.NativeLegActivate { + if request.Type == tunnelv1.NativeLegActivate || request.Type == tunnelv1.NativeLegHandoff { var authErr error if activate == nil || s.nativeActivated || request.NativeLegBinding == nil || !request.NativeLegBinding.Matches(*s.nativeLeg) { authErr = &ProtocolError{Code: tunnelv1.Unauthenticated} } else { - authErr = activate(ctx) + authErr = activate(ctx, *request.NativeLegBinding) if authErr == nil { authErr = context.Cause(ctx) } @@ -88,7 +88,11 @@ func (s *Session) HandleNativeLegControl(ctx context.Context, activate func(cont return errors.Join(authErr, writeErr) } s.nativeActivated = true - if err := tunnelv1.WriteControl(s.control, tunnelv1.Message{Type: tunnelv1.NativeLegActivated, ProtocolVersion: tunnelv1.Version, RequestID: request.RequestID, NativeLegBinding: request.NativeLegBinding}); err != nil { + accepted := tunnelv1.NativeLegActivated + if request.Type == tunnelv1.NativeLegHandoff { + accepted = tunnelv1.NativeLegHandedOff + } + if err := tunnelv1.WriteControl(s.control, tunnelv1.Message{Type: accepted, ProtocolVersion: tunnelv1.Version, RequestID: request.RequestID, NativeLegBinding: request.NativeLegBinding}); err != nil { return err } continue diff --git a/pkg/protocol/tunnelv1/framing_test.go b/pkg/protocol/tunnelv1/framing_test.go index f28ca10a3..a46e58eb0 100644 --- a/pkg/protocol/tunnelv1/framing_test.go +++ b/pkg/protocol/tunnelv1/framing_test.go @@ -30,6 +30,8 @@ var frameReaders = []struct { {"control-native-renewed", MaxControlBytes, readControl}, {"control-native-activate", MaxControlBytes, readControl}, {"control-native-activated", MaxControlBytes, readControl}, + {"control-native-handoff", MaxControlBytes, readControl}, + {"control-native-handed-off", MaxControlBytes, readControl}, {"control-accepted", MaxControlBytes, readControl}, {"native-peer-visitor-entry", MaxHeaderBytes, readNativePeerVisitor}, {"visitor-stream-header", MaxHeaderBytes, readVisitor}, diff --git a/pkg/protocol/tunnelv1/golden_test.go b/pkg/protocol/tunnelv1/golden_test.go index 75a2a7059..bfa79ce00 100644 --- a/pkg/protocol/tunnelv1/golden_test.go +++ b/pkg/protocol/tunnelv1/golden_test.go @@ -35,6 +35,11 @@ func TestGoldenJSONAndFramedWireFixtures(t *testing.T) { assertGoldenFrame(t, "control-native-activate", nativeActivate, WriteControl, ReadControl) nativeActivate.Type, nativeActivate.Role = NativeLegActivated, "" assertGoldenFrame(t, "control-native-activated", nativeActivate, WriteControl, ReadControl) + nativeHandoff := Message{Type: NativeLegHandoff, ProtocolVersion: Version, RequestID: "handoff_1", Role: NativeConnector, + NativeLegBinding: &NativeLegBinding{Nonce: bytes.Repeat([]byte{1}, 32), PreviousNonce: make([]byte, 32)}} + assertGoldenFrame(t, "control-native-handoff", nativeHandoff, WriteControl, ReadControl) + nativeHandoff.Type, nativeHandoff.Role = NativeLegHandedOff, "" + assertGoldenFrame(t, "control-native-handed-off", nativeHandoff, WriteControl, ReadControl) nativeRenew := nativeHello nativeRenew.Type, nativeRenew.RequestID = NativeLegRenew, "renew_1" assertGoldenFrame(t, "control-native-renew", nativeRenew, WriteControl, ReadControl) diff --git a/pkg/protocol/tunnelv1/native_activation.go b/pkg/protocol/tunnelv1/native_activation.go index a894c1faf..0b29b1f87 100644 --- a/pkg/protocol/tunnelv1/native_activation.go +++ b/pkg/protocol/tunnelv1/native_activation.go @@ -8,13 +8,14 @@ import ( // NativeLegBinding selects one authenticated carrier candidate by the nonce in // its signed hello. it neither refreshes authority nor admits an application. type NativeLegBinding struct { - Nonce []byte `json:"nonce"` + Nonce []byte `json:"nonce"` + PreviousNonce []byte `json:"previous_nonce,omitempty"` } var ErrNativeLegBinding = errors.New("native carrier binding is invalid") func (b NativeLegBinding) Validate() error { - if len(b.Nonce) != 32 { + if len(b.Nonce) != 32 || (len(b.PreviousNonce) != 0 && (len(b.PreviousNonce) != 32 || bytes.Equal(b.Nonce, b.PreviousNonce))) { return ErrNativeLegBinding } return nil diff --git a/pkg/protocol/tunnelv1/native_activation_test.go b/pkg/protocol/tunnelv1/native_activation_test.go index c2a271563..9d16611f4 100644 --- a/pkg/protocol/tunnelv1/native_activation_test.go +++ b/pkg/protocol/tunnelv1/native_activation_test.go @@ -19,7 +19,7 @@ func TestNativeLegActivationClosedUnionAndExactNonce(t *testing.T) { if binding.Matches(*hello.NativePairLegProof) { t.Fatal("foreign hello nonce selected this candidate") } - for _, name := range []string{"control-native-activate", "control-native-activated"} { + for _, name := range []string{"control-native-activate", "control-native-activated", "control-native-handoff", "control-native-handed-off"} { var original map[string]json.RawMessage if err := json.Unmarshal(readFixture(t, name+".json"), &original); err != nil { t.Fatal(err) diff --git a/pkg/protocol/tunnelv1/native_handoff_test.go b/pkg/protocol/tunnelv1/native_handoff_test.go new file mode 100644 index 000000000..771ff042d --- /dev/null +++ b/pkg/protocol/tunnelv1/native_handoff_test.go @@ -0,0 +1,41 @@ +package tunnelv1 + +import ( + "bytes" + "encoding/json" + "testing" +) + +func TestNativeLegHandoffRequiresDistinctExactPredecessor(t *testing.T) { + for _, name := range []string{"control-native-handoff", "control-native-handed-off"} { + for _, value := range []string{`null`, `""`, `"AQ=="`, `"AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="`} { + var fields map[string]json.RawMessage + if err := json.Unmarshal(readFixture(t, name+".json"), &fields); err != nil { + t.Fatal(err) + } + var binding map[string]json.RawMessage + if err := json.Unmarshal(fields["native_leg_binding"], &binding); err != nil { + t.Fatal(err) + } + binding["previous_nonce"] = json.RawMessage(value) + fields["native_leg_binding"], _ = json.Marshal(binding) + payload, _ := json.Marshal(fields) + if _, err := ReadControl(frame(string(payload))); err == nil { + t.Fatal("invalid predecessor accepted", name, value) + } + } + } + for _, kind := range []MessageType{NativeLegActivate, NativeLegActivated, NativeLegHandoff, NativeLegHandedOff} { + binding := NativeLegBinding{Nonce: make([]byte, 32)} + if kind == NativeLegActivate || kind == NativeLegActivated { + binding.PreviousNonce = bytes.Repeat([]byte{1}, 32) + } + role := NativeConnector + if kind == NativeLegActivated || kind == NativeLegHandedOff { + role = "" + } + if err := (Message{Type: kind, ProtocolVersion: Version, RequestID: "select", Role: role, NativeLegBinding: &binding}).Validate(); err == nil { + t.Fatal("startup and handoff accepted interchangeable bindings", kind) + } + } +} diff --git a/pkg/protocol/tunnelv1/protocol.go b/pkg/protocol/tunnelv1/protocol.go index d5651cc43..8270675d0 100644 --- a/pkg/protocol/tunnelv1/protocol.go +++ b/pkg/protocol/tunnelv1/protocol.go @@ -40,6 +40,8 @@ const ( NativeLegRenewed MessageType = "native_leg_renewed" NativeLegActivate MessageType = "native_leg_activate" NativeLegActivated MessageType = "native_leg_activated" + NativeLegHandoff MessageType = "native_leg_handoff" + NativeLegHandedOff MessageType = "native_leg_handed_off" ) type PeerRole string @@ -204,12 +206,12 @@ func ReadControl(reader io.Reader) (Message, error) { if err := json.Unmarshal(fields["type"], &kind); err != nil { return Message{}, err } - if kind == NativeLegActivate || kind == NativeLegActivated { + if kind == NativeLegActivate || kind == NativeLegActivated || kind == NativeLegHandoff || kind == NativeLegHandedOff { for name := range fields { switch name { case "type", "protocol_version", "request_id", "native_leg_binding": case "role": - if kind != NativeLegActivate { + if kind != NativeLegActivate && kind != NativeLegHandoff { return Message{}, ErrNativeLegBinding } default: @@ -232,7 +234,15 @@ func ReadControl(reader io.Reader) (Message, error) { } } if binding, present := fields["native_leg_binding"]; present { - if err := json.Unmarshal(fields["type"], &kind); err != nil || (kind != NativeLegActivate && kind != NativeLegActivated) || bytes.Equal(bytes.TrimSpace(binding), []byte("null")) { + if err := json.Unmarshal(fields["type"], &kind); err != nil || (kind != NativeLegActivate && kind != NativeLegActivated && kind != NativeLegHandoff && kind != NativeLegHandedOff) || bytes.Equal(bytes.TrimSpace(binding), []byte("null")) { + return Message{}, ErrNativeLegBinding + } + var selection map[string]json.RawMessage + if err := json.Unmarshal(binding, &selection); err != nil { + return Message{}, ErrNativeLegBinding + } + _, previous := selection["previous_nonce"] + if previous != (kind == NativeLegHandoff || kind == NativeLegHandedOff) { return Message{}, ErrNativeLegBinding } } @@ -317,18 +327,21 @@ func (m Message) Validate() error { if m.RequestID != "" && !validIdentifier(m.RequestID) { return errors.New("tunnelv1: invalid request ID") } - if m.NativeLegBinding != nil && m.Type != NativeLegActivate && m.Type != NativeLegActivated { + if m.NativeLegBinding != nil && m.Type != NativeLegActivate && m.Type != NativeLegActivated && m.Type != NativeLegHandoff && m.Type != NativeLegHandedOff { return ErrNativeLegBinding } switch m.Type { - case NativeLegActivate, NativeLegActivated: + case NativeLegActivate, NativeLegActivated, NativeLegHandoff, NativeLegHandedOff: expectedRole := NativeConnector - if m.Type == NativeLegActivated { + if m.Type == NativeLegActivated || m.Type == NativeLegHandedOff { expectedRole = "" } if !validIdentifier(m.RequestID) || m.Role != expectedRole || m.Credential != "" || m.PublisherConnection != nil || m.ServiceConnection != nil || m.NativePairLegProof != nil || m.NativePairLegAccepted != nil || m.Code != "" || m.NativeLegBinding == nil || m.NativeLegBinding.Validate() != nil { return ErrNativeLegBinding } + if (len(m.NativeLegBinding.PreviousNonce) == 32) != (m.Type == NativeLegHandoff || m.Type == NativeLegHandedOff) { + return ErrNativeLegBinding + } case Hello: if m.NativePairLegAccepted != nil { return errors.New("tunnelv1: native receipt on hello") diff --git a/pkg/protocol/tunnelv1/testdata/control-native-handed-off.frame.hex b/pkg/protocol/tunnelv1/testdata/control-native-handed-off.frame.hex new file mode 100644 index 000000000..ad72cdbd5 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/control-native-handed-off.frame.hex @@ -0,0 +1 @@ +000000dc7b2274797065223a226e61746976655f6c65675f68616e6465645f6f6666222c2270726f746f636f6c5f76657273696f6e223a312c22726571756573745f6964223a2268616e646f66665f31222c226e61746976655f6c65675f62696e64696e67223a7b226e6f6e6365223a22415145424151454241514542415145424151454241514542415145424151454241514542415145424151453d222c2270726576696f75735f6e6f6e6365223a22414141414141414141414141414141414141414141414141414141414141414141414141414141414141413d227d7d diff --git a/pkg/protocol/tunnelv1/testdata/control-native-handed-off.json b/pkg/protocol/tunnelv1/testdata/control-native-handed-off.json new file mode 100644 index 000000000..d4170d146 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/control-native-handed-off.json @@ -0,0 +1,9 @@ +{ + "type": "native_leg_handed_off", + "protocol_version": 1, + "request_id": "handoff_1", + "native_leg_binding": { + "nonce": "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=", + "previous_nonce": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" + } +} diff --git a/pkg/protocol/tunnelv1/testdata/control-native-handoff.frame.hex b/pkg/protocol/tunnelv1/testdata/control-native-handoff.frame.hex new file mode 100644 index 000000000..5714af5d8 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/control-native-handoff.frame.hex @@ -0,0 +1 @@ +000000f37b2274797065223a226e61746976655f6c65675f68616e646f6666222c2270726f746f636f6c5f76657273696f6e223a312c22726571756573745f6964223a2268616e646f66665f31222c22726f6c65223a226e61746976655f636f6e6e6563746f72222c226e61746976655f6c65675f62696e64696e67223a7b226e6f6e6365223a22415145424151454241514542415145424151454241514542415145424151454241514542415145424151453d222c2270726576696f75735f6e6f6e6365223a22414141414141414141414141414141414141414141414141414141414141414141414141414141414141413d227d7d diff --git a/pkg/protocol/tunnelv1/testdata/control-native-handoff.json b/pkg/protocol/tunnelv1/testdata/control-native-handoff.json new file mode 100644 index 000000000..949fabd71 --- /dev/null +++ b/pkg/protocol/tunnelv1/testdata/control-native-handoff.json @@ -0,0 +1,10 @@ +{ + "type": "native_leg_handoff", + "protocol_version": 1, + "request_id": "handoff_1", + "role": "native_connector", + "native_leg_binding": { + "nonce": "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=", + "previous_nonce": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" + } +}