[Heartbeat] Rootless pings (#13795)

This PR attempts to make #11610 by @leopucci, which makes rootless ping possible merge-able (e.g. CI passes). The code looked good, but I think there are some remaining nits. This also changes the way errors are reported to make it cleaner and more testable.

Takes over for #13349 as well.

CHANGELOG.next.asciidoc

@@ -314,6 +314,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support to set the document id in the json reader. {pull}5844[5844]
 
 *Heartbeat*
+- Add non-privileged icmp on linux and darwin(mac). {pull}13795[13795] {issue}11498[11498]
 
 - Enable `add_observer_metadata` processor in default config. {pull}11394[11394]
 - Record HTTP body metadata and optionally contents in `http.response.body.*` fields. {pull}13022[13022]

heartbeat/monitors/active/icmp/icmp.go

@@ -47,12 +47,6 @@ func create(
 		return nil, 0, err
 	}
 
-	// TODO: check icmp is support by OS + check we've
-	// got required credentials (implementation uses RAW socket, requires root +
-	// not supported on all OSes)
-	// TODO: replace icmp package base reader/sender using raw sockets with
-	//       OS specific solution
-
 	ipVersion := config.Mode.Network()
 	if len(config.Hosts) > 0 && ipVersion == "" {
 		err := fmt.Errorf("pinging hosts requires ipv4 or ipv6 mode enabled")
@@ -61,13 +55,14 @@ func create(
 
 	var loopErr error
 	loopInit.Do(func() {
-		debugf("initialize icmp handler")
+		debugf("initializing ICMP loop")
 		loop, loopErr = newICMPLoop()
 	})
 	if loopErr != nil {
-		debugf("Failed to initialize ICMP loop %v", loopErr)
+		logp.Warn("Failed to initialize ICMP loop %v", loopErr)
 		return nil, 0, loopErr
 	}
+	debugf("ICMP loop successfully initialized")
 
 	if err := loop.checkNetworkMode(ipVersion); err != nil {
 		return nil, 0, err

heartbeat/monitors/active/icmp/loop.go

@@ -24,14 +24,14 @@ import (
 	"fmt"
 	"math/rand"
 	"net"
+	"os"
+	"runtime"
 	"sync"
 	"time"
 
 	"golang.org/x/net/icmp"
 	"golang.org/x/net/ipv4"
 	"golang.org/x/net/ipv6"
-
-	"github.com/elastic/beats/libbeat/logp"
 )
 
 type icmpLoop struct {
@@ -85,26 +85,52 @@ var (
 	loop     *icmpLoop
 )
 
+func noPingCapabilityError(message string) error {
+	return fmt.Errorf(fmt.Sprintf("Insufficient privileges to perform ICMP ping. %s", message))
+}
+
 func newICMPLoop() (*icmpLoop, error) {
 	// Log errors at info level, as the loop is setup globally when ICMP module is loaded
 	// first (not yet configured).
 	// With multiple configurations using the icmp loop, we have to postpose
 	// IPv4/IPv6 checking
 	conn4 := createListener("IPv4", "ip4:icmp")
 	conn6 := createListener("IPv6", "ip6:ipv6-icmp")
-
+	unprivilegedPossible := false
 	l := &icmpLoop{
 		conn4:    conn4,
 		conn6:    conn6,
 		recv:     make(chan packet, 16),
 		requests: map[requestID]*requestContext{},
 	}
 
-	if conn4 != nil {
-		go l.runICMPRecv(conn4, protocolICMP)
+	if l.conn4 == nil && l.conn6 == nil {
+		switch runtime.GOOS {
+		case "linux", "darwin":
+			unprivilegedPossible = true
+			//This is non-privileged ICMP, not udp
+			l.conn4 = createListener("Unprivileged IPv4", "udp4")
+			l.conn6 = createListener("Unprivileged IPv6", "udp6")
+		}
 	}
-	if conn6 != nil {
-		go l.runICMPRecv(conn6, protocolIPv6ICMP)
+
+	if l.conn4 != nil {
+		go l.runICMPRecv(l.conn4, protocolICMP)
+	}
+	if l.conn6 != nil {
+		go l.runICMPRecv(l.conn6, protocolIPv6ICMP)
+	}
+
+	if l.conn4 == nil && l.conn6 == nil {
+		if unprivilegedPossible {
+			var buffer bytes.Buffer
+			path, _ := os.Executable()
+			buffer.WriteString("You can run without root by setting cap_net_raw:\n sudo setcap cap_net_raw+eip ")
+			buffer.WriteString(path + " \n")
+			buffer.WriteString("Your system allows the use of unprivileged ping by setting net.ipv4.ping_group_range \n sysctl -w net.ipv4.ping_group_range='<min-uid> <max-uid>' ")
+			return nil, noPingCapabilityError(buffer.String())
+		}
+		return nil, noPingCapabilityError("You must provide the appropriate permissions to this executable")
 	}
 
 	return l, nil
@@ -124,10 +150,10 @@ func (l *icmpLoop) checkNetworkMode(mode string) error {
 	}
 
 	if ip4 && l.conn4 == nil {
-		return errors.New("failed to initiate IPv4 support")
+		return errors.New("failed to initiate IPv4 support. Check log details for permission configuration")
 	}
 	if ip6 && l.conn6 == nil {
-		return errors.New("failed to initiate IPv6 support")
+		return errors.New("failed to initiate IPv6 support. Check log details for permission configuration")
 	}
 
 	return nil
@@ -272,6 +298,7 @@ func (l *icmpLoop) ping(
 	if !success {
 		return 0, requests, timeoutError{}
 	}
+
 	return rtt, requests, nil
 }
 
@@ -344,7 +371,6 @@ func createListener(name, network string) *icmp.PacketConn {
 	//      true, even if error value itself is `nil`. Checking for conn suppresses
 	//      misleading log message.
 	if conn == nil && err != nil {
-		logp.Info("%v ICMP not supported: %v", name, err)
 		return nil
 	}
 	return conn

heartbeat/tests/system/test_icmp.py

@@ -0,0 +1,52 @@
+import os
+import unittest
+import platform
+import socket
+import sys
+from heartbeat import BaseTest
+from elasticsearch import Elasticsearch
+from beat.beat import INTEGRATION_TESTS
+import nose.tools
+import logging
+import subprocess
+import time
+
+
+class Test(BaseTest):
+    def test_base(self):
+        """
+        Basic test with icmp root non privilege ICMP test.
+
+        """
+
+        config = {
+            "monitors": [
+                {
+                    "type": "icmp",
+                    "schedule": "*/5 * * * * * *",
+                    "hosts": ["127.0.0.1"],
+                }
+            ]
+        }
+
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/*",
+            **config
+        )
+
+        proc = self.start_beat()
+
+        def has_started_message(): return self.log_contains("ICMP loop successfully initialized")
+
+        def has_failed_message(): return self.log_contains("Failed to initialize ICMP loop")
+
+        # We don't know if the system tests are running is configured to support or not support ping, but we can at least check that the ICMP loop
+        # was initiated. In the future we should start up VMs with the correct perms configured and be more specific. In addition to that
+        # we should run pings on those machines and make sure they work.
+        self.wait_until(lambda: has_started_message() or has_failed_message(), 30)
+
+        if has_failed_message():
+            proc.check_kill_and_wait(1)
+        else:
+            # Check that documents are moving through
+            self.wait_until(lambda: self.output_has(lines=1))