From 70ef97d5642551d579576aa4cc67a675cf5ee9bf Mon Sep 17 00:00:00 2001 From: Artem Lytkin <146867384+4RH1T3CT0R7@users.noreply.github.com> Date: Mon, 12 Oct 2026 05:05:12 +0300 Subject: [PATCH] TUN inbound: Use Xray-core's built-in lightweight TCP/IP stack instead of gVisor (#7130) https://github.com/XTLS/Xray-core/pull/6771#issuecomment-5752154561 https://github.com/XTLS/Xray-core/pull/6795#issuecomment-6095740152 https://github.com/XTLS/Xray-core/pull/7130#issuecomment-6116258188 --- proxy/tun/handler.go | 48 +- proxy/tun/handler_test.go | 48 ++ proxy/tun/packet.go | 161 ++++ proxy/tun/stack.go | 263 +++++++ proxy/tun/stack_gvisor.go | 267 ------- proxy/tun/stack_gvisor_endpoint.go | 153 ---- proxy/tun/stack_gvisor_icmp_handler.go | 98 --- proxy/tun/stack_test.go | 929 +++++++++++++++++++++++ proxy/tun/tcp_lightweight.go | 999 +++++++++++++++++++++++++ proxy/tun/tun.go | 18 +- proxy/tun/tun_android.go | 24 +- proxy/tun/tun_darwin.go | 112 +-- proxy/tun/tun_default.go | 7 +- proxy/tun/tun_fd.go | 96 +++ proxy/tun/tun_freebsd.go | 94 +-- proxy/tun/tun_linux.go | 71 +- proxy/tun/tun_windows.go | 80 +- proxy/tun/udp_fullcone.go | 55 +- 18 files changed, 2704 insertions(+), 819 deletions(-) create mode 100644 proxy/tun/packet.go delete mode 100644 proxy/tun/stack_gvisor.go delete mode 100644 proxy/tun/stack_gvisor_endpoint.go delete mode 100644 proxy/tun/stack_gvisor_icmp_handler.go create mode 100644 proxy/tun/stack_test.go create mode 100644 proxy/tun/tcp_lightweight.go create mode 100644 proxy/tun/tun_fd.go diff --git a/proxy/tun/handler.go b/proxy/tun/handler.go index c13625b0b..f566c84ad 100644 --- a/proxy/tun/handler.go +++ b/proxy/tun/handler.go @@ -56,6 +56,24 @@ func (w *tunUDPStatsWriter) WriteMultiBuffer(mb buf.MultiBuffer) error { return nil } +// tunTCPWriter is the writing side of a TCP connection for the dispatcher. It has no Close: the outbound closes +// its writer when the other side has no more to send, and the system may still be sending +type tunTCPWriter struct { + writer buf.Writer + counter stats.Counter +} + +func (w *tunTCPWriter) WriteMultiBuffer(mb buf.MultiBuffer) error { + size := mb.Len() + if err := w.writer.WriteMultiBuffer(mb); err != nil { + return err + } + if w.counter != nil { + w.counter.Add(int64(size)) + } + return nil +} + // ConnectionHandler interface with the only method that stack is going to push new connections to type ConnectionHandler interface { HandleConnection(conn net.Conn, destination net.Destination) @@ -143,6 +161,7 @@ func (t *Handler) Start() error { tunStackOptions := StackOptions{ Tun: tunInterface, + MTU: t.config.MTU, IdleTimeout: t.policyManager.ForLevel(t.config.UserLevel).Timeouts.ConnectionIdle, } tunStack, err := NewStack(t.ctx, tunStackOptions, t) @@ -194,22 +213,7 @@ func (t *Handler) HandleConnection(conn net.Conn, destination net.Destination) { defer cancel() ctx = c.ContextWithID(ctx, session.NewID()) - // if the connection is already closed, conn.RemoteAddr() will be nil - // due to gvisor weird behavior - remote := conn.RemoteAddr() - if remote == nil { - errors.LogInfo(t.ctx, "dropped quickly closed connection") - return - } - source := net.DestinationFromAddr(remote) - isUDP := destination.Network == net.Network_UDP - if !isUDP && (t.uplinkCounter != nil || t.downlinkCounter != nil) { - conn = &stat.CounterConnection{ - Connection: conn, - ReadCounter: t.uplinkCounter, - WriteCounter: t.downlinkCounter, - } - } + source := net.DestinationFromAddr(conn.RemoteAddr()) inbound := session.Inbound{ Name: "tun", @@ -235,13 +239,13 @@ func (t *Handler) HandleConnection(conn net.Conn, destination net.Destination) { }) errors.LogInfo(ctx, "processing from ", source, " to ", destination) - reader := &buf.TimeoutWrapperReader{Reader: buf.NewReader(conn)} + // both kinds of connection are buf.Reader and buf.Writer, what they carry is counted on the way + reader := &buf.TimeoutWrapperReader{Reader: buf.NewReader(conn), Counter: t.uplinkCounter} writer := buf.NewWriter(conn) - if isUDP { - reader.Counter = t.uplinkCounter - if t.downlinkCounter != nil { - writer = &tunUDPStatsWriter{writer: writer, counter: t.downlinkCounter} - } + if destination.Network != net.Network_UDP { + writer = &tunTCPWriter{writer: writer, counter: t.downlinkCounter} + } else if t.downlinkCounter != nil { + writer = &tunUDPStatsWriter{writer: writer, counter: t.downlinkCounter} } link := &transport.Link{ diff --git a/proxy/tun/handler_test.go b/proxy/tun/handler_test.go index 25e9117ab..cb7f642d9 100644 --- a/proxy/tun/handler_test.go +++ b/proxy/tun/handler_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/xtls/xray-core/common" "github.com/xtls/xray-core/common/buf" xnet "github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/features/routing" @@ -103,6 +104,53 @@ func (d *testDispatcher) DispatchLink(ctx context.Context, dest xnet.Destination return link.Writer.WriteMultiBuffer(buf.MultiBuffer{buf.FromBytes(d.writePayload)}) } +// closableConn is a connection that is its own writer, as the ones of the stack are +type closableConn struct { + *testConn + closed int32 +} + +func (c *closableConn) WriteMultiBuffer(mb buf.MultiBuffer) error { + buf.ReleaseMulti(mb) + return nil +} + +func (c *closableConn) Close() error { + atomic.AddInt32(&c.closed, 1) + return nil +} + +// closingDispatcher does what an outbound does when the other side has no more to send +type closingDispatcher struct { + testDispatcher + conn *closableConn + closed int32 +} + +func (d *closingDispatcher) DispatchLink(_ context.Context, _ xnet.Destination, link *transport.Link) error { + common.Close(link.Writer) + d.closed = atomic.LoadInt32(&d.conn.closed) + return nil +} + +func TestHandlerClosesTunConnectionWhenBothSidesAreDone(t *testing.T) { + conn := &closableConn{testConn: newTestConn(nil)} + dispatcher := &closingDispatcher{conn: conn} + handler := &Handler{ + ctx: context.Background(), + config: &Config{}, + dispatcher: dispatcher, + } + handler.HandleConnection(conn, xnet.TCPDestination(xnet.LocalHostIP, 443)) + + if dispatcher.closed != 0 { + t.Fatal("the connection was closed with its writer, while the system may still be sending") + } + if got := atomic.LoadInt32(&conn.closed); got != 1 { + t.Fatalf("connection closed %d times, want 1", got) + } +} + func TestHandlerCountsTunConnectionTraffic(t *testing.T) { uplinkCounter := new(testCounter) downlinkCounter := new(testCounter) diff --git a/proxy/tun/packet.go b/proxy/tun/packet.go new file mode 100644 index 000000000..f9933e307 --- /dev/null +++ b/proxy/tun/packet.go @@ -0,0 +1,161 @@ +package tun + +import ( + "encoding/binary" + "math/bits" + "net/netip" +) + +const ( + ipv4HeaderSize = 20 + ipv6HeaderSize = 40 + tcpHeaderSize = 20 + udpHeaderSize = 8 + + protoICMPv4 = 1 + protoTCP = 6 + protoUDP = 17 + protoICMPv6 = 58 +) + +// packet is an IP packet with its header read +type packet struct { + proto byte + src, dst netip.Addr + // payload is what follows the IP header + payload []byte +} + +// parsePacket reads the IP header of b. Fragments are not accepted: what the system sends to the tun fits its MTU +func parsePacket(b []byte) (p packet, ok bool) { + if len(b) < ipv4HeaderSize { + return p, false + } + switch b[0] >> 4 { + case 4: + size := int(b[0]&0xf) * 4 + total := int(binary.BigEndian.Uint16(b[2:])) + if size < ipv4HeaderSize || total < size || total > len(b) || binary.BigEndian.Uint16(b[6:])&0x3fff != 0 { + return p, false + } + p.proto = b[9] + p.src = netip.AddrFrom4([4]byte(b[12:16])) + p.dst = netip.AddrFrom4([4]byte(b[16:20])) + p.payload = b[size:total] + case 6: + if len(b) < ipv6HeaderSize { + return p, false + } + total := ipv6HeaderSize + int(binary.BigEndian.Uint16(b[4:])) + if total > len(b) { + return p, false + } + p.proto = b[6] + p.src = netip.AddrFrom16([16]byte(b[8:24])) + p.dst = netip.AddrFrom16([16]byte(b[24:40])) + p.payload = b[ipv6HeaderSize:total] + // hop-by-hop options, routing, destination options + for p.proto == 0 || p.proto == 43 || p.proto == 60 { + if len(p.payload) < 8 { + return p, false + } + size := int(p.payload[1])*8 + 8 + if size > len(p.payload) { + return p, false + } + p.proto, p.payload = p.payload[0], p.payload[size:] + } + default: + return p, false + } + return p, true +} + +// putIPHeader writes the header of a packet from src to dst into b, without its length, and returns the size of the header +func putIPHeader(b []byte, proto byte, src, dst netip.Addr) int { + if src.Is4() { + s, d := src.As4(), dst.As4() + clear(b[:ipv4HeaderSize]) + b[0] = 0x45 + if proto == protoTCP { + b[6] = 0x40 // don't fragment + } + b[8] = 64 + b[9] = proto + copy(b[12:], s[:]) + copy(b[16:], d[:]) + return ipv4HeaderSize + } + s, d := src.As16(), dst.As16() + clear(b[:ipv6HeaderSize]) + b[0] = 0x60 + b[6] = proto + b[7] = 64 + copy(b[8:], s[:]) + copy(b[24:], d[:]) + return ipv6HeaderSize +} + +// setIPLength completes the header ip of a packet that is size bytes long with it +func setIPLength(ip []byte, size int) { + if len(ip) == ipv4HeaderSize { + binary.BigEndian.PutUint16(ip[2:], uint16(size)) + ip[10], ip[11] = 0, 0 + binary.BigEndian.PutUint16(ip[10:], checksum(sum(ip, 0))) + return + } + binary.BigEndian.PutUint16(ip[4:], uint16(size-ipv6HeaderSize)) +} + +// transportChecksum is the checksum of a TCP or UDP segment, header and payload, of a packet with the header ip. +// The checksum field of header is zero +func transportChecksum(ip []byte, proto byte, header, payload []byte) uint16 { + return checksum(sum(payload, sum(header, pseudoSum(ip, proto, len(header)+len(payload))))) +} + +// pseudoSum is what the header ip of a packet adds to the checksum of its TCP or UDP segment of size bytes +func pseudoSum(ip []byte, proto byte, size int) uint64 { + addresses := ip[12:ipv4HeaderSize] + if len(ip) == ipv6HeaderSize { + addresses = ip[8:] + } + return sum(addresses, uint64(proto)+uint64(size)) +} + +// sum adds the 16-bit words of b to s, with the carry of every addition added back +func sum(b []byte, s uint64) uint64 { + var c uint64 + for len(b) >= 32 { + s, c = bits.Add64(s, binary.BigEndian.Uint64(b), c) + s, c = bits.Add64(s, binary.BigEndian.Uint64(b[8:]), c) + s, c = bits.Add64(s, binary.BigEndian.Uint64(b[16:]), c) + s, c = bits.Add64(s, binary.BigEndian.Uint64(b[24:]), c) + b = b[32:] + } + for len(b) >= 8 { + s, c = bits.Add64(s, binary.BigEndian.Uint64(b), c) + b = b[8:] + } + if len(b) >= 4 { + s, c = bits.Add64(s, uint64(binary.BigEndian.Uint32(b)), c) + b = b[4:] + } + if len(b) >= 2 { + s, c = bits.Add64(s, uint64(binary.BigEndian.Uint16(b)), c) + b = b[2:] + } + if len(b) == 1 { + s, c = bits.Add64(s, uint64(b[0])<<8, c) + } + s, c = bits.Add64(s, 0, c) + return s + c +} + +// checksum folds s into the 16 bits that go into a header +func checksum(s uint64) uint16 { + s = s>>32 + s&0xffffffff + s = s>>16 + s&0xffff + s = s>>16 + s&0xffff + s = s>>16 + s&0xffff + return ^uint16(s) +} diff --git a/proxy/tun/stack.go b/proxy/tun/stack.go index ba67a4885..4efb1c458 100644 --- a/proxy/tun/stack.go +++ b/proxy/tun/stack.go @@ -1,7 +1,16 @@ package tun import ( + "context" + "encoding/binary" + "net/netip" + "sync" + "sync/atomic" "time" + + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/net" ) // Stack interface implement ip protocol stack, bridging raw network packets and data streams @@ -13,5 +22,259 @@ type Stack interface { // StackOptions for the stack implementation type StackOptions struct { Tun Tun + MTU uint32 IdleTimeout time.Duration } + +// flow is what the packets of one connection have in common, src is the side of the system +type flow struct { + src, dst netip.AddrPort +} + +// stack turns the packets that the system sends to the tun into connections and back. +// It has only what the two need to talk through a tun, and a local link when the system passes on +// what others send, see tcpConn +type stack struct { + ctx context.Context + tun Tun + mtu int + // segmenter is the tun if it takes segments that are larger than its MTU + segmenter segmenter + // window is the window of every connection + window int32 + // handle is given every new connection + handle func(conn net.Conn, destination net.Destination) + udp *udpConnectionHandler + + mu sync.Mutex + conns map[flow]*tcpConn + closed bool + // opening is how many of conns wait for the answer to their SYN-ACK + opening atomic.Int32 +} + +// NewStack builds new ip stack +func NewStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { + s := &stack{ + ctx: ctx, + tun: options.Tun, + mtu: int(options.MTU), + window: tcpWindow, + handle: handler.HandleConnection, + conns: make(map[flow]*tcpConn), + } + if t, ok := options.Tun.(segmenter); ok && t.Segments() { + s.segmenter = t + s.window = tcpLargeWindow + } + s.udp = newUdpConnectionHandler(s.handle, s.writeUDP) + return s, nil +} + +// Start is called by Handler to bring stack to life +func (s *stack) Start() error { + go s.read() + return nil +} + +// Close is called by Handler to shut down the stack +func (s *stack) Close() error { + s.mu.Lock() + s.closed = true + conns := make([]*tcpConn, 0, len(s.conns)) + for _, c := range s.conns { + conns = append(conns, c) + } + s.mu.Unlock() + for _, c := range conns { + c.mu.Lock() + if c.err == nil { + c.reset() + } + c.mu.Unlock() + } + return nil +} + +// read takes the packets of the tun until it is closed +func (s *stack) read() { + // a tun that does not split segments gives them as large as a packet can be + b := make([]byte, packetRoom+1<<16) + for { + b, err := s.tun.ReadPacket(b) + if err != nil { + s.mu.Lock() + closed := s.closed + s.mu.Unlock() + if !closed { + errors.LogErrorInner(s.ctx, err, "[tun] failed to read a packet, no more are read") + } + return + } + p, ok := parsePacket(b) + if !ok { + continue + } + switch p.proto { + case protoTCP: + s.handleTCP(p) + case protoUDP: + s.handleUDP(p) + case protoICMPv4, protoICMPv6: + s.handleICMP(p) + } + } +} + +func (s *stack) handleTCP(p packet) { + t := p.payload + if len(t) < tcpHeaderSize { + return + } + size := int(t[12]>>4) * 4 + if size < tcpHeaderSize || size > len(t) { + return + } + f := flow{ + src: netip.AddrPortFrom(p.src, binary.BigEndian.Uint16(t)), + dst: netip.AddrPortFrom(p.dst, binary.BigEndian.Uint16(t[2:])), + } + seg := segment{ + seq: binary.BigEndian.Uint32(t[4:]), + ack: binary.BigEndian.Uint32(t[8:]), + flags: t[13], + window: binary.BigEndian.Uint16(t[14:]), + options: t[tcpHeaderSize:size], + payload: t[size:], + } + + s.mu.Lock() + c := s.conns[f] + s.mu.Unlock() + if c != nil && c.input(seg) { + return + } + if seg.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn { + if seg.flags&tcpRst == 0 { + s.refuse(f, seg) + } + return + } + + if s.opening.Load() >= tcpOpening { + return + } + c = newTCPConn(s, f, seg) + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.conns[f] = c + s.mu.Unlock() + s.opening.Add(1) + c.mu.Lock() + c.sendSynAck() + c.arm(tcpTick) + c.mu.Unlock() +} + +// refuse answers a segment of a connection that is not known with a reset +func (s *stack) refuse(f flow, seg segment) { + var b [ipv6HeaderSize + tcpHeaderSize]byte + n := putIPHeader(b[:], protoTCP, f.dst.Addr(), f.src.Addr()) + t := b[n : n+tcpHeaderSize] + binary.BigEndian.PutUint16(t, f.dst.Port()) + binary.BigEndian.PutUint16(t[2:], f.src.Port()) + t[12] = tcpHeaderSize / 4 << 4 + if seg.flags&tcpAck != 0 { + binary.BigEndian.PutUint32(t[4:], seg.ack) + t[13] = tcpRst + } else { + ack := seg.seq + uint32(len(seg.payload)) + if seg.flags&(tcpSyn|tcpFin) != 0 { + ack++ + } + binary.BigEndian.PutUint32(t[8:], ack) + t[13] = tcpRst | tcpAck + } + setIPLength(b[:n], n+tcpHeaderSize) + binary.BigEndian.PutUint16(t[16:], transportChecksum(b[:n], protoTCP, t, nil)) + _ = s.tun.WritePacket(b[:n+tcpHeaderSize], nil) +} + +// remove forgets a connection. c.mu is held +func (s *stack) remove(c *tcpConn) { + if !c.established { + s.opening.Add(-1) + } + s.mu.Lock() + if s.conns[c.flow] == c { + delete(s.conns, c.flow) + } + s.mu.Unlock() +} + +func (s *stack) handleUDP(p packet) { + u := p.payload + if len(u) < udpHeaderSize { + return + } + size := int(binary.BigEndian.Uint16(u[4:])) + if size < udpHeaderSize || size > len(u) { + return + } + // the packet waits for its reader, longer than read keeps it + b := buf.NewWithSize(int32(size - udpHeaderSize)) + b.Write(u[udpHeaderSize:size]) + src := net.UDPDestination(net.IPAddress(p.src.AsSlice()), net.Port(binary.BigEndian.Uint16(u))) + dst := net.UDPDestination(net.IPAddress(p.dst.AsSlice()), net.Port(binary.BigEndian.Uint16(u[2:]))) + s.udp.HandlePacket(src, dst, b) +} + +// writeUDP writes a packet with the payload from src to dst to the tun +func (s *stack) writeUDP(payload []byte, src net.Destination, dst net.Destination) error { + srcIP, _ := netip.AddrFromSlice(src.Address.IP()) + dstIP, _ := netip.AddrFromSlice(dst.Address.IP()) + srcIP, dstIP = srcIP.Unmap(), dstIP.Unmap() + if !srcIP.IsValid() || srcIP.Is4() != dstIP.Is4() { + return errors.New("failed to write udp packet from ", src, " to ", dst) + } + var b [ipv6HeaderSize + udpHeaderSize]byte + n := putIPHeader(b[:], protoUDP, srcIP, dstIP) + u := b[n : n+udpHeaderSize] + binary.BigEndian.PutUint16(u, uint16(src.Port)) + binary.BigEndian.PutUint16(u[2:], uint16(dst.Port)) + binary.BigEndian.PutUint16(u[4:], uint16(udpHeaderSize+len(payload))) + setIPLength(b[:n], n+udpHeaderSize+len(payload)) + check := transportChecksum(b[:n], protoUDP, u, payload) + if check == 0 { + // zero stands for no checksum + check = 0xffff + } + binary.BigEndian.PutUint16(u[6:], check) + return s.tun.WritePacket(b[:n+udpHeaderSize], payload) +} + +// handleICMP answers an echo request as its destination would +func (s *stack) handleICMP(p packet) { + m := p.payload + request, reply := byte(8), byte(0) + if p.proto == protoICMPv6 { + request, reply = 128, 129 + } + if len(m) < 8 || m[0] != request || m[1] != 0 { + return + } + var b [ipv6HeaderSize]byte + n := putIPHeader(b[:], p.proto, p.dst, p.src) + setIPLength(b[:n], n+len(m)) + // the reply is the request with another type + m[0], m[2], m[3] = reply, 0, 0 + if p.proto == protoICMPv6 { + binary.BigEndian.PutUint16(m[2:], transportChecksum(b[:n], protoICMPv6, m, nil)) + } else { + binary.BigEndian.PutUint16(m[2:], checksum(sum(m, 0))) + } + _ = s.tun.WritePacket(b[:n], m) +} diff --git a/proxy/tun/stack_gvisor.go b/proxy/tun/stack_gvisor.go deleted file mode 100644 index de1735117..000000000 --- a/proxy/tun/stack_gvisor.go +++ /dev/null @@ -1,267 +0,0 @@ -package tun - -import ( - "context" - "time" - - "github.com/xtls/xray-core/common/errors" - "github.com/xtls/xray-core/common/net" - "gvisor.dev/gvisor/pkg/buffer" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" - "gvisor.dev/gvisor/pkg/tcpip/checksum" - "gvisor.dev/gvisor/pkg/tcpip/header" - "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" - "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" - "gvisor.dev/gvisor/pkg/tcpip/stack" - "gvisor.dev/gvisor/pkg/tcpip/transport/icmp" - "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" - "gvisor.dev/gvisor/pkg/tcpip/transport/udp" - "gvisor.dev/gvisor/pkg/waiter" -) - -const ( - defaultNIC tcpip.NICID = 1 - - tcpRXBufMinSize = tcp.MinBufferSize - tcpRXBufDefSize = tcp.DefaultSendBufferSize - tcpRXBufMaxSize = 8 << 20 // 8MiB - - tcpTXBufMinSize = tcp.MinBufferSize - tcpTXBufDefSize = tcp.DefaultReceiveBufferSize - tcpTXBufMaxSize = 6 << 20 // 6MiB -) - -// stackGVisor is ip stack implemented by gVisor package -type stackGVisor struct { - ctx context.Context - tun Tun - idleTimeout time.Duration - handler *Handler - stack *stack.Stack - endpoint stack.LinkEndpoint -} - -// NewStack builds new ip stack (using gVisor) -func NewStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { - gStack := &stackGVisor{ - ctx: ctx, - tun: options.Tun, - idleTimeout: options.IdleTimeout, - handler: handler, - } - - return gStack, nil -} - -// Start is called by Handler to bring stack to life -func (t *stackGVisor) Start() error { - linkEndpoint, err := t.tun.newEndpoint() - if err != nil { - return err - } - - ipStack, err := createStack(linkEndpoint) - if err != nil { - return err - } - - tcpForwarder := tcp.NewForwarder(ipStack, 0, 65535, func(r *tcp.ForwarderRequest) { - go func(r *tcp.ForwarderRequest) { - var wq waiter.Queue - id := r.ID() - - // Perform a TCP three-way handshake. - ep, err := r.CreateEndpoint(&wq) - if err != nil { - errors.LogError(t.ctx, err.String()) - r.Complete(true) - return - } - - options := ep.SocketOptions() - options.SetKeepAlive(false) - options.SetReuseAddress(true) - options.SetReusePort(true) - - t.handler.HandleConnection( - gonet.NewTCPConn(&wq, ep), - // local address on the gVisor side is connection destination - net.TCPDestination(net.IPAddress(id.LocalAddress.AsSlice()), net.Port(id.LocalPort)), - ) - - // close the socket - ep.Close() - // send connection complete upstream - r.Complete(false) - }(r) - }) - ipStack.SetTransportProtocolHandler(tcp.ProtocolNumber, tcpForwarder.HandlePacket) - - // Use custom UDP packet handler, instead of strict gVisor forwarder, for FullCone NAT support - udpForwarder := newUdpConnectionHandler(t.handler.HandleConnection, t.writeRawUDPPacket) - ipStack.SetTransportProtocolHandler(udp.ProtocolNumber, func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { - data := pkt.Data().AsRange().ToSlice() - // if len(data) == 0 { - // return false - // } - // source/destination of the packet we process as incoming, on gVisor side are Remote/Local - // in other terms, src is the side behind tun, dst is the side behind gVisor - // this function handle packets passing from the tun to the gVisor, therefore the src/dst assignement - srcIP := net.IPAddress(id.RemoteAddress.AsSlice()) - dstIP := net.IPAddress(id.LocalAddress.AsSlice()) - if srcIP == nil || dstIP == nil { - panic(id) - } - src := net.UDPDestination(srcIP, net.Port(id.RemotePort)) - dst := net.UDPDestination(dstIP, net.Port(id.LocalPort)) - udpForwarder.HandlePacket(src, dst, data) - return true - }) - ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber4, t.handleICMPv4Packet) - ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber6, t.handleICMPv6Packet) - - t.stack = ipStack - t.endpoint = linkEndpoint - - return nil -} - -func (t *stackGVisor) writeRawUDPPacket(payload []byte, src net.Destination, dst net.Destination) error { - udpLen := header.UDPMinimumSize + len(payload) - srcIP := tcpip.AddrFromSlice(src.Address.IP()) - dstIP := tcpip.AddrFromSlice(dst.Address.IP()) - - // build packet with appropriate IP header size - isIPv4 := dst.Address.Family().IsIPv4() - ipHdrSize := header.IPv6MinimumSize - ipProtocol := header.IPv6ProtocolNumber - if isIPv4 { - ipHdrSize = header.IPv4MinimumSize - ipProtocol = header.IPv4ProtocolNumber - } - - pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ - ReserveHeaderBytes: ipHdrSize + header.UDPMinimumSize, - Payload: buffer.MakeWithData(payload), - }) - defer pkt.DecRef() - - // Build UDP header - udpHdr := header.UDP(pkt.TransportHeader().Push(header.UDPMinimumSize)) - udpHdr.Encode(&header.UDPFields{ - SrcPort: uint16(src.Port), - DstPort: uint16(dst.Port), - Length: uint16(udpLen), - }) - - // Calculate and set UDP checksum - xsum := header.PseudoHeaderChecksum(header.UDPProtocolNumber, srcIP, dstIP, uint16(udpLen)) - udpHdr.SetChecksum(^udpHdr.CalculateChecksum(checksum.Checksum(payload, xsum))) - - // Build IP header - if isIPv4 { - ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) - ipHdr.Encode(&header.IPv4Fields{ - TotalLength: uint16(header.IPv4MinimumSize + udpLen), - TTL: 64, - Protocol: uint8(header.UDPProtocolNumber), - SrcAddr: srcIP, - DstAddr: dstIP, - }) - ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) - } else { - ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) - ipHdr.Encode(&header.IPv6Fields{ - PayloadLength: uint16(udpLen), - TransportProtocol: header.UDPProtocolNumber, - HopLimit: 64, - SrcAddr: srcIP, - DstAddr: dstIP, - }) - } - - // dispatch the packet - err := t.stack.WriteRawPacket(defaultNIC, ipProtocol, buffer.MakeWithView(pkt.ToView())) - if err != nil { - return errors.New("failed to write raw udp packet back to stack", err) - } - - return nil -} - -// Close is called by Handler to shut down the stack -func (t *stackGVisor) Close() error { - if t.stack == nil { - return nil - } - t.endpoint.Attach(nil) - t.stack.Close() - for _, endpoint := range t.stack.CleanupEndpoints() { - endpoint.Abort() - } - - return nil -} - -// createStack configure gVisor ip stack -func createStack(ep stack.LinkEndpoint) (*stack.Stack, error) { - opts := stack.Options{ - NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol}, - TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol4, icmp.NewProtocol6}, - HandleLocal: false, - } - gStack := stack.New(opts) - - err := gStack.CreateNIC(defaultNIC, ep) - if err != nil { - return nil, errors.New(err.String()) - } - - gStack.SetRouteTable([]tcpip.Route{ - {Destination: header.IPv4EmptySubnet, NIC: defaultNIC}, - {Destination: header.IPv6EmptySubnet, NIC: defaultNIC}, - }) - - err = gStack.SetSpoofing(defaultNIC, true) - if err != nil { - return nil, errors.New(err.String()) - } - err = gStack.SetPromiscuousMode(defaultNIC, true) - if err != nil { - return nil, errors.New(err.String()) - } - - cOpt := tcpip.CongestionControlOption("cubic") - gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &cOpt) - sOpt := tcpip.TCPSACKEnabled(true) - gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &sOpt) - mOpt := tcpip.TCPModerateReceiveBufferOption(true) - gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &mOpt) - - // Disable RACK/TLP loss recovery to fix connection stalls under high load - rOpt := tcpip.TCPRecovery(0) - gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &rOpt) - - tcpRXBufOpt := tcpip.TCPReceiveBufferSizeRangeOption{ - Min: tcpRXBufMinSize, - Default: tcpRXBufDefSize, - Max: tcpRXBufMaxSize, - } - err = gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &tcpRXBufOpt) - if err != nil { - return nil, errors.New(err.String()) - } - - tcpTXBufOpt := tcpip.TCPSendBufferSizeRangeOption{ - Min: tcpTXBufMinSize, - Default: tcpTXBufDefSize, - Max: tcpTXBufMaxSize, - } - err = gStack.SetTransportProtocolOption(tcp.ProtocolNumber, &tcpTXBufOpt) - if err != nil { - return nil, errors.New(err.String()) - } - - return gStack, nil -} diff --git a/proxy/tun/stack_gvisor_endpoint.go b/proxy/tun/stack_gvisor_endpoint.go deleted file mode 100644 index 26d8ecabd..000000000 --- a/proxy/tun/stack_gvisor_endpoint.go +++ /dev/null @@ -1,153 +0,0 @@ -package tun - -import ( - "context" - "errors" - - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/header" - "gvisor.dev/gvisor/pkg/tcpip/stack" -) - -var ErrQueueEmpty = errors.New("queue is empty") - -type GVisorDevice interface { - WritePacket(packet *stack.PacketBuffer) tcpip.Error - ReadPacket() (byte, *stack.PacketBuffer, error) - Wait() -} - -// LinkEndpoint implements GVisor stack.LinkEndpoint -var _ stack.LinkEndpoint = (*LinkEndpoint)(nil) - -type LinkEndpoint struct { - deviceMTU uint32 - device GVisorDevice - dispatcherCancel context.CancelFunc -} - -func (e *LinkEndpoint) MTU() uint32 { - return e.deviceMTU -} - -func (e *LinkEndpoint) SetMTU(_ uint32) { - // not Implemented, as it is not expected GVisor will be asking tun device to be modified -} - -func (e *LinkEndpoint) MaxHeaderLength() uint16 { - return 0 -} - -func (e *LinkEndpoint) LinkAddress() tcpip.LinkAddress { - return "" -} - -func (e *LinkEndpoint) SetLinkAddress(_ tcpip.LinkAddress) { - // not Implemented, as it is not expected GVisor will be asking tun device to be modified -} - -func (e *LinkEndpoint) Capabilities() stack.LinkEndpointCapabilities { - return stack.CapabilityRXChecksumOffload -} - -func (e *LinkEndpoint) Attach(dispatcher stack.NetworkDispatcher) { - if e.dispatcherCancel != nil { - e.dispatcherCancel() - e.dispatcherCancel = nil - } - - if dispatcher != nil { - ctx, cancel := context.WithCancel(context.Background()) - go e.dispatchLoop(ctx, dispatcher) - e.dispatcherCancel = cancel - } -} - -func (e *LinkEndpoint) IsAttached() bool { - return e.dispatcherCancel != nil -} - -func (e *LinkEndpoint) Wait() { -} - -func (e *LinkEndpoint) ARPHardwareType() header.ARPHardwareType { - return header.ARPHardwareNone -} - -func (e *LinkEndpoint) AddHeader(buffer *stack.PacketBuffer) { - // tun interface doesn't have link layer header, it will be added by the OS -} - -func (e *LinkEndpoint) ParseHeader(ptr *stack.PacketBuffer) bool { - return true -} - -func (e *LinkEndpoint) Close() { - if e.dispatcherCancel != nil { - e.dispatcherCancel() - e.dispatcherCancel = nil - } -} - -func (e *LinkEndpoint) SetOnCloseAction(_ func()) { -} - -func (e *LinkEndpoint) WritePackets(packetBufferList stack.PacketBufferList) (int, tcpip.Error) { - var n int - var err tcpip.Error - - for _, packetBuffer := range packetBufferList.AsSlice() { - err = e.device.WritePacket(packetBuffer) - if err != nil { - return n, &tcpip.ErrAborted{} - } - n++ - } - - return n, nil -} - -func (e *LinkEndpoint) dispatchLoop(ctx context.Context, dispatcher stack.NetworkDispatcher) { - var networkProtocolNumber tcpip.NetworkProtocolNumber - var version byte - var packet *stack.PacketBuffer - var err error - - for { - select { - case <-ctx.Done(): - return - default: - version, packet, err = e.device.ReadPacket() - // on "queue empty", ask device to yield slightly and continue - if errors.Is(err, ErrQueueEmpty) { - e.device.Wait() - continue - } - // stop dispatcher loop on any other interface failure - if err != nil { - e.Attach(nil) - return - } - - // extract network protocol number from the packet first byte - // (which is returned separately, since it is so incredibly hard to extract one byte from - // stack.PacketBuffer without additional memory allocation and full copying it back and forth) - switch version { - case 4: - networkProtocolNumber = header.IPv4ProtocolNumber - case 6: - networkProtocolNumber = header.IPv6ProtocolNumber - default: - // discard unknown network protocol packet - packet.DecRef() - continue - } - - // dispatch the buffer to the stack - dispatcher.DeliverNetworkPacket(networkProtocolNumber, packet) - // signal the buffer that it can be released - packet.DecRef() - } - } -} diff --git a/proxy/tun/stack_gvisor_icmp_handler.go b/proxy/tun/stack_gvisor_icmp_handler.go deleted file mode 100644 index 17ba692bf..000000000 --- a/proxy/tun/stack_gvisor_icmp_handler.go +++ /dev/null @@ -1,98 +0,0 @@ -package tun - -import ( - "github.com/xtls/xray-core/common/errors" - tunicmp "github.com/xtls/xray-core/proxy/tun/icmp" - "gvisor.dev/gvisor/pkg/buffer" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/header" - "gvisor.dev/gvisor/pkg/tcpip/stack" -) - -func (t *stackGVisor) handleICMPv4Packet(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { - return t.handleICMPEchoPacket(header.IPv4ProtocolNumber, id, pkt) -} - -func (t *stackGVisor) handleICMPv6Packet(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { - return t.handleICMPEchoPacket(header.IPv6ProtocolNumber, id, pkt) -} - -func (t *stackGVisor) handleICMPEchoPacket(netProto tcpip.NetworkProtocolNumber, id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { - srcIP := id.RemoteAddress - dstIP := id.LocalAddress - if srcIP.Len() == 0 || dstIP.Len() == 0 { - return true - } - - message := transportPacketBytes(pkt) - ident, sequence, ok := tunicmp.ParseEchoRequest(netProto, message) - if !ok { - return true - } - - reply, err := tunicmp.BuildLocalEchoReply(netProto, message, dstIP, srcIP) - if err != nil { - errors.LogInfoInner(t.ctx, err, "[tun] failed to build local icmp echo reply") - return true - } - - errors.LogDebug(t.ctx, "[tun][icmp] ", tunicmp.ProtocolLabel(netProto), " local echo reply ", dstIP, " -> ", srcIP, " id=", ident, " seq=", sequence) - if err := t.writeRawICMPPacket(netProto, reply, dstIP, srcIP); err != nil { - errors.LogInfoInner(t.ctx, err, "[tun] failed to write local icmp echo reply") - } - - return true -} - -func (t *stackGVisor) writeRawICMPPacket(netProto tcpip.NetworkProtocolNumber, message []byte, srcIP, dstIP tcpip.Address) error { - ipHeaderSize := header.IPv6MinimumSize - ipProtocol := header.IPv6ProtocolNumber - transportProtocol := header.ICMPv6ProtocolNumber - if netProto == header.IPv4ProtocolNumber { - ipHeaderSize = header.IPv4MinimumSize - ipProtocol = header.IPv4ProtocolNumber - transportProtocol = header.ICMPv4ProtocolNumber - } - - pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ - ReserveHeaderBytes: ipHeaderSize, - Payload: buffer.MakeWithData(message), - }) - defer pkt.DecRef() - - if netProto == header.IPv4ProtocolNumber { - ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) - ipHdr.Encode(&header.IPv4Fields{ - TotalLength: uint16(header.IPv4MinimumSize + len(message)), - TTL: 64, - Protocol: uint8(transportProtocol), - SrcAddr: srcIP, - DstAddr: dstIP, - }) - ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) - } else { - ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) - ipHdr.Encode(&header.IPv6Fields{ - PayloadLength: uint16(len(message)), - TransportProtocol: transportProtocol, - HopLimit: 64, - SrcAddr: srcIP, - DstAddr: dstIP, - }) - } - - if err := t.stack.WriteRawPacket(defaultNIC, ipProtocol, buffer.MakeWithView(pkt.ToView())); err != nil { - return errors.New("failed to write raw icmp packet back to stack", err) - } - - return nil -} - -func transportPacketBytes(pkt *stack.PacketBuffer) []byte { - headerBytes := pkt.TransportHeader().Slice() - payloadBytes := pkt.Data().AsRange().ToSlice() - message := make([]byte, len(headerBytes)+len(payloadBytes)) - copy(message, headerBytes) - copy(message[len(headerBytes):], payloadBytes) - return message -} diff --git a/proxy/tun/stack_test.go b/proxy/tun/stack_test.go new file mode 100644 index 000000000..a68461481 --- /dev/null +++ b/proxy/tun/stack_test.go @@ -0,0 +1,929 @@ +package tun + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/binary" + "io" + mrand "math/rand/v2" + "net/netip" + "sync" + "testing" + "time" + + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/net" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/link/channel" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" + gstack "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" +) + +const testMTU = 1500 + +// link is a tun with the test on its other side +type link struct { + // in is what the stack reads + in chan []byte + // out takes what the stack writes + out func(packet []byte) + // segments is set if the link takes segments that are larger than the MTU, wrong tells what was wrong with one + segments bool + wrong func(args ...any) + done chan struct{} + once sync.Once +} + +func newLink() *link { + return &link{in: make(chan []byte, 512), done: make(chan struct{})} +} + +func (l *link) Start() error { return nil } +func (l *link) Name() (string, error) { return "test", nil } +func (l *link) Index() (int, error) { return 1, nil } + +func (l *link) Close() error { + l.once.Do(func() { close(l.done) }) + return nil +} + +func (l *link) ReadPacket(b []byte) ([]byte, error) { + select { + case packet := <-l.in: + return b[packetRoom : packetRoom+copy(b[packetRoom:], packet)], nil + case <-l.done: + return nil, io.EOF + } +} + +func (l *link) WritePacket(header, payload []byte) error { + l.out(append(append(make([]byte, 0, len(header)+len(payload)), header...), payload...)) + return nil +} + +// Segments implements segmenter +func (l *link) Segments() bool { + return l.segments +} + +// WriteSegment implements segmenter as the system does when it sends a segment on: a packet for every mss bytes +func (l *link) WriteSegment(header []byte, payload [][]byte, mss int) error { + data := bytes.Join(payload, nil) + size := ipv4HeaderSize + if header[0]>>4 == 6 { + size = ipv6HeaderSize + } + if expected := ^checksum(pseudoSum(header[:size], protoTCP, len(header)-size+len(data))); binary.BigEndian.Uint16(header[size+16:]) != expected { + l.wrong("checksum of a segment for the link to split: ", header[size+16:size+18], ", expected ", expected) + } + for sent := 0; sent < len(data); { + n := min(mss, len(data)-sent) + h := bytes.Clone(header) + t := h[size:] + binary.BigEndian.PutUint32(t[4:], binary.BigEndian.Uint32(t[4:])+uint32(sent)) + if sent+n < len(data) { + t[13] &^= tcpFin | tcpPsh + } + t[16], t[17] = 0, 0 + setIPLength(h[:size], len(h)+n) + binary.BigEndian.PutUint16(t[16:], transportChecksum(h[:size], protoTCP, t, data[sent:sent+n])) + l.out(append(h, data[sent:sent+n]...)) + sent += n + } + return nil +} + +func newTestStack(t *testing.T, l *link, handle func(net.Conn, net.Destination)) *stack { + s := &stack{ctx: context.Background(), tun: l, mtu: testMTU, window: tcpWindow, handle: handle, conns: make(map[flow]*tcpConn)} + if l.segments { + s.segmenter, s.window, l.wrong = l, tcpLargeWindow, t.Error + } + s.udp = newUdpConnectionHandler(handle, s.writeUDP) + s.Start() + t.Cleanup(func() { + s.Close() + l.Close() + }) + return s +} + +// forgotten waits until the stack has no connection left +func forgotten(t *testing.T, s *stack) { + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) { + s.mu.Lock() + n := len(s.conns) + s.mu.Unlock() + if n == 0 { + if opening := s.opening.Load(); opening != 0 { + t.Fatal("connections that are still counted as being opened: ", opening) + } + return + } + if time.Now().After(deadline) { + t.Fatal("connections left in the stack: ", n) + } + } +} + +// echo sends back what it reads, and closes when the other side has +func echo(conn net.Conn, _ net.Destination) { + defer conn.Close() + for { + mb, err := conn.(buf.Reader).ReadMultiBuffer() + if err != nil { + return + } + if conn.(buf.Writer).WriteMultiBuffer(mb) != nil { + return + } + } +} + +// A real TCP on the other side of a tun that loses packets both ways gets back what it sent. +func TestTCPEcho(t *testing.T) { + for _, test := range []struct { + name string + // one packet of lose is lost + lose int + size int + // the tun takes segments that are larger than its MTU + segments bool + }{ + {"lossless", 0, 16 << 20, false}, + {"lossy", 100, 4 << 20, false}, + {"very lossy", 20, 512 << 10, false}, + {"lossless, large segments", 0, 16 << 20, true}, + {"lossy, large segments", 100, 4 << 20, true}, + } { + t.Run(test.name, func(t *testing.T) { + l := newLink() + l.segments = test.segments + random := mrand.New(mrand.NewPCG(1, uint64(test.lose))) + var mu sync.Mutex + lost := func() bool { + mu.Lock() + defer mu.Unlock() + return test.lose != 0 && random.IntN(test.lose) == 0 + } + + // the system on the other side of the tun + system := gstack.New(gstack.Options{ + NetworkProtocols: []gstack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol}, + TransportProtocols: []gstack.TransportProtocolFactory{tcp.NewProtocol}, + }) + defer system.Destroy() + endpoint := channel.New(512, testMTU, "") + if err := system.CreateNIC(1, endpoint); err != nil { + t.Fatal(err) + } + for _, address := range []tcpip.ProtocolAddress{ + {Protocol: ipv4.ProtocolNumber, AddressWithPrefix: tcpip.AddrFrom4([4]byte{10, 0, 0, 1}).WithPrefix()}, + {Protocol: ipv6.ProtocolNumber, AddressWithPrefix: tcpip.AddrFrom16([16]byte{0xfd, 15: 1}).WithPrefix()}, + } { + if err := system.AddProtocolAddress(1, address, gstack.AddressProperties{}); err != nil { + t.Fatal(err) + } + } + system.SetRouteTable([]tcpip.Route{{Destination: header.IPv4EmptySubnet, NIC: 1}, {Destination: header.IPv6EmptySubnet, NIC: 1}}) + sack := tcpip.TCPSACKEnabled(true) + system.SetTransportProtocolOption(tcp.ProtocolNumber, &sack) + // without RACK: after a tail loss probe of it that is lost too, it has no timer left to send what was lost again + recovery := tcpip.TCPRecovery(0) + system.SetTransportProtocolOption(tcp.ProtocolNumber, &recovery) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + for { + packet := endpoint.ReadContext(ctx) + if packet == nil { + return + } + var b []byte + for _, part := range packet.AsSlices() { + b = append(b, part...) + } + packet.DecRef() + if !lost() { + select { + case l.in <- b: + default: + // as a tun does when its queue is full + } + } + } + }() + l.out = func(b []byte) { + if lost() { + return + } + protocol := ipv4.ProtocolNumber + if b[0]>>4 == 6 { + protocol = ipv6.ProtocolNumber + } + packet := gstack.NewPacketBuffer(gstack.PacketBufferOptions{Payload: buffer.MakeWithData(b)}) + endpoint.InjectInbound(protocol, packet) + packet.DecRef() + } + s := newTestStack(t, l, echo) + + for _, to := range []struct { + address tcpip.Address + protocol tcpip.NetworkProtocolNumber + }{ + {tcpip.AddrFrom4([4]byte{10, 0, 0, 2}), ipv4.ProtocolNumber}, + {tcpip.AddrFrom16([16]byte{0xfd, 15: 2}), ipv6.ProtocolNumber}, + } { + conn, err := gonet.DialTCP(system, tcpip.FullAddress{NIC: 1, Addr: to.address, Port: 80}, to.protocol) + if err != nil { + t.Fatal(err) + } + sent := make([]byte, test.size) + rand.Read(sent) + go func() { + conn.Write(sent) + conn.CloseWrite() + }() + conn.SetReadDeadline(time.Now().Add(2 * time.Minute)) + received, err := io.ReadAll(conn) + if err != nil || !bytes.Equal(received, sent) { + t.Fatal("received ", len(received), " of ", len(sent), " bytes, equal: ", bytes.Equal(received, sent), ", error: ", err) + } + conn.Close() + forgotten(t, s) + } + }) + } +} + +// peer is the system's end of one connection, with every segment made by the test +type peer struct { + t *testing.T + l *link + sent chan []byte + src, dst netip.AddrPort + // seq is the next that the peer sends, ack the next that it expects + seq, ack uint32 + window uint16 +} + +func newPeer(t *testing.T, src, dst string) (*peer, *link) { + l := newLink() + p := &peer{t: t, l: l, sent: make(chan []byte, 1024), src: netip.MustParseAddrPort(src), dst: netip.MustParseAddrPort(dst), seq: 1000, window: 0xffff} + l.out = func(b []byte) { p.sent <- b } + return p, l +} + +// send gives the stack a segment of the peer +func (p *peer) send(seq uint32, flags byte, options, payload []byte) { + var b [ipv6HeaderSize + tcpHeaderSize + 40]byte + n := putIPHeader(b[:], protoTCP, p.src.Addr(), p.dst.Addr()) + t := b[n : n+tcpHeaderSize+len(options)] + binary.BigEndian.PutUint16(t, p.src.Port()) + binary.BigEndian.PutUint16(t[2:], p.dst.Port()) + binary.BigEndian.PutUint32(t[4:], seq) + binary.BigEndian.PutUint32(t[8:], p.ack) + t[12] = byte(len(t)/4) << 4 + t[13] = flags + binary.BigEndian.PutUint16(t[14:], p.window) + copy(t[tcpHeaderSize:], options) + setIPLength(b[:n], n+len(t)+len(payload)) + binary.BigEndian.PutUint16(t[16:], transportChecksum(b[:n], protoTCP, t, payload)) + p.l.in <- append(b[:n+len(t):n+len(t)], payload...) +} + +// packet takes the next packet that the stack wrote +func (p *peer) packet() []byte { + p.t.Helper() + select { + case b := <-p.sent: + return b + case <-time.After(5 * time.Second): + p.t.Fatal("no packet from the stack") + return nil + } +} + +// receive takes the next segment that the stack wrote and checks its addresses and checksums +func (p *peer) receive() segment { + p.t.Helper() + b := p.packet() + size := ipv4HeaderSize + if b[0]>>4 == 6 { + size = ipv6HeaderSize + } else if checksum(sum(b[:size], 0)) != 0 { + p.t.Fatal("wrong checksum of the ip header") + } + packet, ok := parsePacket(b) + if !ok || packet.proto != protoTCP || packet.src != p.dst.Addr() || packet.dst != p.src.Addr() { + p.t.Fatal("not a segment of the connection: ", b) + } + t := packet.payload + if transportChecksum(b[:size], protoTCP, t, nil) != 0 { + p.t.Fatal("wrong checksum of the segment") + } + if binary.BigEndian.Uint16(t) != p.dst.Port() || binary.BigEndian.Uint16(t[2:]) != p.src.Port() { + p.t.Fatal("wrong ports: ", t[:4]) + } + offset := int(t[12]>>4) * 4 + return segment{ + seq: binary.BigEndian.Uint32(t[4:]), + ack: binary.BigEndian.Uint32(t[8:]), + flags: t[13], + window: binary.BigEndian.Uint16(t[14:]), + options: t[tcpHeaderSize:offset], + payload: t[offset:], + } +} + +// silent fails if the stack writes something within d +func (p *peer) silent(d time.Duration) { + p.t.Helper() + select { + case b := <-p.sent: + p.t.Fatal("unexpected packet from the stack: ", b) + case <-time.After(d): + } +} + +// connect opens the connection with the given options in the SYN and returns the options of the answer +func (p *peer) connect(options []byte) []byte { + p.t.Helper() + p.send(p.seq, tcpSyn, options, nil) + p.seq++ + s := p.receive() + if s.flags != tcpSyn|tcpAck || s.ack != p.seq { + p.t.Fatal("not an answer to the SYN: flags ", s.flags, ", ack ", s.ack) + } + p.ack = s.seq + 1 + p.send(p.seq, tcpAck, nil, nil) + return s.options +} + +// accepted is what a test gets of the connections that the stack hands over +type accepted struct { + conn net.Conn + dest net.Destination +} + +func accept(conns chan accepted) func(net.Conn, net.Destination) { + return func(conn net.Conn, dest net.Destination) { conns <- accepted{conn, dest} } +} + +func take(t *testing.T, conns chan accepted) accepted { + t.Helper() + select { + case a := <-conns: + return a + case <-time.After(5 * time.Second): + t.Fatal("no connection from the stack") + return accepted{} + } +} + +func write(t *testing.T, conn net.Conn, payload []byte) { + t.Helper() + if err := conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, payload)); err != nil { + t.Fatal(err) + } +} + +func read(t *testing.T, conn net.Conn, expected string) { + t.Helper() + mb, err := conn.(buf.Reader).ReadMultiBuffer() + if err != nil || mb.String() != expected { + t.Fatal("read ", mb.String(), ", expected ", expected, ", error: ", err) + } + buf.ReleaseMulti(mb) +} + +// The options of the SYN decide what the answer has, and both ends close the connection in turn. +func TestTCPHandshakeAndClose(t *testing.T) { + for _, test := range []struct { + src, dst string + options, expected []byte + }{ + {"10.0.0.1:1234", "10.0.0.2:80", nil, []byte{2, 4, 5, 0xb4}}, + {"10.0.0.1:1234", "10.0.0.2:80", []byte{2, 4, 2, 0, 4, 2, 1, 3, 3, 7, 0, 0}, []byte{2, 4, 5, 0xb4, 1, 1, 4, 2, 1, 3, 3, tcpWindowShift}}, + {"[fd00::1]:1234", "[fd00::2]:80", []byte{2, 4, 5, 0xa0, 1, 3, 3, 7}, []byte{2, 4, 5, 0xa0, 1, 3, 3, tcpWindowShift}}, + } { + conns := make(chan accepted, 1) + p, l := newPeer(t, test.src, test.dst) + s := newTestStack(t, l, accept(conns)) + if options := p.connect(test.options); !bytes.Equal(options, test.expected) { + t.Fatal("options ", options, ", expected ", test.expected) + } + a := take(t, conns) + if a.dest != net.TCPDestination(net.ParseAddress(p.dst.Addr().String()), net.Port(80)) || a.conn.RemoteAddr().String() != test.src { + t.Fatal("connection from ", a.conn.RemoteAddr(), " to ", a.dest) + } + + // the system closes its side, what it sent is read first + p.send(p.seq, tcpAck|tcpPsh|tcpFin, nil, []byte("abc")) + p.seq += 4 + if s := p.receive(); s.flags != tcpAck || s.ack != p.seq { + t.Fatal("the FIN is not acknowledged: flags ", s.flags, ", ack ", s.ack) + } + read(t, a.conn, "abc") + if _, err := a.conn.(buf.Reader).ReadMultiBuffer(); err != io.EOF { + t.Fatal("no end of the stream: ", err) + } + // the other side is still open + write(t, a.conn, []byte("def")) + if s := p.receive(); string(s.payload) != "def" || s.seq != p.ack || s.flags != tcpAck|tcpPsh { + t.Fatal("wrong segment: seq ", s.seq, ", flags ", s.flags, ", payload ", s.payload) + } + p.ack += 3 + a.conn.Close() + if s := p.receive(); s.flags != tcpFin|tcpAck || s.seq != p.ack { + t.Fatal("not a FIN: flags ", s.flags, ", seq ", s.seq) + } + p.ack++ + // what the system sends again is not something that nobody reads anymore + p.send(p.seq-4, tcpAck|tcpPsh|tcpFin, nil, []byte("abc")) + if s := p.receive(); s.flags != tcpAck || s.ack != p.seq { + t.Fatal("what was sent again is not acknowledged: flags ", s.flags, ", ack ", s.ack) + } + p.send(p.seq, tcpAck, nil, nil) + forgotten(t, s) + // what comes for it now is refused + p.send(p.seq, tcpAck, nil, nil) + if s := p.receive(); s.flags != tcpRst || s.seq != p.ack { + t.Fatal("not a reset: flags ", s.flags, ", seq ", s.seq) + } + } +} + +// What comes after a lost segment is kept and told to the system, and is read once the lost one comes. +func TestTCPOutOfOrder(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect([]byte{2, 4, 5, 0xb4, 4, 2, 0, 0}) + a := take(t, conns) + + sack := func(s segment, blocks ...uint32) { + t.Helper() + expected := []byte{1, 1, 5, byte(2 + 4*len(blocks))} + for _, edge := range blocks { + expected = binary.BigEndian.AppendUint32(expected, edge) + } + if s.flags != tcpAck || s.ack != p.seq || !bytes.Equal(s.options, expected) { + t.Fatal("ack ", s.ack, " with options ", s.options, ", expected ", p.seq, " with ", expected) + } + } + // "abc" is lost, then "ghi" and "def" come, each told with the block that it is in first + p.send(p.seq+6, tcpAck, nil, []byte("ghi")) + sack(p.receive(), p.seq+6, p.seq+9) + p.send(p.seq+12, tcpAck, nil, []byte("mno")) + sack(p.receive(), p.seq+12, p.seq+15, p.seq+6, p.seq+9) + p.send(p.seq+3, tcpAck, nil, []byte("def")) + sack(p.receive(), p.seq+3, p.seq+9, p.seq+12, p.seq+15) + // the same again changes nothing + p.send(p.seq+3, tcpAck, nil, []byte("defg")) + sack(p.receive(), p.seq+3, p.seq+9, p.seq+12, p.seq+15) + + p.send(p.seq, tcpAck, nil, []byte("abc")) + p.seq += 9 + sack(p.receive(), p.seq+3, p.seq+6) + read(t, a.conn, "abcdefghi") + p.send(p.seq, tcpAck, nil, []byte("jklm")) + p.seq += 6 + if s := p.receive(); s.ack != p.seq || len(s.options) != 0 { + t.Fatal("ack ", s.ack, " with options ", s.options) + } + read(t, a.conn, "jklmno") +} + +// A segment is as large as the system makes it when the tun does not split what it reads. +func TestTCPLargeSegment(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect(nil) + a := take(t, conns) + + sent := make([]byte, 60000) + rand.Read(sent) + p.send(p.seq, tcpAck|tcpPsh, nil, sent) + var received []byte + for len(received) < len(sent) { + mb, err := a.conn.(buf.Reader).ReadMultiBuffer() + if err != nil { + t.Fatal(err) + } + for _, b := range mb { + received = append(received, b.Bytes()...) + } + buf.ReleaseMulti(mb) + } + if !bytes.Equal(received, sent) { + t.Fatal("read something else than what was sent") + } +} + +// What the system does not acknowledge is sent again: at once when it says what it has past it, or after a while. +func TestTCPRetransmit(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect([]byte{2, 4, 0, 100, 4, 2, 0, 0}) + a := take(t, conns) + + sent := make([]byte, 450) + rand.Read(sent) + go func() { + if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, sent)); err != nil { + t.Error(err) + } + }() + for i := 0; i < 5; i++ { + limit := 100 + if i == 4 { + limit = 50 + } + if s := p.receive(); s.seq != p.ack+uint32(100*i) || !bytes.Equal(s.payload, sent[100*i:100*i+limit]) { + t.Fatal("segment ", i, ": seq ", s.seq-p.ack, " with ", len(s.payload), " bytes") + } + } + // the second and the third segment are lost + p.ack += 100 + p.send(p.seq, tcpAck, nil, nil) + block := func(from, to uint32) []byte { + b := []byte{1, 1, 5, 10} + return binary.BigEndian.AppendUint32(binary.BigEndian.AppendUint32(b, from), to) + } + p.send(p.seq, tcpAck, block(p.ack+200, p.ack+350), nil) + for i := 1; i < 3; i++ { + if s := p.receive(); s.seq != p.ack+uint32(100*(i-1)) || !bytes.Equal(s.payload, sent[100*i:100*i+100]) { + t.Fatal("segment ", i, " sent again: seq ", s.seq-p.ack, " with ", len(s.payload), " bytes") + } + } + // said again, it is not sent again + p.send(p.seq, tcpAck, block(p.ack+200, p.ack+350), nil) + p.silent(tcpTick) + // the one that was sent again came, the other one did not: it is sent after a while, without what the system has + p.ack += 100 + p.send(p.seq, tcpAck, block(p.ack+100, p.ack+250), nil) + p.silent(tcpTick) + if s := p.receive(); s.seq != p.ack || !bytes.Equal(s.payload, sent[200:300]) { + t.Fatal("segment 2 sent again later: seq ", s.seq-p.ack, " with ", len(s.payload), " bytes") + } + p.ack += 250 + p.send(p.seq, tcpAck, nil, nil) + p.silent(2 * tcpRetransmitTicks * tcpTick) +} + +// No more is sent at once than has arrived before: twice as much when all of it arrives, half of it after a loss. +func TestTCPCongestion(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect([]byte{2, 4, 0, 100, 4, 2, 0, 0}) + a := take(t, conns) + + go a.conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, make([]byte, 8000))) + // next is where what is new starts + next := p.ack + segments := func(seqs ...uint32) { + t.Helper() + for _, seq := range seqs { + if s := p.receive(); s.seq != seq || len(s.payload) != 100 { + t.Fatal("seq ", s.seq-p.ack, " with ", len(s.payload), " bytes, expected ", seq-p.ack, " with 100") + } + } + p.silent(tcpTick / 2) + } + fresh := func(n int) (seqs []uint32) { + for ; n > 0; n-- { + seqs = append(seqs, next) + next += 100 + } + return + } + segments(fresh(tcpStart)...) + p.ack = next + p.send(p.seq, tcpAck, nil, nil) + segments(fresh(2 * tcpStart)...) + // the first of them is lost: it is sent again, and with what is new there is half as much on its way as before + p.send(p.seq, tcpAck, binary.BigEndian.AppendUint32(binary.BigEndian.AppendUint32([]byte{1, 1, 5, 10}, p.ack+100), next), nil) + segments(append([]uint32{p.ack}, fresh(tcpStart-1)...)...) + // from then on a segment more for as much as there was on its way, and never a part of one + p.ack = next + p.send(p.seq, tcpAck, nil, nil) + segments(fresh(tcpStart + 2)...) +} + +// No more is sent than the system takes, and the writer goes on when it takes more. +// The FIN that comes with a segment before the one in front of it is taken when that one has come. +func TestTCPEndOutOfOrder(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect(nil) + a := take(t, conns) + p.send(p.seq+3, tcpAck|tcpFin, nil, []byte("def")) + if s := p.receive(); s.ack != p.seq { + t.Fatal("ack ", s.ack, ", expected ", p.seq) + } + p.send(p.seq, tcpAck, nil, []byte("abc")) + p.seq += 7 + if s := p.receive(); s.ack != p.seq { + t.Fatal("ack ", s.ack, ", expected ", p.seq, " with the end") + } + read(t, a.conn, "abcdef") + if _, err := a.conn.(buf.Reader).ReadMultiBuffer(); err != io.EOF { + t.Fatal("no end after what was read: ", err) + } +} + +// The answer to the SYN-ACK tells how far the system is, and a segment is not sent again before +// its acknowledgment had the time to come from there. +func TestTCPRoundTrip(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.send(p.seq, tcpSyn, nil, nil) + p.seq++ + p.ack = p.receive().seq + 1 + time.Sleep(150 * time.Millisecond) + p.send(p.seq, tcpAck, nil, nil) + a := take(t, conns) + write(t, a.conn, []byte("abc")) + for range 2 { + if s := p.receive(); s.seq != p.ack || string(s.payload) != "abc" { + t.Fatal("seq ", s.seq, " with ", s.payload) + } + p.silent(350 * time.Millisecond) + } +} + +// A peer that says its segments are tiny is not believed all the way. +func TestTCPSmallestSegment(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.connect([]byte{2, 4, 0, 1}) + a := take(t, conns) + write(t, a.conn, make([]byte, 100)) + if s := p.receive(); len(s.payload) != tcpSmallest { + t.Fatal("a segment of ", len(s.payload), " bytes") + } +} + +func TestTCPWindow(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + newTestStack(t, l, accept(conns)) + p.window = 250 + p.connect(nil) + a := take(t, conns) + + sent := make([]byte, 1000) + rand.Read(sent) + done := make(chan struct{}) + go func() { + if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, sent)); err != nil { + t.Error(err) + } + close(done) + }() + var received []byte + for len(received) < len(sent) { + for space := 250; space > 0; { + s := p.receive() + if s.seq != p.ack || len(s.payload) > space { + t.Fatal("seq ", s.seq, " with ", len(s.payload), " bytes, expected ", p.ack, " with ", space, " at most") + } + received = append(received, s.payload...) + p.ack += uint32(len(s.payload)) + space -= len(s.payload) + } + // nothing more before the window opens + p.silent(tcpTick / 2) + p.send(p.seq, tcpAck, nil, nil) + } + <-done + if !bytes.Equal(received, sent) { + t.Fatal("received something else than what was written") + } + + // the time that the window was closed is not time without an acknowledgment of what is sent after it. + // It is closed with a keepalive: that is answered, so the stack has seen it before something is written + p.window = 0 + p.send(p.seq-1, tcpAck, nil, nil) + p.receive() + done = make(chan struct{}) + go func() { + if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, []byte("abc"))); err != nil { + t.Error(err) + } + close(done) + }() + p.silent(4 * tcpTick) + p.window = 250 + p.send(p.seq, tcpAck, nil, nil) + if s := p.receive(); s.seq != p.ack || string(s.payload) != "abc" { + t.Fatal("seq ", s.seq, " with ", s.payload) + } + <-done + p.silent(tcpTick + tcpTick/2) +} + +// A keepalive is answered, and not before a segment that was made before it is written. +func TestTCPKeepalive(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + // the write of the segment with what is written below takes until it is let go + out, writing, written := l.out, make(chan struct{}), make(chan struct{}) + l.out = func(b []byte) { + if b[ipv4HeaderSize+13]&tcpPsh != 0 { + close(writing) + <-written + } + out(b) + } + newTestStack(t, l, accept(conns)) + p.connect(nil) + a := take(t, conns) + + p.send(p.seq-1, tcpAck, nil, nil) + if s := p.receive(); s.flags != tcpAck || s.seq != p.ack || s.ack != p.seq { + t.Fatal("not an answer to the keepalive: flags ", s.flags, ", seq ", s.seq, ", ack ", s.ack) + } + + go func() { + if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MergeBytes(nil, []byte("abc"))); err != nil { + t.Error(err) + } + }() + <-writing + p.send(p.seq-1, tcpAck, nil, nil) + p.silent(tcpTick / 2) + close(written) + if s := p.receive(); string(s.payload) != "abc" || s.seq != p.ack { + t.Fatal("wrong segment: seq ", s.seq, ", flags ", s.flags, ", payload ", s.payload) + } + p.ack += 3 + if s := p.receive(); s.flags != tcpAck || s.seq != p.ack || s.ack != p.seq || len(s.payload) != 0 { + t.Fatal("not an answer to the keepalive: flags ", s.flags, ", seq ", s.seq, ", ack ", s.ack) + } + p.send(p.seq, tcpAck, nil, nil) +} + +// A reset ends the connection for its reader, and a connection that is closed with something unread is reset. +func TestTCPReset(t *testing.T) { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + s := newTestStack(t, l, accept(conns)) + p.connect(nil) + a := take(t, conns) + // a reset from somewhere else than where the connection is changes nothing + p.send(p.seq+tcpWindow+1, tcpRst, nil, nil) + p.send(p.seq-1, tcpRst, nil, nil) + p.send(p.seq, tcpAck|tcpPsh, nil, []byte("abc")) + p.seq += 3 + read(t, a.conn, "abc") + if s := p.receive(); s.flags != tcpAck || s.ack != p.seq { + t.Fatal("not acknowledged: flags ", s.flags, ", ack ", s.ack) + } + p.send(p.seq, tcpRst, nil, nil) + if _, err := a.conn.(buf.Reader).ReadMultiBuffer(); err == nil || err == io.EOF { + t.Fatal("no error after the reset: ", err) + } + a.conn.Close() + forgotten(t, s) + + // what came before the reset is read, to its end + p.src = netip.MustParseAddrPort("10.0.0.1:1236") + p.connect(nil) + a = take(t, conns) + p.send(p.seq, tcpAck|tcpPsh|tcpFin, nil, []byte("abc")) + p.seq += 4 + if s := p.receive(); s.flags != tcpAck || s.ack != p.seq { + t.Fatal("not acknowledged: flags ", s.flags, ", ack ", s.ack) + } + // with the number of the FIN, as the answer to a segment that was made before the FIN came + p.send(p.seq-1, tcpRst, nil, nil) + forgotten(t, s) + read(t, a.conn, "abc") + if _, err := a.conn.(buf.Reader).ReadMultiBuffer(); err != io.EOF { + t.Fatal("no end after what was read: ", err) + } + a.conn.Close() + + p.src = netip.MustParseAddrPort("10.0.0.1:1235") + p.connect(nil) + a = take(t, conns) + p.send(p.seq, tcpAck|tcpPsh, nil, []byte("abc")) + p.seq += 3 + // the timer acknowledges what nobody reads + if s := p.receive(); s.flags != tcpAck || s.ack != p.seq { + t.Fatal("not acknowledged: flags ", s.flags, ", ack ", s.ack) + } + a.conn.Close() + if s := p.receive(); s.flags != tcpRst|tcpAck || s.seq != p.ack { + t.Fatal("not a reset: flags ", s.flags, ", seq ", s.seq) + } + forgotten(t, s) + + // one that the system gives up before it is open is not one that is being opened anymore + p.src = netip.MustParseAddrPort("10.0.0.1:1237") + p.send(p.seq, tcpSyn, nil, nil) + p.seq++ + p.receive() + p.send(p.seq, tcpRst, nil, nil) + forgotten(t, s) +} + +// A UDP packet starts a connection, which answers from any address, and an echo request is answered. +func TestUDPAndICMP(t *testing.T) { + for _, test := range []struct{ src, dst, other string }{ + {"10.0.0.1:1234", "10.0.0.2:53", "10.0.0.3:5353"}, + {"[fd00::1]:1234", "[fd00::2]:53", "[fd00::3]:5353"}, + } { + conns := make(chan accepted, 1) + p, l := newPeer(t, test.src, test.dst) + newTestStack(t, l, accept(conns)) + + var b [ipv6HeaderSize + udpHeaderSize]byte + n := putIPHeader(b[:], protoUDP, p.src.Addr(), p.dst.Addr()) + u := b[n : n+udpHeaderSize] + binary.BigEndian.PutUint16(u, p.src.Port()) + binary.BigEndian.PutUint16(u[2:], p.dst.Port()) + binary.BigEndian.PutUint16(u[4:], udpHeaderSize+3) + setIPLength(b[:n], n+udpHeaderSize+3) + l.in <- append(b[:n+udpHeaderSize:n+udpHeaderSize], "abc"...) + a := take(t, conns) + mb, err := a.conn.(buf.Reader).ReadMultiBuffer() + if err != nil || mb.String() != "abc" || *mb[0].UDP != a.dest || a.dest.NetAddr() != test.dst || a.conn.RemoteAddr().String() != test.src { + t.Fatal("read ", mb.String(), " from ", a.conn.RemoteAddr(), " to ", a.dest, ", error: ", err) + } + buf.ReleaseMulti(mb) + + for _, from := range []string{test.dst, test.other} { + answer := buf.FromBytes([]byte("defg")) + source := net.UDPDestination(net.ParseAddress(netip.MustParseAddrPort(from).Addr().String()), net.Port(netip.MustParseAddrPort(from).Port())) + answer.UDP = &source + if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MultiBuffer{answer}); err != nil { + t.Fatal(err) + } + raw := p.packet() + packet, ok := parsePacket(raw) + u := packet.payload + // the system may have to split it on the way to where it goes + if !ok || raw[0]>>4 == 4 && raw[6] != 0 || packet.proto != protoUDP || netip.AddrPortFrom(packet.src, binary.BigEndian.Uint16(u)).String() != from || + netip.AddrPortFrom(packet.dst, binary.BigEndian.Uint16(u[2:])) != p.src || string(u[udpHeaderSize:]) != "defg" { + t.Fatal("wrong packet from ", from, ": ", packet) + } + var ip [ipv6HeaderSize]byte + if size := putIPHeader(ip[:], protoUDP, packet.src, packet.dst); transportChecksum(ip[:size], protoUDP, u, nil) != 0 { + t.Fatal("wrong checksum of the packet from ", from) + } + } + a.conn.Close() + + proto, request, reply := byte(protoICMPv4), byte(8), byte(0) + if p.src.Addr().Is6() { + proto, request, reply = protoICMPv6, 128, 129 + } + n = putIPHeader(b[:], proto, p.src.Addr(), p.dst.Addr()) + setIPLength(b[:n], n+12) + l.in <- append(b[:n:n], request, 0, 0, 0, 1, 2, 3, 4, 'a', 'b', 'c', 'd') + packet, ok := parsePacket(p.packet()) + m := packet.payload + if !ok || packet.proto != proto || packet.src != p.dst.Addr() || packet.dst != p.src.Addr() || m[0] != reply || string(m[4:]) != "\x01\x02\x03\x04abcd" { + t.Fatal("wrong echo reply: ", packet) + } + var ip [ipv6HeaderSize]byte + size := putIPHeader(ip[:], proto, packet.src, packet.dst) + if proto == protoICMPv4 && checksum(sum(m, 0)) != 0 || proto == protoICMPv6 && transportChecksum(ip[:size], proto, m, nil) != 0 { + t.Fatal("wrong checksum of the echo reply") + } + } +} + +// The checksum is that of the 16-bit words added one by one. +func TestChecksum(t *testing.T) { + b := make([]byte, 5000) + rand.Read(b) + for size := 0; size < len(b); size += 1 + size/64 { + expected := uint32(size) + for i := 0; i < size; i += 2 { + expected += uint32(b[i]) << 8 + if i+1 < size { + expected += uint32(b[i+1]) + } + expected = expected>>16 + expected&0xffff + } + if got := checksum(sum(b[:size], uint64(size))); got != ^uint16(expected) { + t.Fatal("checksum of ", size, " bytes: ", got, ", expected ", ^uint16(expected)) + } + } +} diff --git a/proxy/tun/tcp_lightweight.go b/proxy/tun/tcp_lightweight.go new file mode 100644 index 000000000..1a5964d44 --- /dev/null +++ b/proxy/tun/tcp_lightweight.go @@ -0,0 +1,999 @@ +package tun + +import ( + "encoding/binary" + "io" + "math" + "math/rand/v2" + "slices" + "sync" + "time" + + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/net" +) + +const ( + tcpFin = 1 << iota + tcpSyn + tcpRst + tcpPsh + tcpAck +) + +const ( + // tcpWindow is what a connection takes from the system before its reader has to take it, + // and the most that it sends to the system before that is acknowledged. More than that, for several + // connections at once, is more packets than the queue of a tun holds (500 on Linux) + tcpWindow = 256 << 10 + // tcpLargeWindow is the same for a tun that takes and gives segments that are larger than its MTU: those are + // few, and a way with a longer round trip, as a router has to what is behind it, is not held back by it + tcpLargeWindow = 1 << 20 + // tcpWindowShift makes them fit the 16 bits of a window + tcpWindowShift = 5 + // tcpHeld is how many segments wait for one that was lost before them, + // and of how many parts what the system says it has is known + tcpHeld = 1024 + // tcpSmallest is the smallest mss that the system is believed, the one of Linux + tcpSmallest = 48 + // tcpOpening is how many connections can be waiting for the answer to their SYN-ACK + tcpOpening = 65535 + // tcpStart is how many segments are sent before any of them is acknowledged + tcpStart = 10 + // tcpSegment is the largest payload of a segment that the tun splits by itself, + // tcpParts of how many parts of what was written it is made at most + tcpSegment = math.MaxUint16 - ipv6HeaderSize - tcpHeaderSize + tcpParts = 7 + + // tcpTick is how often a connection that waits for something looks at what it waits for + tcpTick = 100 * time.Millisecond + // ticks without anything acknowledged before it is sent again, twice as many every next time, a minute at most + tcpRetransmitTicks = 3 + tcpRetransmitTicksMost = 600 + // how many times it is sent again before the connection is given up: that is after four minutes, as what is + // behind a wireless link may be away for a while, and after twenty seconds for one that never answered + tcpRetransmits = 10 + tcpSynRetransmits = 5 + // ticks between the probes of a window that stays closed + tcpProbeTicks = 10 + // tcpLinger is how long a closed connection waits for the system to close its side + tcpLinger = 30 * time.Second +) + +var ( + errTCPReset = errors.New("connection reset") + errTCPTimeout = errors.New("connection timed out") +) + +// segment is a TCP segment that came from the tun +type segment struct { + seq, ack uint32 + flags byte + window uint16 + options []byte + payload []byte +} + +// nextOption returns the first option of o and what follows it, kind is zero when there is none +func nextOption(o []byte) (kind byte, value, rest []byte) { + for len(o) > 0 && o[0] == 1 { + o = o[1:] + } + if len(o) < 2 || int(o[1]) < 2 || int(o[1]) > len(o) { + return 0, nil, nil + } + return o[0], o[2:o[1]], o[o[1]:] +} + +// held is a segment that came before the one in front of it +type held struct { + seq uint32 + data *buf.Buffer +} + +// tcpConn is a TCP connection that the system opened through the tun. +// Mostly the two are on the same machine, with nothing but the queue of the tun in between, which drops what +// does not fit. But the system may also pass on what others send, as a router does, and then the link to them +// is in between. So a connection has windows that are small enough, keeps what came after a lost segment, +// tells the system what it has (SACK), sends again what the system misses, and sends no more at once +// than what has arrived before (slow start and congestion avoidance as in RFC 5681) +type tcpConn struct { + stack *stack + flow flow + src, dst net.Destination + // ip is the IP header of what is sent, without the length + ip [ipv6HeaderSize]byte + // ipSize is the size of that header + ipSize int + // mss is the largest payload of a segment that is sent + mss int + // window is tcpWindow or tcpLargeWindow + window int32 + // sndShift and rcvShift are the window scale of what the system and we advertise + sndShift, rcvShift uint8 + // sack is set if the system takes selective acknowledgments + sack bool + + mu sync.Mutex + readable sync.Cond + writable sync.Cond + + established bool + // closed is set when Close is called + closed bool + // err is why the connection is gone + err error + + rcvNxt uint32 + // rcvAcked is the last acknowledgment number sent + rcvAcked uint32 + // rcvAdvertised is the last window sent, in bytes + rcvAdvertised int32 + rcvQueue buf.MultiBuffer + rcvLen int32 + rcvFin bool + // rcvEnd is where what the system sends ends, if rcvEnds: its FIN came with a segment that waits in rcvHeld + rcvEnd uint32 + rcvEnds bool + // rcvHeld is what came after something that did not, in the order of seq and without overlaps + rcvHeld []held + // rcvLast is the seq of the last one that was put into rcvHeld + rcvLast uint32 + + sndUna, sndNxt uint32 + sndWnd uint32 + // sndQueue is what was sent and is not acknowledged + sndQueue buf.MultiBuffer + // sndBase is the sequence number of the first byte of sndQueue + sndBase uint32 + sndFin bool + // writing is how many of the last of sndQueue WriteMultiBuffer is sending, it is the only one to release them + writing int + // blocked is set while WriteMultiBuffer waits for the window + blocked bool + // sending is set while WriteMultiBuffer writes a segment without mu, ackWanted if sendAck was called meanwhile + sending, ackWanted bool + dupAcks int + // cwnd is how much the way to the system takes at once, as far as that is known: it grows while what is sent + // arrives, by as much as arrives below ssthresh, and is cut when something is lost + cwnd, ssthresh uint32 + // sacked is what the system says it has past sndUna, in order and apart from each other + sacked [][2]uint32 + // recovering is set from a sign of loss until what was sent by then, up to recover, is acknowledged + recovering bool + recover uint32 + // lost is the end of what is lost past the last of sacked, resent the end of what was sent again since then + lost, resent uint32 + // timed is the end of a segment that was sent at timedAt and not again, to see how long its acknowledgment takes. + // The first one is the SYN-ACK + timed uint32 + timedAt time.Time + // srtt is how long an acknowledgment takes, rttvar how much that varies (RFC 6298) + srtt, rttvar time.Duration + + timer *time.Timer + armed bool + // stalled counts ticks without an acknowledgment of what waits for one + stalled, retransmitTicks, retransmits int + lingered bool +} + +func newTCPConn(s *stack, f flow, syn segment) *tcpConn { + c := &tcpConn{ + stack: s, + flow: f, + src: net.TCPDestination(net.IPAddress(f.src.Addr().AsSlice()), net.Port(f.src.Port())), + dst: net.TCPDestination(net.IPAddress(f.dst.Addr().AsSlice()), net.Port(f.dst.Port())), + rcvNxt: syn.seq + 1, + sndWnd: uint32(syn.window), + window: s.window, + ssthresh: uint32(s.window), + timedAt: time.Now(), + retransmitTicks: tcpRetransmitTicks, + } + c.readable.L = &c.mu + c.writable.L = &c.mu + c.ipSize = putIPHeader(c.ip[:], protoTCP, f.dst.Addr(), f.src.Addr()) + c.mss = s.mtu - c.ipSize - tcpHeaderSize + for kind, value, o := nextOption(syn.options); kind != 0; kind, value, o = nextOption(o) { + switch { + case kind == 2 && len(value) == 2: + c.mss = min(c.mss, int(binary.BigEndian.Uint16(value))) + case kind == 3 && len(value) == 1: + c.sndShift = min(value[0], 14) + c.rcvShift = tcpWindowShift + case kind == 4: + c.sack = true + } + } + c.mss = max(c.mss, tcpSmallest) + c.cwnd = tcpStart * uint32(c.mss) + c.sndUna = rand.Uint32() + c.sndNxt = c.sndUna + 1 + return c +} + +// header writes the headers of a segment with size bytes of payload into b and returns their size. +// The checksum of the segment is left to write. c.mu is held +func (c *tcpConn) header(b []byte, seq uint32, flags byte, options []byte, size int) int { + n := copy(b, c.ip[:c.ipSize]) + t := b[n : n+tcpHeaderSize+len(options)] + binary.BigEndian.PutUint16(t, c.flow.dst.Port()) + binary.BigEndian.PutUint16(t[2:], c.flow.src.Port()) + binary.BigEndian.PutUint32(t[4:], seq) + binary.BigEndian.PutUint32(t[8:], c.rcvNxt) + t[12] = byte(len(t)/4) << 4 + t[13] = flags + window := max(c.window-c.rcvLen, 0) + if flags&tcpSyn != 0 { + // the window of a SYN is not scaled + binary.BigEndian.PutUint16(t[14:], uint16(min(window, math.MaxUint16))) + } else { + binary.BigEndian.PutUint16(t[14:], uint16(min(window>>c.rcvShift, math.MaxUint16))) + } + t[16], t[17], t[18], t[19] = 0, 0, 0, 0 + copy(t[tcpHeaderSize:], options) + setIPLength(b[:n], n+len(t)+size) + c.rcvAcked = c.rcvNxt + c.rcvAdvertised = window + return n + len(t) +} + +// write completes the headers h with the checksum of the segment and writes it to the tun. +// A segment that is not written is one that is lost: it is sent again when it is not acknowledged +func (c *tcpConn) write(h, payload []byte) { + t := h[c.ipSize:] + binary.BigEndian.PutUint16(t[16:], transportChecksum(h[:c.ipSize], protoTCP, t, payload)) + _ = c.stack.tun.WritePacket(h, payload) +} + +// send writes a segment to the tun. c.mu is held +func (c *tcpConn) send(seq uint32, flags byte, options, payload []byte) { + var b [ipv6HeaderSize + tcpHeaderSize + 40]byte + c.write(b[:c.header(b[:], seq, flags, options, len(payload))], payload) +} + +func (c *tcpConn) sendSynAck() { + options := [12]byte{2, 4} + binary.BigEndian.PutUint16(options[2:], uint16(c.stack.mtu-c.ipSize-tcpHeaderSize)) + n := 4 + if c.sack { + n += copy(options[n:], []byte{1, 1, 4, 2}) + } + if c.rcvShift != 0 { + n += copy(options[n:], []byte{1, 3, 3, c.rcvShift}) + } + c.send(c.sndUna, tcpSyn|tcpAck, options[:n], nil) +} + +// sendAck acknowledges what was received, and tells what of rcvHeld waits for a lost segment +func (c *tcpConn) sendAck() { + if c.sending { + // after the segment that is being written: it was made before, and what the system is told goes in that order + c.ackWanted = true + return + } + if !c.sack || len(c.rcvHeld) == 0 { + c.send(c.sndNxt, tcpAck, nil, nil) + return + } + // the block with the segment that came last goes first + var blocks [4][2]uint32 + n, last := 1, false + for i := 0; i < len(c.rcvHeld); { + start, end, recent := c.rcvHeld[i].seq, c.rcvHeld[i].seq, false + for ; i < len(c.rcvHeld) && c.rcvHeld[i].seq == end; i++ { + recent = recent || end == c.rcvLast + end += uint32(c.rcvHeld[i].data.Len()) + } + if recent { + blocks[0], last = [2]uint32{start, end}, true + } else if n < len(blocks) { + blocks[n] = [2]uint32{start, end} + n++ + } + } + list := blocks[:n] + if !last { + list = list[1:] + } + options := [36]byte{1, 1, 5, byte(2 + 8*len(list))} + for i, block := range list { + binary.BigEndian.PutUint32(options[4+8*i:], block[0]) + binary.BigEndian.PutUint32(options[8+8*i:], block[1]) + } + c.send(c.sndNxt, tcpAck, options[:4+8*len(list)], nil) +} + +// input takes a segment of the connection. It returns false if the connection is gone +// and the segment is the start of another one +func (c *tcpConn) input(s segment) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.err != nil { + return s.flags&(tcpSyn|tcpAck|tcpRst) != tcpSyn + } + if s.flags&tcpRst != 0 { + // from where the system is, or was before its FIN, not from someone who knows no more than the addresses + if s.seq-c.rcvNxt <= uint32(c.window) || c.rcvFin && s.seq+1 == c.rcvNxt { + c.abort(errTCPReset) + } + return true + } + if s.flags&tcpSyn != 0 { + switch { + case !c.established && s.seq+1 == c.rcvNxt: + c.sendSynAck() + c.timedAt = time.Time{} + case c.closed && s.flags&tcpAck == 0: + c.abort(errTCPReset) + return false + default: + c.sendAck() + } + return true + } + if s.flags&tcpAck == 0 { + return true + } + if !c.established { + if s.ack != c.sndNxt { + return true + } + c.established = true + c.sndUna = s.ack + c.sndBase = s.ack + c.sndWnd = uint32(s.window) << c.sndShift + c.stalled, c.retransmits = 0, 0 + if !c.timedAt.IsZero() { + c.clocked() + } + c.stack.opening.Add(-1) + go c.stack.handle(c, c.dst) + } else { + c.acknowledge(s) + } + c.receive(s) + if c.sndFin && c.rcvFin && c.sndUna == c.sndNxt { + // both sides have closed it and nothing waits to be acknowledged + c.abort(io.ErrClosedPipe) + } + return true +} + +// acknowledge takes what a segment says about what was sent. c.mu is held +func (c *tcpConn) acknowledge(s segment) { + acked, sent := s.ack-c.sndUna, c.sndNxt-c.sndUna + if acked > sent { + // older than the last one, or of what was not sent + return + } + window := uint32(s.window) << c.sndShift + if acked != 0 { + c.sndUna = s.ack + c.dupAcks, c.stalled, c.retransmits = 0, 0, 0 + c.release() + n := 0 + for n < len(c.sacked) && int32(c.sacked[n][0]-c.sndUna) <= 0 { + n++ + } + c.sacked = slices.Delete(c.sacked, 0, n) + if int32(c.resent-c.sndUna) < 0 { + c.resent = c.sndUna + } + if !c.timedAt.IsZero() && int32(c.sndUna-c.timed) >= 0 { + c.clocked() + } + switch { + case c.recovering && int32(c.sndUna-c.recover) >= 0: + c.recovering = false + case c.recovering: + // what was sent again has arrived, and what was sent before it has not: the next segment is lost as well + if end := c.sndUna + min(uint32(c.mss), c.recover-c.sndUna); int32(end-c.lost) > 0 { + c.lost = end + } + } + // the way takes more if what it was given has arrived, as long as it was given about as much as it takes + if sent >= c.cwnd/2 { + if c.cwnd < c.ssthresh { + c.cwnd = min(c.cwnd+acked, c.ssthresh) + } else if !c.recovering { + // a segment more for every cwnd that arrives + c.cwnd = min(c.cwnd+max(uint32(uint64(c.mss)*uint64(acked)/uint64(c.cwnd)), 1), uint32(c.window)) + } + } + } else if sent != 0 && window == c.sndWnd && len(s.payload) == 0 && s.flags&tcpFin == 0 { + c.dupAcks++ + } + if c.sack { + for kind, value, o := nextOption(s.options); kind != 0; kind, value, o = nextOption(o) { + for ; kind == 5 && len(value) >= 8; value = value[8:] { + // of what was sent and is not acknowledged + start, end := binary.BigEndian.Uint32(value)-c.sndUna, binary.BigEndian.Uint32(value[4:])-c.sndUna + if start != 0 && start < end && end <= c.sndNxt-c.sndUna { + c.sackAdd(c.sndUna+start, c.sndUna+end) + } + } + } + } + if !c.recovering && c.sndUna != c.sndNxt && (len(c.sacked) > 0 || c.dupAcks >= 3) { + // the system has something past sndUna, or says the same for the third time: the segment at sndUna is lost + c.congested() + c.lost = c.sndUna + min(uint32(c.mss), c.sndNxt-c.sndUna) + } + if c.recovering { + c.resend() + } + c.sndWnd = window + // the writer goes on when it can write a good part of the window, not for every segment that is acknowledged. + // Unless the way to the system is what holds it back: that takes what the acknowledgments say has left it, + // and what is sent as they come does not come to it all at once + limit := min(c.sndWnd, uint32(c.window)) + enough := limit / 2 + if c.cwnd < limit { + enough = uint32(c.mss) + } + if c.blocked && (c.sndUna == c.sndNxt || c.space() >= enough) { + c.writable.Broadcast() + } +} + +// clocked takes the time since timedAt as how long an acknowledgment takes. The next ones are waited for as long +// as that says and no less than a few ticks. Until there is such a time again, after something was sent again, +// they are waited for as long as the last time. c.mu is held +func (c *tcpConn) clocked() { + d := time.Since(c.timedAt) + c.timedAt = time.Time{} + if c.srtt == 0 { + c.srtt, c.rttvar = d, d/2 + } else { + c.rttvar += ((c.srtt - d).Abs() - c.rttvar) / 4 + c.srtt += (d - c.srtt) / 8 + } + c.retransmitTicks = max(tcpRetransmitTicks, int((c.srtt+4*c.rttvar)/tcpTick)+2) +} + +// sackAdd puts what the system says it has into sacked. c.mu is held +func (c *tcpConn) sackAdd(start, end uint32) { + i := 0 + for i < len(c.sacked) && int32(c.sacked[i][1]-start) < 0 { + i++ + } + j := i + for ; j < len(c.sacked) && int32(c.sacked[j][0]-end) <= 0; j++ { + if int32(c.sacked[j][0]-start) < 0 { + start = c.sacked[j][0] + } + if int32(c.sacked[j][1]-end) > 0 { + end = c.sacked[j][1] + } + } + if i == j && len(c.sacked) == tcpHeld { + return + } + c.sacked = slices.Replace(c.sacked, i, j, [2]uint32{start, end}) +} + +// congested is called when something is lost: the way to the system takes half of what it was given, +// and what is lost is sent again from sndUna. c.mu is held +func (c *tcpConn) congested() { + c.ssthresh = max(min(c.cwnd, c.sndNxt-c.sndUna)/2, 2*uint32(c.mss)) + c.cwnd = c.ssthresh + c.recovering, c.recover, c.resent = true, c.sndNxt, c.sndUna + c.timedAt = time.Time{} +} + +// hole returns the first part at or after seq of what is lost: what the system does not have +// before something that it says it has, or before lost. There is none if it ends where it starts. c.mu is held +func (c *tcpConn) hole(seq uint32) (uint32, uint32) { + end := c.lost + for _, b := range c.sacked { + if int32(b[0]-seq) > 0 { + end = b[0] + break + } + if int32(b[1]-seq) > 0 { + seq = b[1] + } + } + if int32(end-seq) < 0 { + end = seq + } + return seq, end +} + +// flight is how much of what was sent is on its way: not what the system says it has, +// and not what is lost and was not sent again. c.mu is held +func (c *tcpConn) flight() uint32 { + n := c.sndNxt - c.sndUna + if !c.recovering { + return n + } + seq := c.resent + for _, b := range c.sacked { + n -= b[1] - b[0] + if int32(b[0]-seq) > 0 { + n -= b[0] - seq + } + if int32(b[1]-seq) > 0 { + seq = b[1] + } + } + if int32(c.lost-seq) > 0 { + n -= c.lost - seq + } + return n +} + +// resend sends again what is lost and was not yet: the segment that the system waits for at once, +// the rest as the way to it takes it. c.mu is held +func (c *tcpConn) resend() { + for { + start, end := c.hole(c.resent) + if start == end { + return + } + room := uint32(c.mss) + if start != c.sndUna { + flight := c.flight() + if flight+room > c.cwnd { + return + } + room = c.cwnd - flight + } + end = start + min(end-start, room) + c.retransmit(start, end) + c.resent = end + } +} + +// space is how much can be sent now: what the system takes, and what the way to it takes. c.mu is held +func (c *tcpConn) space() uint32 { + window, sent, flight := min(c.sndWnd, uint32(c.window)), c.sndNxt-c.sndUna, c.flight() + if sent >= window || flight >= c.cwnd { + return 0 + } + return min(window-sent, c.cwnd-flight) +} + +// receive takes the payload of a segment. c.mu is held +func (c *tcpConn) receive(s segment) { + seq, payload, fin := s.seq, s.payload, s.flags&tcpFin != 0 + if len(payload) == 0 && !fin { + if int32(c.rcvNxt-seq) > 0 { + // a keepalive, or a probe of a window that the system has as closed: answered with what is expected + c.sendAck() + } + return + } + if before := int32(c.rcvNxt - seq); before > 0 { + if int(before) > len(payload) || int(before) == len(payload) && !fin { + // sent again, or a keepalive: answered with what is expected + c.sendAck() + return + } + seq, payload = c.rcvNxt, payload[before:] + } + if c.rcvFin { + c.sendAck() + return + } + if c.closed && len(payload) > 0 { + // nobody reads it anymore + c.reset() + return + } + // the system is told at once when something is wrong, or it sends more of the same + now := len(c.rcvHeld) > 0 + if room := int(c.window - c.rcvLen - int32(seq-c.rcvNxt)); len(payload) > room { + payload, fin, now = payload[:max(room, 0)], false, true + } + if seq != c.rcvNxt { + // something before it was lost, it waits for that, and so does the end if this is it + c.hold(seq, payload) + if fin { + c.rcvEnd, c.rcvEnds = seq+uint32(len(payload)), true + } + c.sendAck() + return + } + + c.rcvQueue = buf.MergeBytes(c.rcvQueue, payload) + c.rcvLen += int32(len(payload)) + c.rcvNxt += uint32(len(payload)) + for !fin && len(c.rcvHeld) > 0 && int32(c.rcvNxt-c.rcvHeld[0].seq) >= 0 { + h := c.rcvHeld[0] + if before := int32(c.rcvNxt - h.seq); before < h.data.Len() { + c.rcvQueue = buf.MergeBytes(c.rcvQueue, h.data.BytesFrom(before)) + c.rcvLen += h.data.Len() - before + c.rcvNxt += uint32(h.data.Len() - before) + } + h.data.Release() + c.rcvHeld = c.rcvHeld[1:] + } + if fin = fin || c.rcvEnds && c.rcvNxt == c.rcvEnd; fin { + c.rcvFin = true + c.rcvNxt++ + } + c.readable.Signal() + // the reader acknowledges what it takes, with one segment for all that came by then. + // It is not waited for when it does not read for long + if fin || now || c.rcvNxt-c.rcvAcked >= tcpWindow/4 { + c.sendAck() + } else { + c.arm(tcpTick) + } +} + +// release releases what of sndQueue is acknowledged. c.mu is held +func (c *tcpConn) release() { + for len(c.sndQueue) > c.writing { + end := c.sndBase + uint32(c.sndQueue[0].Len()) + if int32(c.sndUna-end) < 0 { + break + } + c.sndQueue[0].Release() + c.sndQueue[0] = nil + c.sndQueue = c.sndQueue[1:] + c.sndBase = end + } +} + +// hold keeps a segment that is not the next one. c.mu is held +func (c *tcpConn) hold(seq uint32, payload []byte) { + i := 0 + for i < len(c.rcvHeld) && int32(c.rcvHeld[i].seq-seq) < 0 { + i++ + } + // without what the one before and the one after have + if i > 0 { + if over := int32(c.rcvHeld[i-1].seq + uint32(c.rcvHeld[i-1].data.Len()) - seq); over > 0 { + if int(over) >= len(payload) { + return + } + seq, payload = seq+uint32(over), payload[over:] + } + } + if i < len(c.rcvHeld) { + if over := int32(seq + uint32(len(payload)) - c.rcvHeld[i].seq); over > 0 { + if int(over) >= len(payload) { + return + } + payload = payload[:len(payload)-int(over)] + } + } + if len(payload) == 0 || len(c.rcvHeld) == tcpHeld { + return + } + b := buf.NewWithSize(int32(len(payload))) + b.Write(payload) + c.rcvHeld = slices.Insert(c.rcvHeld, i, held{seq, b}) + c.rcvLast = seq +} + +// retransmit sends again what was sent from seq to end. c.mu is held +func (c *tcpConn) retransmit(seq, end uint32) { + fin := false + if int32(end-c.sndNxt) >= 0 { + end, fin = c.sndNxt, c.sndFin + if fin { + end-- + } + } + for int32(end-seq) > 0 { + var part [1][]byte + _, n := c.queued(part[:0], seq, min(c.mss, int(end-seq))) + if n == 0 { + break + } + c.send(seq, tcpAck, nil, part[0]) + seq += uint32(n) + } + if fin && seq == end { + c.send(end, tcpFin|tcpAck, nil, nil) + } +} + +// queued adds to parts what of sndQueue starts at seq, size bytes at most and as many parts as it has room for, +// and returns them with their size. c.mu is held +func (c *tcpConn) queued(parts [][]byte, seq uint32, size int) ([][]byte, int) { + skip, n := int32(seq-c.sndBase), 0 + for _, b := range c.sndQueue { + if skip >= b.Len() { + skip -= b.Len() + continue + } + if n == size || len(parts) == cap(parts) { + break + } + p := b.BytesFrom(skip) + skip = 0 + p = p[:min(len(p), size-n)] + parts = append(parts, p) + n += len(p) + } + return parts, n +} + +// arm makes tick run after d, unless it is going to run already. c.mu is held +func (c *tcpConn) arm(d time.Duration) { + if c.armed { + return + } + c.armed = true + if c.timer == nil { + c.timer = time.AfterFunc(d, c.tick) + } else { + c.timer.Reset(d) + } +} + +// tick does what was put off and what has to be done again +func (c *tcpConn) tick() { + c.mu.Lock() + defer c.mu.Unlock() + c.armed = false + if c.err != nil { + return + } + if c.rcvAcked != c.rcvNxt { + c.sendAck() + } + switch { + case c.sndUna != c.sndNxt: + if c.stalled++; c.stalled >= c.retransmitTicks { + if c.retransmits++; c.retransmits > tcpRetransmits || !c.established && c.retransmits > tcpSynRetransmits { + c.abort(errTCPTimeout) + return + } + c.stalled = 0 + c.retransmitTicks = min(c.retransmitTicks*2, tcpRetransmitTicksMost) + if !c.established { + c.sendSynAck() + c.timedAt = time.Time{} + } else { + // what the system does not say it has is lost, and the way to it is tried with a segment again + c.congested() + c.cwnd, c.lost = uint32(c.mss), c.sndNxt + c.resend() + } + } + c.arm(tcpTick) + case c.blocked: + // the segment that opens the window may have been lost, this one is answered with the window + if c.stalled++; c.stalled%tcpProbeTicks == 0 { + c.send(c.sndUna-1, tcpAck, nil, nil) + } + c.arm(tcpTick) + case c.closed: + if c.lingered { + c.reset() + return + } + c.lingered = true + c.arm(tcpLinger) + } +} + +// reset tells the system that the connection is gone. c.mu is held +func (c *tcpConn) reset() { + c.send(c.sndNxt, tcpRst|tcpAck, nil, nil) + c.abort(io.ErrClosedPipe) +} + +// abort ends the connection without another word to the system. What it has sent is still there to read. +// c.mu is held +func (c *tcpConn) abort(err error) { + if c.err != nil { + return + } + c.err = err + if c.timer != nil { + c.timer.Stop() + } + for _, h := range c.rcvHeld { + h.data.Release() + } + c.rcvHeld = nil + for _, b := range c.sndQueue[:len(c.sndQueue)-c.writing] { + b.Release() + } + c.sndQueue = nil + c.readable.Broadcast() + c.writable.Broadcast() + c.stack.remove(c) +} + +// wait waits for something to read. c.mu is held +func (c *tcpConn) wait() error { + for c.rcvLen == 0 && !c.rcvFin && !c.closed && c.err == nil { + c.readable.Wait() + } + switch { + case c.rcvLen > 0: + return nil + case c.err != nil && !c.rcvFin: + return c.err + } + return io.EOF +} + +// taken is called when size bytes have been read. c.mu is held +func (c *tcpConn) taken(size int32) { + c.rcvLen -= size + // the system may also have stopped at the window that it was told + if c.err == nil && (c.rcvAcked != c.rcvNxt || c.rcvAdvertised < c.window/2) { + c.sendAck() + } +} + +// ReadMultiBuffer implements buf.Reader +func (c *tcpConn) ReadMultiBuffer() (buf.MultiBuffer, error) { + c.mu.Lock() + defer c.mu.Unlock() + if err := c.wait(); err != nil { + return nil, err + } + mb := c.rcvQueue + c.rcvQueue = nil + c.taken(c.rcvLen) + return mb, nil +} + +// Read implements net.Conn +func (c *tcpConn) Read(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + if err := c.wait(); err != nil { + return 0, err + } + var n int + c.rcvQueue, n = buf.SplitBytes(c.rcvQueue, p) + c.taken(int32(n)) + return n, nil +} + +// WriteMultiBuffer implements buf.Writer. It is not called again before it returns +func (c *tcpConn) WriteMultiBuffer(mb buf.MultiBuffer) error { + c.mu.Lock() + defer c.mu.Unlock() + // without the empty ones + n, size := 0, uint32(0) + for _, b := range mb { + if b.IsEmpty() { + b.Release() + continue + } + mb[n] = b + n++ + size += uint32(b.Len()) + } + mb = mb[:n] + if c.err != nil || c.closed { + buf.ReleaseMulti(mb) + return io.ErrClosedPipe + } + // what is written is in sndQueue until it is acknowledged, and is not released before all of it is sent + c.sndQueue = append(c.sndQueue, mb...) + c.writing = len(mb) + // a segment is one part of a Buffer, or as much as there is to send if the tun splits it by itself + var room [tcpParts][]byte + limit, parts := c.mss, room[:0:1] + if c.stack.segmenter != nil { + limit, parts = tcpSegment, room[:0] + } + for end := c.sndNxt + size; c.sndNxt != end; { + // not a part of a segment when all of it can go later, with what is acknowledged by then + for c.err == nil { + if space := c.space(); space >= min(end-c.sndNxt, uint32(c.mss)) || space != 0 && c.sndUna == c.sndNxt { + break + } + c.blocked = true + c.arm(tcpTick) + c.writable.Wait() + c.blocked = false + } + if c.err != nil { + break + } + most := min(limit, int(c.space())) + if most > c.mss { + // the same for what the tun splits + most -= most % c.mss + } + p, n := c.queued(parts, c.sndNxt, most) + flags := byte(tcpAck) + if c.sndNxt+uint32(n) == end { + flags |= tcpPsh + } + var b [ipv6HeaderSize + tcpHeaderSize]byte + h := b[:c.header(b[:], c.sndNxt, flags, nil, n)] + if c.sndUna == c.sndNxt { + // the ticks without an acknowledgment are counted from the first segment that waits for one + c.stalled = 0 + } + c.sndNxt += uint32(n) + if c.timedAt.IsZero() { + c.timed, c.timedAt = c.sndNxt, time.Now() + } + c.arm(tcpTick) + // segments that come in the meantime are taken while this one is written, and acknowledged after it + c.sending = true + c.mu.Unlock() + if s := c.stack.segmenter; s != nil { + t := h[c.ipSize:] + binary.BigEndian.PutUint16(t[16:], ^checksum(pseudoSum(h[:c.ipSize], protoTCP, len(t)+n))) + _ = s.WriteSegment(h, p, c.mss) + } else { + c.write(h, p[0]) + } + c.mu.Lock() + c.sending = false + if c.ackWanted && c.err == nil { + c.sendAck() + } + c.ackWanted = false + } + c.writing = 0 + if c.err != nil { + buf.ReleaseMulti(mb) + return io.ErrClosedPipe + } + c.release() + return nil +} + +// Write implements net.Conn +func (c *tcpConn) Write(p []byte) (int, error) { + if err := c.WriteMultiBuffer(buf.MergeBytes(nil, p)); err != nil { + return 0, err + } + return len(p), nil +} + +// Close implements net.Conn +func (c *tcpConn) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return nil + } + c.closed = true + unread := c.rcvLen > 0 + c.rcvQueue, c.rcvLen = buf.ReleaseMulti(c.rcvQueue), 0 + if c.err != nil { + return nil + } + if unread || c.writing != 0 { + // something was not read or not written, the system has to know + c.reset() + return nil + } + c.readable.Broadcast() + c.send(c.sndNxt, tcpFin|tcpAck, nil, nil) + c.sndFin = true + c.sndNxt++ + c.arm(tcpTick) + return nil +} + +func (c *tcpConn) LocalAddr() net.Addr { + return c.dst.RawNetAddr() +} + +func (c *tcpConn) RemoteAddr() net.Addr { + return c.src.RawNetAddr() +} + +func (c *tcpConn) SetDeadline(time.Time) error { + return nil +} + +func (c *tcpConn) SetReadDeadline(time.Time) error { + return nil +} + +func (c *tcpConn) SetWriteDeadline(time.Time) error { + return nil +} diff --git a/proxy/tun/tun.go b/proxy/tun/tun.go index 48592b7de..62c4c7c9d 100644 --- a/proxy/tun/tun.go +++ b/proxy/tun/tun.go @@ -1,6 +1,7 @@ package tun -import "gvisor.dev/gvisor/pkg/tcpip/stack" +// packetRoom is what a tun may need in front of a packet in the buffer that it is read into +const packetRoom = 16 // Tun interface implements tun interface interaction type Tun interface { @@ -8,5 +9,18 @@ type Tun interface { Close() error Name() (string, error) Index() (int, error) - newEndpoint() (stack.LinkEndpoint, error) + // ReadPacket waits for a packet, reads it into b and returns it + ReadPacket(b []byte) ([]byte, error) + // WritePacket writes a packet that is in two parts, the second of which may be empty + WritePacket(header, payload []byte) error +} + +// segmenter is a tun whose other side takes TCP segments that are larger than the MTU, +// splits them only if it has to send them on, and completes their checksum then +type segmenter interface { + // Segments reports whether WriteSegment can be used + Segments() bool + // WriteSegment writes a packet with a TCP segment whose payload is in parts, to be sent on with mss bytes + // of it at most in one. Its checksum is that of the pseudo-header only, not inverted + WriteSegment(header []byte, payload [][]byte, mss int) error } diff --git a/proxy/tun/tun_android.go b/proxy/tun/tun_android.go index f287bb1ea..1017458f4 100644 --- a/proxy/tun/tun_android.go +++ b/proxy/tun/tun_android.go @@ -10,12 +10,10 @@ import ( "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/platform" "golang.org/x/sys/unix" - "gvisor.dev/gvisor/pkg/tcpip/link/fdbased" - "gvisor.dev/gvisor/pkg/tcpip/stack" ) type AndroidTun struct { - tunFd int + *fdTun options *Config } @@ -33,8 +31,14 @@ func NewTun(options *Config) (Tun, error) { return nil, err } + // the descriptor stays with the application, Close ends the reading of a duplicate + fd, err = unix.FcntlInt(uintptr(fd), unix.F_DUPFD_CLOEXEC, 0) + if err != nil { + return nil, err + } + return &AndroidTun{ - tunFd: fd, + fdTun: newFdTun(fd, 0), options: options, }, nil } @@ -44,7 +48,7 @@ func (t *AndroidTun) Start() error { } func (t *AndroidTun) Close() error { - return nil + return t.closeFile() } func (t *AndroidTun) Name() (string, error) { @@ -52,7 +56,7 @@ func (t *AndroidTun) Name() (string, error) { if err != nil { return "", err } - if err = unix.IoctlIfreq(t.tunFd, unix.TUNGETIFF, ifr); err != nil { + if err = unix.IoctlIfreq(t.fd, unix.TUNGETIFF, ifr); err != nil { return "", err } return ifr.Name(), nil @@ -70,14 +74,6 @@ func (t *AndroidTun) Index() (int, error) { return iface.Index, nil } -func (t *AndroidTun) newEndpoint() (stack.LinkEndpoint, error) { - return fdbased.New(&fdbased.Options{ - FDs: []int{t.tunFd}, - MTU: t.options.MTU, - RXChecksumOffload: true, - }) -} - func setinterface(network, address string, fd uintptr, iface *net.Interface) error { return unix.BindToDevice(int(fd), iface.Name) } diff --git a/proxy/tun/tun_darwin.go b/proxy/tun/tun_darwin.go index bcca838db..299720cbd 100644 --- a/proxy/tun/tun_darwin.go +++ b/proxy/tun/tun_darwin.go @@ -20,9 +20,6 @@ import ( "github.com/xtls/xray-core/common/platform" "golang.org/x/net/route" "golang.org/x/sys/unix" - "gvisor.dev/gvisor/pkg/buffer" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/stack" ) const ( @@ -47,8 +44,8 @@ type DarwinTun struct { ownsFd bool // true for macOS (we created the fd), false for iOS (fd from system) // Genuinely blocks Wait() until tunFd is readable, instead of the - // previous procyield-only busy-spin (dispatchLoop in - // stack_gvisor_endpoint.go calls ReadPacket() then Wait() in a tight + // previous procyield-only busy-spin (ReadPacket() reads and then calls + // Wait() in a tight // loop with no other throttling whenever the queue is empty -- with // only procyield(1), that pins a full CPU core for as long as the // tunnel is up, observed causing severe device heating/thermal @@ -66,7 +63,7 @@ type DarwinTun struct { // waitKqueue owns a kqueue fd used by DarwinTun.Wait() to block on // read-readiness. Closing and waiting can race from different goroutines // (Close() from the caller that tears down the tunnel, Wait() from -// dispatchLoop's own goroutine) -- reviewer feedback on XTLS/Xray-core#6580 +// the goroutine that reads packets) -- reviewer feedback on XTLS/Xray-core#6580 // found that a bare `int` fd field let Close() race Wait()'s use of the // same fd number, and on Darwin a closed fd number can be reused by an // unrelated concurrent open() before Wait() gets to call Kevent on it, @@ -110,9 +107,9 @@ func newWaitKqueue(fd int) *waitKqueue { // wait blocks until the registered fd is readable, timeout elapses, or a // benign interrupt occurs -- all three are "this kqueue is still healthy, // the caller should just try again" and return true; the caller -// (DarwinTun.Wait) doesn't need to distinguish them since it always calls -// ReadPacket() right after anyway, and that already handles "nothing was -// actually there" via ErrQueueEmpty. Returns false only when the kqueue +// (DarwinTun.Wait) doesn't need to distinguish them since ReadPacket() +// reads right after anyway, and that already handles "nothing was +// actually there" by waiting again. Returns false only when the kqueue // itself is no longer usable -- already closed, or the kevent syscall // failed for a reason other than EINTR -- see its own call site in // DarwinTun.Wait for why a persistent failure must not be silently @@ -145,10 +142,7 @@ func (w *waitKqueue) close() { }) } -var ( - _ Tun = (*DarwinTun)(nil) - _ GVisorDevice = (*DarwinTun)(nil) -) +var _ Tun = (*DarwinTun)(nil) func NewTun(options *Config) (Tun, error) { // Check if fd is provided via environment (iOS mode) @@ -274,73 +268,43 @@ func (t *DarwinTun) Index() (int, error) { return iface.Index, nil } -// WritePacket implements GVisorDevice method to write one packet to the tun device -func (t *DarwinTun) WritePacket(packet *stack.PacketBuffer) tcpip.Error { - // request memory to write from reusable buffer pool - b := buf.NewWithSize(int32(t.options.MTU) + utunHeaderSize) - defer b.Release() - - // prepare Darwin specific packet header - _, _ = b.Write([]byte{0x0, 0x0, 0x0, 0x0}) - // copy the bytes of slices that compose the packet into the allocated buffer - for _, packetElement := range packet.AsSlices() { - _, _ = b.Write(packetElement) - } - // fill Darwin specific header from the first raw packet byte, that we can access now +// WritePacket implements Tun: the two parts are copied behind the header that the system wants in front of a packet +func (t *DarwinTun) WritePacket(header, payload []byte) error { var family byte - switch b.Byte(4) >> 4 { + switch header[0] >> 4 { case 4: family = unix.AF_INET case 6: family = unix.AF_INET6 default: - return &tcpip.ErrAborted{} + return xerrors.New("not an ip packet") } - b.SetByte(3, family) - if _, err := t.tunFile.Write(b.Bytes()); err != nil { - if errors.Is(err, unix.EAGAIN) { - return &tcpip.ErrWouldBlock{} - } - return &tcpip.ErrAborted{} - } - return nil + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(utunHeaderSize + len(header) + len(payload))) + defer b.Release() + b.Extend(utunHeaderSize)[3] = family + _, _ = b.Write(header) + _, _ = b.Write(payload) + + _, err := t.tunFile.Write(b.Bytes()) + return err } -// ReadPacket implements GVisorDevice method to read one packet from the tun device -// It is expected that the method will not block, rather return ErrQueueEmpty when there is nothing on the line, -// which will make the stack call Wait which should implement desired push-back -func (t *DarwinTun) ReadPacket() (byte, *stack.PacketBuffer, error) { - // request memory to write from reusable buffer pool - b := buf.NewWithSize(int32(t.options.MTU) + utunHeaderSize) - - // read the bytes to the interface file - n, err := b.ReadFrom(t.tunFile) - if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) { - b.Release() - return 0, nil, ErrQueueEmpty +// ReadPacket implements Tun: it reads into b, the packet follows the header that the system puts in front of it +func (t *DarwinTun) ReadPacket(b []byte) ([]byte, error) { + for { + n, err := t.tunFile.Read(b) + if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) || err == nil && n <= utunHeaderSize { + // nothing on the line, or an empty packet + t.Wait() + continue + } + if err != nil { + return nil, err + } + return b[utunHeaderSize:n], nil } - if err != nil { - b.Release() - return 0, nil, err - } - - // discard empty or sub-empty packets - if n <= utunHeaderSize { - b.Release() - return 0, nil, ErrQueueEmpty - } - - // network protocol version from first byte of the raw packet, the one that follows Darwin specific header - version := b.Byte(utunHeaderSize) >> 4 - packetBuffer := buffer.MakeWithData(b.BytesFrom(utunHeaderSize)) - return version, stack.NewPacketBuffer(stack.PacketBufferOptions{ - Payload: packetBuffer, - IsForwardedPacket: true, - OnRelease: func() { - b.Release() - }, - }), nil } // Wait blocks until tunFd is readable (or a short timeout elapses), rather @@ -349,8 +313,8 @@ func (t *DarwinTun) ReadPacket() (byte, *stack.PacketBuffer, error) { // happens to race a call already parked here. // // Reviewer feedback (XTLS/Xray-core#6580, P2): the original version -// discarded every error from the underlying kevent syscall. dispatchLoop -// (stack_gvisor_endpoint.go) calls ReadPacket() then Wait() in an +// discarded every error from the underlying kevent syscall. ReadPacket() +// reads and then calls Wait() in an // unconditional tight loop -- if kevent started failing at runtime for a // persistent reason (not just a benign EINTR), Wait() returning // immediately every time reintroduces exactly the busy-spin this whole @@ -378,14 +342,10 @@ func (t *DarwinTun) Wait() { // real Darwin systems, or having just failed permanently above) // instead of always -- a genuine bounded sleep actually yields the CPU // instead of being a near-instant scheduler hint that lets the tight - // dispatchLoop caller spin just as hot as before. + // loop of ReadPacket() spin just as hot as before. time.Sleep(time.Millisecond) } -func (t *DarwinTun) newEndpoint() (stack.LinkEndpoint, error) { - return &LinkEndpoint{deviceMTU: t.options.MTU, device: t}, nil -} - // open the interface, by creating new utunN if in the system and returning its file descriptor func open(name string) (*os.File, error) { ifIndex := -1 diff --git a/proxy/tun/tun_default.go b/proxy/tun/tun_default.go index e26b7cf92..ea5f24060 100644 --- a/proxy/tun/tun_default.go +++ b/proxy/tun/tun_default.go @@ -6,7 +6,6 @@ import ( "net" "github.com/xtls/xray-core/common/errors" - "gvisor.dev/gvisor/pkg/tcpip/stack" ) type DefaultTun struct{} @@ -35,10 +34,14 @@ func (t *DefaultTun) Index() (int, error) { return 0, errors.New("Tun is not supported on your platform") } -func (t *DefaultTun) newEndpoint() (stack.LinkEndpoint, error) { +func (t *DefaultTun) ReadPacket([]byte) ([]byte, error) { return nil, errors.New("Tun is not supported on your platform") } +func (t *DefaultTun) WritePacket(header, payload []byte) error { + return errors.New("Tun is not supported on your platform") +} + func setinterface(string, string, uintptr, *net.Interface) error { return errors.New("Tun is not supported on your platform") } diff --git a/proxy/tun/tun_fd.go b/proxy/tun/tun_fd.go new file mode 100644 index 000000000..2df7d0b3b --- /dev/null +++ b/proxy/tun/tun_fd.go @@ -0,0 +1,96 @@ +//go:build linux + +package tun + +import ( + "encoding/binary" + "os" + "sync" + + "golang.org/x/sys/unix" +) + +// vnetSize is the size of the virtio_net_hdr that a tun with IFF_VNET_HDR has in front of every packet, +// unless it was told another one +const vnetSize = 10 + +// fdTun reads and writes the packets of a tun that is a file descriptor +type fdTun struct { + fd int + // file is what packets are read from: a read waits in the poller of the runtime and ends when file is closed + file *os.File + // vnet is the size of the virtio_net_hdr in front of every packet, zero if the tun has none. + // With it the system does not split what it sends to the tun and takes what is not split from it + vnet int + + writeLock sync.RWMutex + closed bool +} + +// newFdTun takes a file descriptor that does not block +func newFdTun(fd, vnet int) *fdTun { + return &fdTun{fd: fd, file: os.NewFile(uintptr(fd), "/dev/net/tun"), vnet: vnet} +} + +func (t *fdTun) ReadPacket(b []byte) ([]byte, error) { + n, err := t.file.Read(b) + // what the virtio_net_hdr says is of no use here: no checksum is looked at, and a segment may have any size + return b[min(t.vnet, n):n], err +} + +func (t *fdTun) WritePacket(header, payload []byte) error { + // a virtio_net_hdr of zeros leaves nothing to the system + var vnet [packetRoom]byte + return t.writev(vnet[:t.vnet], header, payload) +} + +// Segments implements segmenter +func (t *fdTun) Segments() bool { + return t.vnet != 0 +} + +// WriteSegment implements segmenter +func (t *fdTun) WriteSegment(header []byte, payload [][]byte, mss int) error { + // the virtio_net_hdr and the headers go as one part + var b [packetRoom + ipv6HeaderSize + 60]byte + ip, gso := ipv4HeaderSize, byte(unix.VIRTIO_NET_HDR_GSO_TCPV4) + if header[0]>>4 == 6 { + ip, gso = ipv6HeaderSize, unix.VIRTIO_NET_HDR_GSO_TCPV6 + } + size := 0 + for _, p := range payload { + size += len(p) + } + b[0] = unix.VIRTIO_NET_HDR_F_NEEDS_CSUM + if size > mss { + b[1] = gso + binary.NativeEndian.PutUint16(b[4:], uint16(mss)) + } + binary.NativeEndian.PutUint16(b[2:], uint16(len(header))) + // where the checksum starts, and where it is put from there + binary.NativeEndian.PutUint16(b[6:], uint16(ip)) + binary.NativeEndian.PutUint16(b[8:], 16) + var parts [8][]byte + return t.writev(append(append(parts[:0], b[:t.vnet+copy(b[t.vnet:], header)]), payload...)...) +} + +func (t *fdTun) writev(parts ...[]byte) error { + t.writeLock.RLock() + defer t.writeLock.RUnlock() + if t.closed { + return os.ErrClosed + } + for { + _, err := unix.Writev(t.fd, parts) + if err != unix.EINTR { + return err + } + } +} + +func (t *fdTun) closeFile() error { + t.writeLock.Lock() + defer t.writeLock.Unlock() + t.closed = true + return t.file.Close() +} diff --git a/proxy/tun/tun_freebsd.go b/proxy/tun/tun_freebsd.go index 078475031..90f5a5446 100644 --- a/proxy/tun/tun_freebsd.go +++ b/proxy/tun/tun_freebsd.go @@ -13,9 +13,6 @@ import ( "unsafe" "golang.zx2c4.com/wireguard/tun" - "gvisor.dev/gvisor/pkg/buffer" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/stack" "golang.org/x/net/route" "golang.org/x/sys/unix" @@ -59,10 +56,7 @@ type escapeRoute struct { gateway netip.Addr } -var ( - _ Tun = (*FreeBSDTun)(nil) - _ GVisorDevice = (*FreeBSDTun)(nil) -) +var _ Tun = (*FreeBSDTun)(nil) // NewTun builds new tun interface handler func NewTun(options *Config) (Tun, error) { @@ -242,73 +236,43 @@ func (t *FreeBSDTun) Index() (int, error) { return t.tunIndex, nil } -// WritePacket implements GVisorDevice method to write one packet to the tun device -func (t *FreeBSDTun) WritePacket(packet *stack.PacketBuffer) tcpip.Error { - // request memory to write from reusable buffer pool - b := buf.NewWithSize(int32(t.options.MTU) + tunHeaderSize) - defer b.Release() - - // prepare Unix specific packet header - _, _ = b.Write([]byte{0x0, 0x0, 0x0, 0x0}) - // copy the bytes of slices that compose the packet into the allocated buffer - for _, packetElement := range packet.AsSlices() { - _, _ = b.Write(packetElement) - } - // fill Unix specific header from the first raw packet byte, that we can access now +// WritePacket implements Tun: the two parts are copied behind the header that the system wants in front of a packet +func (t *FreeBSDTun) WritePacket(header, payload []byte) error { var family byte - switch b.Byte(4) >> 4 { + switch header[0] >> 4 { case 4: family = unix.AF_INET case 6: family = unix.AF_INET6 default: - return &tcpip.ErrAborted{} + return xerrors.New("not an ip packet") } - b.SetByte(3, family) - if _, err := t.device.File().Write(b.Bytes()); err != nil { - if errors.Is(err, unix.EAGAIN) { - return &tcpip.ErrWouldBlock{} - } - return &tcpip.ErrAborted{} - } - return nil + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(tunHeaderSize + len(header) + len(payload))) + defer b.Release() + b.Extend(tunHeaderSize)[3] = family + _, _ = b.Write(header) + _, _ = b.Write(payload) + + _, err := t.device.File().Write(b.Bytes()) + return err } -// ReadPacket implements GVisorDevice method to read one packet from the tun device -// It is expected that the method will not block, rather return ErrQueueEmpty when there is nothing on the line, -// which will make the stack call Wait which should implement desired push-back -func (t *FreeBSDTun) ReadPacket() (byte, *stack.PacketBuffer, error) { - // request memory to write from reusable buffer pool - b := buf.NewWithSize(int32(t.options.MTU) + tunHeaderSize) - - // read the bytes to the interface file - n, err := b.ReadFrom(t.device.File()) - if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) { - b.Release() - return 0, nil, ErrQueueEmpty +// ReadPacket implements Tun: it reads into b, the packet follows the header that the system puts in front of it +func (t *FreeBSDTun) ReadPacket(b []byte) ([]byte, error) { + for { + n, err := t.device.File().Read(b) + if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) || err == nil && n <= tunHeaderSize { + // nothing on the line, or an empty packet + t.Wait() + continue + } + if err != nil { + return nil, err + } + return b[tunHeaderSize:n], nil } - if err != nil { - b.Release() - return 0, nil, err - } - - // discard empty or sub-empty packets - if n <= tunHeaderSize { - b.Release() - return 0, nil, ErrQueueEmpty - } - - // network protocol version from first byte of the raw packet, the one that follows Unix specific header - version := b.Byte(tunHeaderSize) >> 4 - packetBuffer := buffer.MakeWithData(b.BytesFrom(tunHeaderSize)) - return version, stack.NewPacketBuffer(stack.PacketBufferOptions{ - Payload: packetBuffer, - IsForwardedPacket: true, - OnRelease: func() { - b.Release() - }, - }), nil } // Wait some cpu cycles @@ -316,10 +280,6 @@ func (t *FreeBSDTun) Wait() { procyield(1) } -func (t *FreeBSDTun) newEndpoint() (stack.LinkEndpoint, error) { - return &LinkEndpoint{deviceMTU: t.options.MTU, device: t}, nil -} - const ( IN6_IFF_NODAD = 0x0020 // netinet6/in6_var.h ND6_INFINITE_LIFETIME = 0xFFFFFFFF // netinet6/nd6.h diff --git a/proxy/tun/tun_linux.go b/proxy/tun/tun_linux.go index 78e2dd400..16dc02841 100644 --- a/proxy/tun/tun_linux.go +++ b/proxy/tun/tun_linux.go @@ -25,15 +25,13 @@ import ( routingsession "github.com/xtls/xray-core/features/routing/session" "github.com/xtls/xray-core/proxy/dns" "golang.org/x/sys/unix" - "gvisor.dev/gvisor/pkg/tcpip/link/fdbased" - "gvisor.dev/gvisor/pkg/tcpip/stack" ) // LinuxTun is an object that handles tun network interface on linux // current version is heavily stripped to do nothing more, -// then create a network interface, to be provided as file descriptor to gVisor ip stack +// then create a network interface, whose packets are read and written through its file descriptor type LinuxTun struct { - tunFd int + *fdTun tunLink netlink.Link options *Config ownsTun bool @@ -287,19 +285,30 @@ var _ Tun = (*LinuxTun)(nil) // NewTun builds new tun interface handler (linux specific) func NewTun(options *Config) (Tun, error) { - tunFd, tunLink, fdProvided, err := openFromEnv(options.Name) + tunFd, vnet, tunLink, fdProvided, err := openFromEnv(options.Name) if err != nil { return nil, err } if fdProvided { return &LinuxTun{ - tunFd: tunFd, + fdTun: newFdTun(tunFd, vnet), tunLink: tunLink, options: options, }, nil } - tunFd, err = open(options.Name) + // with a virtio_net_hdr on every packet the system leaves its TCP segments as large as they are, + // with their checksums not completed, and takes such segments. Without it where that is refused + vnet = vnetSize + tunFd, err = open(options.Name, unix.IFF_VNET_HDR) + if err == nil && unix.IoctlSetInt(tunFd, unix.TUNSETOFFLOAD, unix.TUN_F_CSUM|unix.TUN_F_TSO4|unix.TUN_F_TSO6) != nil { + _ = unix.Close(tunFd) + err = unix.EINVAL + } + if err != nil { + vnet = 0 + tunFd, err = open(options.Name, 0) + } if err != nil { return nil, err } @@ -311,7 +320,7 @@ func NewTun(options *Config) (Tun, error) { } linuxTun := &LinuxTun{ - tunFd: tunFd, + fdTun: newFdTun(tunFd, vnet), tunLink: tunLink, options: options, ownsTun: true, @@ -320,55 +329,62 @@ func NewTun(options *Config) (Tun, error) { return linuxTun, nil } -func openFromEnv(expectedName string) (int, netlink.Link, bool, error) { +// openFromEnv takes the tun that was given to the process, with the size of the virtio_net_hdr of its packets +func openFromEnv(expectedName string) (int, int, netlink.Link, bool, error) { fdStr := platform.NewEnvFlag(platform.TunFdKey).GetValue(func() string { return "" }) if fdStr == "" { - return -1, nil, false, nil + return -1, 0, nil, false, nil } fd, err := strconv.Atoi(fdStr) if err != nil { - return -1, nil, true, errors.New("invalid ", platform.TunFdKey).Base(err) + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey).Base(err) } if fd < 3 { - return -1, nil, true, errors.New("invalid ", platform.TunFdKey, ": file descriptor must be >= 3") + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey, ": file descriptor must be >= 3") } ifr, err := unix.NewIfreq("") if err != nil { - return -1, nil, true, err + return -1, 0, nil, true, err } if err = unix.IoctlIfreq(fd, unix.TUNGETIFF, ifr); err != nil { - return -1, nil, true, err + return -1, 0, nil, true, err } flags := ifr.Uint16() if flags&unix.IFF_TUN == 0 { - return -1, nil, true, errors.New("invalid ", platform.TunFdKey, ": file descriptor is not a TUN device") + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey, ": file descriptor is not a TUN device") } if flags&unix.IFF_NO_PI == 0 { - return -1, nil, true, errors.New("invalid ", platform.TunFdKey, ": TUN device must use IFF_NO_PI") + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey, ": TUN device must use IFF_NO_PI") + } + vnet := 0 + if flags&unix.IFF_VNET_HDR != 0 { + if vnet, err = unix.IoctlGetInt(fd, unix.TUNGETVNETHDRSZ); err != nil || vnet < vnetSize || vnet > packetRoom { + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey, ": virtio_net_hdr of ", vnet, " bytes").Base(err) + } } actualName := ifr.Name() if expectedName != "" && actualName != expectedName { - return -1, nil, true, errors.New("invalid ", platform.TunFdKey, ": TUN device name ", actualName, " does not match configured name ", expectedName) + return -1, 0, nil, true, errors.New("invalid ", platform.TunFdKey, ": TUN device name ", actualName, " does not match configured name ", expectedName) } tunLink, err := netlink.LinkByName(actualName) if err != nil { - return -1, nil, true, err + return -1, 0, nil, true, err } if err = unix.SetNonblock(fd, true); err != nil { - return -1, nil, true, err + return -1, 0, nil, true, err } - return fd, tunLink, true, nil + return fd, vnet, tunLink, true, nil } // open the file that implements tun interface in the OS -func open(name string) (int, error) { +func open(name string, features int) (int, error) { fd, err := unix.Open("/dev/net/tun", unix.O_RDWR, 0) if err != nil { return -1, err @@ -380,7 +396,7 @@ func open(name string) (int, error) { return 0, err } - flags := unix.IFF_TUN | unix.IFF_NO_PI + flags := unix.IFF_TUN | unix.IFF_NO_PI | features ifr.SetUint16(uint16(flags)) err = unix.IoctlIfreq(fd, unix.TUNSETIFF, ifr) if err != nil { @@ -457,7 +473,7 @@ func (t *LinuxTun) Close() error { if t.ownsTun { _ = netlink.LinkSetDown(t.tunLink) } - _ = unix.Close(t.tunFd) + _ = t.closeFile() return nil } @@ -470,15 +486,6 @@ func (t *LinuxTun) Index() (int, error) { return t.tunLink.Attrs().Index, nil } -// newEndpoint builds new gVisor stack.LinkEndpoint from the tun interface file descriptor -func (t *LinuxTun) newEndpoint() (stack.LinkEndpoint, error) { - return fdbased.New(&fdbased.Options{ - FDs: []int{t.tunFd}, - MTU: t.options.MTU, - RXChecksumOffload: true, - }) -} - func setinterface(network, address string, fd uintptr, iface *net.Interface) error { return unix.BindToDevice(int(fd), iface.Name) } diff --git a/proxy/tun/tun_windows.go b/proxy/tun/tun_windows.go index 2045efa7b..e0bf9f1b9 100644 --- a/proxy/tun/tun_windows.go +++ b/proxy/tun/tun_windows.go @@ -25,9 +25,6 @@ import ( "golang.org/x/sys/windows" "golang.zx2c4.com/wintun" "golang.zx2c4.com/wireguard/windows/tunnel/winipcfg" - "gvisor.dev/gvisor/pkg/buffer" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/stack" ) //go:linkname procyield runtime.procyield @@ -35,7 +32,7 @@ func procyield(cycles uint32) // WindowsTun is an object that handles tun network interface on Windows // current version is heavily stripped to do nothing more, -// then create a network interface, to be provided as endpoint to gVisor ip stack +// then create a network interface, whose packets are read and written through a Wintun session type WindowsTun struct { sync.RWMutex @@ -57,9 +54,6 @@ type WindowsTun struct { // WindowsTun implements Tun var _ Tun = (*WindowsTun)(nil) -// WindowsTun implements GVisorDevice -var _ GVisorDevice = (*WindowsTun)(nil) - // NewTun creates a Wintun interface with the given name. Should a Wintun // interface with the same name exist, it tried to be reused. func NewTun(options *Config) (Tun, error) { @@ -368,6 +362,8 @@ func (t *WindowsTun) Close() error { flushDNSCache() } if t.session != (wintun.Session{}) { + // ReadPacket may be waiting for a packet + _ = windows.SetEvent(t.readWait) t.session.End() } if t.adapter != nil { @@ -507,62 +503,46 @@ func (t *WindowsTun) Index() (int, error) { return int(row.InterfaceIndex), nil } -// WritePacket implements GVisorDevice method to write one packet to the tun device -func (t *WindowsTun) WritePacket(packetBuffer *stack.PacketBuffer) tcpip.Error { +// WritePacket implements Tun: the two parts are copied into the ring of Wintun +func (t *WindowsTun) WritePacket(header, payload []byte) error { t.RLock() defer t.RUnlock() if t.closed { - return &tcpip.ErrClosedForSend{} + return windows.ERROR_HANDLE_EOF } - // request buffer from Wintun - packet, err := t.session.AllocateSendPacket(packetBuffer.Size()) + packet, err := t.session.AllocateSendPacket(len(header) + len(payload)) if err != nil { - return &tcpip.ErrAborted{} + return err } - - // copy the bytes of slices that compose the packet into the allocated buffer - var index int - for _, packetElement := range packetBuffer.AsSlices() { - index += copy(packet[index:], packetElement) - } - - // signal Wintun to send that buffer as the packet + copy(packet[copy(packet, header):], payload) t.session.SendPacket(packet) return nil } -// ReadPacket implements GVisorDevice method to read one packet from the tun device -// It is expected that the method will not block, rather return ErrQueueEmpty when there is nothing on the line, -// which will make the stack call Wait which should implement desired push-back -func (t *WindowsTun) ReadPacket() (byte, *stack.PacketBuffer, error) { - packet, err := t.session.ReceivePacket() - if go_errors.Is(err, windows.ERROR_NO_MORE_ITEMS) { - return 0, nil, ErrQueueEmpty - } - if err != nil { - return 0, nil, err - } - - version := packet[0] >> 4 - packetBuffer := buffer.MakeWithView(buffer.NewViewWithData(packet)) - return version, stack.NewPacketBuffer(stack.PacketBufferOptions{ - Payload: packetBuffer, - IsForwardedPacket: true, - OnRelease: func() { +// ReadPacket implements Tun: it waits until Wintun has a packet and copies it out of the ring +func (t *WindowsTun) ReadPacket(b []byte) ([]byte, error) { + for { + t.RLock() + if t.closed { + t.RUnlock() + return nil, windows.ERROR_HANDLE_EOF + } + packet, err := t.session.ReceivePacket() + if err == nil { + n := copy(b, packet) t.session.ReleaseReceivePacket(packet) - }, - }), nil -} - -func (t *WindowsTun) Wait() { - procyield(1) - _, _ = windows.WaitForSingleObject(t.readWait, windows.INFINITE) -} - -func (t *WindowsTun) newEndpoint() (stack.LinkEndpoint, error) { - return &LinkEndpoint{deviceMTU: t.options.MTU, device: t}, nil + t.RUnlock() + return b[:n], nil + } + t.RUnlock() + if !go_errors.Is(err, windows.ERROR_NO_MORE_ITEMS) { + return nil, err + } + procyield(1) + _, _ = windows.WaitForSingleObject(t.readWait, windows.INFINITE) + } } const ( diff --git a/proxy/tun/udp_fullcone.go b/proxy/tun/udp_fullcone.go index 480bb4bb7..82f115b37 100644 --- a/proxy/tun/udp_fullcone.go +++ b/proxy/tun/udp_fullcone.go @@ -11,11 +11,6 @@ import ( "github.com/xtls/xray-core/common/net" ) -type packet struct { - data []byte - dest *net.Destination -} - // sub-handler specifically for udp connections under main handler type udpConnectionHandler struct { sync.RWMutex @@ -38,17 +33,17 @@ func newUdpConnectionHandler(handleConnection func(conn net.Conn, dest net.Desti // HandlePacket handles UDP packets coming from tun, to forward to the dispatcher // this custom handler support FullCone NAT of returning packets, binding connection only by the source addr:port -func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destination, data []byte) { +func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destination, data *buf.Buffer) { + data.UDP = &dst + u.RLock() conn, found := u.udpConns[src] if found { select { - case conn.egress <- &packet{ - data: data, - dest: &dst, - }: + case conn.egress <- data: default: - errors.LogDebug(context.Background(), "drop udp with size ", len(data), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full") + errors.LogDebug(context.Background(), "drop udp with size ", data.Len(), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full") + data.Release() } u.RUnlock() return @@ -60,7 +55,7 @@ func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destina conn, found = u.udpConns[src] if !found { - egress := make(chan *packet, 1024) + egress := make(chan *buf.Buffer, 1024) conn = &udpConn{handler: u, egress: egress, src: src, dst: dst} u.udpConns[src] = conn @@ -69,12 +64,10 @@ func (u *udpConnectionHandler) HandlePacket(src net.Destination, dst net.Destina // send packet data to the egress channel, if it has buffer, or discard select { - case conn.egress <- &packet{ - data: data, - dest: &dst, - }: + case conn.egress <- data: default: - errors.LogDebug(context.Background(), "drop udp with size ", len(data), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full 2") + errors.LogDebug(context.Background(), "drop udp with size ", data.Len(), " to ", dst.NetAddr(), " original ", conn.dst.NetAddr(), " > queue full 2") + data.Release() } } @@ -92,38 +85,28 @@ func (u *udpConnectionHandler) connectionFinished(conn *udpConn) { type udpConn struct { handler *udpConnectionHandler - egress chan *packet + egress chan *buf.Buffer src net.Destination dst net.Destination } func (c *udpConn) ReadMultiBuffer() (buf.MultiBuffer, error) { - for { - e, ok := <-c.egress - if !ok { - return nil, io.EOF - } - - b := buf.New() - if _, err := b.Write(e.data); err != nil { - errors.LogErrorInner(context.Background(), err, "drop packet to ", e.dest, " with size ", len(e.data)) - b.Release() - continue - } - b.UDP = e.dest - - return buf.MultiBuffer{b}, nil + b, ok := <-c.egress + if !ok { + return nil, io.EOF } + return buf.MultiBuffer{b}, nil } // Read packets from the connection func (c *udpConn) Read(p []byte) (int, error) { - e, ok := <-c.egress + b, ok := <-c.egress if !ok { return 0, io.EOF } - n := copy(p, e.data) - if n != len(e.data) { + defer b.Release() + n := copy(p, b.Bytes()) + if n != int(b.Len()) { return 0, io.ErrShortBuffer } return n, nil