Merge pull request #1 from libp2p/implementation

Extract service implementation from go-libp2p-autonat

.gx/lastpubver

@@ -0,0 +1 @@
+1.0.0: QmeYNWTPm2TdZfSepXp1Su22UXgbsPPN1vTXB8H3nFJVMD

README.md

@@ -0,0 +1,24 @@
+# go-libp2p-autonat-svc
+
+[![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](http://ipn.io)
+[![](https://img.shields.io/badge/project-libp2p-blue.svg?style=flat-square)](http://libp2p.io/)
+[![](https://img.shields.io/badge/freenode-%23ipfs-blue.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23ipfs)
+[![standard-readme compliant](https://img.shields.io/badge/standard--readme-OK-green.svg?style=flat-square)](https://github.com/RichardLitt/standard-readme)
+
+> AutoNAT service implementation
+
+This package provides an implementation of the AutoNATService; see [autonat](https://github.com/libp2p/go-libp2p-autonat).
+
+## Documentation
+
+See https://godoc.org/github.com/libp2p/go-libp2p-autonat-svc.
+
+## Contribute
+
+Feel free to join in. All welcome. Open an [issue](https://github.com/libp2p/go-libp2p-discovery/issues)!
+
+This repository falls under the IPFS [Code of Conduct](https://github.com/ipfs/community/blob/master/code-of-conduct.md).
+
+## License
+
+MIT

ci/Jenkinsfile

@@ -0,0 +1 @@
+golang()

package.json

@@ -0,0 +1,28 @@
+{
+  "author": "vyzo",
+  "bugs": {},
+  "gx": {
+    "dvcsimport": "github.com/libp2p/go-libp2p-autonat-svc"
+  },
+  "gxDependencies": [
+    {
+      "author": "whyrusleeping",
+      "hash": "QmUDTcnDp2WssbmiDLC6aYurUeyt7QeRakHUQMxA2mZ5iB",
+      "name": "go-libp2p",
+      "version": "6.0.23"
+    },
+    {
+      "author": "vyzo",
+      "hash": "QmUn8mtaf4tTFwKnFRzkNYYLc8XEo3yz6qBfp5ShVB1HYZ",
+      "name": "go-libp2p-autonat",
+      "version": "1.0.1"
+    }
+  ],
+  "gxVersion": "0.12.1",
+  "language": "go",
+  "license": "",
+  "name": "go-libp2p-autonat-svc",
+  "releaseCmd": "git commit -a -m \"gx publish $VERSION\"",
+  "version": "1.0.0"
+}
+

proto.go

@@ -0,0 +1,24 @@
+package autonat
+
+import (
+	pb "github.com/libp2p/go-libp2p-autonat/pb"
+
+	logging "github.com/ipfs/go-log"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+var log = logging.Logger("autonat-svc")
+
+func newDialResponseOK(addr ma.Multiaddr) *pb.Message_DialResponse {
+	dr := new(pb.Message_DialResponse)
+	dr.Status = pb.Message_OK.Enum()
+	dr.Addr = addr.Bytes()
+	return dr
+}
+
+func newDialResponseError(status pb.Message_ResponseStatus, text string) *pb.Message_DialResponse {
+	dr := new(pb.Message_DialResponse)
+	dr.Status = status.Enum()
+	dr.StatusText = &text
+	return dr
+}

svc.go

@@ -0,0 +1,216 @@
+package autonat
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	pb "github.com/libp2p/go-libp2p-autonat/pb"
+
+	ggio "github.com/gogo/protobuf/io"
+	libp2p "github.com/libp2p/go-libp2p"
+	autonat "github.com/libp2p/go-libp2p-autonat"
+	host "github.com/libp2p/go-libp2p-host"
+	inet "github.com/libp2p/go-libp2p-net"
+	peer "github.com/libp2p/go-libp2p-peer"
+	pstore "github.com/libp2p/go-libp2p-peerstore"
+	ma "github.com/multiformats/go-multiaddr"
+	manet "github.com/multiformats/go-multiaddr-net"
+)
+
+const P_CIRCUIT = 290
+
+var (
+	AutoNATServiceDialTimeout   = 42 * time.Second
+	AutoNATServiceResetInterval = 1 * time.Minute
+
+	AutoNATServiceThrottle = 3
+)
+
+// AutoNATService provides NAT autodetection services to other peers
+type AutoNATService struct {
+	ctx    context.Context
+	dialer host.Host
+
+	// rate limiter
+	mx   sync.Mutex
+	reqs map[peer.ID]int
+}
+
+// NewAutoNATService creates a new AutoNATService instance attached to a host
+func NewAutoNATService(ctx context.Context, h host.Host, opts ...libp2p.Option) (*AutoNATService, error) {
+	opts = append(opts, libp2p.NoListenAddrs)
+	dialer, err := libp2p.New(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	as := &AutoNATService{
+		ctx:    ctx,
+		dialer: dialer,
+		reqs:   make(map[peer.ID]int),
+	}
+	h.SetStreamHandler(autonat.AutoNATProto, as.handleStream)
+
+	go as.resetRateLimiter()
+
+	return as, nil
+}
+
+func (as *AutoNATService) handleStream(s inet.Stream) {
+	defer s.Close()
+
+	pid := s.Conn().RemotePeer()
+	log.Debugf("New stream from %s", pid.Pretty())
+
+	r := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
+	w := ggio.NewDelimitedWriter(s)
+
+	var req pb.Message
+	var res pb.Message
+
+	err := r.ReadMsg(&req)
+	if err != nil {
+		log.Debugf("Error reading message from %s: %s", pid.Pretty(), err.Error())
+		s.Reset()
+		return
+	}
+
+	t := req.GetType()
+	if t != pb.Message_DIAL {
+		log.Debugf("Unexpected message from %s: %s (%d)", pid.Pretty(), t.String(), t)
+		s.Reset()
+		return
+	}
+
+	dr := as.handleDial(pid, s.Conn().RemoteMultiaddr(), req.GetDial().GetPeer())
+	res.Type = pb.Message_DIAL_RESPONSE.Enum()
+	res.DialResponse = dr
+
+	err = w.WriteMsg(&res)
+	if err != nil {
+		log.Debugf("Error writing response to %s: %s", pid.Pretty(), err.Error())
+		s.Reset()
+		return
+	}
+}
+
+func (as *AutoNATService) handleDial(p peer.ID, obsaddr ma.Multiaddr, mpi *pb.Message_PeerInfo) *pb.Message_DialResponse {
+	if mpi == nil {
+		return newDialResponseError(pb.Message_E_BAD_REQUEST, "missing peer info")
+	}
+
+	mpid := mpi.GetId()
+	if mpid != nil {
+		mp, err := peer.IDFromBytes(mpid)
+		if err != nil {
+			return newDialResponseError(pb.Message_E_BAD_REQUEST, "bad peer id")
+		}
+
+		if mp != p {
+			return newDialResponseError(pb.Message_E_BAD_REQUEST, "peer id mismatch")
+		}
+	}
+
+	addrs := make([]ma.Multiaddr, 0)
+	seen := make(map[string]struct{})
+
+	// add observed addr to the list of addresses to dial
+	if !as.skipDial(obsaddr) {
+		addrs = append(addrs, obsaddr)
+		seen[obsaddr.String()] = struct{}{}
+	}
+
+	for _, maddr := range mpi.GetAddrs() {
+		addr, err := ma.NewMultiaddrBytes(maddr)
+		if err != nil {
+			log.Debugf("Error parsing multiaddr: %s", err.Error())
+			continue
+		}
+
+		if as.skipDial(addr) {
+			continue
+		}
+
+		str := addr.String()
+		_, ok := seen[str]
+		if ok {
+			continue
+		}
+
+		addrs = append(addrs, addr)
+		seen[str] = struct{}{}
+	}
+
+	if len(addrs) == 0 {
+		return newDialResponseError(pb.Message_E_DIAL_ERROR, "no dialable addresses")
+	}
+
+	return as.doDial(pstore.PeerInfo{ID: p, Addrs: addrs})
+}
+
+func (as *AutoNATService) skipDial(addr ma.Multiaddr) bool {
+	// skip relay addresses
+	_, err := addr.ValueForProtocol(P_CIRCUIT)
+	if err == nil {
+		return true
+	}
+
+	// skip private network (unroutable) addresses
+	if !manet.IsPublicAddr(addr) {
+		return true
+	}
+
+	return false
+}
+
+func (as *AutoNATService) doDial(pi pstore.PeerInfo) *pb.Message_DialResponse {
+	// rate limit check
+	as.mx.Lock()
+	count := as.reqs[pi.ID]
+	if count >= AutoNATServiceThrottle {
+		as.mx.Unlock()
+		return newDialResponseError(pb.Message_E_DIAL_REFUSED, "too many dials")
+	}
+	as.reqs[pi.ID] = count + 1
+	as.mx.Unlock()
+
+	ctx, cancel := context.WithTimeout(as.ctx, AutoNATServiceDialTimeout)
+	defer cancel()
+
+	err := as.dialer.Connect(ctx, pi)
+	if err != nil {
+		log.Debugf("error dialing %s: %s", pi.ID.Pretty(), err.Error())
+		// wait for the context to timeout to avoid leaking timing information
+		// this renders the service ineffective as a port scanner
+		<-ctx.Done()
+		return newDialResponseError(pb.Message_E_DIAL_ERROR, "dial failed")
+	}
+
+	conns := as.dialer.Network().ConnsToPeer(pi.ID)
+	if len(conns) == 0 {
+		log.Errorf("supposedly connected to %s, but no connection to peer", pi.ID.Pretty())
+		return newDialResponseError(pb.Message_E_INTERNAL_ERROR, "internal service error")
+	}
+
+	ra := conns[0].RemoteMultiaddr()
+	as.dialer.Network().ClosePeer(pi.ID)
+	return newDialResponseOK(ra)
+}
+
+func (as *AutoNATService) resetRateLimiter() {
+	ticker := time.NewTicker(AutoNATServiceResetInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			as.mx.Lock()
+			as.reqs = make(map[peer.ID]int)
+			as.mx.Unlock()
+
+		case <-as.ctx.Done():
+			return
+		}
+	}
+}