Artem Lytkin
2026-10-12 02:18:56 +00:00
committed by RPRX
parent 701af60772
commit 70ef97d564
18 changed files with 2704 additions and 819 deletions
+26 -22
View File
@@ -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{
+48
View File
@@ -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)
+161
View File
@@ -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)
}
+263
View File
@@ -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)
}
-267
View File
@@ -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
}
-153
View File
@@ -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()
}
}
}
-98
View File
@@ -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
}
+929
View File
@@ -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))
}
}
}
+999
View File
@@ -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
}
+16 -2
View File
@@ -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
}
+10 -14
View File
@@ -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)
}
+36 -76
View File
@@ -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
+5 -2
View File
@@ -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")
}
+96
View File
@@ -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()
}
+27 -67
View File
@@ -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
+39 -32
View File
@@ -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)
}
+30 -50
View File
@@ -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 (
+19 -36
View File
@@ -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