Test sending data

wsnet/conn.go

@@ -0,0 +1,45 @@
+package wsnet
+
+import (
+	"net"
+	"time"
+
+	"github.com/pion/datachannel"
+)
+
+type conn struct {
+	addr *net.UnixAddr
+	rw   datachannel.ReadWriteCloser
+}
+
+func (c *conn) Read(b []byte) (n int, err error) {
+	return c.rw.Read(b)
+}
+
+func (c *conn) Write(b []byte) (n int, err error) {
+	return c.rw.Write(b)
+}
+
+func (c *conn) Close() error {
+	return c.rw.Close()
+}
+
+func (c *conn) LocalAddr() net.Addr {
+	return c.addr
+}
+
+func (c *conn) RemoteAddr() net.Addr {
+	return c.addr
+}
+
+func (c *conn) SetDeadline(t time.Time) error {
+	return nil
+}
+
+func (c *conn) SetReadDeadline(t time.Time) error {
+	return nil
+}
+
+func (c *conn) SetWriteDeadline(t time.Time) error {
+	return nil
+}

wsnet/dial.go

@@ -9,6 +9,7 @@ import (
 	"net"
 
 	"cdr.dev/coder-cli/coder-sdk"
+	"github.com/pion/datachannel"
 	"github.com/pion/webrtc/v3"
 	"nhooyr.io/websocket"
 )
@@ -22,7 +23,9 @@ type DialConfig struct {
 //
 func Dial(ctx context.Context, broker string, config *DialConfig) (*Dialer, error) {
 	if config == nil {
-		config = &DialConfig{}
+		config = &DialConfig{
+			ICEServers: []webrtc.ICEServer{},
+		}
 	}
 
 	conn, resp, err := websocket.Dial(ctx, broker, nil)
@@ -92,15 +95,17 @@ func Dial(ctx context.Context, broker string, config *DialConfig) (*Dialer, erro
 			_ = conn.Close(websocket.StatusAbnormalClosure, "timeout")
 			return
 		}
+		dialer.ctrlrw, _ = ctrl.Detach()
 		_ = conn.Close(websocket.StatusNormalClosure, "connected")
 	}()
 
 	return dialer, dialer.negotiate(nconn)
 }
 
 type Dialer struct {
-	ctrl *webrtc.DataChannel
-	rtc  *webrtc.PeerConnection
+	ctrl   *webrtc.DataChannel
+	ctrlrw datachannel.ReadWriteCloser
+	rtc    *webrtc.PeerConnection
 }
 
 func (d *Dialer) negotiate(nconn net.Conn) (err error) {
@@ -142,14 +147,45 @@ func (d *Dialer) negotiate(nconn net.Conn) (err error) {
 	return nil
 }
 
+// Close closes the RTC connection.
+// All data channels dialed will be closed.
 func (d *Dialer) Close() error {
-	return nil
+	return d.rtc.Close()
 }
 
+// Ping sends a ping through the control channel.
 func (d *Dialer) Ping(ctx context.Context) error {
-	return nil
+	_, err := d.ctrlrw.Write([]byte{'a'})
+	if err != nil {
+		return fmt.Errorf("write: %w", err)
+	}
+	b := make([]byte, 4)
+	_, err = d.ctrlrw.Read(b)
+	return err
 }
 
+// DialContext dials the network and address on the remote listener.
 func (d *Dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
-	return nil, nil
+	dc, err := d.rtc.CreateDataChannel("proxy", &webrtc.DataChannelInit{
+		Ordered:  boolPtr(network != "udp"),
+		Protocol: stringPtr(fmt.Sprintf("%s:%s", network, address)),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create data channel: %w", err)
+	}
+	err = waitForDataChannelOpen(ctx, dc)
+	if err != nil {
+		return nil, fmt.Errorf("wait for open: %w", err)
+	}
+	rw, err := dc.Detach()
+	if err != nil {
+		return nil, fmt.Errorf("detach: %w", err)
+	}
+	return &conn{
+		addr: &net.UnixAddr{
+			Name: address,
+			Net:  network,
+		},
+		rw: rw,
+	}, nil
 }

wsnet/dial_test.go

@@ -3,6 +3,8 @@ package wsnet
 import (
 	"context"
 	"errors"
+	"fmt"
+	"testing"
 
 	"github.com/pion/webrtc/v3"
 )
@@ -39,3 +41,48 @@ func ExampleDial_basic() {
 	defer conn.Close()
 	// You now have access to the proxied remote port in `conn`.
 }
+
+func TestDial(t *testing.T) {
+	t.Run("Ping", func(t *testing.T) {
+		connectAddr, listenAddr := createDumbBroker(t)
+		_, err := Listen(context.Background(), listenAddr)
+		if err != nil {
+			t.Error(err)
+		}
+		dialer, err := Dial(context.Background(), connectAddr, nil)
+		if err != nil {
+			t.Error(err)
+		}
+		err = dialer.Ping(context.Background())
+		if err != nil {
+			t.Error(err)
+		}
+
+	})
+
+	t.Run("Pipe", func(t *testing.T) {
+		connectAddr, listenAddr := createDumbBroker(t)
+		listener, err := Listen(context.Background(), listenAddr)
+		if err != nil {
+			t.Error(err)
+		}
+		dialer, err := Dial(context.Background(), connectAddr, nil)
+		if err != nil {
+			t.Error(err)
+		}
+		go func() {
+			conn, err := dialer.DialContext(context.Background(), "tcp", "localhost:40000")
+			if err != nil {
+				t.Error(err)
+			}
+			conn.Write([]byte("hello"))
+		}()
+		conn, err := listener.Accept()
+		if err != nil {
+			t.Error(err)
+		}
+		b := make([]byte, 5)
+		_, _ = conn.Read(b)
+		fmt.Printf("WE LEGIT GOT IT! %s\n", b)
+	})
+}

wsnet/listen.go

@@ -5,10 +5,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
+	"strings"
 
 	"cdr.dev/coder-cli/coder-sdk"
 	"github.com/hashicorp/yamux"
-	"github.com/pion/datachannel"
 	"github.com/pion/webrtc/v3"
 	"nhooyr.io/websocket"
 )
@@ -149,15 +149,43 @@ func (l *listener) negotiate(conn net.Conn) {
 }
 
 func (l *listener) handle(dc *webrtc.DataChannel) {
-	// if dc.Protocol() == controlChannel {
-	// 	return
-	// }
-
-	fmt.Printf("GOT CHANNEL %s\n", dc.Protocol())
+	if dc.Protocol() == controlChannel {
+		// The control channel handles pings.
+		dc.OnOpen(func() {
+			rw, err := dc.Detach()
+			if err != nil {
+				return
+			}
+			// We'll read and write back a single byte for ping/pongin'.
+			d := make([]byte, 1)
+			for {
+				_, err = rw.Read(d)
+				if err != nil {
+					continue
+				}
+				_, _ = rw.Write(d)
+			}
+		})
+		return
+	}
 
-	// dc.OnOpen(func() {
-	// 	rw, err := dc.Detach()
-	// })
+	dc.OnOpen(func() {
+		rw, err := dc.Detach()
+		if err != nil {
+			return
+		}
+		parts := strings.SplitN(dc.Protocol(), ":", 2)
+		network := parts[0]
+		addr := parts[1]
+
+		l.conns <- &conn{
+			addr: &net.UnixAddr{
+				Name: addr,
+				Net:  network,
+			},
+			rw: rw,
+		}
+	})
 }
 
 // Accept accepts a new connection.
@@ -176,28 +204,3 @@ func (l *listener) Close() error {
 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
-}

wsnet/rtc.go

@@ -20,6 +20,10 @@ import (
 var (
 	ErrMismatchedProtocol = errors.New("mismatched protocols")
 	ErrInvalidCredentials = errors.New("invalid credentials")
+
+	DefaultPublicSTUN = webrtc.ICEServer{
+		URLs: []string{"stun:stun.l.google.com:19302"},
+	}
 )
 
 const (