Implement E2E test for Client/Server

Relates to #11

pkg/dtls/conn.go

@@ -186,7 +186,7 @@ func (c *Conn) Write(p []byte) (int, error) {
 // Close closes the connection.
 func (c *Conn) Close() error {
 	c.notify(alertLevelFatal, alertCloseNotify)
-	c.stopWithError(errConnClosed)
+	c.stopWithError(ErrConnClosed)
 	return c.nextConn.Close()
 }
 
@@ -347,8 +347,9 @@ func (c *Conn) notify(level alertLevel, desc alertDescription) {
 
 func (c *Conn) signalHandshakeComplete() {
 	select {
-	case c.handshakeCompleted <- true:
+	case <-c.handshakeCompleted:
 	default:
+		close(c.handshakeCompleted)
 	}
 }
 
@@ -360,6 +361,8 @@ func (c *Conn) startHandshakeOutbound() {
 				err        error
 			)
 			select {
+			case <-c.handshakeCompleted:
+				return
 			case <-c.workerTicker.C:
 				isFinished, err = c.flightHandler(c)
 			case <-c.currFlight.workerTrigger:

pkg/dtls/errors.go

@@ -2,8 +2,10 @@ package dtls
 
 import "errors"
 
+// Typed errors
 var (
-	errConnClosed                        = errors.New("dtls: conn is closed")
+	ErrConnClosed = errors.New("dtls: conn is closed")
+
 	errBufferTooSmall                    = errors.New("dtls: buffer is too small")
 	errCertificateUnset                  = errors.New("dtls: handshakeMessageCertificate can not be marshalled without a certificate")
 	errCipherSuiteNoIntersection         = errors.New("dtls: Client+Server do not support any shared cipher suites")

pkg/dtls/internal/udp/conn.go

@@ -90,7 +90,6 @@ func Listen(network string, laddr *net.UDPAddr) (*Listener, error) {
 func (l *Listener) readLoop() {
 	buf := make([]byte, receiveMTU)
 
-readLoop:
 	for {
 		n, raddr, err := l.pConn.ReadFrom(buf)
 		if err != nil {
@@ -105,7 +104,7 @@ readLoop:
 			n = copy(cBuf, buf[:n])
 			conn.sizeCh <- n
 		case <-conn.doneCh:
-			continue readLoop
+			return
 		}
 	}
 }

test/e2e/pion/simple_test.go

@@ -0,0 +1,142 @@
+package pion
+
+import (
+	"io"
+	"net"
+	"os"
+	"runtime"
+	"runtime/pprof"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pions/dtls/pkg/dtls"
+)
+
+const testMessage = "Hello World"
+const serverPort = 5555
+const testTimeLimit = 5 * time.Second
+const messageRetry = 200 * time.Millisecond
+
+// Counter to make sure both sides got a message
+var messageRecvCount uint64
+
+func simpleReadWrite(errChan chan error, outChan chan string, conn io.ReadWriteCloser) {
+	go func() {
+		buffer := make([]byte, 8192)
+		n, err := conn.Read(buffer)
+		if err != nil {
+			errChan <- err
+			return
+		}
+
+		outChan <- string(buffer[:n])
+		atomic.AddUint64(&messageRecvCount, 1)
+	}()
+
+	for {
+		if atomic.LoadUint64(&messageRecvCount) == 2 {
+			break
+		} else if _, err := conn.Write([]byte(testMessage)); err != nil {
+			errChan <- err
+			break
+		}
+
+		time.Sleep(messageRetry)
+	}
+
+	if err := conn.Close(); err != nil {
+		select {
+		case errChan <- err: // Do we care about these errors?
+		default:
+		}
+	}
+}
+
+/*
+  Simple DTLS Client/Server can communicate
+    - Assert that you can send messages both ways
+	- Assert that Close() on both ends work
+	- Assert that no Goroutines are leaked
+*/
+func TestPionE2ESimple(t *testing.T) {
+	expectedGoRoutineCount := runtime.NumGoroutine()
+	errChan := make(chan error)
+	clientChan := make(chan string)
+	serverChan := make(chan string)
+
+	cert, key, err := createSelfSigned()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// DTLS Client
+	go func() {
+		conn, err := dtls.Dial("udp",
+			&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: serverPort},
+			&dtls.Config{Certificate: cert, PrivateKey: key},
+		)
+		if err != nil {
+			errChan <- err
+			return
+		}
+		simpleReadWrite(errChan, clientChan, conn)
+	}()
+
+	// DTLS Server
+	go func() {
+		listener, err := dtls.Listen("udp",
+			&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: serverPort},
+			&dtls.Config{Certificate: cert, PrivateKey: key},
+		)
+		if err != nil {
+			errChan <- err
+			return
+		}
+		conn, err := listener.Accept()
+		if err != nil {
+			errChan <- err
+			return
+		}
+
+		simpleReadWrite(errChan, serverChan, conn)
+	}()
+
+	func() {
+		seenClient, seenServer := false, false
+		for {
+			select {
+			case err := <-errChan:
+				t.Fatal(err)
+			case <-time.After(testTimeLimit):
+				t.Fatalf("Test timeout, seenClient %t seenServer %t", seenClient, seenServer)
+			case clientMsg := <-clientChan:
+				if clientMsg != testMessage {
+					t.Fatalf("clientMsg does not equal test message: %s %s", clientMsg, testMessage)
+				}
+
+				seenClient = true
+				if seenClient && seenServer {
+					return
+				}
+			case serverMsg := <-serverChan:
+				if serverMsg != testMessage {
+					t.Fatalf("serverMsg does not equal test message: %s %s", serverMsg, testMessage)
+				}
+
+				seenServer = true
+				if seenClient && seenServer {
+					return
+				}
+			}
+		}
+	}()
+
+	// We have seen Client+Server communication! Now count goroutines to make sure we haven't leaked
+	time.Sleep(time.Second) // TODO racey
+	goRoutineCount := runtime.NumGoroutine()
+	if goRoutineCount != expectedGoRoutineCount {
+		pprof.Lookup("goroutine").WriteTo(os.Stderr, 1)
+		t.Fatalf("goRoutineCount != expectedGoRoutineCount, possible leak: %d %d", goRoutineCount, expectedGoRoutineCount)
+	}
+}

test/e2e/pion/util.go

@@ -0,0 +1,58 @@
+package pion
+
+import (
+	"crypto"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/hex"
+	"math/big"
+	"time"
+)
+
+func createSelfSigned() (*x509.Certificate, crypto.PrivateKey, error) {
+	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	origin := make([]byte, 16)
+
+	// Max random value, a 130-bits integer, i.e 2^130 - 1
+	maxBigInt := new(big.Int)
+	/* #nosec */
+	maxBigInt.Exp(big.NewInt(2), big.NewInt(130), nil).Sub(maxBigInt, big.NewInt(1))
+	serialNumber, err := rand.Int(rand.Reader, maxBigInt)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	template := x509.Certificate{
+		ExtKeyUsage: []x509.ExtKeyUsage{
+			x509.ExtKeyUsageClientAuth,
+			x509.ExtKeyUsageServerAuth,
+		},
+		BasicConstraintsValid: true,
+		NotBefore:             time.Now(),
+		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
+		NotAfter:              time.Now().AddDate(0, 1, 0),
+		SerialNumber:          serialNumber,
+		Version:               2,
+		Subject:               pkix.Name{CommonName: hex.EncodeToString(origin)},
+		IsCA:                  true,
+	}
+
+	raw, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	cert, err := x509.ParseCertificate(raw)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return cert, priv, nil
+}