From 98f64ea13d0693261d87143eec29ec88562c008b Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Wed, 7 Oct 2026 09:28:32 +0000 Subject: [PATCH] feat(tun): drop p2p stream on refused registration, bound first-inbound wait MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hub that refuses a spoke registration (bad passphrase, allowlist, self-loop) used to stay silent and keep the stream: the spoke sat 'running' with zero traffic and nothing to look at. Return ErrRegistrationRefused instead so the stream is torn down — the spoke redials and re-registers, which also recovers on its own once the hub is fixed. Malformed frames and bare echoes still return nil and leave the stream up, so transport noise cannot turn into a reconnect storm. Without a keepalive period the client reader had no deadline at all and waited forever on a silent link. Bound it with a 30s first-inbound one-shot (firstInboundTimeout, cleared by any inbound byte); the keepalive>0 path is untouched. --- handler/tun/client.go | 29 +++++++++ handler/tun/client_test.go | 75 ++++++++++++++++++++++ handler/tun/compat_test.go | 12 ++-- handler/tun/p2p.go | 28 ++++++++- handler/tun/p2p_test.go | 124 ++++++++++++++++++------------------- 5 files changed, 200 insertions(+), 68 deletions(-) diff --git a/handler/tun/client.go b/handler/tun/client.go index 05b393c..cddbf42 100644 --- a/handler/tun/client.go +++ b/handler/tun/client.go @@ -27,6 +27,15 @@ var ( magicHeader = []byte("GOST") ) +// firstInboundTimeout bounds the wait for the first inbound byte when no +// keepalive period is configured. A hub that never answers — not started, +// or silently refusing — would otherwise leave the spoke "running" forever +// on a link that will never deliver. Expiring makes the caller redial and +// re-register, which is also how a fixed hub recovers on its own. Any +// inbound byte clears it: from then on the link has proven itself alive. +// A variable, not a constant, so tests can shrink the wait. +var firstInboundTimeout = 30 * time.Second + func (h *tunHandler) handleClient(ctx context.Context, conn net.Conn, network string, raddr string, config *tun_util.Config, log logger.Logger) error { var ips []net.IP for _, net := range config.Net { @@ -181,6 +190,20 @@ func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, con go func() { var b [MaxMessageSize]byte + // The wait for the first inbound byte is bounded even without a + // keepalive period (the tun entrypoint configures none): silence + // means the far side never answered the handshake, and waiting + // forever is exactly the silent-"running" this guards against. With + // a period the window matches the handshake's own 3×period; without + // one the fixed one-shot applies until the first inbound clears it. + // (keepalive() sets the same 3×period deadline for its path; the + // two agree, and this one is what the read loop below relies on.) + oneShot := h.md.keepAlivePeriod <= 0 + if oneShot { + conn.SetReadDeadline(time.Now().Add(firstInboundTimeout)) + } else { + conn.SetReadDeadline(time.Now().Add(h.md.keepAlivePeriod * 3)) + } for { select { case <-c.Done(): @@ -201,6 +224,12 @@ func (h *tunHandler) transportClient(ctx context.Context, tun io.ReadWriter, con // a healthy tunnel after 3×period. if h.md.keepAlivePeriod > 0 { conn.SetReadDeadline(time.Now().Add(h.md.keepAlivePeriod * 3)) + } else if oneShot { + // The one-shot has served: the link proved itself alive, + // so clear it and read without a deadline from here on, + // exactly as before this change. + oneShot = false + conn.SetReadDeadline(time.Time{}) } // A registration handshake or keepalive echo is protocol, not a diff --git a/handler/tun/client_test.go b/handler/tun/client_test.go index b34bdc8..35b5dbd 100644 --- a/handler/tun/client_test.go +++ b/handler/tun/client_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/binary" + "errors" "net" "testing" "time" @@ -87,3 +88,77 @@ func TestTransportClientDropsKeepaliveFrames(t *testing.T) { <-errCh // client loop exits on the closed link } + +// Without a keepalive period the client sets no read deadline at all, so a +// hub that never answers — not started, or silently refusing — leaves the +// spoke "running" forever on a link that will never deliver. The transport +// must give up waiting for the first inbound byte so the caller redials and +// re-registers; any inbound byte clears the one-shot. +func TestNoKeepaliveClientTimesOutWaitingForFirstInbound(t *testing.T) { + old := firstInboundTimeout + firstInboundTimeout = 200 * time.Millisecond + defer func() { firstInboundTimeout = old }() + + device, kernelSide := net.Pipe() + defer device.Close() + defer kernelSide.Close() + ccA, ccB := net.Pipe() + defer ccB.Close() // the link stays open: silence, not closure, must end it + + h := newTestHandler(nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }() + + select { + case err := <-errCh: + var nerr net.Error + if !errors.As(err, &nerr) || !nerr.Timeout() { + t.Fatalf("transportClient = %v, want a read timeout", err) + } + case <-time.After(15 * time.Second): + t.Fatal("transportClient waited forever on a silent link") + } + // Note: transportClient itself takes ~5s more to return — collectFirstError + // waits out the device reader, which blocks on an idle device with no + // context-aware read. That wait is pre-existing; the timeout above is what + // proves the silent link no longer waits forever. +} + +// The mirror: one inbound byte — even protocol chatter, which never reaches +// the device — proves the link alive and clears the one-shot, so a healthy +// but quiet tunnel is never redialed. +func TestNoKeepaliveClientFirstInboundClearsTimeout(t *testing.T) { + old := firstInboundTimeout + firstInboundTimeout = 200 * time.Millisecond + defer func() { firstInboundTimeout = old }() + + device, kernelSide := net.Pipe() + defer device.Close() + defer kernelSide.Close() + ccA, ccB := net.Pipe() + + h := newTestHandler(nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- h.transportClient(ctx, device, ccA, xlogger.Nop()) }() + + if _, err := ccB.Write(bareEcho()); err != nil { + t.Fatal(err) + } + + // Three windows of silence past the one-shot: a cleared deadline stays + // cleared, and the transport must not report anything. + time.Sleep(3 * firstInboundTimeout) + select { + case err := <-errCh: + t.Fatalf("transportClient reported %v on a healthy quiet link", err) + default: + } + + ccB.Close() // unblock the reader; the test proved what it had to +} diff --git a/handler/tun/compat_test.go b/handler/tun/compat_test.go index 954e0ef..78ee3e8 100644 --- a/handler/tun/compat_test.go +++ b/handler/tun/compat_test.go @@ -168,11 +168,13 @@ func compatCases() []struct { t.Fatalf("a refused registration was answered with % x", reply) } - // Proof the frame was otherwise acceptable: the same registration for a - // peer address registers. Without it the refusal above is - // indistinguishable from some other check refusing it, and the guard - // would go untested. - hub.register(spoke, keepAliveFrame("secret", compatPeerA)) + // Proof the frame was otherwise acceptable: a fresh spoke registering + // a peer address registers. Fresh because a refusal may drop the + // link — for a real spoke that is the redial — so reusing the + // refused link would be driving a redial that never happened. + // Without the control the refusal above is indistinguishable from + // some other check refusing it, and the guard would go untested. + hub.register(hub.spoke(t), keepAliveFrame("secret", compatPeerA)) if _, ok := hub.resolve(compatPeerA); !ok { t.Fatal("the control registration did not take, so the test proves nothing") } diff --git a/handler/tun/p2p.go b/handler/tun/p2p.go index 7373090..32be832 100644 --- a/handler/tun/p2p.go +++ b/handler/tun/p2p.go @@ -22,6 +22,17 @@ import ( // packet is dropped, the hub keeps reading the device. var ErrNoRoute = errors.New("tun: no route") +// ErrRegistrationRefused reports a spoke registration the hub's policy +// rejected — the authorizer, the passphrase, or the self-loop guard. It is +// returned — never echoed — so the stream is torn down: the spoke redials +// and re-registers (a fixed hub recovers on its own) instead of sitting +// "running" with zero traffic, and the failure surfaces in its counters. +// Nothing is answered, so a refused peer learns nothing except that the +// stream died. Malformed frames and bare echoes are not this: they never +// reach policy, and killing a stream over transport noise would turn one +// stray datagram into a reconnect storm. +var ErrRegistrationRefused = errors.New("tun: registration refused") + // peerStream is the hub's half of one inbound stream from one peer: the peer's // name, the transport's write side, and the serialization that makes writing // one packet at a time true. @@ -342,12 +353,27 @@ func (h *p2pHub) fromSpoke(s *peerStream, pkt []byte) error { // timeout, with nothing to look at in any log. func (h *p2pHub) answerKeepalive(s *peerStream, pkt []byte) error { if _, ok := h.router.table.onKeepalive(s.ctx, pkt, s.key, h.ownNets); !ok { - return nil + if !isRegistrationFrame(pkt) { + // Transport noise — a bare echo, a fragment — that never + // reached policy. Drop it and leave the stream up. + return nil + } + return ErrRegistrationRefused } return s.write(keepAliveReply(s.key)) } +// isRegistrationFrame reports whether pkt is shaped like a registration — +// the parse gates onKeepalive applies before any policy — so the hub can tell +// a refused claim (drop the stream) from transport noise (drop the frame). +func isRegistrationFrame(pkt []byte) bool { + if len(pkt) <= keepAliveHeaderLength || !bytes.Equal(pkt[:4], magicHeader) { + return false + } + return len(pkt[keepAliveHeaderLength:])%net.IPv6len == 0 +} + // peerGone tears a stream down: close it, and reclaim its routes only if it is // still the stream holding its key. A peer whose reconnect already installed a // replacement keeps both its routes and its new stream — see withdraw. diff --git a/handler/tun/p2p_test.go b/handler/tun/p2p_test.go index 2b748aa..5f5cf0e 100644 --- a/handler/tun/p2p_test.go +++ b/handler/tun/p2p_test.go @@ -434,31 +434,6 @@ func TestP2PKeepaliveIsAnsweredAndRegisters(t *testing.T) { } } -// A refused registration is not answered: echoing a frame the auther rejected -// would tell the peer it is registered. -func TestP2PRefusedKeepaliveIsNotAnswered(t *testing.T) { - _, peerEnd := newDatagramPipe(4) - - table := newPeerTable(newMultiAuther("secret", "10.10.0.3"), 0, "tun-service", nil) - hub := newP2PHub(nopDevice{}, newPeerRouter(table), nil, nil) - s := newPeerStream(context.Background(), "peer-key", peerEnd) - hub.router.install(s) - - frame := keepAliveFrame("wrong", net.ParseIP("10.10.0.3")) - if err := hub.fromSpoke(s, frame); err != nil { - t.Fatalf("fromSpoke: %v", err) - } - - select { - case pkt := <-peerEnd.queue: - t.Fatalf("a refused registration was answered with % x", pkt) - default: - } - if _, ok := table.lookup(net.ParseIP("10.10.0.3")); ok { - t.Fatal("a refused registration left a route behind") - } -} - // newP2PAuthorizerHub builds the hub a deployment gets — NewP2PHandler over the // tun conn the listener produced — and returns it as the compat suite drives one, // so a case writes a registration on a real stream and reads the answer off the @@ -529,43 +504,6 @@ func TestP2PHandleRegistersAnAuthorizedClaim(t *testing.T) { } } -// End to end through the handler: a claim the authorizer refuses registers -// nothing and is not answered. -// -// The route is the load-bearing half of the assertion. A hub that merely stayed -// quiet would pass a reply-only check while still handing its traffic to a peer -// that has no right to the address, so the refusal is asserted where it is -// observable — in the table a packet's destination is resolved against. -func TestP2PHandleDoesNotRegisterARefusedClaim(t *testing.T) { - assigned := net.ParseIP("10.10.0.3") - neighbour := net.ParseIP("10.10.0.4") - hub := newP2PAuthorizerHub(t, newAssigningAuthorizer(map[string][]net.IP{ - "peer-key-1": {assigned}, - })) - - s := hub.spoke(t) - _, reply := hub.register(s, keepAliveFrame("secret", neighbour)) - - if name, ok := hub.resolve(neighbour); ok { - t.Fatalf("a refused claim left the route %q behind for %s", name, neighbour) - } - if reply != nil { - t.Fatalf("a refused registration was answered with % x", reply) - } - - // Proof the frame was otherwise acceptable: the same peer claiming the one - // address it was assigned registers. Without it the refusal above would be - // indistinguishable from a hub that refuses everything, and the authorizer - // would go untested. - _, reply = hub.register(s, keepAliveFrame("secret", assigned)) - if !isKeepaliveFrame(reply) { - t.Fatalf("the control registration was not answered: % x", reply) - } - if name, ok := hub.resolve(assigned); !ok || name != s.name { - t.Fatalf("the control registration did not take: lookup = %q, %v, want %q", name, ok, s.name) - } -} - // An unroutable packet is warned about *by destination*. A spoke configured with // keepalive:0 never registers on a p2p link — the handshake is gated on // network == "udp" and a p2p link is "ip" — so without the address in the @@ -952,3 +890,65 @@ var ( _ io.ReadWriter = (*writeRecorder)(nil) _ io.WriteCloser = discard{} ) + +// A refused registration is not answered *and* drops the stream: staying +// quiet would leave the spoke "running" with zero traffic and nothing to +// look at, while closing makes the spoke redial — which re-registers, so a +// fixed hub recovers on its own — and surfaces the failure in its counters. +func TestP2PRefusedKeepaliveIsRefusedAndDropped(t *testing.T) { + _, peerEnd := newDatagramPipe(4) + + table := newPeerTable(newMultiAuther("secret", "10.10.0.3"), 0, "tun-service", nil) + hub := newP2PHub(nopDevice{}, newPeerRouter(table), nil, nil) + s := newPeerStream(context.Background(), "peer-key", peerEnd) + hub.router.install(s) + + frame := keepAliveFrame("wrong", net.ParseIP("10.10.0.3")) + err := hub.fromSpoke(s, frame) + if !errors.Is(err, ErrRegistrationRefused) { + t.Fatalf("fromSpoke(refused keepalive) = %v, want %v", err, ErrRegistrationRefused) + } + + select { + case pkt := <-peerEnd.queue: + t.Fatalf("a refused registration was answered with % x", pkt) + default: + } + if _, ok := table.lookup(net.ParseIP("10.10.0.3")); ok { + t.Fatal("a refused registration left a route behind") + } +} + +// End to end through the handler: a refused claim tears the stream down, and +// a reconnect with the authorized claim registers — the redial path a real +// spoke takes after the drop, which is how a fixed hub recovers on its own. +func TestP2PRefusedClaimDropsStreamAndReconnectRecovers(t *testing.T) { + assigned := net.ParseIP("10.10.0.3") + neighbour := net.ParseIP("10.10.0.4") + hub := newP2PAuthorizerHub(t, newAssigningAuthorizer(map[string][]net.IP{ + "peer-key-1": {assigned}, + })) + + s := hub.spoke(t) + _, reply := hub.register(s, keepAliveFrame("secret", neighbour)) + if reply != nil { + t.Fatalf("a refused registration was answered with % x", reply) + } + if name, ok := hub.resolve(neighbour); ok { + t.Fatalf("a refused claim left the route %q behind for %s", name, neighbour) + } + select { + case <-s.conn.done: + case <-time.After(compatReadTimeout): + t.Fatal("the refused stream stayed open: the spoke would sit silent") + } + + s2 := hub.reconnect(s, t) + _, reply = hub.register(s2, keepAliveFrame("secret", assigned)) + if !isKeepaliveFrame(reply) { + t.Fatalf("the reconnect registration was not answered: % x", reply) + } + if name, ok := hub.resolve(assigned); !ok || name != s2.name { + t.Fatalf("the reconnect registration did not take: lookup = %q, %v, want %q", name, ok, s2.name) + } +}