test/ping/probing/icmp/packetconn.go

package icmp

import (
	"encoding/binary"
	"fmt"
	"net"
	"sync"
	"syscall"
	"time"

	"github.com/google/uuid"
	"golang.org/x/net/icmp"
	"golang.org/x/net/ipv4"
	"golang.org/x/net/ipv6"
)

const (
	protocolICMP     = 1
	protocolIPv6ICMP = 58
)

var ENOLISTENER = fmt.Errorf("no listener")

type Type icmp.Type

type PacketConn interface {
	Close() error
	ICMPRequestType() Type
	ReadFrom(b []byte) (n int, ttl int, src net.Addr, err error)
	SetFlagTTL() error
	SetReadDeadline(t time.Time) error
	WriteTo(b []byte, dst net.Addr) (int, error)
	SetTTL(ttl int)
}

var (
	ipv4Proto = map[string]string{"icmp": "ip4:icmp", "udp": "udp4"}
	ipv6Proto = map[string]string{"icmp": "ip6:ipv6-icmp", "udp": "udp6"}
)

// Packet represents a received and processed ICMP echo packet.
type Packet struct {
	// IPAddr is the address of the host being pinged.
	IPAddr *net.IPAddr
	// ID is the ICMP identifier.
	ID int
	// Seq is the ICMP sequence number.
	Seq int
	// UUID
	UUID uuid.UUID
	// TTL is the Time To Live on the packet.
	TTL int
	// NBytes is the number of bytes in the message.
	Nbytes int
	// SendTime
	SendTime time.Time
	// Rtt is the round-trip time it took to ping.
	Rtt time.Duration
	//
	TimeoutTimer *time.Timer
}

type recvPkt struct {
	recvtime time.Time
	addr     net.Addr
	bytes    []byte
	nbytes   int
	ttl      int
}

type MPacketConn struct {
	IPV4         bool
	Protocol     string
	Source       string
	Backlog      int
	TTL          int
	OnRecvPacket func(pkt *Packet)
	OnError      func(error)
	mutex        sync.Mutex
	conn         PacketConn
	done         chan interface{}
	recvbuf      chan *recvPkt
}

func (mp *MPacketConn) Listen() error {
	mp.mutex.Lock()
	defer mp.mutex.Unlock()

	if mp.conn != nil {
		return nil
	}

	conn, err := mp.listen()
	if err != nil {
		return err
	}

	mp.done = make(chan interface{})
	mp.recvbuf = make(chan *recvPkt, mp.Backlog)
	mp.conn = conn
	go mp.recvICMP()
	go mp.processRecvPacket()

	return nil
}

func (mp *MPacketConn) listen() (conn PacketConn, err error) {
	if mp.IPV4 {
		var c icmpv4Conn
		c.c, err = icmp.ListenPacket(ipv4Proto[mp.Protocol], mp.Source)
		conn = &c
	} else {
		var c icmpV6Conn
		c.c, err = icmp.ListenPacket(ipv6Proto[mp.Protocol], mp.Source)
		conn = &c
	}
	if err != nil {
		return nil, err
	}
	conn.SetTTL(mp.TTL)
	if err := conn.SetFlagTTL(); err != nil {
		conn.Close()
		return nil, err
	}
	return conn, nil
}

func (mp *MPacketConn) Close() error {
	mp.mutex.Lock()
	defer mp.mutex.Unlock()

	open := true
	select {
	case _, open = <-mp.done:
	default:
	}

	if open {
		close(mp.done)
	}

	if mp.conn != nil {
		mp.conn.Close()
		mp.conn = nil
	}

	return nil
}

func (mp *MPacketConn) recvICMP() {
	bytes := make([]byte, 65536)
	for {
		select {
		case <-mp.done:
			return
		default:
			conn := mp.conn
			if conn == nil {
				return
			}
			var n, ttl int
			var addr net.Addr
			var err error
			n, ttl, addr, err = conn.ReadFrom(bytes)
			if err != nil {
				if neterr, ok := err.(*net.OpError); ok {
					if neterr.Timeout() {
						// Read timeout
						continue
					}
				}
				if mp.OnError != nil {
					mp.OnError(err)
				} else {
					fmt.Println("ReadFrom Error:", err)
				}
			}
			bs := make([]byte, n)
			copy(bs, bytes[:n])
			select {
			case <-mp.done:
				return
			case mp.recvbuf <- &recvPkt{recvtime: time.Now(), addr: addr, bytes: bs, nbytes: n, ttl: ttl}:
			}
		}
	}
}

func (mp *MPacketConn) SendPacket(pkt *Packet) error {
	conn := mp.conn
	if conn == nil {
		return ENOLISTENER
	}
	var dst net.Addr = pkt.IPAddr
	if mp.Protocol == "udp" {
		dst = &net.UDPAddr{IP: pkt.IPAddr.IP, Zone: pkt.IPAddr.Zone}
	}
	for {
		select {
		case <-mp.done:
			return nil
		default:
		}
		msgBytes, err := pkt.BuildEchoRequestMessage(conn.ICMPRequestType())
		if err != nil {
			return err
		}
		if _, err := conn.WriteTo(msgBytes, dst); err != nil {
			if neterr, ok := err.(*net.OpError); ok {
				if neterr.Err == syscall.ENOBUFS {
					if mp.OnError != nil {
						mp.OnError(neterr.Err)
					} else {
						// 运行时默认忽略缓存不够错误
						// fmt.Println("缓存不够，发送失败，重发")
					}
					continue
				}
			}
			return err
		} else {
			return nil
		}
	}
}

var max_receive_buffer_used = 0

func MaxReceiveBufferUsed() int {
	return max_receive_buffer_used
}

func (mp *MPacketConn) processRecvPacket() {
	for pkt := range mp.recvbuf {
		if len(mp.recvbuf) > max_receive_buffer_used {
			max_receive_buffer_used = len(mp.recvbuf)
		}
		err := mp.processPacket(pkt)
		if err != nil {
			if mp.OnError != nil {
				mp.OnError(err)
			} else {
				// 运行时默认忽略接收数据格式不符错误
				// fmt.Println(err)
			}
		}
	}
}

var count = 0

func (mp *MPacketConn) processPacket(recv *recvPkt) error {
	var proto int
	if mp.IPV4 {
		proto = protocolICMP
	} else {
		proto = protocolIPv6ICMP
	}

	// fmt.Println(count, "from", recv.addr.String(), "bytes", recv.bytes)

	var m *icmp.Message
	var err error
	if m, err = icmp.ParseMessage(proto, recv.bytes); err != nil {
		return fmt.Errorf("error parsing icmp message: %w", err)
	}

	if m.Type != ipv4.ICMPTypeEchoReply && m.Type != ipv6.ICMPTypeEchoReply {
		// Not an echo reply, ignore it
		return nil
	}

	switch pkt := m.Body.(type) {
	case *icmp.Echo:
		return mp.processEchoReply(pkt, recv)
	default:
		// Very bad, not sure how this can happen
		return fmt.Errorf("invalid ICMP echo reply; type: '%T', '%v'", pkt, pkt)
	}
}

func (mp *MPacketConn) processEchoReply(pkt *icmp.Echo, recv *recvPkt) error {
	if len(pkt.Data) < 40 {
		return nil
	}

	sendtime := int64(binary.BigEndian.Uint64(pkt.Data[:8]))
	fullseq := int(binary.BigEndian.Uint64(pkt.Data[8:16]))
	fullid := int(binary.BigEndian.Uint64(pkt.Data[16:24]))
	pktuuid := uuid.Must(uuid.FromBytes(pkt.Data[24:40]))

	// Linux 下 UDP 方式，接收的 EchoReply.ID 与发送的 Echo.ID 是不一致的
	// if fullid%65536 != pkt.ID || fullseq%65536 != pkt.Seq {
	// 	return nil
	// }

	// fmt.Printf("%s %d bytes from %s: icmp_seq=%d time=%v\n",
	// 	time.Now().Format("15:04:05.000"), recv.nbytes, recv.addr, fullseq, recv.recvtime.Sub(time.Unix(0, sendtime)))

	if mp.OnRecvPacket != nil {
		mp.OnRecvPacket(&Packet{
			IPAddr:   netAddrToIPAddr(recv.addr),
			ID:       fullid,
			Seq:      fullseq,
			UUID:     pktuuid,
			Nbytes:   recv.nbytes,
			TTL:      recv.ttl,
			SendTime: time.Unix(0, sendtime),
			Rtt:      recv.recvtime.Sub(time.Unix(0, sendtime)),
		})
	}
	return nil
}