Rename proxy-go/ directory to proxy/

Now that the web proxies are in a different repository, no need to distinguish the two.
2025-10-13 20:11:19 -04:00 · 2020-04-16 10:02:11 -04:00 · 2020-04-16 10:02:11 -04:00 · 20180dcb04
commit 20180dcb04
parent 3ff04c3c65
4 changed files with 3 additions and 3 deletions
--- a/proxy/README.md
+++ b/proxy/README.md
@ -0,0 +1,3 @@
+This is a standalone (not browser-based) version of the Snowflake proxy.
+
+Usage: ./proxy
--- a/proxy/proxy-go_test.go
+++ b/proxy/proxy-go_test.go
@ -0,0 +1,397 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"testing"
+
+	"git.torproject.org/pluggable-transports/snowflake.git/common/messages"
+	"git.torproject.org/pluggable-transports/snowflake.git/common/util"
+	"github.com/pion/webrtc/v2"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+// Set up a mock broker to communicate with
+type MockTransport struct {
+	statusOverride int
+	body           []byte
+}
+
+// Just returns a response with fake SDP answer.
+func (m *MockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	s := ioutil.NopCloser(bytes.NewReader(m.body))
+	r := &http.Response{
+		StatusCode: m.statusOverride,
+		Body:       s,
+	}
+	return r, nil
+}
+
+// Set up a mock faulty transport
+type FaultyTransport struct {
+	statusOverride int
+	body           []byte
+}
+
+// Just returns a response with fake SDP answer.
+func (f *FaultyTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	return nil, fmt.Errorf("TransportFailed")
+}
+
+func TestRemoteIPFromSDP(t *testing.T) {
+	tests := []struct {
+		sdp      string
+		expected net.IP
+	}{
+		// https://tools.ietf.org/html/rfc4566#section-5
+		{`v=0
+o=jdoe 2890844526 2890842807 IN IP4 10.47.16.5
+s=SDP Seminar
+i=A Seminar on the session description protocol
+u=http://www.example.com/seminars/sdp.pdf
+e=j.doe@example.com (Jane Doe)
+c=IN IP4 224.2.17.12/127
+t=2873397496 2873404696
+a=recvonly
+m=audio 49170 RTP/AVP 0
+m=video 51372 RTP/AVP 99
+a=rtpmap:99 h263-1998/90000
+`, net.ParseIP("224.2.17.12")},
+		// Missing c= line
+		{`v=0
+o=jdoe 2890844526 2890842807 IN IP4 10.47.16.5
+s=SDP Seminar
+i=A Seminar on the session description protocol
+u=http://www.example.com/seminars/sdp.pdf
+e=j.doe@example.com (Jane Doe)
+t=2873397496 2873404696
+a=recvonly
+m=audio 49170 RTP/AVP 0
+m=video 51372 RTP/AVP 99
+a=rtpmap:99 h263-1998/90000
+`, nil},
+		// Single line, IP address only
+		{`c=IN IP4 224.2.1.1
+`, net.ParseIP("224.2.1.1")},
+		// Same, with TTL
+		{`c=IN IP4 224.2.1.1/127
+`, net.ParseIP("224.2.1.1")},
+		// Same, with TTL and multicast addresses
+		{`c=IN IP4 224.2.1.1/127/3
+`, net.ParseIP("224.2.1.1")},
+		// IPv6, address only
+		{`c=IN IP6 FF15::101
+`, net.ParseIP("ff15::101")},
+		// Same, with multicast addresses
+		{`c=IN IP6 FF15::101/3
+`, net.ParseIP("ff15::101")},
+		// Multiple c= lines
+		{`c=IN IP4 1.2.3.4
+c=IN IP4 5.6.7.8
+`, net.ParseIP("1.2.3.4")},
+		// Modified from SDP sent by snowflake-client.
+		{`v=0
+o=- 7860378660295630295 2 IN IP4 127.0.0.1
+s=-
+t=0 0
+a=group:BUNDLE data
+a=msid-semantic: WMS
+m=application 54653 DTLS/SCTP 5000
+c=IN IP4 1.2.3.4
+a=candidate:3581707038 1 udp 2122260223 192.168.0.1 54653 typ host generation 0 network-id 1 network-cost 50
+a=candidate:2617212910 1 tcp 1518280447 192.168.0.1 59673 typ host tcptype passive generation 0 network-id 1 network-cost 50
+a=candidate:2082671819 1 udp 1686052607 1.2.3.4 54653 typ srflx raddr 192.168.0.1 rport 54653 generation 0 network-id 1 network-cost 50
+a=ice-ufrag:IBdf
+a=ice-pwd:G3lTrrC9gmhQx481AowtkhYz
+a=fingerprint:sha-256 53:F8:84:D9:3C:1F:A0:44:AA:D6:3C:65:80:D3:CB:6F:23:90:17:41:06:F9:9C:10:D8:48:4A:A8:B6:FA:14:A1
+a=setup:actpass
+a=mid:data
+a=sctpmap:5000 webrtc-datachannel 1024
+`, net.ParseIP("1.2.3.4")},
+		// Improper character within IPv4
+		{`c=IN IP4 224.2z.1.1
+`, nil},
+		// Improper character within IPv6
+		{`c=IN IP6 ff15:g::101
+`, nil},
+		// Bogus "IP7" addrtype
+		{`c=IN IP7 1.2.3.4
+`, nil},
+	}
+
+	for _, test := range tests {
+		// https://tools.ietf.org/html/rfc4566#section-5: "The sequence
+		// CRLF (0x0d0a) is used to end a record, although parsers
+		// SHOULD be tolerant and also accept records terminated with a
+		// single newline character." We represent the test cases with
+		// LF line endings for convenience, and test them both that way
+		// and with CRLF line endings.
+		lfSDP := test.sdp
+		crlfSDP := strings.Replace(lfSDP, "\n", "\r\n", -1)
+
+		ip := remoteIPFromSDP(lfSDP)
+		if !ip.Equal(test.expected) {
+			t.Errorf("expected %q, got %q from %q", test.expected, ip, lfSDP)
+		}
+		ip = remoteIPFromSDP(crlfSDP)
+		if !ip.Equal(test.expected) {
+			t.Errorf("expected %q, got %q from %q", test.expected, ip, crlfSDP)
+		}
+	}
+}
+
+func TestSessionDescriptions(t *testing.T) {
+	Convey("Session description deserialization", t, func() {
+		for _, test := range []struct {
+			msg string
+			ret *webrtc.SessionDescription
+		}{
+			{
+				"test",
+				nil,
+			},
+			{
+				`{"type":"answer"}`,
+				nil,
+			},
+			{
+				`{"sdp":"test"}`,
+				nil,
+			},
+			{
+				`{"type":"test", "sdp":"test"}`,
+				nil,
+			},
+			{
+				`{"type":"answer", "sdp":"test"}`,
+				&webrtc.SessionDescription{
+					Type: webrtc.SDPTypeAnswer,
+					SDP:  "test",
+				},
+			},
+			{
+				`{"type":"pranswer", "sdp":"test"}`,
+				&webrtc.SessionDescription{
+					Type: webrtc.SDPTypePranswer,
+					SDP:  "test",
+				},
+			},
+			{
+				`{"type":"rollback", "sdp":"test"}`,
+				&webrtc.SessionDescription{
+					Type: webrtc.SDPTypeRollback,
+					SDP:  "test",
+				},
+			},
+			{
+				`{"type":"offer", "sdp":"test"}`,
+				&webrtc.SessionDescription{
+					Type: webrtc.SDPTypeOffer,
+					SDP:  "test",
+				},
+			},
+		} {
+			desc := util.DeserializeSessionDescription(test.msg)
+			So(desc, ShouldResemble, test.ret)
+		}
+	})
+	Convey("Session description serialization", t, func() {
+		for _, test := range []struct {
+			desc *webrtc.SessionDescription
+			ret  string
+		}{
+			{
+				&webrtc.SessionDescription{
+					Type: webrtc.SDPTypeOffer,
+					SDP:  "test",
+				},
+				`{"type":"offer","sdp":"test"}`,
+			},
+		} {
+			msg := util.SerializeSessionDescription(test.desc)
+			So(msg, ShouldResemble, test.ret)
+		}
+	})
+}
+
+func TestBrokerInteractions(t *testing.T) {
+	const sampleSDP = `"v=0\r\no=- 4358805017720277108 2 IN IP4 8.8.8.8\r\ns=-\r\nt=0 0\r\na=group:BUNDLE data\r\na=msid-semantic: WMS\r\nm=application 56688 DTLS/SCTP 5000\r\nc=IN IP4 8.8.8.8\r\na=candidate:3769337065 1 udp 2122260223 8.8.8.8 56688 typ host generation 0 network-id 1 network-cost 50\r\na=candidate:2921887769 1 tcp 1518280447 8.8.8.8 35441 typ host tcptype passive generation 0 network-id 1 network-cost 50\r\na=ice-ufrag:aMAZ\r\na=ice-pwd:jcHb08Jjgrazp2dzjdrvPPvV\r\na=ice-options:trickle\r\na=fingerprint:sha-256 C8:88:EE:B9:E7:02:2E:21:37:ED:7A:D1:EB:2B:A3:15:A2:3B:5B:1C:3D:D4:D5:1F:06:CF:52:40:03:F8:DD:66\r\na=setup:actpass\r\na=mid:data\r\na=sctpmap:5000 webrtc-datachannel 1024\r\n"`
+
+	const sampleOffer = `{"type":"offer","sdp":` + sampleSDP + `}`
+	const sampleAnswer = `{"type":"answer","sdp":` + sampleSDP + `}`
+
+	Convey("Proxy connections to broker", t, func() {
+		broker := new(Broker)
+		broker.url, _ = url.Parse("localhost")
+
+		//Mock peerConnection
+		config = webrtc.Configuration{
+			ICEServers: []webrtc.ICEServer{
+				{
+					URLs: []string{"stun:stun.l.google.com:19302"},
+				},
+			},
+		}
+		pc, _ := webrtc.NewPeerConnection(config)
+		offer := util.DeserializeSessionDescription(sampleOffer)
+		pc.SetRemoteDescription(*offer)
+		answer, _ := pc.CreateAnswer(nil)
+		pc.SetLocalDescription(answer)
+
+		Convey("polls broker correctly", func() {
+			var err error
+
+			b, err := messages.EncodePollResponse(sampleOffer, true)
+			So(err, ShouldEqual, nil)
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				b,
+			}
+
+			sdp := broker.pollOffer(sampleOffer)
+			expectedSDP, _ := strconv.Unquote(sampleSDP)
+			So(sdp.SDP, ShouldResemble, expectedSDP)
+		})
+		Convey("handles poll error", func() {
+			var err error
+
+			b := []byte("test")
+			So(err, ShouldEqual, nil)
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				b,
+			}
+
+			sdp := broker.pollOffer(sampleOffer)
+			So(sdp, ShouldBeNil)
+		})
+		Convey("sends answer to broker", func() {
+			var err error
+
+			b, err := messages.EncodeAnswerResponse(true)
+			So(err, ShouldEqual, nil)
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				b,
+			}
+
+			err = broker.sendAnswer(sampleAnswer, pc)
+			So(err, ShouldEqual, nil)
+
+			b, err = messages.EncodeAnswerResponse(false)
+			So(err, ShouldEqual, nil)
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				b,
+			}
+
+			err = broker.sendAnswer(sampleAnswer, pc)
+			So(err, ShouldNotBeNil)
+		})
+		Convey("handles answer error", func() {
+			//Error if faulty transport
+			broker.transport = &FaultyTransport{}
+			err := broker.sendAnswer(sampleAnswer, pc)
+			So(err, ShouldNotBeNil)
+
+			//Error if status code is not ok
+			broker.transport = &MockTransport{
+				http.StatusGone,
+				[]byte(""),
+			}
+			err = broker.sendAnswer("test", pc)
+			So(err, ShouldNotEqual, nil)
+			So(err.Error(), ShouldResemble, "broker returned 410")
+
+			//Error if we can't parse broker message
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				[]byte("test"),
+			}
+			err = broker.sendAnswer("test", pc)
+			So(err, ShouldNotBeNil)
+
+			//Error if broker message surpasses read limit
+			broker.transport = &MockTransport{
+				http.StatusOK,
+				make([]byte, 100001),
+			}
+			err = broker.sendAnswer("test", pc)
+			So(err, ShouldNotBeNil)
+		})
+	})
+}
+
+func TestUtilityFuncs(t *testing.T) {
+	Convey("LimitedRead", t, func() {
+		c, s := net.Pipe()
+		Convey("Successful read", func() {
+			go func() {
+				bytes := make([]byte, 50)
+				c.Write(bytes)
+				c.Close()
+			}()
+			bytes, err := limitedRead(s, 60)
+			So(len(bytes), ShouldEqual, 50)
+			So(err, ShouldBeNil)
+		})
+		Convey("Large read", func() {
+			go func() {
+				bytes := make([]byte, 50)
+				c.Write(bytes)
+				c.Close()
+			}()
+			bytes, err := limitedRead(s, 49)
+			So(len(bytes), ShouldEqual, 49)
+			So(err, ShouldEqual, io.ErrUnexpectedEOF)
+		})
+		Convey("Failed read", func() {
+			s.Close()
+			bytes, err := limitedRead(s, 49)
+			So(len(bytes), ShouldEqual, 0)
+			So(err, ShouldEqual, io.ErrClosedPipe)
+		})
+	})
+	Convey("Tokens", t, func() {
+		tokens = make(chan bool, 2)
+		for i := uint(0); i < 2; i++ {
+			tokens <- true
+		}
+		So(len(tokens), ShouldEqual, 2)
+		getToken()
+		So(len(tokens), ShouldEqual, 1)
+		retToken()
+		So(len(tokens), ShouldEqual, 2)
+	})
+	Convey("SessionID Generation", t, func() {
+		sid1 := genSessionID()
+		sid2 := genSessionID()
+		So(sid1, ShouldNotEqual, sid2)
+	})
+	Convey("CopyLoop", t, func() {
+		c1, s1 := net.Pipe()
+		c2, s2 := net.Pipe()
+		go CopyLoop(s1, s2)
+		go func() {
+			bytes := []byte("Hello!")
+			c1.Write(bytes)
+		}()
+		bytes := make([]byte, 6)
+		n, err := c2.Read(bytes)
+		So(n, ShouldEqual, 6)
+		So(err, ShouldEqual, nil)
+		So(bytes, ShouldResemble, []byte("Hello!"))
+		s1.Close()
+
+		//Check that copy loop has closed other connection
+		_, err = s2.Write(bytes)
+		So(err, ShouldNotBeNil)
+	})
+}
--- a/proxy/snowflake.go
+++ b/proxy/snowflake.go
@ -0,0 +1,484 @@
+package main
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/base64"
+	"flag"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"regexp"
+	"strings"
+	"sync"
+	"time"
+
+	"git.torproject.org/pluggable-transports/snowflake.git/common/messages"
+	"git.torproject.org/pluggable-transports/snowflake.git/common/safelog"
+	"git.torproject.org/pluggable-transports/snowflake.git/common/util"
+	"git.torproject.org/pluggable-transports/snowflake.git/common/websocketconn"
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v2"
+)
+
+const defaultBrokerURL = "https://snowflake-broker.bamsoftware.com/"
+const defaultRelayURL = "wss://snowflake.bamsoftware.com/"
+const defaultSTUNURL = "stun:stun.l.google.com:19302"
+const pollInterval = 5 * time.Second
+
+//amount of time after sending an SDP answer before the proxy assumes the
+//client is not going to connect
+const dataChannelTimeout = 20 * time.Second
+
+const readLimit = 100000 //Maximum number of bytes to be read from an HTTP request
+
+var broker *Broker
+var relayURL string
+
+const (
+	sessionIDLength = 16
+)
+
+var (
+	tokens chan bool
+	config webrtc.Configuration
+	client http.Client
+)
+
+var remoteIPPatterns = []*regexp.Regexp{
+	/* IPv4 */
+	regexp.MustCompile(`(?m)^c=IN IP4 ([\d.]+)(?:(?:\/\d+)?\/\d+)?(:? |\r?\n)`),
+	/* IPv6 */
+	regexp.MustCompile(`(?m)^c=IN IP6 ([0-9A-Fa-f:.]+)(?:\/\d+)?(:? |\r?\n)`),
+}
+
+// https://tools.ietf.org/html/rfc4566#section-5.7
+func remoteIPFromSDP(sdp string) net.IP {
+	for _, pattern := range remoteIPPatterns {
+		m := pattern.FindStringSubmatch(sdp)
+		if m != nil {
+			// Ignore parsing errors, ParseIP returns nil.
+			return net.ParseIP(m[1])
+		}
+	}
+	return nil
+}
+
+type Broker struct {
+	url                *url.URL
+	transport          http.RoundTripper
+	keepLocalAddresses bool
+}
+
+type webRTCConn struct {
+	dc *webrtc.DataChannel
+	pc *webrtc.PeerConnection
+	pr *io.PipeReader
+
+	lock sync.Mutex // Synchronization for DataChannel destruction
+	once sync.Once  // Synchronization for PeerConnection destruction
+}
+
+func (c *webRTCConn) Read(b []byte) (int, error) {
+	return c.pr.Read(b)
+}
+
+func (c *webRTCConn) Write(b []byte) (int, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.dc != nil {
+		c.dc.Send(b)
+	}
+	return len(b), nil
+}
+
+func (c *webRTCConn) Close() (err error) {
+	c.once.Do(func() {
+		err = c.pc.Close()
+	})
+	return
+}
+
+func (c *webRTCConn) LocalAddr() net.Addr {
+	return nil
+}
+
+func (c *webRTCConn) RemoteAddr() net.Addr {
+	//Parse Remote SDP offer and extract client IP
+	clientIP := remoteIPFromSDP(c.pc.RemoteDescription().SDP)
+	if clientIP == nil {
+		return nil
+	}
+	return &net.IPAddr{IP: clientIP, Zone: ""}
+}
+
+func (c *webRTCConn) SetDeadline(t time.Time) error {
+	// nolint: golint
+	return fmt.Errorf("SetDeadline not implemented")
+}
+
+func (c *webRTCConn) SetReadDeadline(t time.Time) error {
+	// nolint: golint
+	return fmt.Errorf("SetReadDeadline not implemented")
+}
+
+func (c *webRTCConn) SetWriteDeadline(t time.Time) error {
+	// nolint: golint
+	return fmt.Errorf("SetWriteDeadline not implemented")
+}
+
+func getToken() {
+	<-tokens
+}
+
+func retToken() {
+	tokens <- true
+}
+
+func genSessionID() string {
+	buf := make([]byte, sessionIDLength)
+	_, err := rand.Read(buf)
+	if err != nil {
+		panic(err.Error())
+	}
+	return strings.TrimRight(base64.StdEncoding.EncodeToString(buf), "=")
+}
+
+func limitedRead(r io.Reader, limit int64) ([]byte, error) {
+	p, err := ioutil.ReadAll(&io.LimitedReader{R: r, N: limit + 1})
+	if err != nil {
+		return p, err
+	} else if int64(len(p)) == limit+1 {
+		return p[0:limit], io.ErrUnexpectedEOF
+	}
+	return p, err
+}
+
+func (b *Broker) pollOffer(sid string) *webrtc.SessionDescription {
+	brokerPath := b.url.ResolveReference(&url.URL{Path: "proxy"})
+	timeOfNextPoll := time.Now()
+	for {
+		// Sleep until we're scheduled to poll again.
+		now := time.Now()
+		time.Sleep(timeOfNextPoll.Sub(now))
+		// Compute the next time to poll -- if it's in the past, that
+		// means that the POST took longer than pollInterval, so we're
+		// allowed to do another one immediately.
+		timeOfNextPoll = timeOfNextPoll.Add(pollInterval)
+		if timeOfNextPoll.Before(now) {
+			timeOfNextPoll = now
+		}
+
+		body, err := messages.EncodePollRequest(sid, "standalone")
+		if err != nil {
+			log.Printf("Error encoding poll message: %s", err.Error())
+			return nil
+		}
+		req, _ := http.NewRequest("POST", brokerPath.String(), bytes.NewBuffer(body))
+		resp, err := b.transport.RoundTrip(req)
+		if err != nil {
+			log.Printf("error polling broker: %s", err)
+		} else {
+			defer resp.Body.Close()
+			if resp.StatusCode != http.StatusOK {
+				log.Printf("broker returns: %d", resp.StatusCode)
+			} else {
+				body, err := limitedRead(resp.Body, readLimit)
+				if err != nil {
+					log.Printf("error reading broker response: %s", err)
+				} else {
+
+					offer, err := messages.DecodePollResponse(body)
+					if err != nil {
+						log.Printf("error reading broker response: %s", err.Error())
+						log.Printf("body: %s", body)
+						return nil
+					}
+					if offer != "" {
+						return util.DeserializeSessionDescription(offer)
+					}
+				}
+			}
+		}
+	}
+}
+
+func (b *Broker) sendAnswer(sid string, pc *webrtc.PeerConnection) error {
+	brokerPath := b.url.ResolveReference(&url.URL{Path: "answer"})
+	ld := pc.LocalDescription()
+	if !b.keepLocalAddresses {
+		ld = &webrtc.SessionDescription{
+			Type: ld.Type,
+			SDP:  util.StripLocalAddresses(ld.SDP),
+		}
+	}
+	answer := string([]byte(util.SerializeSessionDescription(ld)))
+	body, err := messages.EncodeAnswerRequest(answer, sid)
+	if err != nil {
+		return err
+	}
+	req, _ := http.NewRequest("POST", brokerPath.String(), bytes.NewBuffer(body))
+	resp, err := b.transport.RoundTrip(req)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("broker returned %d", resp.StatusCode)
+	}
+
+	body, err = limitedRead(resp.Body, readLimit)
+	if err != nil {
+		return fmt.Errorf("error reading broker response: %s", err)
+	}
+	success, err := messages.DecodeAnswerResponse(body)
+	if err != nil {
+		return err
+	}
+	if !success {
+		return fmt.Errorf("broker returned client timeout")
+	}
+
+	return nil
+}
+
+func CopyLoop(c1 io.ReadWriteCloser, c2 io.ReadWriteCloser) {
+	var wg sync.WaitGroup
+	copyer := func(dst io.ReadWriteCloser, src io.ReadWriteCloser) {
+		defer wg.Done()
+		if _, err := io.Copy(dst, src); err != nil {
+			log.Printf("io.Copy inside CopyLoop generated an error: %v", err)
+		}
+		dst.Close()
+		src.Close()
+	}
+	wg.Add(2)
+	go copyer(c1, c2)
+	go copyer(c2, c1)
+	wg.Wait()
+}
+
+// We pass conn.RemoteAddr() as an additional parameter, rather than calling
+// conn.RemoteAddr() inside this function, as a workaround for a hang that
+// otherwise occurs inside of conn.pc.RemoteDescription() (called by
+// RemoteAddr). https://bugs.torproject.org/18628#comment:8
+func datachannelHandler(conn *webRTCConn, remoteAddr net.Addr) {
+	defer conn.Close()
+	defer retToken()
+
+	u, err := url.Parse(relayURL)
+	if err != nil {
+		log.Fatalf("invalid relay url: %s", err)
+	}
+
+	// Retrieve client IP address
+	if remoteAddr != nil {
+		// Encode client IP address in relay URL
+		q := u.Query()
+		clientIP := remoteAddr.String()
+		q.Set("client_ip", clientIP)
+		u.RawQuery = q.Encode()
+	} else {
+		log.Printf("no remote address given in websocket")
+	}
+
+	ws, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	if err != nil {
+		log.Printf("error dialing relay: %s", err)
+		return
+	}
+	wsConn := websocketconn.New(ws)
+	log.Printf("connected to relay")
+	defer wsConn.Close()
+	CopyLoop(conn, wsConn)
+	log.Printf("datachannelHandler ends")
+}
+
+// Create a PeerConnection from an SDP offer. Blocks until the gathering of ICE
+// candidates is complete and the answer is available in LocalDescription.
+// Installs an OnDataChannel callback that creates a webRTCConn and passes it to
+// datachannelHandler.
+func makePeerConnectionFromOffer(sdp *webrtc.SessionDescription, config webrtc.Configuration, dataChan chan struct{}) (*webrtc.PeerConnection, error) {
+	pc, err := webrtc.NewPeerConnection(config)
+	if err != nil {
+		return nil, fmt.Errorf("accept: NewPeerConnection: %s", err)
+	}
+	pc.OnDataChannel(func(dc *webrtc.DataChannel) {
+		log.Println("OnDataChannel")
+		close(dataChan)
+
+		pr, pw := io.Pipe()
+		conn := &webRTCConn{pc: pc, dc: dc, pr: pr}
+
+		dc.OnOpen(func() {
+			log.Println("OnOpen channel")
+		})
+		dc.OnClose(func() {
+			conn.lock.Lock()
+			defer conn.lock.Unlock()
+			log.Println("OnClose channel")
+			conn.dc = nil
+			dc.Close()
+			pw.Close()
+		})
+		dc.OnMessage(func(msg webrtc.DataChannelMessage) {
+			var n int
+			n, err = pw.Write(msg.Data)
+			if err != nil {
+				if inerr := pw.CloseWithError(err); inerr != nil {
+					log.Printf("close with error generated an error: %v", inerr)
+				}
+			}
+			if n != len(msg.Data) {
+				panic("short write")
+			}
+		})
+
+		go datachannelHandler(conn, conn.RemoteAddr())
+	})
+
+	err = pc.SetRemoteDescription(*sdp)
+	if err != nil {
+		if inerr := pc.Close(); inerr != nil {
+			log.Printf("unable to call pc.Close after pc.SetRemoteDescription with error: %v", inerr)
+		}
+		return nil, fmt.Errorf("accept: SetRemoteDescription: %s", err)
+	}
+	log.Println("sdp offer successfully received.")
+
+	log.Println("Generating answer...")
+	answer, err := pc.CreateAnswer(nil)
+	// blocks on ICE gathering. we need to add a timeout if needed
+	// not putting this in a separate go routine, because we need
+	// SetLocalDescription(answer) to be called before sendAnswer
+	if err != nil {
+		if inerr := pc.Close(); inerr != nil {
+			log.Printf("ICE gathering has generated an error when calling pc.Close: %v", inerr)
+		}
+		return nil, err
+	}
+
+	err = pc.SetLocalDescription(answer)
+	if err != nil {
+		if err = pc.Close(); err != nil {
+			log.Printf("pc.Close after setting local description returned : %v", err)
+		}
+		return nil, err
+	}
+
+	return pc, nil
+}
+
+func runSession(sid string) {
+	offer := broker.pollOffer(sid)
+	if offer == nil {
+		log.Printf("bad offer from broker")
+		retToken()
+		return
+	}
+	dataChan := make(chan struct{})
+	pc, err := makePeerConnectionFromOffer(offer, config, dataChan)
+	if err != nil {
+		log.Printf("error making WebRTC connection: %s", err)
+		retToken()
+		return
+	}
+	err = broker.sendAnswer(sid, pc)
+	if err != nil {
+		log.Printf("error sending answer to client through broker: %s", err)
+		if inerr := pc.Close(); inerr != nil {
+			log.Printf("error calling pc.Close: %v", inerr)
+		}
+		retToken()
+		return
+	}
+	// Set a timeout on peerconnection. If the connection state has not
+	// advanced to PeerConnectionStateConnected in this time,
+	// destroy the peer connection and return the token.
+	select {
+	case <-dataChan:
+		log.Println("Connection successful.")
+	case <-time.After(dataChannelTimeout):
+		log.Println("Timed out waiting for client to open data channel.")
+		if err := pc.Close(); err != nil {
+			log.Printf("error calling pc.Close: %v", err)
+		}
+		retToken()
+	}
+}
+
+func main() {
+	var capacity uint
+	var stunURL string
+	var logFilename string
+	var rawBrokerURL string
+	var unsafeLogging bool
+	var keepLocalAddresses bool
+
+	flag.UintVar(&capacity, "capacity", 10, "maximum concurrent clients")
+	flag.StringVar(&rawBrokerURL, "broker", defaultBrokerURL, "broker URL")
+	flag.StringVar(&relayURL, "relay", defaultRelayURL, "websocket relay URL")
+	flag.StringVar(&stunURL, "stun", defaultSTUNURL, "stun URL")
+	flag.StringVar(&logFilename, "log", "", "log filename")
+	flag.BoolVar(&unsafeLogging, "unsafe-logging", false, "prevent logs from being scrubbed")
+	flag.BoolVar(&keepLocalAddresses, "keep-local-addresses", false, "keep local LAN address ICE candidates")
+	flag.Parse()
+
+	var logOutput io.Writer = os.Stderr
+	log.SetFlags(log.LstdFlags | log.LUTC)
+	if logFilename != "" {
+		f, err := os.OpenFile(logFilename, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer f.Close()
+		logOutput = io.MultiWriter(os.Stderr, f)
+	}
+	if unsafeLogging {
+		log.SetOutput(logOutput)
+	} else {
+		// We want to send the log output through our scrubber first
+		log.SetOutput(&safelog.LogScrubber{Output: logOutput})
+	}
+
+	log.Println("starting")
+
+	var err error
+	broker = new(Broker)
+	broker.keepLocalAddresses = keepLocalAddresses
+	broker.url, err = url.Parse(rawBrokerURL)
+	if err != nil {
+		log.Fatalf("invalid broker url: %s", err)
+	}
+	_, err = url.Parse(stunURL)
+	if err != nil {
+		log.Fatalf("invalid stun url: %s", err)
+	}
+	_, err = url.Parse(relayURL)
+	if err != nil {
+		log.Fatalf("invalid relay url: %s", err)
+	}
+
+	broker.transport = http.DefaultTransport.(*http.Transport)
+	config = webrtc.Configuration{
+		ICEServers: []webrtc.ICEServer{
+			{
+				URLs: []string{stunURL},
+			},
+		},
+	}
+	tokens = make(chan bool, capacity)
+	for i := uint(0); i < capacity; i++ {
+		tokens <- true
+	}
+
+	for {
+		getToken()
+		sessionID := genSessionID()
+		runSession(sessionID)
+	}
+}