Implementing Peer Discovery Strategies

go.mod

@@ -3,13 +3,16 @@ module github.com/projectdiscovery/pd-agent
 go 1.24.1
 
 require (
+	github.com/google/gopacket v1.1.19
 	github.com/projectdiscovery/gcache v0.0.0-20241015120333-12546c6e3f4c
 	github.com/projectdiscovery/goflags v0.1.74
 	github.com/projectdiscovery/gologger v1.1.61
+	github.com/projectdiscovery/mapcidr v1.1.97
 	github.com/projectdiscovery/utils v0.7.3
 	github.com/rs/xid v1.6.0
 	github.com/shirou/gopsutil/v3 v3.23.7
 	github.com/tidwall/gjson v1.18.0
+	golang.org/x/net v0.47.0
 	golang.org/x/sys v0.38.0
 	k8s.io/apimachinery v0.34.2
 	k8s.io/client-go v0.34.2
@@ -77,7 +80,6 @@ require (
 	go4.org v0.0.0-20230225012048-214862532bf5 // indirect
 	golang.org/x/exp v0.0.0-20250813145105-42675adae3e6 // indirect
 	golang.org/x/mod v0.29.0 // indirect
-	golang.org/x/net v0.47.0 // indirect
 	golang.org/x/oauth2 v0.27.0 // indirect
 	golang.org/x/sync v0.18.0 // indirect
 	golang.org/x/term v0.37.0 // indirect

go.sum

@@ -95,6 +95,8 @@ github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -185,6 +187,8 @@ github.com/projectdiscovery/goflags v0.1.74 h1:n85uTRj5qMosm0PFBfsvOL24I7TdWRcWq
 github.com/projectdiscovery/goflags v0.1.74/go.mod h1:UMc9/7dFz2oln+10tv6cy+7WZKTHf9UGhaNkF95emh4=
 github.com/projectdiscovery/gologger v1.1.61 h1:+jJ0Z0x6X9s69IRjbtsnOfMD8YTFTVADHMKFNu6dUGg=
 github.com/projectdiscovery/gologger v1.1.61/go.mod h1:EfuwZ1lQX7kH4rgNo0nzk5XPh2j2gpYEQUi9tkoJDJw=
+github.com/projectdiscovery/mapcidr v1.1.97 h1:7FkxNNVXp+m1rIu5Nv/2SrF9k4+LwP8QuWs2puwy+2w=
+github.com/projectdiscovery/mapcidr v1.1.97/go.mod h1:9dgTJh1SP02gYZdpzMjm6vtYFkEHQHoTyaVNvaeJ7lA=
 github.com/projectdiscovery/utils v0.7.3 h1:kX+77AA58yK6EZgkTRJEnK9V/7AZYzlXdcu/o/kJhFs=
 github.com/projectdiscovery/utils v0.7.3/go.mod h1:uDdQ3/VWomai98l+a3Ye/srDXdJ4xUIar/mSXlQ9gBM=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -284,6 +288,7 @@ golang.org/x/lint v0.0.0-20190909230951-414d861bb4ac/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f/go.mod h1:5qLYkcX4OjUUV8bRuDixDT3tpyyb+LUpUlRWLxfhWrs=
 golang.org/x/lint v0.0.0-20200130185559-910be7a94367/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
+golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
 golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCcRqshq8CkpyQDoeVncDDYHnLhea+o=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=

pkg/peerdiscovery/arp/arp.go

@@ -0,0 +1,183 @@
+package arp
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/projectdiscovery/mapcidr"
+	"github.com/projectdiscovery/pd-agent/pkg/peerdiscovery/common"
+	mapsutil "github.com/projectdiscovery/utils/maps"
+	syncutil "github.com/projectdiscovery/utils/sync"
+)
+
+// Peer represents a discovered ARP peer
+type Peer struct {
+	IP  net.IP
+	MAC net.HardwareAddr
+}
+
+// DiscoverPeers retrieves all ARP peers by first reading the local ARP table,
+// then scanning the network in parallel to discover additional peers.
+func DiscoverPeers(ctx context.Context) ([]Peer, error) {
+	peers := mapsutil.NewSyncLockMap[string, *Peer]()
+
+	// Read local ARP table
+	localPeers, err := readLocalARPTable()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read local ARP table: %w", err)
+	}
+
+	for _, peer := range localPeers {
+		key := peer.IP.String()
+		peerCopy := peer
+		_ = peers.Set(key, &peerCopy)
+	}
+
+	// Get /24 network ranges from local interfaces
+	networks, err := common.GetLocalNetworks24()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get local networks: %w", err)
+	}
+
+	// Scan networks sequentially (no hurry)
+	for _, network := range networks {
+		select {
+		case <-ctx.Done():
+			goto done
+		default:
+		}
+
+		discovered, err := scanNetwork24(ctx, network)
+		if err != nil {
+			continue
+		}
+
+		for _, peer := range discovered {
+			key := peer.IP.String()
+			if _, exists := peers.Get(key); !exists {
+				peerCopy := peer
+				_ = peers.Set(key, &peerCopy)
+			}
+		}
+	}
+
+done:
+	// Convert map to slice
+	var result []Peer
+	_ = peers.Iterate(func(key string, peer *Peer) error {
+		if peer != nil {
+			result = append(result, *peer)
+		}
+		return nil
+	})
+
+	return result, nil
+}
+
+// scanNetwork24 scans a /24 network range to discover ARP peers
+// Uses UDP connections to trigger OS ARP requests and monitors the ARP table
+func scanNetwork24(ctx context.Context, network *net.IPNet) ([]Peer, error) {
+	// Verify it's a /24 network
+	ones, bits := network.Mask.Size()
+	if ones != 24 || bits != 32 {
+		return nil, fmt.Errorf("network %s is not a /24 network", network.String())
+	}
+
+	// Get initial ARP table state
+	initialPeers, err := readLocalARPTable()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read initial ARP table: %w", err)
+	}
+
+	initialSet := make(map[string]struct{})
+	for _, peer := range initialPeers {
+		if network.Contains(peer.IP) {
+			initialSet[peer.IP.String()] = struct{}{}
+		}
+	}
+
+	// Expand CIDR to get all IPs in /24 range
+	cidrStr := network.String()
+	ips, err := mapcidr.IPAddresses(cidrStr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to expand CIDR %s: %w", cidrStr, err)
+	}
+
+	if len(ips) == 0 {
+		return []Peer{}, nil
+	}
+
+	// Use adaptive waitgroup with low parallelism (no hurry)
+	awg, err := syncutil.New(syncutil.WithSize(5))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create adaptive waitgroup: %w", err)
+	}
+
+	// Trigger ARP resolution for each IP using UDP connections
+	for _, ipStr := range ips {
+		select {
+		case <-ctx.Done():
+			goto done
+		default:
+		}
+
+		ip := net.ParseIP(ipStr)
+		if ip == nil {
+			continue
+		}
+
+		// Skip network and broadcast addresses
+		if common.IsNetworkOrBroadcast(ip, network) {
+			continue
+		}
+
+		awg.Add()
+		go func(targetIP net.IP) {
+			defer awg.Done()
+
+			// Send UDP packet to trigger ARP resolution
+			// The OS will handle the ARP request for us
+			conn, err := net.DialTimeout("udp", net.JoinHostPort(targetIP.String(), "12345"), 50*time.Millisecond)
+			if err != nil {
+				// Connection will fail, but ARP resolution may occur
+				return
+			}
+			if conn != nil {
+				_ = conn.Close()
+			}
+		}(ip)
+
+		// Small delay between requests to avoid overwhelming
+		time.Sleep(10 * time.Millisecond)
+	}
+
+done:
+	awg.Wait()
+
+	// Wait for OS ARP requests to complete and ARP table to update
+	// Give it time since we're not in a hurry
+	time.Sleep(2 * time.Second)
+
+	// Read ARP table again to find new entries
+	finalPeers, err := readLocalARPTable()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read final ARP table: %w", err)
+	}
+
+	// Find newly discovered peers
+	var discovered []Peer
+	for _, peer := range finalPeers {
+		if !network.Contains(peer.IP) {
+			continue
+		}
+
+		// Check if this is a new peer
+		if _, exists := initialSet[peer.IP.String()]; !exists {
+			discovered = append(discovered, peer)
+		}
+	}
+
+	return discovered, nil
+}

pkg/peerdiscovery/arp/arp_unix.go

@@ -0,0 +1,145 @@
+//go:build !windows
+
+package arp
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+
+	osutils "github.com/projectdiscovery/utils/os"
+)
+
+// readLocalARPTable reads the local ARP table (Linux and macOS)
+func readLocalARPTable() ([]Peer, error) {
+	if osutils.IsLinux() {
+		return readLinuxARPTable()
+	} else if osutils.IsOSX() {
+		return readDarwinARPTable()
+	}
+	return nil, fmt.Errorf("unsupported OS")
+}
+
+// readLinuxARPTable reads ARP table from /proc/net/arp
+func readLinuxARPTable() ([]Peer, error) {
+	data, err := os.ReadFile("/proc/net/arp")
+	if err != nil {
+		return nil, err
+	}
+
+	var peers []Peer
+	scanner := bufio.NewScanner(strings.NewReader(string(data)))
+
+	// Skip header line
+	if !scanner.Scan() {
+		return peers, nil
+	}
+
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+
+		fields := strings.Fields(line)
+		if len(fields) < 6 {
+			continue
+		}
+
+		// Format: IP address HW type Flags HW address Mask Device
+		ipStr := fields[0]
+		macStr := fields[3]
+
+		// Skip incomplete entries
+		if macStr == "00:00:00:00:00:00" || macStr == "<incomplete>" {
+			continue
+		}
+
+		ip := net.ParseIP(ipStr)
+		if ip == nil || ip.To4() == nil {
+			continue
+		}
+
+		mac, err := net.ParseMAC(macStr)
+		if err != nil {
+			continue
+		}
+
+		peers = append(peers, Peer{
+			IP:  ip,
+			MAC: mac,
+		})
+	}
+
+	return peers, scanner.Err()
+}
+
+// readDarwinARPTable reads ARP table using 'arp -a' command on macOS
+func readDarwinARPTable() ([]Peer, error) {
+	cmd := exec.Command("arp", "-a")
+	output, err := cmd.Output()
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute arp -a: %w", err)
+	}
+
+	var peers []Peer
+	scanner := bufio.NewScanner(strings.NewReader(string(output)))
+
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+
+		// macOS arp -a format: "hostname (192.168.1.1) at aa:bb:cc:dd:ee:ff [ethernet] on en0"
+		// or: "? (192.168.1.1) at aa:bb:cc:dd:ee:ff [ethernet] on en0"
+
+		// Extract IP address (between parentheses)
+		ipStart := strings.Index(line, "(")
+		ipEnd := strings.Index(line, ")")
+		if ipStart == -1 || ipEnd == -1 || ipStart >= ipEnd {
+			continue
+		}
+		ipStr := line[ipStart+1 : ipEnd]
+
+		// Extract MAC address (after "at ")
+		atIndex := strings.Index(line, " at ")
+		if atIndex == -1 {
+			continue
+		}
+		macStart := atIndex + 4
+		macEnd := strings.Index(line[macStart:], " ")
+		if macEnd == -1 {
+			macEnd = strings.Index(line[macStart:], "[")
+		}
+		if macEnd == -1 {
+			macEnd = len(line) - macStart
+		}
+		macStr := strings.TrimSpace(line[macStart : macStart+macEnd])
+
+		// Skip incomplete entries
+		if macStr == "(incomplete)" || macStr == "" {
+			continue
+		}
+
+		ip := net.ParseIP(ipStr)
+		if ip == nil || ip.To4() == nil {
+			continue
+		}
+
+		mac, err := net.ParseMAC(macStr)
+		if err != nil {
+			continue
+		}
+
+		peers = append(peers, Peer{
+			IP:  ip,
+			MAC: mac,
+		})
+	}
+
+	return peers, scanner.Err()
+}