/
peerdiscovery.go
403 lines (347 loc) · 9.89 KB
/
peerdiscovery.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
package peerdiscovery
import (
"fmt"
"net"
"strconv"
"sync"
"time"
"golang.org/x/net/ipv4"
"golang.org/x/net/ipv6"
)
// IPVersion specifies the version of the Internet Protocol to be used.
type IPVersion uint
const (
IPv4 IPVersion = 4
IPv6 IPVersion = 6
)
// Discovered is the structure of the discovered peers,
// which holds their local address (port removed) and
// a payload if there is one.
type Discovered struct {
// Address is the local address of a discovered peer.
Address string
// Payload is the associated payload from discovered peer.
Payload []byte
}
func (d Discovered) String() string {
return fmt.Sprintf("address: %s, payload: %s", d.Address, d.Payload)
}
// Settings are the settings that can be specified for
// doing peer discovery.
type Settings struct {
// Limit is the number of peers to discover, use < 1 for unlimited.
Limit int
// Port is the port to broadcast on (the peers must also broadcast using the same port).
// The default port is 9999.
Port string
// MulticastAddress specifies the multicast address.
// You should be able to use any of 224.0.0.0/4 or ff00::/8.
// By default it uses the Simple Service Discovery Protocol
// address (239.255.255.250 for IPv4 or ff02::c for IPv6).
MulticastAddress string
// Payload is the bytes that are sent out with each broadcast. Must be short.
Payload []byte
// Delay is the amount of time between broadcasts. The default delay is 1 second.
Delay time.Duration
// TimeLimit is the amount of time to spend discovering, if the limit is not reached.
// A negative limit indiciates scanning until the limit was reached or, if an
// unlimited scanning was requested, no timeout.
// The default time limit is 10 seconds.
TimeLimit time.Duration
// StopChan is a channel to stop the peer discvoery immediatley after reception.
StopChan chan struct{}
// AllowSelf will allow discovery the local machine (default false)
AllowSelf bool
// DisableBroadcast will not allow sending out a broadcast
DisableBroadcast bool
// IPVersion specifies the version of the Internet Protocol (default IPv4)
IPVersion IPVersion
// Notify will be called each time a new peer was discovered.
// The default is nil, which means no notification whatsoever.
Notify func(Discovered)
portNum int
multicastAddressNumbers net.IP
}
// peerDiscovery is the object that can do the discovery for finding LAN peers.
type peerDiscovery struct {
settings Settings
received map[string][]byte
sync.RWMutex
}
// initialize returns a new peerDiscovery object which can be used to discover peers.
// The settings are optional. If any setting is not supplied, then defaults are used.
// See the Settings for more information.
func initialize(settings Settings) (p *peerDiscovery, err error) {
p = new(peerDiscovery)
p.Lock()
defer p.Unlock()
// initialize settings
p.settings = settings
// defaults
if p.settings.Port == "" {
p.settings.Port = "9999"
}
if p.settings.IPVersion == 0 {
p.settings.IPVersion = IPv4
}
if p.settings.MulticastAddress == "" {
if p.settings.IPVersion == IPv4 {
p.settings.MulticastAddress = "239.255.255.250"
} else {
p.settings.MulticastAddress = "ff02::c"
}
}
if len(p.settings.Payload) == 0 {
p.settings.Payload = []byte("hi")
}
if p.settings.Delay == 0 {
p.settings.Delay = 1 * time.Second
}
if p.settings.TimeLimit == 0 {
p.settings.TimeLimit = 10 * time.Second
}
if p.settings.StopChan == nil {
p.settings.StopChan = make(chan struct{})
}
p.received = make(map[string][]byte)
p.settings.multicastAddressNumbers = net.ParseIP(p.settings.MulticastAddress)
if p.settings.multicastAddressNumbers == nil {
err = fmt.Errorf("Multicast Address %s could not be converted to an IP",
p.settings.MulticastAddress)
return
}
p.settings.portNum, err = strconv.Atoi(p.settings.Port)
if err != nil {
return
}
return
}
type NetPacketConn interface {
JoinGroup(ifi *net.Interface, group net.Addr) error
SetMulticastInterface(ini *net.Interface) error
SetMulticastTTL(int) error
ReadFrom(buf []byte) (int, net.Addr, error)
WriteTo(buf []byte, dst net.Addr) (int, error)
}
// Discover will use the created settings to scan for LAN peers. It will return
// an array of the discovered peers and their associate payloads. It will not
// return broadcasts sent to itself.
func Discover(settings ...Settings) (discoveries []Discovered, err error) {
s := Settings{}
if len(settings) > 0 {
s = settings[0]
}
p, err := initialize(s)
if err != nil {
return
}
p.RLock()
address := net.JoinHostPort(p.settings.MulticastAddress, p.settings.Port)
portNum := p.settings.portNum
payload := p.settings.Payload
tickerDuration := p.settings.Delay
timeLimit := p.settings.TimeLimit
p.RUnlock()
// get interfaces
ifaces, err := net.Interfaces()
if err != nil {
return
}
// Open up a connection
c, err := net.ListenPacket(fmt.Sprintf("udp%d", p.settings.IPVersion), address)
if err != nil {
return
}
defer c.Close()
group := p.settings.multicastAddressNumbers
// ipv{4,6} have an own PacketConn, which does not implement net.PacketConn
var p2 NetPacketConn
if p.settings.IPVersion == IPv4 {
p2 = PacketConn4{ipv4.NewPacketConn(c)}
} else {
p2 = PacketConn6{ipv6.NewPacketConn(c)}
}
for i := range ifaces {
p2.JoinGroup(&ifaces[i], &net.UDPAddr{IP: group, Port: portNum})
}
go p.listen()
ticker := time.NewTicker(tickerDuration)
defer ticker.Stop()
start := time.Now()
for {
exit := false
p.RLock()
if len(p.received) >= p.settings.Limit && p.settings.Limit > 0 {
exit = true
}
p.RUnlock()
if !s.DisableBroadcast {
// write to multicast
broadcast(p2, payload, ifaces, &net.UDPAddr{IP: group, Port: portNum})
}
select {
case <-p.settings.StopChan:
exit = true
case <-ticker.C:
}
if exit || timeLimit > 0 && time.Since(start) > timeLimit {
break
}
}
if !s.DisableBroadcast {
// send out broadcast that is finished
broadcast(p2, payload, ifaces, &net.UDPAddr{IP: group, Port: portNum})
}
p.RLock()
discoveries = make([]Discovered, len(p.received))
i := 0
for ip, payload := range p.received {
discoveries[i] = Discovered{
Address: ip,
Payload: payload,
}
i++
}
p.RUnlock()
return
}
func broadcast(p2 NetPacketConn, payload []byte, ifaces []net.Interface, dst net.Addr) {
for i := range ifaces {
if errMulticast := p2.SetMulticastInterface(&ifaces[i]); errMulticast != nil {
continue
}
p2.SetMulticastTTL(2)
if _, errMulticast := p2.WriteTo([]byte(payload), dst); errMulticast != nil {
continue
}
}
}
const (
// https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
maxDatagramSize = 66507
)
// Listen binds to the UDP address and port given and writes packets received
// from that address to a buffer which is passed to a hander
func (p *peerDiscovery) listen() (recievedBytes []byte, err error) {
p.RLock()
address := net.JoinHostPort(p.settings.MulticastAddress, p.settings.Port)
portNum := p.settings.portNum
allowSelf := p.settings.AllowSelf
notify := p.settings.Notify
p.RUnlock()
localIPs := getLocalIPs()
// get interfaces
ifaces, err := net.Interfaces()
if err != nil {
return
}
// log.Println(ifaces)
// Open up a connection
c, err := net.ListenPacket(fmt.Sprintf("udp%d", p.settings.IPVersion), address)
if err != nil {
return
}
defer c.Close()
group := p.settings.multicastAddressNumbers
var p2 NetPacketConn
if p.settings.IPVersion == IPv4 {
p2 = PacketConn4{ipv4.NewPacketConn(c)}
} else {
p2 = PacketConn6{ipv6.NewPacketConn(c)}
}
for i := range ifaces {
p2.JoinGroup(&ifaces[i], &net.UDPAddr{IP: group, Port: portNum})
}
// Loop forever reading from the socket
for {
buffer := make([]byte, maxDatagramSize)
var (
n int
src net.Addr
errRead error
)
n, src, errRead = p2.ReadFrom(buffer)
if errRead != nil {
err = errRead
return
}
srcHost, _, _ := net.SplitHostPort(src.String())
if _, ok := localIPs[srcHost]; ok && !allowSelf {
continue
}
// log.Println(src, hex.Dump(buffer[:n]))
p.Lock()
if _, ok := p.received[srcHost]; !ok {
p.received[srcHost] = buffer[:n]
}
p.Unlock()
if notify != nil {
notify(Discovered{
Address: srcHost,
Payload: buffer[:n],
})
}
p.RLock()
if len(p.received) >= p.settings.Limit && p.settings.Limit > 0 {
p.RUnlock()
break
}
p.RUnlock()
}
return
}
// getLocalIPs returns the local ip address
func getLocalIPs() (ips map[string]struct{}) {
ips = make(map[string]struct{})
ips["localhost"] = struct{}{}
ips["127.0.0.1"] = struct{}{}
ips["::1"] = struct{}{}
ifaces, err := net.Interfaces()
if err != nil {
return
}
for _, iface := range ifaces {
addrs, err := iface.Addrs()
if err != nil {
continue
}
for _, address := range addrs {
ip, _, err := net.ParseCIDR(address.String())
if err != nil {
// log.Printf("Failed to parse %s: %v", address.String(), err)
continue
}
ips[ip.String()+"%"+iface.Name] = struct{}{}
ips[ip.String()] = struct{}{}
}
}
return
}
type PacketConn4 struct {
*ipv4.PacketConn
}
// ReadFrom wraps the ipv4 ReadFrom without a control message
func (pc4 PacketConn4) ReadFrom(buf []byte) (int, net.Addr, error) {
n, _, addr, err := pc4.PacketConn.ReadFrom(buf)
return n, addr, err
}
// WriteTo wraps the ipv4 WriteTo without a control message
func (pc4 PacketConn4) WriteTo(buf []byte, dst net.Addr) (int, error) {
return pc4.PacketConn.WriteTo(buf, nil, dst)
}
type PacketConn6 struct {
*ipv6.PacketConn
}
// ReadFrom wraps the ipv6 ReadFrom without a control message
func (pc6 PacketConn6) ReadFrom(buf []byte) (int, net.Addr, error) {
n, _, addr, err := pc6.PacketConn.ReadFrom(buf)
return n, addr, err
}
// WriteTo wraps the ipv6 WriteTo without a control message
func (pc6 PacketConn6) WriteTo(buf []byte, dst net.Addr) (int, error) {
return pc6.PacketConn.WriteTo(buf, nil, dst)
}
// SetMulticastTTL wraps the hop limit of ipv6
func (pc6 PacketConn6) SetMulticastTTL(i int) error {
return pc6.SetMulticastHopLimit(i)
}