spb/internetpressure

1// Package ping implements the "ping" check (ICMP echo through an unprivileged datagram socket) and the "tcp"2// check (sequential TCP connects), which is also the fallback when ICMP sockets are not available.3//4// Unprivileged ICMP: golang.org/x/net/icmp "udp4"/"udp6" endpoints work out of the box on macOS; on Linux5// they require the probe's group to be within net.ipv4.ping_group_range (see README). When the socket cannot be6// opened the check falls back to kind "tcp" and reports the reason "icmp_unavailable".7package ping89import (10	"context"11	"crypto/rand"12	"encoding/binary"13	"errors"14	"math"15	"net"16	"os"17	"runtime"18	"sort"19	"time"2021	"golang.org/x/net/icmp"22	"golang.org/x/net/ipv4"23	"golang.org/x/net/ipv6"2425	"internetpressure.io/probe-agent/internal/protocol"26)2728const (29	// Count is the number of echoes / connects per check.30	Count = 531	// Spacing between consecutive echoes.32	Spacing = 200 * time.Millisecond33	// EchoTimeout is the wait for each reply.34	EchoTimeout = 2 * time.Second35	// ResolveTimeout bounds the one hostname lookup per run.36	ResolveTimeout = 3 * time.Second3738	protoICMP   = 1  // iana.ProtocolICMP39	protoICMPv6 = 58 // iana.ProtocolIPv6ICMP40)4142// Checker runs ping / tcp checks.43type Checker struct {44	// ICMP is true when an unprivileged ICMP socket could be opened at start-up (see Detect).45	ICMP bool46}4748// Detect reports whether an unprivileged ICMPv4 datagram socket can be opened on this host.49func Detect() bool {50	c, err := icmp.ListenPacket("udp4", "0.0.0.0")51	if err != nil {52		return false53	}54	c.Close()55	return true56}5758// Resolve picks the address to ping: the pinned IP if any, else the first IPv4 (then IPv6) of the hostname,59// resolved once with the system resolver.60func Resolve(ctx context.Context, t protocol.Target) (net.IP, error) {61	if ip := t.FixedIP(); ip != "" {62		parsed := net.ParseIP(ip)63		if parsed == nil {64			return nil, errors.New("invalid pinned ip")65		}66		return parsed, nil67	}68	ctx, cancel := context.WithTimeout(ctx, ResolveTimeout)69	defer cancel()70	addrs, err := net.DefaultResolver.LookupIPAddr(ctx, t.Hostname)71	if err != nil {72		return nil, err73	}74	for _, a := range addrs {75		if a.IP.To4() != nil {76			return a.IP, nil77		}78	}79	if len(addrs) > 0 {80		return addrs[0].IP, nil81	}82	return nil, errors.New("no address")83}8485// Run performs the ping check for t: 5 ICMP echoes 200 ms apart, or the tcp fallback.86func (c *Checker) Run(ctx context.Context, t protocol.Target) protocol.Measurement {87	m := protocol.Measurement{TS: protocol.FormatTime(time.Now()), TargetID: t.TargetID, Kind: "ping"}88	ip, err := Resolve(ctx, t)89	if err != nil {90		m.Error = protocol.ErrDNSFail91		m.Sent, m.Received, m.PacketLoss = protocol.I(0), protocol.I(0), protocol.F(1)92		return m93	}94	m.ResolvedIP = ip.String()9596	rtts, sent, icmpErr := echo(ctx, ip)97	if icmpErr != nil {98		// Socket could not be opened → tcp fallback with the reason kept in error.99		port := t.Port100		if port == 0 {101			port = 443102		}103		rtts, sent = connects(ctx, ip, port)104		m.Kind = "tcp"105		summarise(&m, rtts, sent)106		if m.OK {107			m.Error = protocol.ErrICMPUnavailable108		}109		return m110	}111	summarise(&m, rtts, sent)112	return m113}114115// RunTCP performs the "tcp" check (5 sequential connects to t.Port, default 443).116func (c *Checker) RunTCP(ctx context.Context, t protocol.Target) protocol.Measurement {117	m := protocol.Measurement{TS: protocol.FormatTime(time.Now()), TargetID: t.TargetID, Kind: "tcp"}118	ip, err := Resolve(ctx, t)119	if err != nil {120		m.Error = protocol.ErrDNSFail121		m.Sent, m.Received, m.PacketLoss = protocol.I(0), protocol.I(0), protocol.F(1)122		return m123	}124	m.ResolvedIP = ip.String()125	port := t.Port126	if port == 0 {127		port = 443128	}129	rtts, sent := connects(ctx, ip, port)130	summarise(&m, rtts, sent)131	return m132}133134// summarise fills sent/received/loss/rtt stats/jitter and ok/error.135func summarise(m *protocol.Measurement, rtts []time.Duration, sent int) {136	m.Sent = protocol.I(sent)137	m.Received = protocol.I(len(rtts))138	if sent > 0 {139		m.PacketLoss = protocol.F(math.Round((1-float64(len(rtts))/float64(sent))*1000) / 1000)140	} else {141		m.PacketLoss = protocol.F(1)142	}143	if len(rtts) == 0 {144		m.Error = protocol.ErrUnreachable145		return146	}147	m.OK = true148	// Jitter = mean absolute successive difference, in arrival order.149	var jitter float64150	for i := 1; i < len(rtts); i++ {151		jitter += math.Abs(protocol.Ms(rtts[i]) - protocol.Ms(rtts[i-1]))152	}153	if len(rtts) > 1 {154		jitter /= float64(len(rtts) - 1)155	}156	m.JitterMs = protocol.F(round1(jitter))157158	sorted := append([]time.Duration(nil), rtts...)159	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })160	var sum time.Duration161	for _, r := range sorted {162		sum += r163	}164	m.RTTMinMs = protocol.F(protocol.Ms(sorted[0]))165	m.RTTMaxMs = protocol.F(protocol.Ms(sorted[len(sorted)-1]))166	m.RTTAvgMs = protocol.F(round1(protocol.Ms(sum) / float64(len(sorted))))167}168169func round1(v float64) float64 { return math.Round(v*1000) / 1000 }170171// echo sends Count ICMP echo requests, one at a time, waiting up to EchoTimeout for each reply and keeping172// Spacing between sends. It returns (rtts of received replies, number sent, socket error).173func echo(ctx context.Context, dst net.IP) ([]time.Duration, int, error) {174	network, laddr, proto := "udp4", "0.0.0.0", protoICMP175	var typ, replyType icmp.Type = ipv4.ICMPTypeEcho, ipv4.ICMPTypeEchoReply176	if dst.To4() == nil {177		network, laddr, proto = "udp6", "::", protoICMPv6178		typ, replyType = ipv6.ICMPTypeEchoRequest, ipv6.ICMPTypeEchoReply179	}180	conn, err := icmp.ListenPacket(network, laddr)181	if err != nil {182		return nil, 0, err183	}184	defer conn.Close()185186	var token [8]byte187	_, _ = rand.Read(token[:])188	id := os.Getpid() & 0xffff189	peer := &net.UDPAddr{IP: dst}190	buf := make([]byte, 1500)191	var rtts []time.Duration192	sent := 0193194	for seq := 1; seq <= Count; seq++ {195		if ctx.Err() != nil {196			break197		}198		payload := make([]byte, 16)199		copy(payload, token[:])200		binary.BigEndian.PutUint64(payload[8:], uint64(time.Now().UnixNano()))201		msg := icmp.Message{Type: typ, Code: 0, Body: &icmp.Echo{ID: id, Seq: seq, Data: payload}}202		wb, err := msg.Marshal(nil)203		if err != nil {204			return rtts, sent, nil205		}206		sendAt := time.Now()207		if _, err := conn.WriteTo(wb, peer); err != nil {208			sent++209			sleepUntil(ctx, sendAt.Add(Spacing))210			continue211		}212		sent++213		deadline := sendAt.Add(EchoTimeout)214		if d, ok := ctx.Deadline(); ok && d.Before(deadline) {215			deadline = d216		}217		for {218			_ = conn.SetReadDeadline(deadline)219			n, from, err := conn.ReadFrom(buf)220			if err != nil {221				break // timeout for this echo222			}223			if !sameIP(from, dst) {224				continue225			}226			body, ok := parseReply(buf[:n], proto, replyType)227			if !ok || body.Seq != seq || len(body.Data) < 8 || string(body.Data[:8]) != string(token[:]) {228				continue229			}230			rtts = append(rtts, time.Since(sendAt))231			break232		}233		if seq < Count {234			sleepUntil(ctx, sendAt.Add(Spacing))235		}236	}237	return rtts, sent, nil238}239240// parseReply decodes an ICMP packet, tolerating the IPv4 header that macOS prepends on udp4 sockets.241func parseReply(b []byte, proto int, want icmp.Type) (*icmp.Echo, bool) {242	try := func(p []byte) (*icmp.Echo, bool) {243		msg, err := icmp.ParseMessage(proto, p)244		if err != nil || msg.Type != want {245			return nil, false246		}247		e, ok := msg.Body.(*icmp.Echo)248		return e, ok249	}250	if e, ok := try(b); ok {251		return e, true252	}253	if proto == protoICMP && len(b) > ipv4.HeaderLen && b[0]>>4 == 4 {254		hl := int(b[0]&0x0f) * 4255		if hl >= ipv4.HeaderLen && len(b) > hl {256			return try(b[hl:])257		}258	}259	return nil, false260}261262func sameIP(a net.Addr, ip net.IP) bool {263	switch v := a.(type) {264	case *net.UDPAddr:265		return v.IP.Equal(ip)266	case *net.IPAddr:267		return v.IP.Equal(ip)268	}269	return false270}271272// connects performs Count sequential TCP connects to ip:port and returns connect durations.273func connects(ctx context.Context, ip net.IP, port int) ([]time.Duration, int) {274	addr := net.JoinHostPort(ip.String(), itoa(port))275	var rtts []time.Duration276	sent := 0277	d := &net.Dialer{Timeout: EchoTimeout}278	for i := 0; i < Count; i++ {279		if ctx.Err() != nil {280			break281		}282		start := time.Now()283		sent++284		conn, err := d.DialContext(ctx, "tcp", addr)285		if err == nil {286			rtts = append(rtts, time.Since(start))287			conn.Close()288		}289		if i < Count-1 {290			sleepUntil(ctx, start.Add(Spacing))291		}292	}293	return rtts, sent294}295296func sleepUntil(ctx context.Context, t time.Time) {297	d := time.Until(t)298	if d <= 0 {299		return300	}301	select {302	case <-ctx.Done():303	case <-time.After(d):304	}305}306307func itoa(n int) string {308	if n == 0 {309		return "0"310	}311	var b [20]byte312	i := len(b)313	for n > 0 {314		i--315		b[i] = byte('0' + n%10)316		n /= 10317	}318	return string(b[i:])319}320321// LinuxHint is logged when ICMP is unavailable on Linux.322func LinuxHint() string {323	if runtime.GOOS == "linux" {324		return "set `sysctl -w net.ipv4.ping_group_range=\"0 2147483647\"` to enable unprivileged ICMP"325	}326	return ""327}328