Merge pull request #4 from marten-seemann/feat/refactor

use the new transport interfaces

README.md

@@ -5,3 +5,7 @@
 [![Code Coverage](https://img.shields.io/codecov/c/github/marten-seemann/libp2p-quic-transport/master.svg?style=flat-square)](https://codecov.io/gh/marten-seemann/libp2p-quic-transport/)
 
 This is an implementation of the [libp2p transport](https://github.com/libp2p/go-libp2p-transport/blob/master/transport.go) and the [libp2p stream muxer](https://github.com/libp2p/go-stream-muxer) using QUIC.
+
+## Known limitations
+
+* currently only works with RSA host keys

conn.go

 package libp2pquic
 
 import (
-	"fmt"
 	"net"
-	"sync"
 
+	ic "github.com/libp2p/go-libp2p-crypto"
+	peer "github.com/libp2p/go-libp2p-peer"
 	tpt "github.com/libp2p/go-libp2p-transport"
 	smux "github.com/libp2p/go-stream-muxer"
 	quic "github.com/lucas-clemente/quic-go"
@@ -12,96 +12,78 @@ import (
 	manet "github.com/multiformats/go-multiaddr-net"
 )
 
-type quicConn struct {
-	mutex sync.RWMutex
-
+type conn struct {
 	sess      quic.Session
 	transport tpt.Transport
 
-	laddr ma.Multiaddr
-	raddr ma.Multiaddr
+	localPeer      peer.ID
+	privKey        ic.PrivKey
+	localMultiaddr ma.Multiaddr
 
-	closed bool
+	remotePeerID    peer.ID
+	remotePubKey    ic.PubKey
+	remoteMultiaddr ma.Multiaddr
 }
 
-var _ tpt.Conn = &quicConn{}
-var _ tpt.MultiplexConn = &quicConn{}
-
-func newQuicConn(sess quic.Session, t tpt.Transport) (*quicConn, error) {
-	// analogues to manet.WrapNetConn
-	laddr, err := quicMultiAddress(sess.LocalAddr())
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert nconn.LocalAddr: %s", err)
-	}
+var _ tpt.Conn = &conn{}
 
-	// analogues to manet.WrapNetConn
-	raddr, err := quicMultiAddress(sess.RemoteAddr())
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert nconn.RemoteAddr: %s", err)
-	}
-
-	c := &quicConn{
-		sess:      sess,
-		laddr:     laddr,
-		raddr:     raddr,
-		transport: t,
-	}
-	go c.watchClosed()
-
-	return c, nil
+func (c *conn) Close() error {
+	return c.sess.Close(nil)
 }
 
-func (c *quicConn) AcceptStream() (smux.Stream, error) {
-	str, err := c.sess.AcceptStream()
-	return &stream{str}, err
+// IsClosed returns whether a connection is fully closed.
+func (c *conn) IsClosed() bool {
+	return c.sess.Context().Err() != nil
 }
 
-// OpenStream opens a new stream
-// It blocks until a new stream can be opened (when limited by the QUIC maximum stream limit)
-func (c *quicConn) OpenStream() (smux.Stream, error) {
-	str, err := c.sess.OpenStreamSync()
-	return &stream{str}, err
+// OpenStream creates a new stream.
+func (c *conn) OpenStream() (smux.Stream, error) {
+	qstr, err := c.sess.OpenStreamSync()
+	return &stream{Stream: qstr}, err
 }
 
-func (c *quicConn) Close() error {
-	return c.sess.Close(nil)
+// AcceptStream accepts a stream opened by the other side.
+func (c *conn) AcceptStream() (smux.Stream, error) {
+	qstr, err := c.sess.AcceptStream()
+	return &stream{Stream: qstr}, err
 }
 
-func (c *quicConn) watchClosed() {
-	<-c.sess.Context().Done()
-	c.mutex.Lock()
-	c.closed = true
-	c.mutex.Unlock()
+// LocalPeer returns our peer ID
+func (c *conn) LocalPeer() peer.ID {
+	return c.localPeer
 }
 
-func (c *quicConn) IsClosed() bool {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-	return c.closed
+// LocalPrivateKey returns our private key
+func (c *conn) LocalPrivateKey() ic.PrivKey {
+	return c.privKey
 }
 
-func (c *quicConn) LocalAddr() net.Addr {
-	return c.sess.LocalAddr()
+// RemotePeer returns the peer ID of the remote peer.
+func (c *conn) RemotePeer() peer.ID {
+	return c.remotePeerID
 }
 
-func (c *quicConn) LocalMultiaddr() ma.Multiaddr {
-	return c.laddr
+// RemotePublicKey returns the public key of the remote peer.
+func (c *conn) RemotePublicKey() ic.PubKey {
+	return c.remotePubKey
 }
 
-func (c *quicConn) RemoteAddr() net.Addr {
-	return c.sess.RemoteAddr()
+// LocalMultiaddr returns the local Multiaddr associated
+func (c *conn) LocalMultiaddr() ma.Multiaddr {
+	return c.localMultiaddr
 }
 
-func (c *quicConn) RemoteMultiaddr() ma.Multiaddr {
-	return c.raddr
+// RemoteMultiaddr returns the remote Multiaddr associated
+func (c *conn) RemoteMultiaddr() ma.Multiaddr {
+	return c.remoteMultiaddr
 }
 
-func (c *quicConn) Transport() tpt.Transport {
+func (c *conn) Transport() tpt.Transport {
 	return c.transport
 }
 
 // TODO: there must be a better way to do this
-func quicMultiAddress(na net.Addr) (ma.Multiaddr, error) {
+func quicMultiaddr(na net.Addr) (ma.Multiaddr, error) {
 	udpMA, err := manet.FromNetAddr(na)
 	if err != nil {
 		return nil, err

conn_test.go

@@ -2,133 +2,113 @@ package libp2pquic
 
 import (
 	"context"
-	"errors"
-	"net"
-	"time"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"io/ioutil"
 
-	quic "github.com/lucas-clemente/quic-go"
+	ic "github.com/libp2p/go-libp2p-crypto"
+	peer "github.com/libp2p/go-libp2p-peer"
+	tpt "github.com/libp2p/go-libp2p-transport"
+	ma "github.com/multiformats/go-multiaddr"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 )
 
-type mockStream struct {
-	id quic.StreamID
-}
-
-func (s *mockStream) Close() error                     { return nil }
-func (s *mockStream) Reset(error)                      { return }
-func (s *mockStream) Read([]byte) (int, error)         { return 0, nil }
-func (s *mockStream) Write([]byte) (int, error)        { return 0, nil }
-func (s *mockStream) StreamID() quic.StreamID          { return s.id }
-func (s *mockStream) SetReadDeadline(time.Time) error  { panic("not implemented") }
-func (s *mockStream) SetWriteDeadline(time.Time) error { panic("not implemented") }
-func (s *mockStream) SetDeadline(time.Time) error      { panic("not implemented") }
-func (s *mockStream) Context() context.Context         { panic("not implemented") }
-
-var _ quic.Stream = &mockStream{}
-
-type mockQuicSession struct {
-	closed  bool
-	context context.Context
-
-	localAddr  net.Addr
-	remoteAddr net.Addr
-
-	streamToAccept  quic.Stream
-	streamAcceptErr error
-
-	streamToOpen  quic.Stream
-	streamOpenErr error
-}
-
-var _ quic.Session = &mockQuicSession{}
-
-func (s *mockQuicSession) AcceptStream() (quic.Stream, error) {
-	return s.streamToAccept, s.streamAcceptErr
-}
-func (s *mockQuicSession) OpenStream() (quic.Stream, error) { return s.streamToOpen, s.streamOpenErr }
-func (s *mockQuicSession) OpenStreamSync() (quic.Stream, error) {
-	return s.streamToOpen, s.streamOpenErr
-}
-func (s *mockQuicSession) Close(error) error        { s.closed = true; return nil }
-func (s *mockQuicSession) LocalAddr() net.Addr      { return s.localAddr }
-func (s *mockQuicSession) RemoteAddr() net.Addr     { return s.remoteAddr }
-func (s *mockQuicSession) Context() context.Context { return s.context }
-
-var _ = Describe("Conn", func() {
+var _ = Describe("Connection", func() {
 	var (
-		conn      *quicConn
-		sess      *mockQuicSession
-		ctxCancel context.CancelFunc
+		serverKey, clientKey ic.PrivKey
+		serverID, clientID   peer.ID
 	)
 
-	BeforeEach(func() {
-		var ctx context.Context
-		ctx, ctxCancel = context.WithCancel(context.Background())
-		sess = &mockQuicSession{
-			localAddr:  &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1337},
-			remoteAddr: &net.UDPAddr{IP: net.IPv4(192, 168, 13, 37), Port: 1234},
-			context:    ctx,
-		}
-		var err error
-		conn, err = newQuicConn(sess, nil)
+	createPeer := func() ic.PrivKey {
+		key, err := rsa.GenerateKey(rand.Reader, 1024)
 		Expect(err).ToNot(HaveOccurred())
-	})
-
-	It("has the correct local address", func() {
-		Expect(conn.LocalAddr()).To(Equal(sess.localAddr))
-		Expect(conn.LocalMultiaddr().String()).To(Equal("/ip4/127.0.0.1/udp/1337/quic"))
-	})
+		priv, err := ic.UnmarshalRsaPrivateKey(x509.MarshalPKCS1PrivateKey(key))
+		Expect(err).ToNot(HaveOccurred())
+		return priv
+	}
 
-	It("has the correct remote address", func() {
-		Expect(conn.RemoteAddr()).To(Equal(sess.remoteAddr))
-		Expect(conn.RemoteMultiaddr().String()).To(Equal("/ip4/192.168.13.37/udp/1234/quic"))
-	})
+	runServer := func() (<-chan ma.Multiaddr, <-chan tpt.Conn) {
+		serverTransport, err := NewTransport(serverKey)
+		Expect(err).ToNot(HaveOccurred())
+		addrChan := make(chan ma.Multiaddr)
+		connChan := make(chan tpt.Conn)
+		go func() {
+			defer GinkgoRecover()
+			addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/0/quic")
+			Expect(err).ToNot(HaveOccurred())
+			ln, err := serverTransport.Listen(addr)
+			Expect(err).ToNot(HaveOccurred())
+			addrChan <- ln.Multiaddr()
+			conn, err := ln.Accept()
+			Expect(err).ToNot(HaveOccurred())
+			connChan <- conn
+		}()
+		return addrChan, connChan
+	}
 
-	It("closes", func() {
-		err := conn.Close()
+	BeforeEach(func() {
+		var err error
+		serverKey = createPeer()
+		serverID, err = peer.IDFromPrivateKey(serverKey)
+		Expect(err).ToNot(HaveOccurred())
+		clientKey = createPeer()
+		clientID, err = peer.IDFromPrivateKey(clientKey)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(sess.closed).To(BeTrue())
 	})
 
-	It("says if it is closed", func() {
-		Consistently(func() bool { return conn.IsClosed() }).Should(BeFalse())
-		ctxCancel()
-		Eventually(func() bool { return conn.IsClosed() }).Should(BeTrue())
+	It("handshakes", func() {
+		serverAddrChan, serverConnChan := runServer()
+		clientTransport, err := NewTransport(clientKey)
+		Expect(err).ToNot(HaveOccurred())
+		serverAddr := <-serverAddrChan
+		conn, err := clientTransport.Dial(context.Background(), serverAddr, serverID)
+		Expect(err).ToNot(HaveOccurred())
+		serverConn := <-serverConnChan
+		Expect(conn.LocalPeer()).To(Equal(clientID))
+		Expect(conn.LocalPrivateKey()).To(Equal(clientKey))
+		Expect(conn.RemotePeer()).To(Equal(serverID))
+		Expect(conn.RemotePublicKey()).To(Equal(serverKey.GetPublic()))
+		Expect(serverConn.LocalPeer()).To(Equal(serverID))
+		Expect(serverConn.LocalPrivateKey()).To(Equal(serverKey))
+		Expect(serverConn.RemotePeer()).To(Equal(clientID))
+		Expect(serverConn.RemotePublicKey()).To(Equal(clientKey.GetPublic()))
 	})
 
-	Context("opening streams", func() {
-		It("opens streams", func() {
-			s := &mockStream{id: 1337}
-			sess.streamToOpen = s
-			str, err := conn.OpenStream()
-			Expect(err).ToNot(HaveOccurred())
-			Expect(str.(*stream).Stream).To(Equal(s))
-		})
+	It("opens and accepts streams", func() {
+		serverAddrChan, serverConnChan := runServer()
+		clientTransport, err := NewTransport(clientKey)
+		Expect(err).ToNot(HaveOccurred())
+		conn, err := clientTransport.Dial(context.Background(), <-serverAddrChan, serverID)
+		Expect(err).ToNot(HaveOccurred())
+		serverConn := <-serverConnChan
 
-		It("errors when it can't open a stream", func() {
-			testErr := errors.New("stream open err")
-			sess.streamOpenErr = testErr
-			_, err := conn.OpenStream()
-			Expect(err).To(MatchError(testErr))
-		})
+		str, err := conn.OpenStream()
+		Expect(err).ToNot(HaveOccurred())
+		_, err = str.Write([]byte("foobar"))
+		Expect(err).ToNot(HaveOccurred())
+		str.Close()
+		sstr, err := serverConn.AcceptStream()
+		Expect(err).ToNot(HaveOccurred())
+		data, err := ioutil.ReadAll(sstr)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(data).To(Equal([]byte("foobar")))
 	})
 
-	Context("accepting streams", func() {
-		It("accepts streams", func() {
-			s := &mockStream{id: 1337}
-			sess.streamToAccept = s
-			str, err := conn.AcceptStream()
-			Expect(err).ToNot(HaveOccurred())
-			Expect(str.(*stream).Stream).To(Equal(s))
-		})
+	It("fails if the peer ID doesn't match", func() {
+		thirdPartyID, err := peer.IDFromPrivateKey(createPeer())
+		Expect(err).ToNot(HaveOccurred())
 
-		It("errors when it can't open a stream", func() {
-			testErr := errors.New("stream open err")
-			sess.streamAcceptErr = testErr
-			_, err := conn.AcceptStream()
-			Expect(err).To(MatchError(testErr))
-		})
+		serverAddrChan, serverConnChan := runServer()
+		clientTransport, err := NewTransport(clientKey)
+		Expect(err).ToNot(HaveOccurred())
+		serverAddr := <-serverAddrChan
+		// dial, but expect the wrong peer ID
+		_, err = clientTransport.Dial(context.Background(), serverAddr, thirdPartyID)
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("TLS handshake error: bad certificate"))
+		Consistently(serverConnChan).ShouldNot(Receive())
 	})
 })

crypto.go

+package libp2pquic
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"math/big"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	ic "github.com/libp2p/go-libp2p-crypto"
+	pb "github.com/libp2p/go-libp2p-crypto/pb"
+)
+
+// mint certificate selection is broken.
+const hostname = "quic.ipfs"
+
+func generateConfig(privKey ic.PrivKey) (*tls.Config, error) {
+	key, hostCert, err := keyToCertificate(privKey)
+	if err != nil {
+		return nil, err
+	}
+	// The ephemeral key used just for a couple of connections (or a limited time).
+	ephemeralKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, err
+	}
+	// Sign the ephemeral key using the host key.
+	// This is the only time that the host's private key of the peer is needed.
+	// Note that this step could be done asynchronously, such that a running node doesn't need access its private key at all.
+	certTemplate := &x509.Certificate{
+		DNSNames:     []string{hostname},
+		SerialNumber: big.NewInt(1),
+		NotBefore:    time.Now().Add(-24 * time.Hour),
+		NotAfter:     time.Now().Add(30 * 24 * time.Hour),
+	}
+	certDER, err := x509.CreateCertificate(rand.Reader, certTemplate, hostCert, ephemeralKey.Public(), key)
+	if err != nil {
+		return nil, err
+	}
+	cert, err := x509.ParseCertificate(certDER)
+	if err != nil {
+		return nil, err
+	}
+	return &tls.Config{
+		ServerName:         hostname,
+		InsecureSkipVerify: true, // This is not insecure here. We will verify the cert chain ourselves.
+		ClientAuth:         tls.RequireAnyClientCert,
+		Certificates: []tls.Certificate{{
+			Certificate: [][]byte{cert.Raw, hostCert.Raw},
+			PrivateKey:  ephemeralKey,
+		}},
+	}, nil
+}
+
+func getRemotePubKey(chain []*x509.Certificate) (ic.PubKey, error) {
+	if len(chain) != 2 {
+		return nil, errors.New("expected 2 certificates in the chain")
+	}
+	pool := x509.NewCertPool()
+	pool.AddCert(chain[1])
+	if _, err := chain[0].Verify(x509.VerifyOptions{Roots: pool}); err != nil {
+		return nil, err
+	}
+	remotePubKey, err := x509.MarshalPKIXPublicKey(chain[1].PublicKey)
+	if err != nil {
+		return nil, err
+	}
+	return ic.UnmarshalRsaPublicKey(remotePubKey)
+}
+
+func keyToCertificate(sk ic.PrivKey) (interface{}, *x509.Certificate, error) {
+	sn, err := rand.Int(rand.Reader, big.NewInt(1<<62))
+	if err != nil {
+		return nil, nil, err
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: sn,
+		NotBefore:    time.Now().Add(-24 * time.Hour),
+		NotAfter:     time.Now().Add(30 * 24 * time.Hour),
+		IsCA:         true,
+		BasicConstraintsValid: true,
+	}
+
+	var publicKey, privateKey interface{}
+	keyBytes, err := sk.Bytes()
+	if err != nil {
+		return nil, nil, err
+	}
+	pbmes := new(pb.PrivateKey)
+	if err := proto.Unmarshal(keyBytes, pbmes); err != nil {
+		return nil, nil, err
+	}
+	switch pbmes.GetType() {
+	case pb.KeyType_RSA:
+		k, err := x509.ParsePKCS1PrivateKey(pbmes.GetData())
+		if err != nil {
+			return nil, nil, err
+		}
+		publicKey = &k.PublicKey
+		privateKey = k
+	// TODO: add support for ECDSA
+	default:
+		return nil, nil, errors.New("unsupported key type for TLS")
+	}
+	certDER, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, publicKey, privateKey)
+	if err != nil {
+		return nil, nil, err
+	}
+	cert, err := x509.ParseCertificate(certDER)
+	if err != nil {
+		return nil, nil, err
+	}
+	return privateKey, cert, nil
+}

dialer.go

-package libp2pquic
-
-import (
-	"context"
-	"crypto/tls"
-
-	tpt "github.com/libp2p/go-libp2p-transport"
-	quic "github.com/lucas-clemente/quic-go"
-	ma "github.com/multiformats/go-multiaddr"
-	manet "github.com/multiformats/go-multiaddr-net"
-	"github.com/whyrusleeping/mafmt"
-)
-
-type dialer struct {
-	transport tpt.Transport
-}
-
-var _ tpt.Dialer = &dialer{}
-
-func newDialer(transport tpt.Transport) (*dialer, error) {
-	return &dialer{
-		transport: transport,
-	}, nil
-}
-
-func (d *dialer) Dial(raddr ma.Multiaddr) (tpt.Conn, error) {
-	// TODO: check that raddr is a QUIC address
-	_, host, err := manet.DialArgs(raddr)
-	if err != nil {
-		return nil, err
-	}
-
-	qsess, err := quic.DialAddr(host, &tls.Config{InsecureSkipVerify: true}, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	return newQuicConn(qsess, d.transport)
-}
-
-func (d *dialer) DialContext(ctx context.Context, raddr ma.Multiaddr) (tpt.Conn, error) {
-	// TODO: implement the ctx
-	return d.Dial(raddr)
-}
-
-func (d *dialer) Matches(a ma.Multiaddr) bool {
-	return mafmt.QUIC.Matches(a)
-}

dialer_test.go

-package libp2pquic
-
-import (
-	tpt "github.com/libp2p/go-libp2p-transport"
-	ma "github.com/multiformats/go-multiaddr"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-)
-
-var _ = Describe("Listener", func() {
-	var (
-		d         *dialer
-		transport tpt.Transport
-	)
-
-	BeforeEach(func() {
-		var err error
-		transport = &QuicTransport{}
-		d, err = newDialer(transport)
-		Expect(err).ToNot(HaveOccurred())
-	})
-
-	It("dials", func() {
-		addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/8888")
-		Expect(err).ToNot(HaveOccurred())
-
-		// start a listener to connect to
-		var ln *listener
-		go func() {
-			defer GinkgoRecover()
-			ln, err = newListener(addr, transport)
-			Expect(err).ToNot(HaveOccurred())
-			_, err = ln.Accept()
-			Expect(err).ToNot(HaveOccurred())
-		}()
-
-		Eventually(func() *listener { return ln }).ShouldNot(BeNil())
-		conn, err := d.Dial(addr)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(conn.Transport()).To(Equal(d.transport))
-	})
-
-	It("matches", func() {
-		invalidAddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234")
-		Expect(err).ToNot(HaveOccurred())
-		validAddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234/quic")
-		Expect(err).ToNot(HaveOccurred())
-		Expect(d.Matches(invalidAddr)).To(BeFalse())
-		Expect(d.Matches(validAddr)).To(BeTrue())
-	})
-})

listener.go

 package libp2pquic
 
 import (
-	"crypto/rand"
-	"crypto/rsa"
 	"crypto/tls"
-	"crypto/x509"
-	"encoding/pem"
-	"math/big"
 	"net"
 
+	ic "github.com/libp2p/go-libp2p-crypto"
+	peer "github.com/libp2p/go-libp2p-peer"
 	tpt "github.com/libp2p/go-libp2p-transport"
 	quic "github.com/lucas-clemente/quic-go"
 	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr-net"
 )
 
+var quicListenAddr = quic.ListenAddr
+
+// A listener listens for QUIC connections.
 type listener struct {
-	laddr        ma.Multiaddr
 	quicListener quic.Listener
+	transport    tpt.Transport
+
+	acceptQueue chan tpt.Conn
 
-	transport tpt.Transport
+	privKey        ic.PrivKey
+	localPeer      peer.ID
+	localMultiaddr ma.Multiaddr
 }
 
 var _ tpt.Listener = &listener{}
 
-func newListener(laddr ma.Multiaddr, t tpt.Transport) (*listener, error) {
-	_, host, err := manet.DialArgs(laddr)
+func newListener(addr ma.Multiaddr, transport tpt.Transport, localPeer peer.ID, key ic.PrivKey, tlsConf *tls.Config) (tpt.Listener, error) {
+	_, host, err := manet.DialArgs(addr)
 	if err != nil {
 		return nil, err
 	}
-	tlsConf, err := generateTLSConfig()
+	ln, err := quicListenAddr(host, tlsConf, quicConfig)
 	if err != nil {
 		return nil, err
 	}
-	qln, err := quic.ListenAddr(host, tlsConf, nil)
+	localMultiaddr, err := quicMultiaddr(ln.Addr())
 	if err != nil {
 		return nil, err
 	}
-	addr, err := quicMultiAddress(qln.Addr())
-	if err != nil {
-		return nil, err
-	}
-
 	return &listener{
-		laddr:        addr,
-		quicListener: qln,
-		transport:    t,
+		quicListener:   ln,
+		transport:      transport,
+		privKey:        key,
+		localPeer:      localPeer,
+		localMultiaddr: localMultiaddr,
 	}, nil
 }
 
+// Accept accepts new connections.
+// TODO(#2): don't accept a connection if the client's peer verification fails
 func (l *listener) Accept() (tpt.Conn, error) {
 	sess, err := l.quicListener.Accept()
 	if err != nil {
 		return nil, err
 	}
-	return newQuicConn(sess, l.transport)
+	remotePubKey, err := getRemotePubKey(sess.ConnectionState().PeerCertificates)
+	if err != nil {
+		return nil, err
+	}
+	remotePeerID, err := peer.IDFromPublicKey(remotePubKey)
+	if err != nil {
+		return nil, err
+	}
+	remoteMultiaddr, err := quicMultiaddr(sess.RemoteAddr())
+	if err != nil {
+		return nil, err
+	}
+	return &conn{
+		sess:            sess,
+		transport:       l.transport,
+		localPeer:       l.localPeer,
+		localMultiaddr:  l.localMultiaddr,
+		privKey:         l.privKey,
+		remoteMultiaddr: remoteMultiaddr,
+		remotePeerID:    remotePeerID,
+		remotePubKey:    remotePubKey,
+	}, nil
 }
 
+// Close closes the listener.
 func (l *listener) Close() error {
 	return l.quicListener.Close()
 }
 
+// Addr returns the address of this listener.
 func (l *listener) Addr() net.Addr {
 	return l.quicListener.Addr()
 }
 
+// Multiaddr returns the multiaddress of this listener.
 func (l *listener) Multiaddr() ma.Multiaddr {
-	return l.laddr
-}
-
-// Generate a bare-bones TLS config for the server.
-// The client doesn't verify the certificate yet.
-func generateTLSConfig() (*tls.Config, error) {
-	key, err := rsa.GenerateKey(rand.Reader, 2048)
-	if err != nil {
-		return nil, err
-	}
-	template := x509.Certificate{SerialNumber: big.NewInt(1)}
-	certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
-	if err != nil {
-		return nil, err
-	}
-	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)})
-	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
-	tlsCert, err := tls.X509KeyPair(certPEM, keyPEM)
-	if err != nil {
-		return nil, err
-	}
-	return &tls.Config{Certificates: []tls.Certificate{tlsCert}}, nil
+	return l.localMultiaddr
 }

listener_test.go

 package libp2pquic
 
 import (
-	"errors"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"fmt"
 	"net"
 
+	ic "github.com/libp2p/go-libp2p-crypto"
 	tpt "github.com/libp2p/go-libp2p-transport"
-	quic "github.com/lucas-clemente/quic-go"
 	ma "github.com/multiformats/go-multiaddr"
-
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 )
 
-type mockQuicListener struct {
-	connToAccept net.Conn
-	serveErr     error
-	closed       bool
-
-	sessionToAccept quic.Session
-	acceptErr       error
-}
-
-var _ quic.Listener = &mockQuicListener{}
-
-func (m *mockQuicListener) Close() error                  { m.closed = true; return nil }
-func (m *mockQuicListener) Accept() (quic.Session, error) { return m.sessionToAccept, m.acceptErr }
-func (m *mockQuicListener) Addr() net.Addr                { panic("not implemented") }
-
 var _ = Describe("Listener", func() {
 	var (
-		l            *listener
-		quicListener *mockQuicListener
-		transport    tpt.Transport
+		t         tpt.Transport
+		localAddr ma.Multiaddr
 	)
 
 	BeforeEach(func() {
-		quicListener = &mockQuicListener{}
-		transport = &QuicTransport{}
-		l = &listener{
-			quicListener: quicListener,
-			transport:    transport,
-		}
-	})
-
-	It("returns its addr", func() {
-		laddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/0/quic")
+		rsaKey, err := rsa.GenerateKey(rand.Reader, 1024)
+		Expect(err).ToNot(HaveOccurred())
+		key, err := ic.UnmarshalRsaPrivateKey(x509.MarshalPKCS1PrivateKey(rsaKey))
+		Expect(err).ToNot(HaveOccurred())
+		t, err = NewTransport(key)
 		Expect(err).ToNot(HaveOccurred())
-		l, err = newListener(laddr, nil)
+		localAddr, err = ma.NewMultiaddr("/ip4/127.0.0.1/udp/0/quic")
 		Expect(err).ToNot(HaveOccurred())
-		as := l.Addr().String()
-		Expect(as).ToNot(Equal("127.0.0.1:0)"))
-		Expect(as).To(ContainSubstring("127.0.0.1:"))
 	})
 
-	It("returns its multiaddr", func() {
-		laddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/0/quic")
+	It("returns the address it is listening on", func() {
+		ln, err := t.Listen(localAddr)
 		Expect(err).ToNot(HaveOccurred())
-		l, err = newListener(laddr, nil)
-		Expect(err).ToNot(HaveOccurred())
-		mas := l.Multiaddr().String()
-		Expect(mas).ToNot(Equal("/ip4/127.0.0.1/udp/0/quic"))
-		Expect(mas).To(ContainSubstring("/ip4/127.0.0.1/udp/"))
-		Expect(mas).To(ContainSubstring("/quic"))
+		netAddr := ln.Addr()
+		Expect(netAddr).To(BeAssignableToTypeOf(&net.UDPAddr{}))
+		port := netAddr.(*net.UDPAddr).Port
+		Expect(port).ToNot(BeZero())
+		Expect(ln.Multiaddr().String()).To(Equal(fmt.Sprintf("/ip4/127.0.0.1/udp/%d/quic", port)))
 	})
 
-	It("closes", func() {
-		err := l.Close()
+	It("returns Accept when it is closed", func() {
+		addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/0/quic")
+		Expect(err).ToNot(HaveOccurred())
+		ln, err := t.Listen(addr)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(quicListener.closed).To(BeTrue())
+		done := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			ln.Accept()
+			close(done)
+		}()
+		Consistently(done).ShouldNot(BeClosed())
+		Expect(ln.Close()).To(Succeed())
+		Eventually(done).Should(BeClosed())
 	})
 
-	Context("accepting", func() {
-		It("errors if no connection can be accepted", func() {
-			testErr := errors.New("test error")
-			quicListener.acceptErr = testErr
-			_, err := l.Accept()
-			Expect(err).To(MatchError(testErr))
-		})
+	It("doesn't accept Accept calls after it is closed", func() {
+		ln, err := t.Listen(localAddr)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(ln.Close()).To(Succeed())
+		_, err = ln.Accept()
+		Expect(err).To(HaveOccurred())
 	})
 })

stream.go

@@ -2,7 +2,7 @@ package libp2pquic
 
 import (
 	smux "github.com/libp2p/go-stream-muxer"
-	"github.com/lucas-clemente/quic-go"
+	quic "github.com/lucas-clemente/quic-go"
 )
 
 type stream struct {
@@ -12,6 +12,8 @@ type stream struct {
 var _ smux.Stream = &stream{}
 
 func (s *stream) Reset() error {
-	s.Stream.Reset(nil)
-	return nil
+	if err := s.Stream.CancelRead(0); err != nil {
+		return err
+	}
+	return s.Stream.CancelWrite(0)
 }

transport.go

 package libp2pquic
 
 import (
-	"fmt"
-	"sync"
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
 
+	ic "github.com/libp2p/go-libp2p-crypto"
+	peer "github.com/libp2p/go-libp2p-peer"
 	tpt "github.com/libp2p/go-libp2p-transport"
+	quic "github.com/lucas-clemente/quic-go"
 	ma "github.com/multiformats/go-multiaddr"
+	manet "github.com/multiformats/go-multiaddr-net"
 	"github.com/whyrusleeping/mafmt"
 )
 
-// QuicTransport implements a QUIC Transport
-type QuicTransport struct {
-	lmutex    sync.Mutex
-	listeners map[string]tpt.Listener
-
-	dmutex  sync.Mutex
-	dialers map[string]tpt.Dialer
+var quicConfig = &quic.Config{
+	MaxReceiveStreamFlowControlWindow:     3 * (1 << 20),   // 3 MB
+	MaxReceiveConnectionFlowControlWindow: 4.5 * (1 << 20), // 4.5 MB
+	Versions: []quic.VersionNumber{101},
 }
 
-var _ tpt.Transport = &QuicTransport{}
+var quicDialAddr = quic.DialAddr
 
-// NewQuicTransport creates a new QUIC Transport
-// it tracks dialers and listeners created
-func NewQuicTransport() *QuicTransport {
-	// utils.SetLogLevel(utils.LogLevelDebug)
-	return &QuicTransport{
-		listeners: make(map[string]tpt.Listener),
-		dialers:   make(map[string]tpt.Dialer),
-	}
+// The Transport implements the tpt.Transport interface for QUIC connections.
+type transport struct {
+	privKey   ic.PrivKey
+	localPeer peer.ID
+	tlsConf   *tls.Config
 }
 
-func (t *QuicTransport) Dialer(laddr ma.Multiaddr, opts ...tpt.DialOpt) (tpt.Dialer, error) {
-	if !t.Matches(laddr) {
-		return nil, fmt.Errorf("quic transport cannot dial %q", laddr)
-	}
-
-	t.dmutex.Lock()
-	defer t.dmutex.Unlock()
+var _ tpt.Transport = &transport{}
 
-	s := laddr.String()
-	d, ok := t.dialers[s]
-	if ok {
-		return d, nil
+// NewTransport creates a new QUIC transport
+func NewTransport(key ic.PrivKey) (tpt.Transport, error) {
+	localPeer, err := peer.IDFromPrivateKey(key)
+	if err != nil {
+		return nil, err
 	}
-
-	// TODO: read opts
-	quicd, err := newDialer(t)
+	tlsConf, err := generateConfig(key)
 	if err != nil {
 		return nil, err
 	}
-	t.dialers[s] = quicd
-	return quicd, nil
+	return &transport{
+		privKey:   key,
+		localPeer: localPeer,
+		tlsConf:   tlsConf,
+	}, nil
 }
 
-// Listen starts listening on laddr
-func (t *QuicTransport) Listen(laddr ma.Multiaddr) (tpt.Listener, error) {
-	if !t.Matches(laddr) {
-		return nil, fmt.Errorf("quic transport cannot listen on %q", laddr)
+// Dial dials a new QUIC connection
+func (t *transport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (tpt.Conn, error) {
+	_, host, err := manet.DialArgs(raddr)
+	if err != nil {
+		return nil, err
 	}
-
-	t.lmutex.Lock()
-	defer t.lmutex.Unlock()
-
-	l, ok := t.listeners[laddr.String()]
-	if ok {
-		return l, nil
+	var remotePubKey ic.PubKey
+	tlsConf := t.tlsConf.Clone()
+	tlsConf.VerifyPeerCertificate = func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
+		chain := make([]*x509.Certificate, len(rawCerts))
+		for i := 0; i < len(rawCerts); i++ {
+			cert, err := x509.ParseCertificate(rawCerts[i])
+			if err != nil {
+				return err
+			}
+			chain[i] = cert
+		}
+		var err error
+		remotePubKey, err = getRemotePubKey(chain)
+		if err != nil {
+			return err
+		}
+		if !p.MatchesPublicKey(remotePubKey) {
+			return errors.New("peer IDs don't match")
+		}
+		return nil
 	}
-
-	ln, err := newListener(laddr, t)
+	sess, err := quicDialAddr(host, tlsConf, quicConfig)
 	if err != nil {
 		return nil, err
 	}
+	localMultiaddr, err := quicMultiaddr(sess.LocalAddr())
+	if err != nil {
+		return nil, err
+	}
+	return &conn{
+		sess:            sess,
+		privKey:         t.privKey,
+		localPeer:       t.localPeer,
+		localMultiaddr:  localMultiaddr,
+		remotePubKey:    remotePubKey,
+		remotePeerID:    p,
+		remoteMultiaddr: raddr,
+	}, nil
+}
 
-	t.listeners[laddr.String()] = ln
-	return ln, nil
+// CanDial determines if we can dial to an address
+func (t *transport) CanDial(addr ma.Multiaddr) bool {
+	return mafmt.QUIC.Matches(addr)
 }
 
-func (t *QuicTransport) Matches(a ma.Multiaddr) bool {
-	return mafmt.QUIC.Matches(a)
+// Listen listens for new QUIC connections on the passed multiaddr.
+func (t *transport) Listen(addr ma.Multiaddr) (tpt.Listener, error) {
+	return newListener(addr, t, t.localPeer, t.privKey, t.tlsConf)
 }
 
-var _ tpt.Transport = &QuicTransport{}
+// Proxy returns true if this transport proxies.
+func (t *transport) Proxy() bool {
+	return false
+}
+
+// Protocols returns the set of protocols handled by this transport.
+func (t *transport) Protocols() []int {
+	return []int{ma.P_QUIC}
+}
+
+func (t *transport) String() string {
+	return "QUIC"
+}

transport_test.go

 package libp2pquic
 
 import (
+	tpt "github.com/libp2p/go-libp2p-transport"
 	ma "github.com/multiformats/go-multiaddr"
 
 	. "github.com/onsi/ginkgo"
@@ -8,75 +9,24 @@ import (
 )
 
 var _ = Describe("Transport", func() {
-	var t *QuicTransport
+	var t tpt.Transport
 
 	BeforeEach(func() {
-		t = NewQuicTransport()
+		t = &transport{}
 	})
 
-	Context("listening", func() {
-		It("creates a new listener", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234/quic")
-			Expect(err).ToNot(HaveOccurred())
-			ln, err := t.Listen(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			Expect(ln.Multiaddr()).To(Equal(maddr))
-		})
-
-		It("returns an existing listener", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1235/quic")
-			Expect(err).ToNot(HaveOccurred())
-			ln, err := t.Listen(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			Expect(ln.Multiaddr()).To(Equal(maddr))
-			ln2, err := t.Listen(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			Expect(ln2).To(Equal(ln))
-			Expect(t.listeners).To(HaveLen(1))
-		})
-
-		It("errors if the address is not a QUIC address", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1235/utp")
-			Expect(err).ToNot(HaveOccurred())
-			_, err = t.Listen(maddr)
-			Expect(err).To(MatchError("quic transport cannot listen on \"/ip4/127.0.0.1/udp/1235/utp\""))
-		})
-	})
-
-	Context("dialing", func() {
-		It("creates a new dialer", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234/quic")
-			Expect(err).ToNot(HaveOccurred())
-			d, err := t.Dialer(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			Expect(d).ToNot(BeNil())
-		})
-
-		It("returns an existing dialer", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1235/quic")
-			Expect(err).ToNot(HaveOccurred())
-			d, err := t.Dialer(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			d2, err := t.Dialer(maddr)
-			Expect(err).ToNot(HaveOccurred())
-			Expect(d2).To(Equal(d))
-			Expect(t.dialers).To(HaveLen(1))
-		})
-
-		It("errors if the address is not a QUIC address", func() {
-			maddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1235/utp")
-			Expect(err).ToNot(HaveOccurred())
-			_, err = t.Dialer(maddr)
-			Expect(err).To(MatchError("quic transport cannot dial \"/ip4/127.0.0.1/udp/1235/utp\""))
-		})
-	})
-
-	It("matches", func() {
+	It("says if it can dial an address", func() {
 		invalidAddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234")
 		Expect(err).ToNot(HaveOccurred())
 		validAddr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234/quic")
 		Expect(err).ToNot(HaveOccurred())
-		Expect(t.Matches(invalidAddr)).To(BeFalse())
-		Expect(t.Matches(validAddr)).To(BeTrue())
+		Expect(t.CanDial(invalidAddr)).To(BeFalse())
+		Expect(t.CanDial(validAddr)).To(BeTrue())
+	})
+
+	It("supports the QUIC protocol", func() {
+		protocols := t.Protocols()
+		Expect(protocols).To(HaveLen(1))
+		Expect(protocols[0]).To(Equal(ma.P_QUIC))
 	})
 })