Skip to content

Commit

Permalink
p2p/nat: limit UPNP request concurrency (ethereum#21390)
Browse files Browse the repository at this point in the history 
This adds a lock around requests because some routers can't handle
concurrent requests. Requests are also rate-limited.
 
The Map function request a new mapping exactly when the map timeout
occurs instead of 5 minutes earlier. This should prevent duplicate mappings.
  • Loading branch information
@fjl @enriquefynn
fjl authored and enriquefynn committed Feb 15, 2021
1 parent 43826e1 commit 4df65d8
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 25 deletions.
9 changes: 4 additions & 5 deletions p2p/nat/nat.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,15 +91,14 @@ func Parse(spec string) (Interface, error) {
}

const (
mapTimeout = 20 * time.Minute
mapUpdateInterval = 15 * time.Minute
mapTimeout = 10 * time.Minute
)

// Map adds a port mapping on m and keeps it alive until c is closed.
// This function is typically invoked in its own goroutine.
func Map(m Interface, c chan struct{}, protocol string, extport, intport int, name string) {
func Map(m Interface, c <-chan struct{}, protocol string, extport, intport int, name string) {
log := log.New("proto", protocol, "extport", extport, "intport", intport, "interface", m)
refresh := time.NewTimer(mapUpdateInterval)
refresh := time.NewTimer(mapTimeout)
defer func() {
refresh.Stop()
log.Debug("Deleting port mapping")
Expand All @@ -121,7 +120,7 @@ func Map(m Interface, c chan struct{}, protocol string, extport, intport int, na
if err := m.AddMapping(protocol, extport, intport, name, mapTimeout); err != nil {
log.Debug("Couldn't add port mapping", "err", err)
}
refresh.Reset(mapUpdateInterval)
refresh.Reset(mapTimeout)
}
}
}
Expand Down
80 changes: 60 additions & 20 deletions p2p/nat/natupnp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,25 @@ import (
"fmt"
"net"
"strings"
"sync"
"time"

"github.com/huin/goupnp"
"github.com/huin/goupnp/dcps/internetgateway1"
"github.com/huin/goupnp/dcps/internetgateway2"
)

const soapRequestTimeout = 3 * time.Second
const (
soapRequestTimeout = 3 * time.Second
rateLimit = 200 * time.Millisecond
)

type upnp struct {
dev *goupnp.RootDevice
service string
client upnpClient
dev *goupnp.RootDevice
service string
client upnpClient
mu sync.Mutex
lastReqTime time.Time
}

type upnpClient interface {
Expand All @@ -43,8 +49,23 @@ type upnpClient interface {
GetNATRSIPStatus() (sip bool, nat bool, err error)
}

func (n *upnp) natEnabled() bool {
var ok bool
var err error
n.withRateLimit(func() error {
_, ok, err = n.client.GetNATRSIPStatus()
return err
})
return err == nil && ok
}

func (n *upnp) ExternalIP() (addr net.IP, err error) {
ipString, err := n.client.GetExternalIPAddress()
var ipString string
n.withRateLimit(func() error {
ipString, err = n.client.GetExternalIPAddress()
return err
})

if err != nil {
return nil, err
}
Expand All @@ -63,7 +84,10 @@ func (n *upnp) AddMapping(protocol string, extport, intport int, desc string, li
protocol = strings.ToUpper(protocol)
lifetimeS := uint32(lifetime / time.Second)
n.DeleteMapping(protocol, extport, intport)
return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS)

return n.withRateLimit(func() error {
return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS)
})
}

func (n *upnp) internalAddress() (net.IP, error) {
Expand All @@ -90,36 +114,51 @@ func (n *upnp) internalAddress() (net.IP, error) {
}

func (n *upnp) DeleteMapping(protocol string, extport, intport int) error {
return n.client.DeletePortMapping("", uint16(extport), strings.ToUpper(protocol))
return n.withRateLimit(func() error {
return n.client.DeletePortMapping("", uint16(extport), strings.ToUpper(protocol))
})
}

func (n *upnp) String() string {
return "UPNP " + n.service
}

func (n *upnp) withRateLimit(fn func() error) error {
n.mu.Lock()
defer n.mu.Unlock()

lastreq := time.Since(n.lastReqTime)
if lastreq < rateLimit {
time.Sleep(rateLimit - lastreq)
}
err := fn()
n.lastReqTime = time.Now()
return err
}

// discoverUPnP searches for Internet Gateway Devices
// and returns the first one it can find on the local network.
func discoverUPnP() Interface {
found := make(chan *upnp, 2)
// IGDv1
go discover(found, internetgateway1.URN_WANConnectionDevice_1, func(dev *goupnp.RootDevice, sc goupnp.ServiceClient) *upnp {
go discover(found, internetgateway1.URN_WANConnectionDevice_1, func(sc goupnp.ServiceClient) *upnp {
switch sc.Service.ServiceType {
case internetgateway1.URN_WANIPConnection_1:
return &upnp{dev, "IGDv1-IP1", &internetgateway1.WANIPConnection1{ServiceClient: sc}}
return &upnp{service: "IGDv1-IP1", client: &internetgateway1.WANIPConnection1{ServiceClient: sc}}
case internetgateway1.URN_WANPPPConnection_1:
return &upnp{dev, "IGDv1-PPP1", &internetgateway1.WANPPPConnection1{ServiceClient: sc}}
return &upnp{service: "IGDv1-PPP1", client: &internetgateway1.WANPPPConnection1{ServiceClient: sc}}
}
return nil
})
// IGDv2
go discover(found, internetgateway2.URN_WANConnectionDevice_2, func(dev *goupnp.RootDevice, sc goupnp.ServiceClient) *upnp {
go discover(found, internetgateway2.URN_WANConnectionDevice_2, func(sc goupnp.ServiceClient) *upnp {
switch sc.Service.ServiceType {
case internetgateway2.URN_WANIPConnection_1:
return &upnp{dev, "IGDv2-IP1", &internetgateway2.WANIPConnection1{ServiceClient: sc}}
return &upnp{service: "IGDv2-IP1", client: &internetgateway2.WANIPConnection1{ServiceClient: sc}}
case internetgateway2.URN_WANIPConnection_2:
return &upnp{dev, "IGDv2-IP2", &internetgateway2.WANIPConnection2{ServiceClient: sc}}
return &upnp{service: "IGDv2-IP2", client: &internetgateway2.WANIPConnection2{ServiceClient: sc}}
case internetgateway2.URN_WANPPPConnection_1:
return &upnp{dev, "IGDv2-PPP1", &internetgateway2.WANPPPConnection1{ServiceClient: sc}}
return &upnp{service: "IGDv2-PPP1", client: &internetgateway2.WANPPPConnection1{ServiceClient: sc}}
}
return nil
})
Expand All @@ -134,7 +173,7 @@ func discoverUPnP() Interface {
// finds devices matching the given target and calls matcher for all
// advertised services of each device. The first non-nil service found
// is sent into out. If no service matched, nil is sent.
func discover(out chan<- *upnp, target string, matcher func(*goupnp.RootDevice, goupnp.ServiceClient) *upnp) {
func discover(out chan<- *upnp, target string, matcher func(goupnp.ServiceClient) *upnp) {
devs, err := goupnp.DiscoverDevices(target)
if err != nil {
out <- nil
Expand All @@ -157,16 +196,17 @@ func discover(out chan<- *upnp, target string, matcher func(*goupnp.RootDevice,
Service: service,
}
sc.SOAPClient.HTTPClient.Timeout = soapRequestTimeout
upnp := matcher(devs[i].Root, sc)
upnp := matcher(sc)
if upnp == nil {
return
}
upnp.dev = devs[i].Root

// check whether port mapping is enabled
if _, nat, err := upnp.client.GetNATRSIPStatus(); err != nil || !nat {
return
if upnp.natEnabled() {
out <- upnp
found = true
}
out <- upnp
found = true
})
}
if !found {
Expand Down

0 comments on commit 4df65d8

Please sign in to comment.