From ffe7158053f3cf22a816a9a30b5da77c86bb2560 Mon Sep 17 00:00:00 2001 From: Derek den Haas Date: Sun, 9 Aug 2026 11:07:33 +0200 Subject: [PATCH] feat(tcp): implement window scaling (RFC 7323) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The window-scale option was parsed for length validation but never applied: the effective receive window was capped at the 16-bit field, which caps throughput at 65535/RTT (about 4MB/s on a 15ms path) no matter how large the receive buffer is. Worse, a receive buffer over 64KiB silently WRAPPED the advertised window on the SYN (uint16 truncation): a 128KiB buffer went on the wire as a near-zero window. Design: scaling lives purely at the wire seam in Handler. The ControlBlock always holds real octet counts; conversion happens on frame read (peer windows shifted up by the peer's offer, never on SYN segments) and frame write (wireWnd: our shift down, SYN never scaled, saturation instead of wrap when the value still does not fit). The local shift derives from the receive buffer size in SetBuffers; every active SYN offers it (a zero shift still lets the peer scale, RFC 7323 §2.5) and a SYN-ACK echoes it only when the peer's SYN carried the option. The ControlBlock's three 2**16 window caps move to the scaled maximum (65535<<14). Tests: on-wire negotiation with asymmetric buffers (shift values, unscaled saturated SYN windows, first scaled advertisement, peer scaling back up); a transfer proving more than 64KiB genuinely in flight without a single ACK, received intact; wire-safety corners (no echo without an offer, saturation not wrap, so the pre-existing 128KiB SYN wrap bug stays pinned). Fuzzers clean: 9.1M TCB execs, 4.9M full-stack HTTP execs, 7.2M TCB actions. --- tcp/control.go | 7 +- tcp/definitions.go | 4 +- tcp/handler.go | 84 ++++++++++++++-- tcp/seqsbug_test.go | 2 +- tcp/wscale_test.go | 237 ++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 318 insertions(+), 16 deletions(-) create mode 100644 tcp/wscale_test.go diff --git a/tcp/control.go b/tcp/control.go index ba258ab..ec9c07b 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -3,7 +3,6 @@ package tcp import ( "io" "log/slog" - "math" "net" "github.com/soypat/lneto/internal" @@ -230,7 +229,7 @@ func (tcb *ControlBlock) Open(iss Value, wnd Size) (err error) { switch { case tcb._state != StateClosed && tcb._state != StateTimeWait: err = errNeedClosedTCBToOpen - case wnd > math.MaxUint16: + case wnd > maxWindow: err = errWindowTooLarge } if err != nil { @@ -535,7 +534,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { switch { case tcb._state == StateClosed && !isFirst: err = io.ErrClosedPipe - case seg.WND > math.MaxUint16: + case seg.WND > maxWindow: err = errWindowTooLarge case hasAck && seg.ACK != tcb.rcv.NXT: err = errAckNotNext @@ -578,7 +577,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { zeroWindowOK := tcb.rcv.WND == 0 && seg.DATALEN == 0 && seg.SEQ == tcb.rcv.NXT // See section 3.4 of RFC 9293 for more on these checks. switch { - case seg.WND > math.MaxUint16: + case seg.WND > maxWindow: err = errWindowOverflow case tcb._state == StateClosed: err = io.ErrClosedPipe diff --git a/tcp/definitions.go b/tcp/definitions.go index 42263ef..5a0ff7d 100644 --- a/tcp/definitions.go +++ b/tcp/definitions.go @@ -14,7 +14,7 @@ import ( var ( errDropSegment error = lneto.ErrPacketDrop - errWindowTooLarge = errors.New("invalid window size > 2**16") + errWindowTooLarge = errors.New("invalid window size > max scaled window") errBufferTooSmall error = lneto.ErrShortBuffer errNeedClosedTCBToOpen = errors.New("need closed TCB to call open") @@ -25,7 +25,7 @@ var ( errBadSegack = errors.New("seqs:bad segack") errFinwaitExpectedACK = errors.New("seqs:finwait1 expected ACK") - errWindowOverflow = newRejectErr("wnd > 2**16") + errWindowOverflow = newRejectErr("wnd > max scaled window") errSeqNotInWindow = newRejectErr("seq not in snd/rcv.wnd") errZeroWindow = newRejectErr("zero window") errLastNotInWindow = newRejectErr("last not in snd/rcv.wnd") diff --git a/tcp/handler.go b/tcp/handler.go index 09eca89..9ab069d 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -29,6 +29,14 @@ type Handler struct { // Read and Write calls belong to the current connection. optcodec OptionCodec + // Window scaling (RFC 7323 §2). wndShiftLocal is derived from the receive + // buffer in [Handler.SetBuffers], wndShiftPeer learned from the peer's offer. + // Scaling lives at the wire seam only: the ControlBlock always holds real + // octet counts, converted on frame read ([Handler.Recv]) and write + // ([Handler.wireWnd]). + wndShiftLocal uint8 + wndShiftPeer uint8 + peerOfferedWS bool // reasm tracks out-of-order segments staged in bufRx's free region. Always // enabled once buffers are set (see [Handler.SetBuffers]). reasm reassembly @@ -68,6 +76,7 @@ func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { h.bufRx.Buf = rxbuf } h.scb.SetRecvWindow(Size(h.bufRx.Size())) + h.wndShiftLocal = wndShiftFor(h.bufRx.Size()) h.bufRx.Reset() h.reasm.reset(maxReasmSegments) return h.bufTx.ResetOrReuse(txbuf, packets, 0) @@ -147,9 +156,10 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { closing: false, shutdownRx: false, // Persist configuration across reopen: - validator: h.validator, - policy: h.policy, - logger: h.logger, + validator: h.validator, + policy: h.policy, + logger: h.logger, + wndShiftLocal: h.wndShiftLocal, // derived from buffers, which persist too // persist memory across repoen: bufTx: h.bufTx, bufRx: h.bufRx, @@ -189,6 +199,12 @@ func (h *Handler) Recv(incomingPacket []byte) error { } payload := tfrm.Payload() segIncoming := tfrm.Segment(len(payload)) + if h.peerOfferedWS && !segIncoming.Flags.HasAny(FlagSYN) { + // Peer windows arrive scaled once both sides offered scaling, but never on + // SYN segments (RFC 7323 §2.2). Restore real octets before the + // ControlBlock sees them. + segIncoming.WND <<= h.wndShiftPeer + } if h.scb.IncomingIsKeepalive(segIncoming) { h.info("tcp.Handler:rx-keepalive", slog.Uint64("port", uint64(h.localPort))) return nil @@ -266,6 +282,11 @@ func (h *Handler) Recv(incomingPacket []byte) error { h.scb.snd.MSS = Size(mss) } } + if kind == OptWindowScale && len(data) == 1 { + // RFC 7323 §2.3: a shift above 14 is clamped, not rejected. + h.peerOfferedWS = true + h.wndShiftPeer = min(data[0], maxWndShift) + } return nil }) if h.remotePort == 0 { @@ -423,8 +444,7 @@ func (h *Handler) Send(b []byte) (int, error) { if awaitingSyn || requeueControl && h.scb.State() == StateSynSent { // Handling init syn segment. segment = ClientSynSegment(h.bufTx.iss, Size(h.bufRx.Size())) - h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) - offset++ + offset += h.putSynOptions(b[sizeHeaderTCP:], mss, false) if requeueControl { h.info("tcp.Handler:requeue-syn", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort))) } @@ -435,8 +455,7 @@ func (h *Handler) Send(b []byte) (int, error) { WND: Size(h.bufRx.Free()), Flags: synack, } - h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) - offset++ + offset += h.putSynOptions(b[sizeHeaderTCP:], mss, true) h.info("tcp.Handler:requeue-synack", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort))) } else if requeueControl { h.requeueControl = false @@ -454,8 +473,7 @@ func (h *Handler) Send(b []byte) (int, error) { // No pending control segment or data to send. Yield. return 0, nil } else if segment.Flags == synack { - h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss) - offset++ + offset += h.putSynOptions(b[sizeHeaderTCP:], mss, true) } else if segment.DATALEN > 0 { n, err := h.bufTx.MakePacket(b[optHead:optHead+int(segment.DATALEN)], segment.SEQ) if err != nil { @@ -477,6 +495,7 @@ func (h *Handler) Send(b []byte) (int, error) { h.requeueControl = false tfrm.SetSourcePort(h.localPort) tfrm.SetDestinationPort(h.remotePort) + segment.WND = h.wireWnd(segment) // wire representation only; scb keeps real octets tfrm.SetSegment(segment, offset) tfrm.SetUrgentPtr(0) datalen := int(offset)*4 + int(segment.DATALEN) @@ -623,6 +642,53 @@ func (h *Handler) recvWindow() Size { return 0 } +// maxWndShift is the RFC 7323 §2.2/§2.3 cap on the window-scale shift count and +// maxWindow the largest window it permits: the 16-bit wire field at that shift. +const ( + maxWndShift = 14 + maxWindow = 0xFFFF << maxWndShift +) + +// wndShiftFor returns the smallest window-scale shift with which a receive +// buffer of bufSize octets can be advertised in the 16-bit window field. +func wndShiftFor(bufSize int) (shift uint8) { + for shift < maxWndShift && bufSize>>shift > 0xFFFF { + shift++ + } + return shift +} + +// putSynOptions writes the option block shared by SYN and SYN-ACK segments: +// MSS always, then the NOP-padded window-scale offer. An active SYN always +// offers scaling, since a zero shift still lets the peer scale its own window +// (RFC 7323 §2.5). A SYN-ACK echoes the offer only when the peer's SYN carried +// it (§2.2). Returns the number of 32-bit header words written. +func (h *Handler) putSynOptions(b []byte, mss uint16, isSynack bool) uint8 { + h.optcodec.PutOption16(b, OptMaxSegmentSize, mss) + words := uint8(1) + if (!isSynack || h.peerOfferedWS) && len(b) >= 8 { + b[4] = byte(OptNop) + h.optcodec.PutOption(b[5:], OptWindowScale, h.wndShiftLocal) + words++ + } + return words +} + +// wireWnd converts a segment's real window to its on-wire representation: SYN +// windows are never scaled (RFC 7323 §2.2), other windows drop the low shift bits +// once both sides offered scaling, and a value still exceeding the 16-bit field +// saturates rather than wrapping to near zero. +func (h *Handler) wireWnd(seg Segment) Size { + wnd := seg.WND + if h.peerOfferedWS && !seg.Flags.HasAny(FlagSYN) { + wnd >>= h.wndShiftLocal + } + if wnd > 0xFFFF { + wnd = 0xFFFF + } + return wnd +} + // AwaitingSynResponse returns true if the Handler is an active client opened with [Handler.OpenActive] and has already sent out the first SYN packet to the remote client. func (h *Handler) AwaitingSynResponse() bool { return h.remotePort != 0 && h.scb.State() == StateSynSent diff --git a/tcp/seqsbug_test.go b/tcp/seqsbug_test.go index f0dfd26..1ea26ed 100644 --- a/tcp/seqsbug_test.go +++ b/tcp/seqsbug_test.go @@ -329,7 +329,7 @@ func TestWindowReject_ChallengeACK(t *testing.T) { SEQ: issB + 512, ACK: issA, Flags: FlagACK, - WND: Size(1 << 17), // > MaxUint16. + WND: Size(maxWindow + 1), // beyond even the largest scaled window (RFC 7323). } err := tcb.Recv(seg) if err == nil { diff --git a/tcp/wscale_test.go b/tcp/wscale_test.go new file mode 100644 index 0000000..5fd4278 --- /dev/null +++ b/tcp/wscale_test.go @@ -0,0 +1,237 @@ +package tcp + +import ( + "bytes" + "math/rand" + "testing" +) + +// synOptions parses the option block of a raw TCP frame and returns the MSS +// and window-scale values found, with ok flags for presence. +func synOptions(t *testing.T, frame []byte) (mss uint16, mssOK bool, shift uint8, shiftOK bool) { + t.Helper() + tfrm, err := NewFrame(frame) + if err != nil { + t.Fatal(err) + } + var oc OptionCodec + err = oc.ForEachOption(tfrm.Options(), func(kind OptionKind, data []byte) error { + switch { + case kind == OptMaxSegmentSize && len(data) == 2: + mss = uint16(data[0])<<8 | uint16(data[1]) + mssOK = true + case kind == OptWindowScale && len(data) == 1: + shift = data[0] + shiftOK = true + } + return nil + }) + if err != nil { + t.Fatal(err) + } + return mss, mssOK, shift, shiftOK +} + +// TestWindowScaleNegotiation walks the three-way handshake with asymmetric +// buffer sizes and verifies RFC 7323 §2 end to end on the wire: the SYN and +// SYN-ACK carry the shift derived from each side's receive buffer, SYN +// windows are never scaled (and saturate rather than wrap the 16-bit field), +// and the first post-handshake segment advertises the scaled window. +func TestWindowScaleNegotiation(t *testing.T) { + const clientBuf = 256 << 10 // shift 3: 256KiB>>3 = 32Ki fits, >>2 does not. + const serverBuf = 1 << 20 // shift 5: 1MiB>>5 = 32Ki fits, >>4 does not. + rng := rand.New(rand.NewSource(1)) + client, server := new(Handler), new(Handler) + err := client.SetBuffers(make([]byte, 2048), make([]byte, clientBuf), 32) + if err != nil { + t.Fatal(err) + } + err = server.SetBuffers(make([]byte, 2048), make([]byte, serverBuf), 32) + if err != nil { + t.Fatal(err) + } + setupClientServer(t, rng, client, server) + + packetBuf := make([]byte, 2048) + + // Client SYN: window-scale offer present, window field unscaled+saturated. + n, err := client.Send(packetBuf) + if err != nil { + t.Fatal(err) + } + _, mssOK, shift, shiftOK := synOptions(t, packetBuf[:n]) + if !mssOK { + t.Fatal("client SYN lacks MSS option") + } + if !shiftOK { + t.Fatal("client SYN lacks window-scale option") + } + if shift != 3 { + t.Errorf("client SYN shift = %d, want 3 (buffer %d)", shift, clientBuf) + } + tfrm, _ := NewFrame(packetBuf[:n]) + if got := tfrm.WindowSize(); got != 0xFFFF { + t.Errorf("client SYN wire window = %d, want 65535 (saturated, never scaled)", got) + } + if err = server.Recv(packetBuf[:n]); err != nil { + t.Fatal(err) + } + + // Server SYN-ACK: echoes its own shift because the SYN offered scaling. + clear(packetBuf) + n, err = server.Send(packetBuf) + if err != nil { + t.Fatal(err) + } + _, _, shift, shiftOK = synOptions(t, packetBuf[:n]) + if !shiftOK { + t.Fatal("server SYN-ACK lacks window-scale option") + } + if shift != 5 { + t.Errorf("server SYN-ACK shift = %d, want 5 (buffer %d)", shift, serverBuf) + } + tfrm, _ = NewFrame(packetBuf[:n]) + if got := tfrm.WindowSize(); got != 0xFFFF { + t.Errorf("server SYN-ACK wire window = %d, want 65535 (saturated, never scaled)", got) + } + if err = client.Recv(packetBuf[:n]); err != nil { + t.Fatal(err) + } + + // Client handshake ACK: the first scaled window on the wire. The client + // advertises its whole free buffer, which only fits the field when + // right-shifted by its offered shift. + clear(packetBuf) + n, err = client.Send(packetBuf) + if err != nil { + t.Fatal(err) + } + tfrm, _ = NewFrame(packetBuf[:n]) + wantWire := uint16(clientBuf >> 3) + if got := tfrm.WindowSize(); got != wantWire { + t.Errorf("client ACK wire window = %d, want %d (%d >> 3)", got, wantWire, clientBuf) + } + if err = server.Recv(packetBuf[:n]); err != nil { + t.Fatal(err) + } + // The server must have scaled the advertisement back up to real octets. + if got := server.scb.snd.WND; got != Size(clientBuf) { + t.Errorf("server snd.WND = %d, want %d (scaled back up)", got, clientBuf) + } +} + +// TestWindowScaleBigTransfer proves the negotiated scale carries real data +// past the unscaled 64KiB ceiling: with 256KiB buffers on both sides the +// server streams segments without receiving a single ACK, and must be able +// to put more than 64KiB in flight before stalling on the send window. The +// client then receives everything intact. +func TestWindowScaleBigTransfer(t *testing.T) { + const bufSize = 256 << 10 + const payload = 200 << 10 + const mtu = 2048 + rng := rand.New(rand.NewSource(2)) + client, server := new(Handler), new(Handler) + err := client.SetBuffers(make([]byte, mtu), make([]byte, bufSize), 32) + if err != nil { + t.Fatal(err) + } + err = server.SetBuffers(make([]byte, bufSize), make([]byte, bufSize), 256) + if err != nil { + t.Fatal(err) + } + setupClientServer(t, rng, client, server) + packetBuf := make([]byte, mtu) + establish(t, client, server, packetBuf) + + data := make([]byte, payload) + rng.Read(data) + nw, err := server.Write(data) + if err != nil { + t.Fatal(err) + } else if nw != payload { + t.Fatalf("server buffered %d of %d", nw, payload) + } + + // Stream server→client WITHOUT delivering anything back: no ACKs, so + // everything sent stays in flight. Past 64KiB in flight is the proof + // that the scaled window governs the sender. + inFlight := 0 + frames := make([][]byte, 0, payload/1024) + for { + clear(packetBuf) + n, err := server.Send(packetBuf) + if err != nil { + t.Fatal(err) + } + if n <= sizeHeaderTCP { + break // window exhausted (or nothing left to send). + } + tfrm, _ := NewFrame(packetBuf[:n]) + inFlight += len(tfrm.Payload()) + frames = append(frames, append([]byte(nil), packetBuf[:n]...)) + if inFlight >= payload { + break + } + } + if inFlight <= 0xFFFF { + t.Fatalf("server stalled at %d bytes in flight; scaled window should allow more than 65535", inFlight) + } + + // Deliver the flight; the client must reassemble the stream intact. + for _, frm := range frames { + if err := client.Recv(frm); err != nil { + t.Fatal(err) + } + } + got := make([]byte, inFlight) + nr, err := client.Read(got) + if err != nil { + t.Fatal(err) + } + if nr != inFlight { + t.Fatalf("client read %d of %d in-flight bytes", nr, inFlight) + } + if !bytes.Equal(got[:nr], data[:nr]) { + t.Fatal("received data differs from sent data") + } +} + +// TestWindowScaleWireSafety covers the wire-conversion corners without a +// peer: no echo of the offer when the peer never gave one, and saturation +// (not wrap-around) of oversized windows for both SYN and non-SYN segments +// when scaling is off. Before window scaling existed a 128KiB receive buffer +// wrapped to a near-zero wire window on the SYN; that regression stays pinned +// here. +func TestWindowScaleWireSafety(t *testing.T) { + h := new(Handler) + err := h.SetBuffers(make([]byte, 2048), make([]byte, 128<<10), 32) + if err != nil { + t.Fatal(err) + } + if h.wndShiftLocal != 2 { + // 128KiB>>1 = 65536 still overflows the field; >>2 = 32768 fits. + t.Errorf("wndShiftLocal = %d, want 2 for 128KiB buffer", h.wndShiftLocal) + } + + var b [16]byte + if words := h.putSynOptions(b[:], 1460, true); words != 1 { + t.Errorf("SYN-ACK echoed window scale without a peer offer (words=%d)", words) + } + if words := h.putSynOptions(b[:], 1460, false); words != 2 { + t.Errorf("active SYN did not offer window scale (words=%d)", words) + } + + // Scaling off: oversized windows saturate the 16-bit field. + if got := h.wireWnd(Segment{WND: 128 << 10, Flags: FlagACK}); got != 0xFFFF { + t.Errorf("unscaled oversized window = %d, want 65535", got) + } + // SYN never scales, even with a negotiated peer shift. + h.peerOfferedWS = true + if got := h.wireWnd(Segment{WND: 128 << 10, Flags: FlagSYN}); got != 0xFFFF { + t.Errorf("SYN window = %d, want 65535 (saturated, unscaled)", got) + } + // Established segment with negotiated scaling: shifted representation. + if got := h.wireWnd(Segment{WND: 128 << 10, Flags: FlagACK}); got != (128<<10)>>2 { + t.Errorf("scaled window = %d, want %d", got, (128<<10)>>2) + } +}