p2p/nat: re-factors nat object
- allow retrieving addresses - allow notifications on mapping changes - allow lifecycle mgmt (future commit will manage it)
Juan Batiz-Benet committed
Jan 26, 2015 at 09:09 UTC
332d350196d4db2a71f08b1b1bb431560456549d
2 files changed
+309
-69
core/core.go
+8
-4
@@ -432,10 +432,14 @@ func constructPeerHost(ctx context.Context, cfg *config.Config, id peer.ID, ps p
432
}
433
log.Infof("Swarm listening at: %s", addrs)
434
435
- mapAddrs := inat.MapAddrs(filteredAddrs)
436
- if len(mapAddrs) > 0 {
437
- log.Infof("NAT mapping addrs: %s", mapAddrs)
438
- addrs = append(addrs, mapAddrs...)
435
+ nat := inat.DiscoverGateway()
436
+ if nat != nil {
437
+ nat.PortMapAddrs(filteredAddrs)
438
+ mapAddrs := nat.ExternalAddrs()
439
+ if len(mapAddrs) > 0 {
440
+ log.Infof("NAT mapping addrs: %s", mapAddrs)
441
+ addrs = append(addrs, mapAddrs...)
442
+ }
443
}
444
445
ps.AddAddresses(id, addrs)
p2p/nat/nat.go
+301
-65
@@ -1,9 +1,11 @@
1
package nat
2
3
import (
4
+ "errors"
5
"fmt"
6
"strconv"
7
"strings"
8
+ "sync"
9
"time"
10
11
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
@@ -11,14 +13,25 @@ import (
13
14
nat "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/fd/go-nat"
15
goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
16
+ periodic "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/periodic"
17
eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
18
+ notifier "github.com/jbenet/go-ipfs/thirdparty/notifier"
19
+)
20
+
21
+var (
22
+ // ErrNoMapping signals no mapping exists for an address
23
+ ErrNoMapping = errors.New("mapping not established")
24
)
25
26
var log = eventlog.Logger("nat")
27
28
+// MappingDuration is a default port mapping duration.
29
+// Port mappings are renewed every (MappingDuration / 3)
30
const MappingDuration = time.Second * 60
31
21
-func DiscoverGateway() nat.NAT {
32
+// DiscoverGateway looks for a NAT device in the network and
33
+// returns an object that can manage port mappings.
34
+func DiscoverGateway() *NAT {
35
nat, err := nat.DiscoverGateway()
36
if err != nil {
37
log.Debug("DiscoverGateway error:", err)
@@ -30,91 +43,215 @@ func DiscoverGateway() nat.NAT {
43
} else {
44
log.Debug("DiscoverGateway address:", addr)
45
}
33
- return nat
46
+ return newNAT(nat)
47
+}
48
+
49
+// NAT is an object that manages address port mappings in
50
+// NATs (Network Address Translators). It is a long-running
51
+// service that will periodically renew port mappings,
52
+// and keep an up-to-date list of all the external addresses.
53
+type NAT struct {
54
+ nat nat.NAT
55
+ proc goprocess.Process // manages nat mappings lifecycle
56
+
57
+ mappingmu sync.RWMutex // guards mappings
58
+ mappings []*mapping
59
+
60
+ Notifier
61
+}
62
+
63
+func newNAT(realNAT nat.NAT) *NAT {
64
+ return &NAT{
65
+ nat: realNAT,
66
+ proc: goprocess.WithParent(goprocess.Background()),
67
+ }
68
+}
69
+
70
+// Close shuts down all port mappings. NAT can no longer be used.
71
+func (nat *NAT) Close() error {
72
+ return nat.proc.Close()
73
+}
74
+
75
+// Notifier is an object that assists NAT in notifying listeners.
76
+// It is implemented using github.com/jbenet/go-ipfs/thirdparty/notifier
77
+type Notifier struct {
78
+ n notifier.Notifier
79
+}
80
+
81
+func (n *Notifier) notifyAll(notify func(n Notifiee)) {
82
+ n.n.NotifyAll(func(n notifier.Notifiee) {
83
+ notify(n.(Notifiee))
84
+ })
85
+}
86
+
87
+// Notify signs up notifiee to listen to NAT events.
88
+func (n *Notifier) Notify(notifiee Notifiee) {
89
+ n.n.Notify(n)
90
+}
91
+
92
+// StopNotify stops signaling events to notifiee.
93
+func (n *Notifier) StopNotify(notifiee Notifiee) {
94
+ n.n.StopNotify(notifiee)
95
+}
96
+
97
+// Notifiee is an interface objects must implement to listen to NAT events.
98
+type Notifiee interface {
99
+
100
+ // Called every time a successful mapping happens
101
+ // Warning: the port mapping may have changed. If that is the
102
+ // case, both MappingSuccess and MappingChanged are called.
103
+ MappingSuccess(nat *NAT, m Mapping)
104
+
105
+ // Called when mapping a port succeeds, but the mapping is
106
+ // with a different port than an earlier success.
107
+ MappingChanged(nat *NAT, m Mapping, oldport int, newport int)
108
+
109
+ // Called when a port mapping fails. NAT will continue attempting after
110
+ // the next period. To stop trying, use: mapping.Close(). After this failure,
111
+ // mapping.ExternalPort() will be zero, and nat.ExternalAddrs() will not
112
+ // return the address for this mapping. With luck, the next attempt will
113
+ // succeed, without the client needing to do anything.
114
+ MappingFailed(nat *NAT, m Mapping, oldport int, err error)
115
}
116
117
+// Mapping represents a port mapping in a NAT.
118
type Mapping interface {
37
- NAT() nat.NAT
119
+ // NAT returns the NAT object this Mapping belongs to.
120
+ NAT() *NAT
121
+
122
+ // Protocol returns the protocol of this port mapping. This is either
123
+ // "tcp" or "udp" as no other protocols are likely to be NAT-supported.
124
Protocol() string
125
+
126
+ // InternalPort returns the internal device port. Mapping will continue to
127
+ // try to map InternalPort() to an external facing port.
128
InternalPort() int
129
+
130
+ // ExternalPort returns the external facing port. If the mapping is not
131
+ // established, port will be 0
132
ExternalPort() int
133
+
134
+ // InternalAddr returns the internal address.
135
+ InternalAddr() ma.Multiaddr
136
+
137
+ // ExternalAddr returns the external facing address. If the mapping is not
138
+ // established, addr will be nil, and and ErrNoMapping will be returned.
139
+ ExternalAddr() (addr ma.Multiaddr, err error)
140
}
141
142
+// keeps republishing
143
type mapping struct {
44
- // keeps republishing
45
- nat nat.NAT
144
+ sync.Mutex // guards all fields
145
+
146
+ nat *NAT
147
proto string
148
intport int
149
extport int
150
+ intaddr ma.Multiaddr
151
proc goprocess.Process
152
}
153
52
-func (m *mapping) NAT() nat.NAT {
154
+func (m *mapping) NAT() *NAT {
155
+ m.Lock()
156
+ defer m.Unlock()
157
return m.nat
158
}
159
+
160
func (m *mapping) Protocol() string {
161
+ m.Lock()
162
+ defer m.Unlock()
163
return m.proto
164
}
165
+
166
func (m *mapping) InternalPort() int {
167
+ m.Lock()
168
+ defer m.Unlock()
169
return m.intport
170
}
171
+
172
func (m *mapping) ExternalPort() int {
173
+ m.Lock()
174
+ defer m.Unlock()
175
return m.extport
176
}
177
65
-// NewMapping attemps to construct a mapping on protocl and internal port
66
-func NewMapping(nat nat.NAT, protocol string, internalPort int) (Mapping, error) {
67
- log.Debugf("Attempting port map: %s/%d", protocol, internalPort)
68
- eport, err := nat.AddPortMapping(protocol, internalPort, "http", MappingDuration)
178
+func (m *mapping) setExternalPort(p int) {
179
+ m.Lock()
180
+ defer m.Unlock()
181
+ m.extport = p
182
+}
183
+
184
+func (m *mapping) InternalAddr() ma.Multiaddr {
185
+ m.Lock()
186
+ defer m.Unlock()
187
+ return m.intaddr
188
+}
189
+
190
+func (m *mapping) ExternalAddr() (ma.Multiaddr, error) {
191
+ if m.ExternalPort() == 0 { // dont even try right now.
192
+ return nil, ErrNoMapping
193
+ }
194
+
195
+ ip, err := m.nat.nat.GetExternalAddress()
196
if err != nil {
197
return nil, err
198
}
199
73
- m := &mapping{
74
- nat: nat,
75
- proto: protocol,
76
- intport: internalPort,
77
- extport: eport,
78
- }
79
-
80
- m.proc = goprocess.Go(func(worker goprocess.Process) {
81
- for {
82
- select {
83
- case <-worker.Closing():
84
- return
85
- case <-time.After(MappingDuration / 3):
86
- eport, err := m.NAT().AddPortMapping(protocol, internalPort, "http", MappingDuration)
87
- if err != nil {
88
- log.Warningf("failed to renew port mapping: %s", err)
89
- continue
90
- }
91
- if eport != m.extport {
92
- log.Warningf("failed to renew same port mapping: ch %d -> %d", m.extport, eport)
93
- }
94
- }
95
- }
96
- })
200
+ ipmaddr, err := manet.FromIP(ip)
201
+ if err != nil {
202
+ return nil, fmt.Errorf("error parsing ip")
203
+ }
204
98
- return m, nil
205
+ // call m.ExternalPort again, as mapping may have changed under our feet. (tocttou)
206
+ extport := m.ExternalPort()
207
+ if extport == 0 {
208
+ return nil, ErrNoMapping
209
+ }
210
+
211
+ tcp, err := ma.NewMultiaddr(fmt.Sprintf("/%s/%d", m.Protocol(), extport))
212
+ if err != nil {
213
+ return nil, err
214
+ }
215
+
216
+ maddr2 := ipmaddr.Encapsulate(tcp)
217
+ return maddr2, nil
218
}
219
220
func (m *mapping) Close() error {
221
return m.proc.Close()
222
}
223
105
-func MapAddr(n nat.NAT, maddr ma.Multiaddr) (ma.Multiaddr, error) {
106
- if n == nil {
107
- return nil, fmt.Errorf("no nat available")
224
+// Mappings returns a slice of all NAT mappings
225
+func (nat *NAT) Mappings() []Mapping {
226
+ nat.mappingmu.Lock()
227
+ maps2 := make([]Mapping, len(nat.mappings))
228
+ for i, m := range nat.mappings {
229
+ maps2[i] = m
230
}
231
+ nat.mappingmu.Unlock()
232
+ return maps2
233
+}
234
110
- ip, err := n.GetExternalAddress()
111
- if err != nil {
112
- return nil, err
113
- }
235
+func (nat *NAT) addMapping(m *mapping) {
236
+ // make mapping automatically close when nat is closed.
237
+ nat.proc.AddChild(m.proc)
238
115
- ipmaddr, err := manet.FromIP(ip)
116
- if err != nil {
117
- return nil, fmt.Errorf("error parsing ip")
239
+ nat.mappingmu.Lock()
240
+ nat.mappings = append(nat.mappings, m)
241
+ nat.mappingmu.Unlock()
242
+}
243
+
244
+// NewMapping attemps to construct a mapping on protocol and internal port
245
+// It will also periodically renew the mapping until the returned Mapping
246
+// -- or its parent NAT -- is Closed.
247
+//
248
+// May not succeed, and mappings may change over time;
249
+// NAT devices may not respect our port requests, and even lie.
250
+// Clients should not store the mapped results, but rather always
251
+// poll our object for the latest mappings.
252
+func (nat *NAT) NewMapping(maddr ma.Multiaddr) (Mapping, error) {
253
+ if nat == nil {
254
+ return nil, fmt.Errorf("no nat available")
255
}
256
257
network, addr, err := manet.DialArgs(maddr)
@@ -131,38 +268,137 @@ func MapAddr(n nat.NAT, maddr ma.Multiaddr) (ma.Multiaddr, error) {
268
return nil, fmt.Errorf("transport not supported by NAT: %s", network)
269
}
270
134
- port := strings.Split(addr, ":")[1]
135
- intport, err := strconv.Atoi(port)
271
+ intports := strings.Split(addr, ":")[1]
272
+ intport, err := strconv.Atoi(intports)
273
if err != nil {
274
return nil, err
275
}
276
140
- m, err := NewMapping(n, "tcp", intport)
141
- if err != nil {
142
- return nil, err
277
+ m := &mapping{
278
+ nat: nat,
279
+ proto: network,
280
+ intport: intport,
281
+ intaddr: maddr,
282
+ }
283
+ m.proc = periodic.Every(MappingDuration/3, func(worker goprocess.Process) {
284
+ nat.establishMapping(m)
285
+ })
286
+ nat.addMapping(m)
287
+ // do it once synchronously, so first mapping is done right away, and before exiting,
288
+ // allowing users -- in the optimistic case -- to use results right after.
289
+ nat.establishMapping(m)
290
+ return m, nil
291
+}
292
+
293
+func (nat *NAT) establishMapping(m *mapping) {
294
+ oldport := m.ExternalPort()
295
+ log.Debugf("Attempting port map: %s/%d", m.Protocol(), m.InternalPort())
296
+ newport, err := nat.nat.AddPortMapping(m.Protocol(), m.InternalPort(), "http", MappingDuration)
297
+
298
+ failure := func() {
299
+ m.setExternalPort(0) // clear mapping
300
+ // TODO: log.Event
301
+ log.Infof("failed to establish port mapping: %s", err)
302
+ nat.Notifier.notifyAll(func(n Notifiee) {
303
+ n.MappingFailed(nat, m, oldport, err)
304
+ })
305
+
306
+ // we do not close if the mapping failed,
307
+ // because it may work again next time.
308
+ }
309
+
310
+ if err != nil || newport == 0 {
311
+ failure()
312
+ return
313
}
314
145
- tcp, err := ma.NewMultiaddr(fmt.Sprintf("/tcp/%d", m.ExternalPort()))
315
+ m.setExternalPort(newport)
316
+ ext, err := m.ExternalAddr()
317
if err != nil {
147
- return nil, err
318
+ log.Debugf("NAT Mapping addr error: %s %s", m.InternalAddr(), err)
319
+ failure()
320
+ return
321
}
322
150
- maddr2 := ipmaddr.Encapsulate(tcp)
151
- log.Debugf("NAT Mapping: %s --> %s", maddr, maddr2)
152
- return maddr2, nil
323
+ log.Debugf("NAT Mapping: %s --> %s", m.InternalAddr(), ext)
324
+ if oldport != 0 && newport != oldport {
325
+ log.Infof("failed to renew same port mapping: ch %d -> %d", oldport, newport)
326
+ nat.Notifier.notifyAll(func(n Notifiee) {
327
+ n.MappingChanged(nat, m, oldport, newport)
328
+ })
329
+ }
330
+
331
+ nat.Notifier.notifyAll(func(n Notifiee) {
332
+ n.MappingSuccess(nat, m)
333
+ })
334
+}
335
+
336
+// PortMapAddrs attempts to open (and continue to keep open)
337
+// port mappings for given addrs. This function blocks until
338
+// all addresses have been tried. This allows clients to
339
+// retrieve results immediately after:
340
+//
341
+// nat.PortMapAddrs(addrs)
342
+// mapped := nat.ExternalAddrs()
343
+//
344
+// Some may not succeed, and mappings may change over time;
345
+// NAT devices may not respect our port requests, and even lie.
346
+// Clients should not store the mapped results, but rather always
347
+// poll our object for the latest mappings.
348
+func (nat *NAT) PortMapAddrs(addrs []ma.Multiaddr) {
349
+ // spin off addr mappings independently.
350
+ var wg sync.WaitGroup
351
+ for _, addr := range addrs {
352
+ // do all of them concurrently
353
+ wg.Add(1)
354
+ go func() {
355
+ defer wg.Done()
356
+ nat.NewMapping(addr)
357
+ }()
358
+ }
359
+ wg.Wait()
360
}
361
155
-func MapAddrs(addrs []ma.Multiaddr) []ma.Multiaddr {
156
- nat := DiscoverGateway()
362
+// MappedAddrs returns address mappings NAT believes have been
363
+// successfully established. Unsuccessful mappings are nil. This is:
364
+//
365
+// map[internalAddr]externalAddr
366
+//
367
+// This set of mappings _may not_ be correct, as NAT devices are finicky.
368
+// Consider this with _best effort_ semantics.
369
+func (nat *NAT) MappedAddrs() map[ma.Multiaddr]ma.Multiaddr {
370
+
371
+ mappings := nat.Mappings()
372
+ addrmap := make(map[ma.Multiaddr]ma.Multiaddr, len(mappings))
373
+
374
+ for _, m := range mappings {
375
+ i := m.InternalAddr()
376
+ e, err := m.ExternalAddr()
377
+ if err != nil {
378
+ addrmap[i] = nil
379
+ } else {
380
+ addrmap[i] = e
381
+ }
382
+ }
383
+ return addrmap
384
+}
385
158
- var advertise []ma.Multiaddr
159
- for _, maddr := range addrs {
160
- maddr2, err := MapAddr(nat, maddr)
161
- if err != nil || maddr2 == nil {
162
- log.Debug("failed to map addr:", maddr, err)
163
- continue
386
+// ExternalAddrs returns a list of addresses that NAT believes have
387
+// been successfully established. Unsuccessful mappings are omitted,
388
+// so nat.ExternalAddrs() may return less addresses than nat.InternalAddrs().
389
+// To see which addresses are mapped, use nat.MappedAddrs().
390
+//
391
+// This set of mappings _may not_ be correct, as NAT devices are finicky.
392
+// Consider this with _best effort_ semantics.
393
+func (nat *NAT) ExternalAddrs() []ma.Multiaddr {
394
+ mappings := nat.Mappings()
395
+ addrs := make([]ma.Multiaddr, 0, len(mappings))
396
+ for _, m := range mappings {
397
+ a, err := m.ExternalAddr()
398
+ if err != nil {
399
+ continue // this mapping not currently successful.
400
}
165
- advertise = append(advertise, maddr2)
401
+ addrs = append(addrs, a)
402
}
167
- return advertise
403
+ return addrs
404
}