Return proper errors from dial ice

Author: kylecarbs

wsnet/broker_test.go

@@ -13,11 +13,13 @@ import (
 	"nhooyr.io/websocket"
 )
 
+// DialConfig provides options to configure the Dial for a connection.
 type DialConfig struct {
-	ICEServers []ICEServer
+	ICEServers []webrtc.ICEServer
 }
 
 // Dial connects to the broker and negotiates a connection to a listener.
+//
 func Dial(ctx context.Context, broker string, config *DialConfig) (*Dialer, error) {
 	if config == nil {
 		config = &DialConfig{}
@@ -46,17 +48,11 @@ func Dial(ctx context.Context, broker string, config *DialConfig) (*Dialer, erro
 	if err != nil {
 		return nil, fmt.Errorf("create peer connection: %w", err)
 	}
-	rtc.OnICEConnectionStateChange(func(is webrtc.ICEConnectionState) {
-		fmt.Printf("WE CONNECTED!: %s\n", is)
-	})
-	rtc.OnICECandidate(func(i *webrtc.ICECandidate) {
-		fmt.Printf("WE GOT ICE: %+v\n", i)
-	})
 
 	flushCandidates := proxyICECandidates(rtc, nconn)
 
-	ctrl, err := rtc.CreateDataChannel("control", &webrtc.DataChannelInit{
-		Protocol: stringPtr("control"),
+	ctrl, err := rtc.CreateDataChannel(controlChannel, &webrtc.DataChannelInit{
+		Protocol: stringPtr(controlChannel),
 		Ordered:  boolPtr(true),
 	})
 	if err != nil {

wsnet/dial.go

@@ -2,35 +2,40 @@ package wsnet
 
 import (
 	"context"
-	"fmt"
-	"testing"
+	"errors"
 
-	"github.com/pion/ice/v2"
-	"github.com/pion/turn/v2"
 	"github.com/pion/webrtc/v3"
 )
 
-func TestDial(t *testing.T) {
-	t.Run("Example", func(t *testing.T) {
-		connectAddr, _ := listenBroker(t)
-		turnAddr := listenTURN(t, ice.ProtoTypeTCP, "wowie", true)
+func ExampleDial_basic() {
+	servers := []webrtc.ICEServer{{
+		URLs:           []string{"turns:master.cdr.dev"},
+		Username:       "kyle",
+		Credential:     "pass",
+		CredentialType: webrtc.ICECredentialTypePassword,
+	}}
 
-		dialer, err := Dial(context.Background(), connectAddr, &DialConfig{
-			[]ICEServer{{
-				URLs:           []string{turnAddr},
-				Username:       "insecure",
-				Credential:     "pass",
-				CredentialType: webrtc.ICECredentialTypePassword,
-			}},
-		})
-		if err != nil {
-			t.Error(err)
+	for _, server := range servers {
+		err := DialICE(server, DefaultICETimeout)
+		if errors.Is(err, ErrInvalidCredentials) {
+			// You could do something...
 		}
-		fmt.Printf("Dialer: %+v\n", dialer)
-	})
-}
-
-func testTURN() {
+		if errors.Is(err, ErrMismatchedProtocol) {
+			// Likely they used TURN when they should have used TURN.
+			// Or they could have used TURN instead of TURNS.
+		}
+	}
 
-	turn.NewServer(turn.ServerConfig{})
+	dialer, err := Dial(context.Background(), "wss://master.cdr.dev/agent/workspace/connect", &DialConfig{
+		ICEServers: servers,
+	})
+	if err != nil {
+		// Do something...
+	}
+	conn, err := dialer.DialContext(context.Background(), "tcp", "localhost:13337")
+	if err != nil {
+		// Something...
+	}
+	defer conn.Close()
+	// You now have access to the proxied remote port in `conn`.
 }

wsnet/dial_test.go

@@ -4,15 +4,19 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"net"
 
 	"cdr.dev/coder-cli/coder-sdk"
 	"github.com/hashicorp/yamux"
+	"github.com/pion/datachannel"
 	"github.com/pion/webrtc/v3"
 	"nhooyr.io/websocket"
 )
 
+// Listen connects to the broker and returns a Listener that's triggered
+// when a new connection is requested from a Dialer.
+//
+// LocalAddr on connections indicates the target specified by the dialer.
 func Listen(ctx context.Context, broker string) (net.Listener, error) {
 	conn, resp, err := websocket.Dial(ctx, broker, nil)
 	if err != nil {
@@ -26,100 +30,174 @@ func Listen(ctx context.Context, broker string) (net.Listener, error) {
 	nconn := websocket.NetConn(ctx, conn, websocket.MessageBinary)
 	session, err := yamux.Server(nconn, nil)
 	if err != nil {
-		return nil, fmt.Errorf("")
+		return nil, fmt.Errorf("create multiplex: %w", err)
 	}
-	return nil, nil
+	l := &listener{
+		ws:    conn,
+		conns: make(chan net.Conn),
+	}
+	go func() {
+		for {
+			conn, err := session.Accept()
+			if err != nil {
+				l.acceptError = err
+				l.Close()
+				return
+			}
+			l.negotiate(conn)
+		}
+	}()
+	return l, nil
 }
 
 type listener struct {
-	session *yamux.Session
-}
+	acceptError error
+	ws          *websocket.Conn
 
-func (l *listener) Accept() (net.Conn, error) {
-	conn, err := l.session.Accept()
-	if err != nil {
-		return nil, err
-	}
+	conns chan net.Conn
+}
 
+// Negotiates the handshake protocol over the connection provided.
+func (l *listener) negotiate(conn net.Conn) {
 	var (
-		decoder    = json.NewDecoder(conn)
-		closeError = func(err error) error {
+		err     error
+		decoder = json.NewDecoder(conn)
+		rtc     *webrtc.PeerConnection
+		// Sends the error provided then closes the connection.
+		// If RTC isn't connected, we'll close it.
+		closeError = func(err error) {
 			d, _ := json.Marshal(&protoMessage{
 				Error: err.Error(),
 			})
 			_, _ = conn.Write(d)
 			_ = conn.Close()
-			return err
+			if rtc != nil {
+				if rtc.ConnectionState() != webrtc.PeerConnectionStateConnected {
+					rtc.Close()
+					rtc = nil
+				}
+			}
 		}
-		rtc *webrtc.PeerConnection
 	)
 
 	for {
 		var msg protoMessage
 		err = decoder.Decode(&msg)
-		if err == io.EOF {
-			break
-		}
 		if err != nil {
-			return nil, err
+			closeError(err)
+			return
 		}
 
 		if msg.Candidate != "" {
 			if rtc == nil {
-				return nil, closeError(fmt.Errorf("Offer must be sent before candidates"))
+				closeError(fmt.Errorf("offer must be sent before candidates"))
+				return
 			}
 
 			err = rtc.AddICECandidate(webrtc.ICECandidateInit{
 				Candidate: msg.Candidate,
 			})
 			if err != nil {
-				return nil, closeError(fmt.Errorf("accept ice candidate: %w", err))
+				closeError(fmt.Errorf("accept ice candidate: %w", err))
+				return
 			}
 		}
 
 		if msg.Offer != nil {
 			if msg.Servers == nil {
-				return nil, closeError(fmt.Errorf("ICEServers must be provided"))
+				closeError(fmt.Errorf("ICEServers must be provided"))
+				return
 			}
 			rtc, err = newPeerConnection(msg.Servers)
 			if err != nil {
-				return nil, closeError(err)
+				closeError(err)
+				return
 			}
+			rtc.OnDataChannel(l.handle)
 			flushCandidates := proxyICECandidates(rtc, conn)
 			err = rtc.SetRemoteDescription(*msg.Offer)
 			if err != nil {
-				return nil, closeError(fmt.Errorf("apply offer: %w", err))
+				closeError(fmt.Errorf("apply offer: %w", err))
+				return
 			}
 			answer, err := rtc.CreateAnswer(nil)
 			if err != nil {
-				return nil, closeError(fmt.Errorf("create answer: %w", err))
+				closeError(fmt.Errorf("create answer: %w", err))
+				return
 			}
 			err = rtc.SetLocalDescription(answer)
 			if err != nil {
-				return nil, closeError(fmt.Errorf("set local answer: %w", err))
+				closeError(fmt.Errorf("set local answer: %w", err))
+				return
 			}
 			flushCandidates()
 
 			data, err := json.Marshal(&protoMessage{
 				Answer: rtc.LocalDescription(),
 			})
 			if err != nil {
-				return nil, closeError(fmt.Errorf("marshal: %w", err))
+				closeError(fmt.Errorf("marshal: %w", err))
+				return
 			}
 			_, err = conn.Write(data)
 			if err != nil {
-				return nil, closeError(fmt.Errorf("write: %w", err))
+				closeError(fmt.Errorf("write: %w", err))
+				return
 			}
 		}
 	}
+}
+
+func (l *listener) handle(dc *webrtc.DataChannel) {
+	// if dc.Protocol() == controlChannel {
+	// 	return
+	// }
+
+	fmt.Printf("GOT CHANNEL %s\n", dc.Protocol())
+
+	// dc.OnOpen(func() {
+	// 	rw, err := dc.Detach()
+	// })
+}
 
-	return nil, nil
+// Accept accepts a new connection.
+func (l *listener) Accept() (net.Conn, error) {
+	return <-l.conns, l.acceptError
 }
 
+// Close closes the broker socket.
 func (l *listener) Close() error {
-	return nil
+	close(l.conns)
+	return l.ws.Close(websocket.StatusNormalClosure, "")
 }
 
+// Since this listener is bound to the WebSocket, we could
+// return that resolved Addr, but until we need it we won't.
 func (l *listener) Addr() net.Addr {
 	return nil
 }
+
+type dataChannelConn struct {
+	rw        datachannel.ReadWriteCloser
+	localAddr net.Addr
+}
+
+func (d *dataChannelConn) Read(b []byte) (n int, err error) {
+	return d.rw.Read(b)
+}
+
+func (d *dataChannelConn) Write(b []byte) (n int, err error) {
+	return d.rw.Write(b)
+}
+
+func (d *dataChannelConn) Close() error {
+	return d.Close()
+}
+
+func (d *dataChannelConn) LocalAddr() net.Addr {
+	return d.localAddr
+}
+
+func (d *dataChannelConn) RemoteAddr() net.Addr {
+	return nil
+}