diff --git a/proxy/tun/packet.go b/proxy/tun/packet.go index a4a97c9c7..697d20eb7 100644 --- a/proxy/tun/packet.go +++ b/proxy/tun/packet.go @@ -41,8 +41,9 @@ const ( type packet struct { proto byte src, dst netip.Addr - // payload is what follows the IP header + // payload is what follows the IP header, without the last rest bytes of it when they are in other buffers payload []byte + rest int // fragment is set if payload is a part of what follows the IP header: the one at offset of the packet // with the number id, which has more after it or not fragment bool @@ -51,8 +52,8 @@ type packet struct { id uint32 } -// parsePacket reads the IP header of b -func parsePacket(b []byte) (p packet, ok bool) { +// parsePacket reads the IP header of a packet that is b and rest bytes after it +func parsePacket(b []byte, rest int) (p packet, ok bool) { if len(b) < ipv4HeaderSize { return p, false } @@ -60,7 +61,7 @@ func parsePacket(b []byte) (p packet, ok bool) { case 4: size := int(b[0]&0xf) * 4 total := int(binary.BigEndian.Uint16(b[2:])) - if size < ipv4HeaderSize || total < size || total > len(b) { + if size < ipv4HeaderSize || total < size || total > len(b)+rest { return p, false } if f := binary.BigEndian.Uint16(b[6:]); f&0x3fff != 0 { @@ -69,19 +70,19 @@ func parsePacket(b []byte) (p packet, ok bool) { 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] + p.payload, p.rest = b[size:min(total, len(b))], max(total-len(b), 0) case 6: if len(b) < ipv6HeaderSize { return p, false } total := ipv6HeaderSize + int(binary.BigEndian.Uint16(b[4:])) - if total > len(b) { + if total > len(b)+rest { 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] + p.payload, p.rest = b[ipv6HeaderSize:min(total, len(b))], max(total-len(b), 0) // hop-by-hop options, routing, destination options for p.proto == 0 || p.proto == 43 || p.proto == 60 { if len(p.payload) < 8 { diff --git a/proxy/tun/stack.go b/proxy/tun/stack.go index 24830c8d4..5048ae94c 100644 --- a/proxy/tun/stack.go +++ b/proxy/tun/stack.go @@ -71,6 +71,10 @@ type stack struct { udp *udpConnectionHandler // addresses are what connections are opened from addresses []netip.Addr + // bufs is what read reads a packet into if the tun is a segmenter, one after the other, the headers in the first. + // One that keep took is nil. joined is for a packet that is in more than one and is needed in one piece + bufs buf.MultiBuffer + joined []byte mu sync.Mutex conns map[flow]*tcpConn @@ -149,26 +153,66 @@ func (s *stack) Close() error { // 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") + var err error + if s.segmenter != nil { + err = s.readSegments() + } else { + // as large as a packet can be, whatever the MTU is + b := make([]byte, packetRoom+1<<16) + for err == nil { + var p []byte + if p, err = s.tun.ReadPacket(b); err == nil { + s.Input(p) } - return } - s.Input(b) } + 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") + } +} + +// readSegments is read for a tun that does not split segments and gives them as large as a packet can be. +// They are read into buffers for a connection to keep what is large as it is, see keep +func (s *stack) readSegments() error { + s.bufs = make(buf.MultiBuffer, (1<<16)/buf.Size) + defer buf.ReleaseMulti(s.bufs) + parts := make([][]byte, len(s.bufs)) + var r room + for { + for i, b := range s.bufs { + if b == nil { + s.bufs[i] = buf.New() + _, _ = s.bufs[i].ReadFrom(&r) + parts[i] = r.b + } + } + n, err := s.segmenter.ReadParts(parts) + if err != nil { + return err + } + s.input(parts[0][:min(n, buf.Size)], max(n-buf.Size, 0)) + } +} + +// room takes what it is given to read into as read: that is the room of a Buffer as it is, not cleared +type room struct{ b []byte } + +func (r *room) Read(b []byte) (int, error) { + r.b = b + return len(b), nil } // Input implements Stack. Several tunnels may hand over their packets at once func (s *stack) Input(b []byte) { - p, ok := parsePacket(b) + s.input(b, 0) +} + +// input takes a packet that is b and rest bytes after it, in the bufs that follow the one that b is in +func (s *stack) input(b []byte, rest int) { + p, ok := parsePacket(b, rest) if !ok || p.src.Is4In6() || p.dst.Is4In6() || p.src.IsLoopback() || p.dst.IsLoopback() || p.dst.IsUnspecified() || p.src.IsMulticast() { // nothing comes from many at once, and an address that stands for this machine itself is not one // that the other side of a tun can ask for. Nor is an IPv6 address that stands for an IPv4 one, which @@ -180,6 +224,10 @@ func (s *stack) Input(b []byte) { } // what was put together may be of two packets with one number, its checksum tells verify := s.verify || p.fragment + if p.rest != 0 && (verify || p.proto != protoTCP) { + // only a segment of TCP that is not looked at can stay in pieces + p.payload, p.rest = s.whole(p.payload, p.rest), 0 + } if p.fragment { p, ok = s.fragments.add(p) } @@ -196,6 +244,37 @@ func (s *stack) Input(b []byte) { } } +// whole returns in one piece what is b and rest bytes after it, in the bufs that follow the one that b is in +func (s *stack) whole(b []byte, rest int) []byte { + if rest == 0 { + return b + } + s.joined = append(s.joined[:0], b...) + for i := 1; rest > 0; i++ { + n := min(rest, buf.Size) + s.joined = append(s.joined, s.bufs[i].BytesTo(int32(n))...) + rest -= n + } + return s.joined +} + +// keep adds to mb what is b and rest bytes after it, in the bufs that follow the one that b is in: nothing is copied, +// the buffers are taken for it and no longer read into +func (s *stack) keep(mb buf.MultiBuffer, b []byte, rest int) buf.MultiBuffer { + // b ends where what was read into the first one does, or before it + first := s.bufs[0] + from := first.Cap() - int32(cap(b)) + first.Resize(from, from+int32(len(b))) + mb, s.bufs[0] = append(mb, first), nil + for i := 1; rest > 0; i++ { + n := min(rest, buf.Size) + s.bufs[i].Resize(0, int32(n)) + mb, s.bufs[i] = append(mb, s.bufs[i]), nil + rest -= n + } + return mb +} + // local returns the address that packets to remote are sent from func (s *stack) local(remote netip.Addr) (netip.Addr, error) { for _, a := range s.addresses { @@ -237,6 +316,7 @@ func (s *stack) handleTCP(p packet) { window: binary.BigEndian.Uint16(t[14:]), options: t[tcpHeaderSize:size], payload: t[size:], + rest: p.rest, } s.mu.Lock() @@ -283,7 +363,7 @@ func (s *stack) refuse(f flow, seg segment) { binary.BigEndian.PutUint32(t[4:], seg.ack) t[13] = tcpRst } else { - ack := seg.seq + uint32(len(seg.payload)) + ack := seg.seq + uint32(len(seg.payload)+seg.rest) if seg.flags&(tcpSyn|tcpFin) != 0 { ack++ } diff --git a/proxy/tun/stack_test.go b/proxy/tun/stack_test.go index fb7bbeef7..e6b7b312c 100644 --- a/proxy/tun/stack_test.go +++ b/proxy/tun/stack_test.go @@ -65,6 +65,20 @@ func (l *link) Segments() bool { return l.segments } +// ReadParts implements segmenter +func (l *link) ReadParts(parts [][]byte) (int, error) { + select { + case packet := <-l.in: + n := 0 + for _, p := range parts { + n += copy(p, packet[n:]) + } + return n, nil + case <-l.done: + return 0, io.EOF + } +} + // 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) @@ -206,7 +220,7 @@ func (p *peer) receive() segment { func (p *peer) dialed() (segment, []byte) { p.t.Helper() b := p.packet() - packet, ok := parsePacket(b) + packet, ok := parsePacket(b, 0) if !ok || packet.proto != protoTCP || len(packet.payload) < tcpHeaderSize { p.t.Fatal("not a segment: ", b) } @@ -223,7 +237,7 @@ func (p *peer) segment(b []byte) segment { } else if checksum(sum(b[:size], 0)) != 0 { p.t.Fatal("wrong checksum of the ip header") } - packet, ok := parsePacket(b) + packet, ok := parsePacket(b, 0) 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) } @@ -464,30 +478,90 @@ func TestTCPOutOfOrder(t *testing.T) { read(t, a.conn, "jklmno") } -// A segment is as large as the system makes it when the tun does not split what it reads. +// A segment is as large as the system makes it when the tun does not split what it reads, and one that is large, +// the next and all taken stays in the buffers that such a tun reads it into. 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) + for _, segments := range []bool{false, true} { + conns := make(chan accepted, 1) + p, l := newPeer(t, "10.0.0.1:1234", "10.0.0.2:80") + l.segments = segments + s := newTestStack(t, l, accept(conns)) + sent := make([]byte, 200000) + rand.Read(sent) + // one of no connection is refused as a whole + p.send(p.seq, tcpPsh, nil, sent[:20000]) + if r := p.receive(); r.flags != tcpRst|tcpAck || r.ack != p.seq+20000 { + t.Fatal("segments: ", segments, ", not refused: flags ", r.flags, ", ack ", r.ack-p.seq, " after its seq") + } + 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) + // one that waits for the one before it, that one, one with a part that came already, small ones around + // one that fits a buffer, and the rest + for _, part := range [][2]int{{60000, 90000}, {0, 60000}, {80000, 110000}, {110000, 110100}, {110100, 115100}, {115100, 118100}, {118100, 170000}, {170000, 200000}} { + p.send(p.seq+uint32(part[0]), tcpAck|tcpPsh, nil, sent[part[0]:part[1]]) } - for _, b := range mb { - received = append(received, b.Bytes()...) + // all of it is taken before anything is read, for the buffers to be as the stack put it in them + for r := p.receive(); r.ack != p.seq+uint32(len(sent)); r = p.receive() { + } + var received []byte + for len(received) < len(sent) { + mb, err := a.conn.(buf.Reader).ReadMultiBuffer() + if err != nil { + t.Fatal(err) + } + // a buffer that was read into starts with the headers, one that was copied to is full + if kept := mb[0].Len() == buf.Size-ipv4HeaderSize-tcpHeaderSize; len(received) == 0 && kept != segments { + t.Fatal("segments: ", segments, ", the first buffer has ", mb[0].Len(), " bytes") + } + for _, b := range mb { + // and a small one is copied to where the one before it ends + if b.Len() < 1000 { + t.Fatal("segments: ", segments, ", a buffer of ", b.Len(), " bytes") + } + received = append(received, b.Bytes()...) + } + buf.ReleaseMulti(mb) + } + if !bytes.Equal(received, sent) { + t.Fatal("segments: ", segments, ", read something else than what was sent") + } + + // no more is taken than the window while nobody reads, of the last one a part, + // and what is taken is acknowledged when a quarter of the window is + p.seq += uint32(len(sent)) + sent = make([]byte, (int(s.window)/50000+1)*50000) + rand.Read(sent) + for i := 0; i < len(sent); i += 50000 { + p.send(p.seq+uint32(i), tcpAck|tcpPsh, nil, sent[i:i+50000]) + } + for acked := p.seq; ; { + r := p.receive() + if taken := int32(r.ack - p.seq); taken > s.window { + t.Fatal("segments: ", segments, ", ", taken, " bytes taken with a window of ", s.window) + } else if taken == s.window { + break + } + // one that acknowledges nothing more tells that the window opened for what was read before + if r.ack != acked && int32(r.ack-acked) < s.window/4 { + t.Fatal("segments: ", segments, ", ", r.ack-acked, " bytes acknowledged with a window of ", s.window) + } + acked = r.ack + } + received = received[:0] + for len(received) < int(s.window) { + 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[:s.window]) { + t.Fatal("segments: ", segments, ", read something else than what the window took") } - buf.ReleaseMulti(mb) - } - if !bytes.Equal(received, sent) { - t.Fatal("read something else than what was sent") } } @@ -1082,13 +1156,20 @@ func TestTCPSynAfterClose(t *testing.T) { } // A UDP packet starts a connection, which answers from any address, and an echo request is answered. +// Both can be larger than a buffer of a tun that reads into more than one. 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"}, + large := make([]byte, 20000) + rand.Read(large) + for _, test := range []struct { + src, dst, other string + segments bool + }{ + {"10.0.0.1:1234", "10.0.0.2:53", "10.0.0.3:5353", false}, + {"[fd00::1]:1234", "[fd00::2]:53", "[fd00::3]:5353", true}, } { conns := make(chan accepted, 1) p, l := newPeer(t, test.src, test.dst) + l.segments = test.segments newTestStack(t, l, accept(conns)) var b [ipv6HeaderSize + udpHeaderSize]byte @@ -1105,6 +1186,13 @@ func TestUDPAndICMP(t *testing.T) { t.Fatal("read ", mb.String(), " from ", a.conn.RemoteAddr(), " to ", a.dest, ", error: ", err) } buf.ReleaseMulti(mb) + binary.BigEndian.PutUint16(u[4:], uint16(udpHeaderSize+len(large))) + setIPLength(b[:n], n+udpHeaderSize+len(large)) + l.in <- append(b[:n+udpHeaderSize:n+udpHeaderSize], large...) + if mb, err = a.conn.(buf.Reader).ReadMultiBuffer(); err != nil || len(mb) != 1 || !bytes.Equal(mb[0].Bytes(), large) { + t.Fatal("read ", mb.Len(), " bytes of a large packet, error: ", err) + } + buf.ReleaseMulti(mb) for _, from := range []string{test.dst, test.other} { answer := buf.FromBytes([]byte("defg")) @@ -1114,7 +1202,7 @@ func TestUDPAndICMP(t *testing.T) { t.Fatal(err) } raw := p.packet() - packet, ok := parsePacket(raw) + packet, ok := parsePacket(raw, 0) 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 || @@ -1133,12 +1221,12 @@ func TestUDPAndICMP(t *testing.T) { 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()) + setIPLength(b[:n], n+8+len(large)) + l.in <- append(append(b[:n:n], request, 0, 0, 0, 1, 2, 3, 4), large...) + packet, ok := parsePacket(p.joined(), 0) 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) + if !ok || packet.proto != proto || packet.src != p.dst.Addr() || packet.dst != p.src.Addr() || m[0] != reply || string(m[4:8]) != "\x01\x02\x03\x04" || !bytes.Equal(m[8:], large) { + t.Fatal("wrong echo reply of ", len(m), " bytes from ", packet.src, " to ", packet.dst) } var ip [ipv6HeaderSize]byte size := putIPHeader(ip[:], proto, packet.src, packet.dst) @@ -1373,7 +1461,7 @@ func TestUDPDial(t *testing.T) { t.Fatal("wrong header of the packet to ", to, ": ", raw[:ipv4HeaderSize]) } numbers[to][binary.BigEndian.Uint16(raw[4:])] = true - packet, ok := parsePacket(raw) + packet, ok := parsePacket(raw, 0) u := packet.payload if !ok || packet.proto != protoUDP || netip.AddrPortFrom(packet.src, binary.BigEndian.Uint16(u)) != local || netip.AddrPortFrom(packet.dst, binary.BigEndian.Uint16(u[2:])) != to || string(u[udpHeaderSize:]) != "abc" { @@ -1395,7 +1483,7 @@ func TestUDPDial(t *testing.T) { if _, err := conn.WriteTo([]byte("abc"), &net.UDPAddr{IP: across.Addr().AsSlice(), Port: int(across.Port())}); err != nil { t.Fatal(err) } - if packet, ok := parsePacket(p.packet()); !ok || packet.proto != protoUDP || packet.dst != across.Addr() || + if packet, ok := parsePacket(p.packet(), 0); !ok || packet.proto != protoUDP || packet.dst != across.Addr() || netip.AddrPortFrom(packet.src, binary.BigEndian.Uint16(packet.payload)) != acrossLocal { t.Fatal("wrong packet to ", across, ": ", packet) } @@ -1407,7 +1495,7 @@ func TestUDPDial(t *testing.T) { // a stack that takes no connections is one host: it answers a ping of its address and of no other l.in <- ping(p.src.Addr(), netip.MustParseAddrPort(test.other).Addr()) l.in <- ping(p.src.Addr(), local.Addr()) - if packet, ok := parsePacket(p.packet()); !ok || packet.src != local.Addr() || packet.dst != p.src.Addr() || packet.payload[0]&^129 != 0 { + if packet, ok := parsePacket(p.packet(), 0); !ok || packet.src != local.Addr() || packet.dst != p.src.Addr() || packet.payload[0]&^129 != 0 { t.Fatal("wrong echo reply: ", packet) } p.silent(20 * time.Millisecond) @@ -1873,7 +1961,7 @@ func TestFragments(t *testing.T) { if err := a.conn.(buf.Writer).WriteMultiBuffer(buf.MultiBuffer{buf.FromBytes(sent)}); err != nil { t.Fatal(err) } - packet, ok := parsePacket(p.joined()) + packet, ok := parsePacket(p.joined(), 0) u := packet.payload if !ok || packet.fragment || packet.proto != protoUDP || netip.AddrPortFrom(packet.src, binary.BigEndian.Uint16(u)) != p.dst || netip.AddrPortFrom(packet.dst, binary.BigEndian.Uint16(u[2:])) != p.src || int(binary.BigEndian.Uint16(u[4:])) != len(u) || !bytes.Equal(u[udpHeaderSize:], sent) { @@ -1898,7 +1986,7 @@ func TestFragments(t *testing.T) { mixed := split(summed(udpPacket(p.src, p.dst, bytes.Repeat([]byte{1}, 2000))), 1024, 11) l.in <- mixed[0] l.in <- split(summed(udpPacket(p.src, p.dst, bytes.Repeat([]byte{2}, 2000))), 1024, 11)[1] - packet, ok = parsePacket(p.joined()) + packet, ok = parsePacket(p.joined(), 0) m = packet.payload if !ok || packet.proto != proto || packet.src != p.dst.Addr() || packet.dst != p.src.Addr() || m[0] != reply || !bytes.Equal(m[8:], sent) { t.Fatal("wrong echo reply: ", packet.src, packet.dst, packet.proto, len(m)) @@ -2040,7 +2128,7 @@ func TestChecksumsAndLoopback(t *testing.T) { l.in <- damaged(echo) p.silent(50 * time.Millisecond) l.in <- echo - if packet, ok := parsePacket(p.packet()); !ok || packet.proto != proto || string(packet.payload[4:]) != "\x01\x02\x03\x04abcd" { + if packet, ok := parsePacket(p.packet(), 0); !ok || packet.proto != proto || string(packet.payload[4:]) != "\x01\x02\x03\x04abcd" { t.Fatal("wrong echo reply: ", packet) } diff --git a/proxy/tun/tcp_lightweight.go b/proxy/tun/tcp_lightweight.go index bb888b684..efef58b97 100644 --- a/proxy/tun/tcp_lightweight.go +++ b/proxy/tun/tcp_lightweight.go @@ -49,6 +49,9 @@ const ( // tcpParts of how many parts of what was written it is made at most tcpSegment = math.MaxUint16 - ipv6HeaderSize - tcpHeaderSize tcpParts = 7 + // tcpKept is how large a payload is at least for the buffers that it was read into to be kept. A smaller one + // is copied to where the one before it ends, for a reader to get buffers that are full + tcpKept = buf.Size / 2 // tcpTick is how often a connection that waits for something looks at what it waits for tcpTick = 100 * time.Millisecond @@ -87,7 +90,9 @@ type segment struct { flags byte window uint16 options []byte - payload []byte + // payload is without its last rest bytes when those are in other buffers of the stack, see stack.whole + payload []byte + rest int } // nextOption returns the first option of o and what follows it, kind is zero when there is none @@ -749,6 +754,12 @@ func (c *tcpConn) space() uint32 { // receive takes the payload of a segment. c.mu is held func (c *tcpConn) receive(s segment) { + // what is large, the next and all taken stays in the buffers that it was read into + size := len(s.payload) + s.rest + keep := c.stack.bufs != nil && size >= tcpKept && s.seq == c.rcvNxt && size <= int(c.window-c.rcvLen) + if !keep { + s.payload = c.stack.whole(s.payload, s.rest) + } seq, payload, fin := s.seq, s.payload, s.flags&tcpFin != 0 if len(payload) == 0 && !fin { if int32(c.rcvNxt-seq) > 0 { @@ -789,9 +800,14 @@ func (c *tcpConn) receive(s segment) { return } - c.rcvQueue = buf.MergeBytes(c.rcvQueue, payload) - c.rcvLen += int32(len(payload)) - c.rcvNxt += uint32(len(payload)) + if keep { + c.rcvQueue = c.stack.keep(c.rcvQueue, payload, s.rest) + } else { + size = len(payload) + c.rcvQueue = buf.MergeBytes(c.rcvQueue, payload) + } + c.rcvLen += int32(size) + c.rcvNxt += uint32(size) 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() { @@ -809,7 +825,7 @@ func (c *tcpConn) receive(s segment) { 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 { + if fin || now || c.rcvNxt-c.rcvAcked >= uint32(c.window/4) { c.sendAck() } else { c.arm(tcpTick) diff --git a/proxy/tun/tun.go b/proxy/tun/tun.go index 62c4c7c9d..fd6777e9a 100644 --- a/proxy/tun/tun.go +++ b/proxy/tun/tun.go @@ -16,10 +16,13 @@ type Tun interface { } // 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 +// splits them only if it has to send them on, and completes their checksum then. +// What it sends comes as it is too, not split to the MTU type segmenter interface { - // Segments reports whether WriteSegment can be used + // Segments reports whether ReadParts and WriteSegment can be used Segments() bool + // ReadParts waits for a packet, reads it into parts, one after the other, and returns its size + ReadParts(parts [][]byte) (int, error) // WriteSegment writes a packet with a TCP segment whose payload is in parts, to be sent on with mss bytes // of it at most in one. Its checksum is that of the pseudo-header only, not inverted WriteSegment(header []byte, payload [][]byte, mss int) error diff --git a/proxy/tun/tun_fd.go b/proxy/tun/tun_fd.go index 2df7d0b3b..230c55824 100644 --- a/proxy/tun/tun_fd.go +++ b/proxy/tun/tun_fd.go @@ -6,6 +6,8 @@ import ( "encoding/binary" "os" "sync" + "syscall" + "unsafe" "golang.org/x/sys/unix" ) @@ -22,6 +24,14 @@ type fdTun struct { // 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 + // raw is file for ReadParts, readv what it calls when there may be a packet, iovecs where that reads to: + // header, which takes the virtio_net_hdr, and the parts. read and readErr are what came of it + raw syscall.RawConn + readv func(fd uintptr) bool + header [packetRoom]byte + iovecs []unix.Iovec + read int + readErr error writeLock sync.RWMutex closed bool @@ -29,7 +39,25 @@ type fdTun struct { // 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} + t := &fdTun{fd: fd, file: os.NewFile(uintptr(fd), "/dev/net/tun"), vnet: vnet} + t.raw, _ = t.file.SyscallConn() + t.readv = func(fd uintptr) bool { + for { + n, _, errno := unix.Syscall(unix.SYS_READV, fd, uintptr(unsafe.Pointer(&t.iovecs[0])), uintptr(len(t.iovecs))) + switch errno { + case 0: + t.read, t.readErr = int(n), nil + case unix.EINTR: + continue + case unix.EAGAIN: + return false + default: + t.read, t.readErr = 0, errno + } + return true + } + } + return t } func (t *fdTun) ReadPacket(b []byte) ([]byte, error) { @@ -38,6 +66,20 @@ func (t *fdTun) ReadPacket(b []byte) ([]byte, error) { return b[min(t.vnet, n):n], err } +// ReadParts implements segmenter +func (t *fdTun) ReadParts(parts [][]byte) (int, error) { + t.iovecs = append(t.iovecs[:0], unix.Iovec{Base: &t.header[0]}) + t.iovecs[0].SetLen(t.vnet) + for i, p := range parts { + t.iovecs = append(t.iovecs, unix.Iovec{Base: &p[0]}) + t.iovecs[i+1].SetLen(len(p)) + } + if err := t.raw.Read(t.readv); err != nil { + return 0, err + } + return max(t.read-t.vnet, 0), t.readErr +} + func (t *fdTun) WritePacket(header, payload []byte) error { // a virtio_net_hdr of zeros leaves nothing to the system var vnet [packetRoom]byte diff --git a/proxy/tun/tun_fd_test.go b/proxy/tun/tun_fd_test.go new file mode 100644 index 000000000..de16d76ec --- /dev/null +++ b/proxy/tun/tun_fd_test.go @@ -0,0 +1,66 @@ +//go:build linux + +package tun + +import ( + "bytes" + "crypto/rand" + "testing" + "time" + + "golang.org/x/sys/unix" +) + +// A tun with a virtio_net_hdr reads one packet at a time into parts, without the header, waits for the next one, +// and stops waiting when it is closed. +func TestReadParts(t *testing.T) { + // what a tun is like: one packet for every read, and a file descriptor that does not block + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_SEQPACKET|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0) + if err != nil { + t.Fatal(err) + } + defer unix.Close(fds[1]) + tun := newFdTun(fds[0], vnetSize) + defer tun.closeFile() + + large := make([]byte, 20000) + rand.Read(large) + parts := [][]byte{make([]byte, 8192), make([]byte, 8192), make([]byte, 8192)} + send := func(packet []byte) { + if _, err := unix.Write(fds[1], append(bytes.Repeat([]byte{0xee}, vnetSize), packet...)); err != nil { + t.Error(err) + } + } + send(large) + send([]byte("small")) + for _, packet := range [][]byte{large, []byte("small")} { + n, err := tun.ReadParts(parts) + if err != nil || n != len(packet) || !bytes.Equal(bytes.Join(parts, nil)[:n], packet) { + t.Fatal("read ", n, " bytes of ", len(packet), ", error: ", err) + } + } + + go func() { + time.Sleep(50 * time.Millisecond) + send([]byte("later")) + }() + if n, err := tun.ReadParts(parts); err != nil || string(parts[0][:n]) != "later" { + t.Fatal("read ", parts[0][:n], " after waiting, error: ", err) + } + + read := make(chan error) + go func() { + _, err := tun.ReadParts(parts) + read <- err + }() + time.Sleep(50 * time.Millisecond) + tun.closeFile() + select { + case err := <-read: + if err == nil { + t.Fatal("read after the tun was closed") + } + case <-time.After(5 * time.Second): + t.Fatal("still waits after the tun was closed") + } +}