From 05921d175537631d5ae23808fb61ec364fa73dac Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 21 Sep 2026 02:37:52 +0000 Subject: [PATCH] Implement lean "system" TCP/IP stack backend for TUN inbound Co-authored-by: RPRX <63339210+RPRX@users.noreply.github.com> --- infra/conf/tun.go | 7 + proxy/tun/config.pb.go | 13 +- proxy/tun/config.proto | 1 + proxy/tun/handler.go | 2 + proxy/tun/stack.go | 37 ++ proxy/tun/stack_gvisor.go | 4 +- proxy/tun/stack_system.go | 367 +++++++++++++++++ proxy/tun/stack_system_tcp.go | 725 ++++++++++++++++++++++++++++++++++ proxy/tun/tun_android.go | 85 ++++ proxy/tun/tun_linux.go | 85 ++++ 10 files changed, 1322 insertions(+), 4 deletions(-) create mode 100644 proxy/tun/stack_system.go create mode 100644 proxy/tun/stack_system_tcp.go diff --git a/infra/conf/tun.go b/infra/conf/tun.go index 73e71a991..1afe6c28d 100644 --- a/infra/conf/tun.go +++ b/infra/conf/tun.go @@ -20,6 +20,7 @@ type TunConfig struct { UserLevel uint32 `json:"userLevel"` AutoSystemRoutingTable []string `json:"autoSystemRoutingTable"` AutoOutboundsInterface *string `json:"autoOutboundsInterface"` + Stack string `json:"stack"` } func (v *TunConfig) Build() (proto.Message, error) { @@ -31,6 +32,7 @@ func (v *TunConfig) Build() (proto.Message, error) { DNS: v.DNS, UserLevel: v.UserLevel, AutoSystemRoutingTable: v.AutoSystemRoutingTable, + Stack: v.Stack, } if v.AutoOutboundsInterface != nil { config.AutoOutboundsInterface = *v.AutoOutboundsInterface @@ -52,6 +54,11 @@ func (v *TunConfig) Build() (proto.Message, error) { if config.MTU == 0 { config.MTU = 1500 } + switch config.Stack { + case "", "gvisor", "system": + default: + return nil, fmt.Errorf("unknown tun stack: %s (must be \"gvisor\" or \"system\")", config.Stack) + } return config, nil } diff --git a/proxy/tun/config.pb.go b/proxy/tun/config.pb.go index 33bc2ba6f..ba7a13d40 100644 --- a/proxy/tun/config.pb.go +++ b/proxy/tun/config.pb.go @@ -32,6 +32,7 @@ type Config struct { AutoSystemRoutingTable []string `protobuf:"bytes,6,rep,name=auto_system_routing_table,json=autoSystemRoutingTable,proto3" json:"auto_system_routing_table,omitempty"` AutoOutboundsInterface string `protobuf:"bytes,7,opt,name=auto_outbounds_interface,json=autoOutboundsInterface,proto3" json:"auto_outbounds_interface,omitempty"` Desc string `protobuf:"bytes,8,opt,name=desc,proto3" json:"desc,omitempty"` + Stack string `protobuf:"bytes,9,opt,name=stack,proto3" json:"stack,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -122,11 +123,18 @@ func (x *Config) GetDesc() string { return "" } +func (x *Config) GetStack() string { + if x != nil { + return x.Stack + } + return "" +} + var File_proxy_tun_config_proto protoreflect.FileDescriptor const file_proxy_tun_config_proto_rawDesc = "" + "\n" + - "\x16proxy/tun/config.proto\x12\x0exray.proxy.tun\"\x82\x02\n" + + "\x16proxy/tun/config.proto\x12\x0exray.proxy.tun\"\x98\x02\n" + "\x06Config\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x10\n" + "\x03MTU\x18\x02 \x01(\rR\x03MTU\x12\x18\n" + @@ -136,7 +144,8 @@ const file_proxy_tun_config_proto_rawDesc = "" + "user_level\x18\x05 \x01(\rR\tuserLevel\x129\n" + "\x19auto_system_routing_table\x18\x06 \x03(\tR\x16autoSystemRoutingTable\x128\n" + "\x18auto_outbounds_interface\x18\a \x01(\tR\x16autoOutboundsInterface\x12\x12\n" + - "\x04desc\x18\b \x01(\tR\x04descBL\n" + + "\x04desc\x18\b \x01(\tR\x04desc\x12\x14\n" + + "\x05stack\x18\t \x01(\tR\x05stackBL\n" + "\x12com.xray.proxy.tunP\x01Z#github.com/xtls/xray-core/proxy/tun\xaa\x02\x0eXray.Proxy.Tunb\x06proto3" var ( diff --git a/proxy/tun/config.proto b/proxy/tun/config.proto index 376ac5af4..0a1dc7e08 100644 --- a/proxy/tun/config.proto +++ b/proxy/tun/config.proto @@ -15,4 +15,5 @@ message Config { repeated string auto_system_routing_table = 6; string auto_outbounds_interface = 7; string desc = 8; + string stack = 9; } diff --git a/proxy/tun/handler.go b/proxy/tun/handler.go index 53c74a5e3..c3064f7c9 100644 --- a/proxy/tun/handler.go +++ b/proxy/tun/handler.go @@ -143,7 +143,9 @@ func (t *Handler) Start() error { tunStackOptions := StackOptions{ Tun: tunInterface, + MTU: t.config.MTU, IdleTimeout: t.policyManager.ForLevel(t.config.UserLevel).Timeouts.ConnectionIdle, + Backend: t.config.Stack, } tunStack, err := NewStack(t.ctx, tunStackOptions, t) if err != nil { diff --git a/proxy/tun/stack.go b/proxy/tun/stack.go index ba67a4885..5ddf5a79f 100644 --- a/proxy/tun/stack.go +++ b/proxy/tun/stack.go @@ -1,7 +1,10 @@ package tun import ( + "context" "time" + + "github.com/xtls/xray-core/common/errors" ) // Stack interface implement ip protocol stack, bridging raw network packets and data streams @@ -13,5 +16,39 @@ type Stack interface { // StackOptions for the stack implementation type StackOptions struct { Tun Tun + MTU uint32 IdleTimeout time.Duration + // Backend selects the concrete Stack implementation, see NewStack. + Backend string +} + +const ( + // StackGVisor selects the full-featured gVisor based stack (default). + StackGVisor = "gvisor" + // StackSystem selects the lightweight, Xray-native stack, see newSystemStack. + StackSystem = "system" +) + +// NewStack builds the ip stack selected by options.Backend. +// +// gVisor (the default/"gvisor" backend) is a general purpose stack, built +// with the semantics needed for a real, lossy public network in mind: +// congestion control, SACK/RACK loss recovery, retransmission timers, etc. +// TUN traffic instead travels over a local, kernel-to-userspace channel that +// neither reorders nor drops packets in normal operation, so none of that +// complexity is actually required to shuffle bytes between it and the +// dispatcher. The "system" backend trades gVisor's generality for a much +// smaller, more direct code path tailored to that trusted, in-order channel: +// no congestion control, no SACK/RACK, minimal buffering, and a plain RTO +// timer as a safety net for the rare real loss, rather than a full +// re-implementation of one. See stack_system.go for details. +func NewStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { + switch options.Backend { + case "", StackGVisor: + return newGVisorStack(ctx, options, handler) + case StackSystem: + return newSystemStack(ctx, options, handler) + default: + return nil, errors.New("unknown tun stack: ", options.Backend) + } } diff --git a/proxy/tun/stack_gvisor.go b/proxy/tun/stack_gvisor.go index 8584616e9..f04518704 100644 --- a/proxy/tun/stack_gvisor.go +++ b/proxy/tun/stack_gvisor.go @@ -42,8 +42,8 @@ type stackGVisor struct { endpoint stack.LinkEndpoint } -// NewStack builds new ip stack (using gVisor) -func NewStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { +// newGVisorStack builds new ip stack (using gVisor) +func newGVisorStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { gStack := &stackGVisor{ ctx: ctx, tun: options.Tun, diff --git a/proxy/tun/stack_system.go b/proxy/tun/stack_system.go new file mode 100644 index 000000000..221185d1f --- /dev/null +++ b/proxy/tun/stack_system.go @@ -0,0 +1,367 @@ +package tun + +import ( + "context" + "crypto/rand" + "encoding/binary" + "errors" + "sync" + "time" + + xerrors "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/net" + tunicmp "github.com/xtls/xray-core/proxy/tun/icmp" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/seqnum" + "gvisor.dev/gvisor/pkg/tcpip/stack" +) + +// stackSystem is the lightweight, Xray-native ip stack, see NewStack. +// +// It reads and parses IPv4/IPv6 packets directly off the tun device (through +// the GVisorDevice interface, already implemented for every supported +// platform), without involving gVisor's stack.Stack, NIC or routing +// machinery. UDP and ICMP echo reuse the exact same handlers as the gVisor +// backend (udpConnectionHandler, tun/icmp) since those were already +// implemented in terms of raw bytes. TCP is handled by a small dedicated +// state machine, see stack_system_tcp.go. +type stackSystem struct { + ctx context.Context + device GVisorDevice + mtu uint32 + idleTimeout time.Duration + handler *Handler + + udp *udpConnectionHandler + + tcpMu sync.Mutex + tcp map[tcpKey]*tcpConn + + cancel context.CancelFunc +} + +const systemStackDefaultMTU = 1500 + +// newSystemStack builds the lightweight "system" ip stack, see NewStack. +func newSystemStack(ctx context.Context, options StackOptions, handler *Handler) (Stack, error) { + device, ok := options.Tun.(GVisorDevice) + if !ok { + return nil, xerrors.New("tun stack \"system\" is not supported by this tun device") + } + mtu := options.MTU + if mtu == 0 { + mtu = systemStackDefaultMTU + } + return &stackSystem{ + ctx: ctx, + device: device, + mtu: mtu, + idleTimeout: options.IdleTimeout, + handler: handler, + tcp: make(map[tcpKey]*tcpConn), + }, nil +} + +// Start is called by Handler to bring the stack to life +func (s *stackSystem) Start() error { + ctx, cancel := context.WithCancel(s.ctx) + s.cancel = cancel + s.udp = newUdpConnectionHandler(s.handler.HandleConnection, s.writeRawUDPPacket) + + go s.dispatchLoop(ctx) + go s.idleReapLoop(ctx) + return nil +} + +// Close is called by Handler to shut down the stack +func (s *stackSystem) Close() error { + if s.cancel != nil { + s.cancel() + } + + s.tcpMu.Lock() + conns := make([]*tcpConn, 0, len(s.tcp)) + for _, c := range s.tcp { + conns = append(conns, c) + } + s.tcp = make(map[tcpKey]*tcpConn) + s.tcpMu.Unlock() + + for _, c := range conns { + c.abort(errStackClosed) + } + + return nil +} + +// dispatchLoop reads and demultiplexes packets off the tun device, until ctx +// is cancelled or the device fails permanently. It mirrors LinkEndpoint's own +// dispatchLoop (stack_gvisor_endpoint.go), reusing the exact same GVisorDevice +// contract, but hands packets to this file's own IPv4/IPv6 parsing instead of +// gVisor's NIC/stack.Stack. +func (s *stackSystem) dispatchLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + } + + version, packet, err := s.device.ReadPacket() + if err != nil { + if errors.Is(err, ErrQueueEmpty) { + s.device.Wait() + continue + } + return + } + + s.handlePacket(version, packet) + packet.DecRef() + } +} + +func (s *stackSystem) handlePacket(version byte, packet *stack.PacketBuffer) { + data := concatSlices(packet.AsSlices()) + if len(data) == 0 { + return + } + + switch version { + case 4: + s.handleIPv4(data) + case 6: + s.handleIPv6(data) + } +} + +func concatSlices(slices [][]byte) []byte { + if len(slices) == 1 { + return slices[0] + } + total := 0 + for _, sl := range slices { + total += len(sl) + } + if total == 0 { + return nil + } + data := make([]byte, 0, total) + for _, sl := range slices { + data = append(data, sl...) + } + return data +} + +func (s *stackSystem) handleIPv4(data []byte) { + hdr := header.IPv4(data) + if !hdr.IsValid(len(data)) { + return + } + // fragmentation is not supported: the tun MTU is expected to keep locally + // generated packets from ever needing it, same as the gVisor backend's + // default configuration + if hdr.More() || hdr.FragmentOffset() != 0 { + return + } + + s.handleTransport(header.IPv4ProtocolNumber, hdr.TransportProtocol(), hdr.SourceAddress(), hdr.DestinationAddress(), hdr.Payload()) +} + +func (s *stackSystem) handleIPv6(data []byte) { + hdr := header.IPv6(data) + if !hdr.IsValid(len(data)) { + return + } + + // only directly-encapsulated transport headers are handled, IPv6 + // extension headers (rare for ordinary locally generated traffic) are not + // walked, same limitation as the fragmentation one above + s.handleTransport(header.IPv6ProtocolNumber, hdr.TransportProtocol(), hdr.SourceAddress(), hdr.DestinationAddress(), hdr.Payload()) +} + +func (s *stackSystem) handleTransport(netProto tcpip.NetworkProtocolNumber, transProto tcpip.TransportProtocolNumber, srcIP, dstIP tcpip.Address, payload []byte) { + switch transProto { + case header.TCPProtocolNumber: + s.handleTCP(netProto, srcIP, dstIP, payload) + case header.UDPProtocolNumber: + s.handleUDP(netProto, srcIP, dstIP, payload) + case header.ICMPv4ProtocolNumber: + if netProto == header.IPv4ProtocolNumber { + s.handleICMP(netProto, srcIP, dstIP, payload) + } + case header.ICMPv6ProtocolNumber: + if netProto == header.IPv6ProtocolNumber { + s.handleICMP(netProto, srcIP, dstIP, payload) + } + } +} + +func (s *stackSystem) handleUDP(netProto tcpip.NetworkProtocolNumber, srcIP, dstIP tcpip.Address, payload []byte) { + if len(payload) < header.UDPMinimumSize { + return + } + udpHdr := header.UDP(payload) + length := udpHdr.Length() + if int(length) < header.UDPMinimumSize || int(length) > len(payload) { + return + } + + // source/destination of the packet we process as incoming are, in other terms, + // src is the side behind tun, dst is the side behind the dispatcher + src := net.UDPDestination(net.IPAddress(srcIP.AsSlice()), net.Port(udpHdr.SourcePort())) + dst := net.UDPDestination(net.IPAddress(dstIP.AsSlice()), net.Port(udpHdr.DestinationPort())) + s.udp.HandlePacket(src, dst, payload[header.UDPMinimumSize:length]) +} + +func (s *stackSystem) handleICMP(netProto tcpip.NetworkProtocolNumber, srcIP, dstIP tcpip.Address, message []byte) { + ident, sequence, ok := tunicmp.ParseEchoRequest(netProto, message) + if !ok { + return + } + + reply, err := tunicmp.BuildLocalEchoReply(netProto, message, dstIP, srcIP) + if err != nil { + xerrors.LogInfoInner(s.ctx, err, "[tun] failed to build local icmp echo reply") + return + } + + xerrors.LogDebug(s.ctx, "[tun][icmp] ", tunicmp.ProtocolLabel(netProto), " local echo reply ", dstIP, " -> ", srcIP, " id=", ident, " seq=", sequence) + + transProto := header.ICMPv4ProtocolNumber + if netProto == header.IPv6ProtocolNumber { + transProto = header.ICMPv6ProtocolNumber + } + if err := s.writeTransportSegment(netProto, tcpip.TransportProtocolNumber(transProto), dstIP, srcIP, reply); err != nil { + xerrors.LogInfoInner(s.ctx, err, "[tun] failed to write local icmp echo reply") + } +} + +func (s *stackSystem) 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()) + + netProto := header.IPv4ProtocolNumber + if !dst.Address.Family().IsIPv4() { + netProto = header.IPv6ProtocolNumber + } + + segment := make([]byte, udpLen) + udpHdr := header.UDP(segment) + udpHdr.Encode(&header.UDPFields{ + SrcPort: uint16(src.Port), + DstPort: uint16(dst.Port), + Length: uint16(udpLen), + }) + copy(segment[header.UDPMinimumSize:], payload) + + xsum := header.PseudoHeaderChecksum(header.UDPProtocolNumber, srcIP, dstIP, uint16(udpLen)) + udpHdr.SetChecksum(^udpHdr.CalculateChecksum(checksum.Checksum(payload, xsum))) + + return s.writeTransportSegment(netProto, header.UDPProtocolNumber, srcIP, dstIP, segment) +} + +// writeTransportSegment wraps a fully built, already checksummed transport +// layer segment (UDP, ICMP or TCP) with an IP header and writes it to the tun +// device. +func (s *stackSystem) writeTransportSegment(netProto tcpip.NetworkProtocolNumber, transProto tcpip.TransportProtocolNumber, srcIP, dstIP tcpip.Address, segment []byte) error { + ipHdrSize := header.IPv4MinimumSize + if netProto == header.IPv6ProtocolNumber { + ipHdrSize = header.IPv6MinimumSize + } + + pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ + ReserveHeaderBytes: ipHdrSize, + Payload: buffer.MakeWithData(segment), + }) + defer pkt.DecRef() + + if netProto == header.IPv4ProtocolNumber { + ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) + ipHdr.Encode(&header.IPv4Fields{ + TotalLength: uint16(header.IPv4MinimumSize + len(segment)), + TTL: 64, + Protocol: uint8(transProto), + SrcAddr: srcIP, + DstAddr: dstIP, + }) + ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) + } else { + ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) + ipHdr.Encode(&header.IPv6Fields{ + PayloadLength: uint16(len(segment)), + TransportProtocol: transProto, + HopLimit: 64, + SrcAddr: srcIP, + DstAddr: dstIP, + }) + } + + if err := s.device.WritePacket(pkt); err != nil { + return xerrors.New("failed to write raw packet: ", err.String()) + } + return nil +} + +// idleReapLoop periodically aborts tcp connections that have seen no traffic +// for longer than idleTimeout, finally putting that option to use (it was +// tracked but never read anywhere before the "system" backend existed). +func (s *stackSystem) idleReapLoop(ctx context.Context) { + if s.idleTimeout <= 0 { + return + } + interval := s.idleTimeout / 4 + if interval < time.Second { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.reapIdleConnections() + } + } +} + +func (s *stackSystem) reapIdleConnections() { + deadline := time.Now().Add(-s.idleTimeout) + + s.tcpMu.Lock() + var idle []*tcpConn + for _, c := range s.tcp { + if c.lastActiveTime().Before(deadline) { + idle = append(idle, c) + } + } + s.tcpMu.Unlock() + + for _, c := range idle { + c.abort(errConnIdleTimeout) + } +} + +func (s *stackSystem) removeTCPConn(key tcpKey, c *tcpConn) { + s.tcpMu.Lock() + if existing, ok := s.tcp[key]; ok && existing == c { + delete(s.tcp, key) + } + s.tcpMu.Unlock() +} + +// randomSequenceNumber returns a random initial sequence number for a new +// connection. It doesn't need to be cryptographically unpredictable (the tun +// channel is local and trusted), just varied enough to avoid confusion with +// prior incarnations of the same 4-tuple. +func randomSequenceNumber() seqnum.Value { + var b [4]byte + _, _ = rand.Read(b[:]) + return seqnum.Value(binary.BigEndian.Uint32(b[:])) +} diff --git a/proxy/tun/stack_system_tcp.go b/proxy/tun/stack_system_tcp.go new file mode 100644 index 000000000..c5ad92281 --- /dev/null +++ b/proxy/tun/stack_system_tcp.go @@ -0,0 +1,725 @@ +package tun + +import ( + "io" + "sync" + "time" + + xerrors "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/net" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/seqnum" +) + +// This file implements a small, dedicated TCP state machine for the "system" +// tun stack, see stack_system.go. It intentionally does not implement window +// scaling, SACK, timestamps, congestion control, fast retransmit or +// out-of-order reassembly: the tun channel only ever carries packets produced +// by the local OS network stack and handed to us directly, so it neither +// reorders nor drops them the way the public internet does; a single RTO +// timer (also used for zero-window probing) is enough to make the connection +// robust against the rare occasions a segment does not make it through. +const ( + minRTO = 300 * time.Millisecond + maxRTO = 30 * time.Second + maxRTORetries = 12 + lingerDuration = 5 * time.Second + + // maxSendBuffer/maxRecvBuffer match the gVisor backend's own default + // buffer sizes (tcp.DefaultSendBufferSize/DefaultReceiveBufferSize), so + // switching between backends does not change buffering expectations. + maxSendBuffer = 1 << 20 + maxRecvBuffer = 1 << 20 +) + +var ( + errStackClosed = xerrors.New("tun stack closed") + errConnReset = xerrors.New("connection reset by peer") + errConnClosed = xerrors.New("use of closed network connection") + errConnIdleTimeout = xerrors.New("connection idle timeout") + errConnTimedOut = xerrors.New("connection timed out") +) + +type tcpState uint8 + +const ( + stateSynRcvd tcpState = iota + stateEstablished + stateCloseWait // peer's FIN was received; we may still send until we close too + stateClosing // our FIN was sent (from Established or CloseWait) + stateTimeWait // both FINs exchanged and acked; short linger before removal + stateClosed // terminal, removed from the connection table +) + +// tcpKey identifies a tcp connection the same way it appears on the wire +// flowing from the app behind the tun device towards its destination. +type tcpKey struct { + netProto tcpip.NetworkProtocolNumber + srcAddr tcpip.Address + srcPort uint16 + dstAddr tcpip.Address + dstPort uint16 +} + +// tcpConn is a minimal TCP endpoint implementing net.Conn. It deliberately +// exposes only plain Read/Write (never ReadMultiBuffer/WriteMultiBuffer) so +// that stat.CounterConnection in handler.go keeps accounting traffic +// correctly, matching the udpConn precedent in udp_fullcone.go. +type tcpConn struct { + stack *stackSystem + key tcpKey + src net.Destination + dst net.Destination + + ourMSS int + + mu sync.Mutex + cond *sync.Cond + + state tcpState + + // send side. sendQueue[0] always holds the byte at sequence sndUna: acked + // bytes are trimmed off the front, so no separate "acked" bookkeeping is + // needed. sendQueue[:unsentOffset] has been transmitted at least once; + // sendQueue[unsentOffset:] never has. + iss seqnum.Value + sndUna seqnum.Value + sndNxt seqnum.Value + sndMSS int + peerWindow uint32 + sendQueue []byte + unsentOffset int + closeCalled bool + finSent bool + finAcked bool + finSeq seqnum.Value + + // receive side. + irs seqnum.Value + rcvNxt seqnum.Value + recvQueue [][]byte + recvOffset int + recvBuffered int + recvClosed bool + + err error + + lastActive time.Time + + rtoTimer *time.Timer + rtoBackoff int + lingerTimer *time.Timer +} + +var _ net.Conn = (*tcpConn)(nil) + +// outgoingMSS returns the MSS we can use without ever needing IP +// fragmentation (unsupported), given the tun device's MTU. +func outgoingMSS(mtu uint32, netProto tcpip.NetworkProtocolNumber) int { + ipHdrSize := header.IPv4MinimumSize + if netProto == header.IPv6ProtocolNumber { + ipHdrSize = header.IPv6MinimumSize + } + mss := int(mtu) - ipHdrSize - header.TCPMinimumSize + const minMSS = 88 + if mss < minMSS { + mss = minMSS + } + return mss +} + +// handleTCP is the tcp entry point from stackSystem.handleTransport. +func (s *stackSystem) handleTCP(netProto tcpip.NetworkProtocolNumber, srcIP, dstIP tcpip.Address, payload []byte) { + if len(payload) < header.TCPMinimumSize { + return + } + tcpHdr := header.TCP(payload) + if _, _, ok := header.TCPValid(tcpHdr, nil, 0, tcpip.Address{}, tcpip.Address{}, true); !ok { + return + } + + key := tcpKey{ + netProto: netProto, + srcAddr: srcIP, + srcPort: tcpHdr.SourcePort(), + dstAddr: dstIP, + dstPort: tcpHdr.DestinationPort(), + } + + s.tcpMu.Lock() + conn, ok := s.tcp[key] + s.tcpMu.Unlock() + + if ok { + conn.handleSegment(tcpHdr) + return + } + + flags := tcpHdr.Flags() + if flags&header.TCPFlagRst != 0 { + return // never generate a reset in response to a reset + } + if flags&header.TCPFlagSyn != 0 && flags&header.TCPFlagAck == 0 { + s.newTCPConn(key, tcpHdr) + return + } + + // any other segment referencing an unknown connection: let the peer know + // promptly it no longer/never existed, same as a real kernel would + s.sendRawTCPReset(key, tcpHdr) +} + +func (s *stackSystem) newTCPConn(key tcpKey, tcpHdr header.TCP) { + synOpts := header.ParseSynOptions(tcpHdr.Options(), false) + + c := &tcpConn{ + stack: s, + key: key, + src: net.TCPDestination(net.IPAddress(key.srcAddr.AsSlice()), net.Port(key.srcPort)), + dst: net.TCPDestination(net.IPAddress(key.dstAddr.AsSlice()), net.Port(key.dstPort)), + state: stateSynRcvd, + } + c.cond = sync.NewCond(&c.mu) + + c.iss = randomSequenceNumber() + c.sndUna = c.iss + c.sndNxt = c.iss.Add(1) + + c.irs = seqnum.Value(tcpHdr.SequenceNumber()) + c.rcvNxt = c.irs.Add(1) + + c.ourMSS = outgoingMSS(s.mtu, key.netProto) + c.sndMSS = int(synOpts.MSS) + if c.sndMSS <= 0 || c.sndMSS > c.ourMSS { + c.sndMSS = c.ourMSS + } + c.lastActive = time.Now() + + s.tcpMu.Lock() + s.tcp[key] = c + s.tcpMu.Unlock() + + c.mu.Lock() + c.sendSynAckLocked() + c.mu.Unlock() +} + +// sendRawTCPReset replies to a segment that does not match any known +// connection, following the rules of RFC 9293 §3.10.7.1. +func (s *stackSystem) sendRawTCPReset(key tcpKey, tcpHdr header.TCP) { + flags := tcpHdr.Flags() + segLen := seqnum.Size(len(tcpHdr.Payload())) + if flags&header.TCPFlagSyn != 0 { + segLen++ + } + if flags&header.TCPFlagFin != 0 { + segLen++ + } + + var seq, ack seqnum.Value + var ackFlag header.TCPFlags + if flags&header.TCPFlagAck != 0 { + seq = seqnum.Value(tcpHdr.AckNumber()) + } else { + ack = seqnum.Value(tcpHdr.SequenceNumber()).Add(segLen) + ackFlag = header.TCPFlagAck + } + + segment := make([]byte, header.TCPMinimumSize) + rst := header.TCP(segment) + rst.Encode(&header.TCPFields{ + SrcPort: key.dstPort, + DstPort: key.srcPort, + SeqNum: uint32(seq), + AckNum: uint32(ack), + DataOffset: header.TCPMinimumSize, + Flags: header.TCPFlagRst | ackFlag, + WindowSize: 0, + }) + xsum := header.PseudoHeaderChecksum(header.TCPProtocolNumber, key.dstAddr, key.srcAddr, uint16(len(segment))) + rst.SetChecksum(^rst.CalculateChecksum(xsum)) + + if err := s.writeTransportSegment(key.netProto, header.TCPProtocolNumber, key.dstAddr, key.srcAddr, segment); err != nil { + xerrors.LogInfoInner(s.ctx, err, "[tun] failed to write tcp reset") + } +} + +func (c *tcpConn) lastActiveTime() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.lastActive +} + +// abort is the externally callable (unlocked) equivalent of abortLocked, +// used by the idle reaper and by Close's callers indirectly through it. +func (c *tcpConn) abort(err error) { + c.mu.Lock() + defer c.mu.Unlock() + c.abortLocked(err) +} + +func (c *tcpConn) abortLocked(err error) { + if c.state == stateClosed { + return + } + c.state = stateClosed + c.stopRTOLocked() + c.stopLingerLocked() + c.err = err + c.cond.Broadcast() + // deliberately does not send an RST: if the peer sends anything else for + // this connection later, it will miss the (now removed) table entry and + // get a fresh, correctly-addressed reset from sendRawTCPReset above. + c.stack.removeTCPConn(c.key, c) +} + +// handleSegment processes one already-demultiplexed incoming segment. +func (c *tcpConn) handleSegment(tcpHdr header.TCP) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.state == stateClosed { + return + } + c.lastActive = time.Now() + + flags := tcpHdr.Flags() + + if flags&header.TCPFlagRst != 0 { + c.abortLocked(errConnReset) + return + } + + if c.state == stateSynRcvd { + c.handleSynRcvdSegmentLocked(tcpHdr) + return + } + + if flags&header.TCPFlagSyn != 0 { + // unexpected SYN on an already-established connection is not + // modeled; treat it like the peer abandoned and reset it + c.abortLocked(errConnReset) + return + } + + if flags&header.TCPFlagAck != 0 { + c.handleAckLocked(seqnum.Value(tcpHdr.AckNumber()), tcpHdr.WindowSize()) + } + + c.acceptInOrderLocked(seqnum.Value(tcpHdr.SequenceNumber()), tcpHdr.Payload(), flags&header.TCPFlagFin != 0) +} + +func (c *tcpConn) handleSynRcvdSegmentLocked(tcpHdr header.TCP) { + flags := tcpHdr.Flags() + + if flags&header.TCPFlagSyn != 0 { + // peer's retransmission of the original SYN, our SYN-ACK likely + // hasn't reached them yet: resend it and rely entirely on their own + // retransmission timer rather than running one on our side too + c.sendSynAckLocked() + return + } + if flags&header.TCPFlagAck == 0 { + return + } + if seqnum.Value(tcpHdr.AckNumber()) != c.sndNxt { + // does not acknowledge our SYN correctly; a well-behaved peer will + // simply retry, so it is safe to just ignore this segment + return + } + + c.state = stateEstablished + go c.stack.handler.HandleConnection(c, c.dst) + + c.handleAckLocked(seqnum.Value(tcpHdr.AckNumber()), tcpHdr.WindowSize()) + c.acceptInOrderLocked(seqnum.Value(tcpHdr.SequenceNumber()), tcpHdr.Payload(), flags&header.TCPFlagFin != 0) +} + +// acceptInOrderLocked handles the data/FIN portion of a segment once it is +// known to be neither a SYN nor a RST. Only strictly in-order segments are +// accepted; anything else is dropped (relying on the peer's retransmission) +// since the tun channel is expected to already deliver packets in order. +func (c *tcpConn) acceptInOrderLocked(seq seqnum.Value, payload []byte, fin bool) { + if seq != c.rcvNxt { + c.sendAckLocked() + return + } + + accept := payload + if room := c.recvWindowLocked(); uint32(len(accept)) > room { + accept = accept[:room] + } + if len(accept) > 0 { + c.enqueueRecvLocked(accept) + c.rcvNxt = c.rcvNxt.Add(seqnum.Size(len(accept))) + } + + finAccepted := false + if fin && len(accept) == len(payload) { + c.onFinLocked() + c.rcvNxt = c.rcvNxt.Add(1) + finAccepted = true + } + + if len(accept) > 0 || finAccepted || len(accept) < len(payload) { + c.sendAckLocked() + } +} + +func (c *tcpConn) onFinLocked() { + if c.recvClosed { + return + } + c.recvClosed = true + c.cond.Broadcast() + if c.state == stateEstablished { + c.state = stateCloseWait + } + c.maybeFinishCloseLocked() +} + +func (c *tcpConn) handleAckLocked(ackNum seqnum.Value, windowSize uint16) { + if ackNum.LessThan(c.sndUna) { + // old/duplicate ack: no fast-retransmit heuristics implemented + c.peerWindow = uint32(windowSize) + c.trySendLocked() + return + } + if c.sndNxt.LessThan(ackNum) { + // acknowledges more than we ever sent: lenient clamp instead of + // rejecting the segment outright + ackNum = c.sndNxt + } + + if advanced := c.sndUna.Size(ackNum); advanced > 0 { + c.sndUna = ackNum + n := int(advanced) + if n > len(c.sendQueue) { + n = len(c.sendQueue) + } + c.sendQueue = c.sendQueue[n:] + c.unsentOffset -= n + if c.unsentOffset < 0 { + c.unsentOffset = 0 + } + c.rtoBackoff = 0 + if c.finSent && c.sndUna == c.sndNxt { + c.finAcked = true + } + c.cond.Broadcast() + } + + c.peerWindow = uint32(windowSize) + c.trySendLocked() + c.maybeFinishCloseLocked() +} + +func (c *tcpConn) maybeFinishCloseLocked() { + if c.state == stateClosing && c.finAcked && c.recvClosed { + c.state = stateTimeWait + c.startLingerLocked() + } +} + +func (c *tcpConn) recvWindowLocked() uint32 { + room := maxRecvBuffer - c.recvBuffered + if room < 0 { + room = 0 + } + if room > 0xffff { + room = 0xffff + } + return uint32(room) +} + +func (c *tcpConn) enqueueRecvLocked(payload []byte) { + data := make([]byte, len(payload)) + copy(data, payload) + c.recvQueue = append(c.recvQueue, data) + c.recvBuffered += len(data) + c.cond.Broadcast() +} + +// sendOneChunkLocked transmits up to maxLen bytes of never-yet-sent data (if +// any remains), advancing sndNxt/unsentOffset. It returns the number of +// bytes sent, 0 if none remained. +func (c *tcpConn) sendOneChunkLocked(maxLen int) int { + remaining := len(c.sendQueue) - c.unsentOffset + if remaining <= 0 { + return 0 + } + if maxLen > remaining { + maxLen = remaining + } + if maxLen > c.sndMSS { + maxLen = c.sndMSS + } + if maxLen <= 0 { + return 0 + } + data := c.sendQueue[c.unsentOffset : c.unsentOffset+maxLen] + c.sendDataSegmentLocked(c.sndNxt, data, false) + c.sndNxt = c.sndNxt.Add(seqnum.Size(maxLen)) + c.unsentOffset += maxLen + return maxLen +} + +func (c *tcpConn) trySendLocked() { + switch c.state { + case stateSynRcvd, stateTimeWait, stateClosed: + return + } + + for { + inFlight := int(c.sndUna.Size(c.sndNxt)) + windowLeft := int(c.peerWindow) - inFlight + if windowLeft <= 0 { + break + } + if c.sendOneChunkLocked(windowLeft) == 0 { + break + } + } + + if c.closeCalled && !c.finSent && c.unsentOffset == len(c.sendQueue) { + c.finSeq = c.sndNxt + c.sendDataSegmentLocked(c.finSeq, nil, true) + c.sndNxt = c.sndNxt.Add(1) + c.finSent = true + } + + c.refreshRTOLocked() +} + +func (c *tcpConn) outstandingLocked() bool { + if c.unsentOffset > 0 { + return true // already-transmitted data pending ack + } + if len(c.sendQueue) > c.unsentOffset && c.peerWindow == 0 { + return true // blocked purely by a zero window; need to probe + } + if c.finSent && !c.finAcked { + return true // FIN transmitted but not yet acked + } + return false +} + +func (c *tcpConn) refreshRTOLocked() { + if c.outstandingLocked() { + c.scheduleRTOLocked() + } else { + c.stopRTOLocked() + } +} + +func (c *tcpConn) rtoDurationLocked() time.Duration { + d := minRTO * time.Duration(uint64(1)< maxRTO || d <= 0 { + d = maxRTO + } + return d +} + +func (c *tcpConn) scheduleRTOLocked() { + d := c.rtoDurationLocked() + if c.rtoTimer == nil { + c.rtoTimer = time.AfterFunc(d, c.onRTOTimerFired) + } else { + c.rtoTimer.Reset(d) + } +} + +func (c *tcpConn) stopRTOLocked() { + if c.rtoTimer != nil { + c.rtoTimer.Stop() + } +} + +func (c *tcpConn) onRTOTimerFired() { + c.mu.Lock() + defer c.mu.Unlock() + c.onRTOFireLocked() +} + +func (c *tcpConn) onRTOFireLocked() { + if c.state == stateClosed || !c.outstandingLocked() { + return + } + if c.rtoBackoff >= maxRTORetries { + c.abortLocked(errConnTimedOut) + return + } + c.rtoBackoff++ + + switch { + case c.unsentOffset > 0: + c.sendDataSegmentLocked(c.sndUna, c.sendQueue[:c.unsentOffset], false) + case len(c.sendQueue) > c.unsentOffset: + // nothing in flight, but blocked by a zero peer window: probe with + // exactly one new byte, per RFC 9293 §3.8.6.1 + c.sendOneChunkLocked(1) + case c.finSent && !c.finAcked: + c.sendDataSegmentLocked(c.finSeq, nil, true) + } + + c.refreshRTOLocked() +} + +func (c *tcpConn) startLingerLocked() { + c.stopRTOLocked() + c.lingerTimer = time.AfterFunc(lingerDuration, func() { + c.mu.Lock() + defer c.mu.Unlock() + c.abortLocked(errConnClosed) + }) +} + +func (c *tcpConn) stopLingerLocked() { + if c.lingerTimer != nil { + c.lingerTimer.Stop() + } +} + +// transmitLocked builds, checksums and writes a single tcp segment. +func (c *tcpConn) transmitLocked(seq, ack seqnum.Value, flags header.TCPFlags, payload []byte, options []byte) { + headerLen := header.TCPMinimumSize + len(options) + segment := make([]byte, headerLen+len(payload)) + tcpHdr := header.TCP(segment) + tcpHdr.Encode(&header.TCPFields{ + SrcPort: c.key.dstPort, + DstPort: c.key.srcPort, + SeqNum: uint32(seq), + AckNum: uint32(ack), + DataOffset: uint8(headerLen), + Flags: flags, + WindowSize: uint16(c.recvWindowLocked()), + }) + copy(tcpHdr.Options(), options) + copy(segment[headerLen:], payload) + + xsum := header.PseudoHeaderChecksum(header.TCPProtocolNumber, c.key.dstAddr, c.key.srcAddr, uint16(len(segment))) + xsum = checksum.Checksum(payload, xsum) + tcpHdr.SetChecksum(^tcpHdr.CalculateChecksum(xsum)) + + if err := c.stack.writeTransportSegment(c.key.netProto, header.TCPProtocolNumber, c.key.dstAddr, c.key.srcAddr, segment); err != nil { + xerrors.LogInfoInner(c.stack.ctx, err, "[tun] failed to write tcp segment") + } +} + +func (c *tcpConn) sendSynAckLocked() { + var optBuf [header.TCPOptionMSSLength]byte + n := header.EncodeMSSOption(uint32(c.ourMSS), optBuf[:]) + c.transmitLocked(c.iss, c.rcvNxt, header.TCPFlagSyn|header.TCPFlagAck, nil, optBuf[:n]) +} + +func (c *tcpConn) sendAckLocked() { + c.transmitLocked(c.sndNxt, c.rcvNxt, header.TCPFlagAck, nil, nil) +} + +func (c *tcpConn) sendDataSegmentLocked(seq seqnum.Value, payload []byte, fin bool) { + flags := header.TCPFlagAck + if fin { + flags |= header.TCPFlagFin + } + c.transmitLocked(seq, c.rcvNxt, flags, payload, nil) +} + +// Read implements net.Conn. +func (c *tcpConn) Read(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + + for len(c.recvQueue) == 0 && c.err == nil && !c.recvClosed { + c.cond.Wait() + } + if c.err != nil { + return 0, c.err + } + if len(c.recvQueue) == 0 { + return 0, io.EOF + } + + before := c.recvWindowLocked() + + chunk := c.recvQueue[0] + n := copy(p, chunk[c.recvOffset:]) + c.recvOffset += n + c.recvBuffered -= n + if c.recvOffset == len(chunk) { + c.recvQueue = c.recvQueue[1:] + c.recvOffset = 0 + } + + // let the peer know promptly if reading just freed up a previously + // exhausted window, instead of waiting for it to probe us for an update + if after := c.recvWindowLocked(); before == 0 && after > 0 { + c.sendAckLocked() + } + + return n, nil +} + +// Write implements net.Conn. +func (c *tcpConn) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.closeCalled { + return 0, errConnClosed + } + + total := 0 + for total < len(p) { + if c.err != nil { + return total, c.err + } + if c.closeCalled { + return total, errConnClosed + } + room := maxSendBuffer - len(c.sendQueue) + if room <= 0 { + c.cond.Wait() + continue + } + n := len(p) - total + if n > room { + n = room + } + c.sendQueue = append(c.sendQueue, p[total:total+n]...) + total += n + } + + c.trySendLocked() + return total, nil +} + +// Close implements net.Conn. +func (c *tcpConn) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.closeCalled { + return nil + } + c.closeCalled = true + c.cond.Broadcast() + + switch c.state { + case stateEstablished, stateCloseWait: + c.state = stateClosing + default: + return nil + } + + c.trySendLocked() + 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(t time.Time) error { return nil } +func (c *tcpConn) SetReadDeadline(t time.Time) error { return nil } +func (c *tcpConn) SetWriteDeadline(t time.Time) error { return nil } diff --git a/proxy/tun/tun_android.go b/proxy/tun/tun_android.go index f287bb1ea..b250ba090 100644 --- a/proxy/tun/tun_android.go +++ b/proxy/tun/tun_android.go @@ -7,9 +7,12 @@ import ( "net" "strconv" + "github.com/xtls/xray-core/common/buf" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/platform" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/link/fdbased" "gvisor.dev/gvisor/pkg/tcpip/stack" ) @@ -22,6 +25,22 @@ type AndroidTun struct { // DefaultTun implements Tun var _ Tun = (*AndroidTun)(nil) +// AndroidTun implements GVisorDevice, used by the "system" (lite) ip stack +var _ GVisorDevice = (*AndroidTun)(nil) + +// fdReadWriter adapts a raw, already non-blocking file descriptor to io.Reader/io.Writer, +// so it can be used with buf.Buffer.ReadFrom, without the ownership/finalizer overhead of +// wrapping it in an *os.File (the fd is owned and closed elsewhere). +type fdReadWriter int + +func (f fdReadWriter) Read(p []byte) (int, error) { + return unix.Read(int(f), p) +} + +func (f fdReadWriter) Write(p []byte) (int, error) { + return unix.Write(int(f), p) +} + // NewTun builds new tun interface handler func NewTun(options *Config) (Tun, error) { fd, err := strconv.Atoi(platform.NewEnvFlag(platform.TunFdKey).GetValue(func() string { return "0" })) @@ -78,6 +97,72 @@ func (t *AndroidTun) newEndpoint() (stack.LinkEndpoint, error) { }) } +// ReadPacket implements GVisorDevice method to read one packet from the tun device, used by +// the "system" (lite) ip stack. The gVisor backed stack instead talks to the fd directly through +// fdbased.New above, for lower overhead batched IO, bypassing GVisorDevice entirely. +// 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 *AndroidTun) ReadPacket() (byte, *stack.PacketBuffer, error) { + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(t.options.MTU)) + + // read the bytes from the interface file descriptor, which is already non-blocking + n, err := b.ReadFrom(fdReadWriter(t.tunFd)) + if err == unix.EAGAIN || err == unix.EWOULDBLOCK || err == unix.EINTR { + b.Release() + return 0, nil, ErrQueueEmpty + } + if err != nil { + b.Release() + return 0, nil, err + } + + // discard empty packets + if n == 0 { + b.Release() + return 0, nil, ErrQueueEmpty + } + + // network protocol version from the first nibble of the raw packet + version := b.Byte(0) >> 4 + packetBuffer := buffer.MakeWithData(b.Bytes()) + return version, stack.NewPacketBuffer(stack.PacketBufferOptions{ + Payload: packetBuffer, + IsForwardedPacket: true, + OnRelease: func() { + b.Release() + }, + }), nil +} + +// WritePacket implements GVisorDevice method to write one packet to the tun device +func (t *AndroidTun) WritePacket(packet *stack.PacketBuffer) tcpip.Error { + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(t.options.MTU)) + defer b.Release() + + // copy the bytes of slices that compose the packet into the allocated buffer, no + // extra header is needed here, unlike Darwin/FreeBSD's utun devices + for _, packetElement := range packet.AsSlices() { + _, _ = b.Write(packetElement) + } + + if _, err := fdReadWriter(t.tunFd).Write(b.Bytes()); err != nil { + if err == unix.EAGAIN || err == unix.EWOULDBLOCK { + return &tcpip.ErrWouldBlock{} + } + return &tcpip.ErrAborted{} + } + return nil +} + +// Wait blocks until the tun fd is likely readable again, rather than spinning the CPU. +// A bounded timeout keeps this responsive to a Close() racing a call already parked here. +func (t *AndroidTun) Wait() { + fds := []unix.PollFd{{Fd: int32(t.tunFd), Events: unix.POLLIN}} + _, _ = unix.Poll(fds, 1000) +} + func setinterface(network, address string, fd uintptr, iface *net.Interface) error { return unix.BindToDevice(int(fd), iface.Name) } diff --git a/proxy/tun/tun_linux.go b/proxy/tun/tun_linux.go index b2c5d35a1..b05a0e819 100644 --- a/proxy/tun/tun_linux.go +++ b/proxy/tun/tun_linux.go @@ -10,9 +10,12 @@ import ( "sync" "github.com/vishvananda/netlink" + "github.com/xtls/xray-core/common/buf" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/platform" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/link/fdbased" "gvisor.dev/gvisor/pkg/tcpip/stack" ) @@ -35,6 +38,22 @@ type LinuxTun struct { // LinuxTun implements Tun var _ Tun = (*LinuxTun)(nil) +// LinuxTun implements GVisorDevice, used by the "system" (lite) ip stack +var _ GVisorDevice = (*LinuxTun)(nil) + +// fdReadWriter adapts a raw, already non-blocking file descriptor to io.Reader/io.Writer, +// so it can be used with buf.Buffer.ReadFrom, without the ownership/finalizer overhead of +// wrapping it in an *os.File (the fd is owned and closed elsewhere, see LinuxTun.Close). +type fdReadWriter int + +func (f fdReadWriter) Read(p []byte) (int, error) { + return unix.Read(int(f), p) +} + +func (f fdReadWriter) Write(p []byte) (int, error) { + return unix.Write(int(f), p) +} + // NewTun builds new tun interface handler (linux specific) func NewTun(options *Config) (Tun, error) { tunFd, tunLink, fdProvided, err := openFromEnv(options.Name) @@ -228,6 +247,72 @@ func (t *LinuxTun) newEndpoint() (stack.LinkEndpoint, error) { }) } +// ReadPacket implements GVisorDevice method to read one packet from the tun device, used by +// the "system" (lite) ip stack. The gVisor backed stack instead talks to the fd directly through +// fdbased.New above, for lower overhead batched IO, bypassing GVisorDevice entirely. +// 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 *LinuxTun) ReadPacket() (byte, *stack.PacketBuffer, error) { + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(t.options.MTU)) + + // read the bytes from the interface file descriptor, which is already non-blocking + n, err := b.ReadFrom(fdReadWriter(t.tunFd)) + if err == unix.EAGAIN || err == unix.EWOULDBLOCK || err == unix.EINTR { + b.Release() + return 0, nil, ErrQueueEmpty + } + if err != nil { + b.Release() + return 0, nil, err + } + + // discard empty packets + if n == 0 { + b.Release() + return 0, nil, ErrQueueEmpty + } + + // network protocol version from the first nibble of the raw packet + version := b.Byte(0) >> 4 + packetBuffer := buffer.MakeWithData(b.Bytes()) + return version, stack.NewPacketBuffer(stack.PacketBufferOptions{ + Payload: packetBuffer, + IsForwardedPacket: true, + OnRelease: func() { + b.Release() + }, + }), nil +} + +// WritePacket implements GVisorDevice method to write one packet to the tun device +func (t *LinuxTun) WritePacket(packet *stack.PacketBuffer) tcpip.Error { + // request memory to write from reusable buffer pool + b := buf.NewWithSize(int32(t.options.MTU)) + defer b.Release() + + // copy the bytes of slices that compose the packet into the allocated buffer, no + // Linux specific header is needed here, unlike Darwin/FreeBSD's utun devices + for _, packetElement := range packet.AsSlices() { + _, _ = b.Write(packetElement) + } + + if _, err := fdReadWriter(t.tunFd).Write(b.Bytes()); err != nil { + if err == unix.EAGAIN || err == unix.EWOULDBLOCK { + return &tcpip.ErrWouldBlock{} + } + return &tcpip.ErrAborted{} + } + return nil +} + +// Wait blocks until the tun fd is likely readable again, rather than spinning the CPU. +// A bounded timeout keeps this responsive to a Close() racing a call already parked here. +func (t *LinuxTun) Wait() { + fds := []unix.PollFd{{Fd: int32(t.tunFd), Events: unix.POLLIN}} + _, _ = unix.Poll(fds, 1000) +} + func setinterface(network, address string, fd uintptr, iface *net.Interface) error { return unix.BindToDevice(int(fd), iface.Name) }