Encode client-broker messages as json in HTTP body

Send the client poll request and response in a json-encoded format in
the HTTP request body rather than sending the data in HTTP headers. This
will pave the way for using domain-fronting alternatives for the
Snowflake rendezvous.

broker/broker.go

@@ -6,6 +6,7 @@ SessionDescriptions in order to negotiate a WebRTC connection.
 package main
 
 import (
+	"bytes"
 	"container/heap"
 	"crypto/tls"
 	"flag"
@@ -39,6 +40,16 @@ const (
 	NATUnrestricted = "unrestricted"
 )
 
+// We support two client message formats. The legacy format is for backwards
+// combatability and relies heavily on HTTP headers and status codes to convey
+// information.
+type clientVersion int
+
+const (
+	v0 clientVersion = iota //legacy version
+	v1
+)
+
 type BrokerContext struct {
 	snowflakes           *SnowflakeHeap
 	restrictedSnowflakes *SnowflakeHeap
@@ -90,7 +101,7 @@ type MetricsHandler struct {
 
 func (sh SnowflakeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Access-Control-Allow-Origin", "*")
-	w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Session-ID, Snowflake-NAT-Type")
+	w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Session-ID")
 	// Return early if it's CORS preflight.
 	if "OPTIONS" == r.Method {
 		return
@@ -170,7 +181,7 @@ func (ctx *BrokerContext) AddSnowflake(id string, proxyType string, natType stri
 	snowflake.proxyType = proxyType
 	snowflake.natType = natType
 	snowflake.offerChannel = make(chan *ClientOffer)
-	snowflake.answerChannel = make(chan []byte)
+	snowflake.answerChannel = make(chan string)
 	ctx.snowflakeLock.Lock()
 	if natType == NATUnrestricted {
 		heap.Push(ctx.snowflakes, snowflake)
@@ -245,6 +256,20 @@ type ClientOffer struct {
 	sdp     []byte
 }
 
+// Sends an encoded response to the client and an
+// HTTP server error if the response encoding fails
+func sendClientResponse(resp *messages.ClientPollResponse, w http.ResponseWriter) {
+	data, err := resp.EncodePollResponse()
+	if err != nil {
+		log.Printf("error encoding answer")
+		w.WriteHeader(http.StatusInternalServerError)
+	} else {
+		if _, err := w.Write([]byte(data)); err != nil {
+			log.Printf("unable to write answer with error: %v", err)
+		}
+	}
+}
+
 /*
 Expects a WebRTC SDP offer in the Request to give to an assigned
 snowflake proxy, which responds with the SDP answer to be sent in
@@ -252,19 +277,55 @@ the HTTP response back to the client.
 */
 func clientOffers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) {
 	var err error
+	var version clientVersion
 
 	startTime := time.Now()
-	offer := &ClientOffer{}
-	offer.sdp, err = ioutil.ReadAll(http.MaxBytesReader(w, r.Body, readLimit))
-	if nil != err {
-		log.Println("Invalid data.")
-		w.WriteHeader(http.StatusBadRequest)
+	body, err := ioutil.ReadAll(http.MaxBytesReader(w, r.Body, readLimit))
+	if err != nil {
+		log.Printf("Error reading client request: %s", err.Error())
+		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
+	if len(body) > 0 && body[0] == '{' {
+		version = v0
+	} else {
+		parts := bytes.SplitN(body, []byte("\n"), 2)
+		if len(parts) < 2 {
+			// no version number found
+			err := fmt.Errorf("unsupported message version")
+			sendClientResponse(&messages.ClientPollResponse{Error: err.Error()}, w)
+			return
+		}
+		body = parts[1]
+		if string(parts[0]) == "1.0" {
+			version = v1
 
-	offer.natType = r.Header.Get("Snowflake-NAT-Type")
-	if offer.natType == "" {
-		offer.natType = NATUnknown
+		} else {
+			err := fmt.Errorf("unsupported message version")
+			sendClientResponse(&messages.ClientPollResponse{Error: err.Error()}, w)
+			return
+		}
+	}
+
+	var offer *ClientOffer
+	switch version {
+	case v0:
+		offer = &ClientOffer{
+			natType: r.Header.Get("Snowflake-NAT-Type"),
+			sdp:     body,
+		}
+	case v1:
+		req, err := messages.DecodeClientPollRequest(body)
+		if err != nil {
+			sendClientResponse(&messages.ClientPollResponse{Error: err.Error()}, w)
+			return
+		}
+		offer = &ClientOffer{
+			natType: req.NAT,
+			sdp:     []byte(req.Offer),
+		}
+	default:
+		panic("unknown version")
 	}
 
 	// Only hand out known restricted snowflakes to unrestricted clients
@@ -289,7 +350,15 @@ func clientOffers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) {
 			ctx.metrics.clientRestrictedDeniedCount++
 		}
 		ctx.metrics.lock.Unlock()
-		w.WriteHeader(http.StatusServiceUnavailable)
+		switch version {
+		case v0:
+			w.WriteHeader(http.StatusServiceUnavailable)
+		case v1:
+			resp := &messages.ClientPollResponse{Error: "no snowflake proxies currently available"}
+			sendClientResponse(resp, w)
+		default:
+			panic("unknown version")
+		}
 		return
 	}
 	// Otherwise, find the most available snowflake proxy, and pass the offer to it.
@@ -306,17 +375,36 @@ func clientOffers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) {
 		ctx.metrics.clientProxyMatchCount++
 		ctx.metrics.promMetrics.ClientPollTotal.With(prometheus.Labels{"nat": offer.natType, "status": "matched"}).Inc()
 		ctx.metrics.lock.Unlock()
-		if _, err := w.Write(answer); err != nil {
-			log.Printf("unable to write answer with error: %v", err)
+		switch version {
+		case v0:
+			if _, err := w.Write([]byte(answer)); err != nil {
+				log.Printf("unable to write answer with error: %v", err)
+			}
+		case v1:
+			resp := &messages.ClientPollResponse{Answer: answer}
+			sendClientResponse(resp, w)
+		default:
+			panic("unknown version")
 		}
 		// Initial tracking of elapsed time.
 		ctx.metrics.clientRoundtripEstimate = time.Since(startTime) /
 			time.Millisecond
 	case <-time.After(time.Second * ClientTimeout):
 		log.Println("Client: Timed out.")
-		w.WriteHeader(http.StatusGatewayTimeout)
-		if _, err := w.Write([]byte("timed out waiting for answer!")); err != nil {
-			log.Printf("unable to write timeout error, failed with error: %v", err)
+		switch version {
+		case v0:
+			w.WriteHeader(http.StatusGatewayTimeout)
+			if _, err := w.Write(
+				[]byte("timed out waiting for answer!")); err != nil {
+				log.Printf("unable to write timeout error, failed with error: %v",
+					err)
+			}
+		case v1:
+			resp := &messages.ClientPollResponse{
+				Error: "timed out waiting for answer!"}
+			sendClientResponse(resp, w)
+		default:
+			panic("unknown version")
 		}
 	}
 
@@ -364,7 +452,7 @@ func proxyAnswers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) {
 	w.Write(b)
 
 	if success {
-		snowflake.answerChannel <- []byte(answer)
+		snowflake.answerChannel <- answer
 	}
 
 }

broker/snowflake-broker_test.go

@@ -70,10 +70,59 @@ func TestBroker(t *testing.T) {
 
 		Convey("Responds to client offers...", func() {
 			w := httptest.NewRecorder()
-			data := bytes.NewReader([]byte("test"))
+			data := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unknown\"}"))
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
 			So(err, ShouldBeNil)
 
+			Convey("with error when no snowflakes are available.", func() {
+				clientOffers(ctx, w, r)
+				So(w.Code, ShouldEqual, http.StatusOK)
+				So(w.Body.String(), ShouldEqual, `{"error":"no snowflake proxies currently available"}`)
+			})
+
+			Convey("with a proxy answer if available.", func() {
+				done := make(chan bool)
+				// Prepare a fake proxy to respond with.
+				snowflake := ctx.AddSnowflake("fake", "", NATUnrestricted)
+				go func() {
+					clientOffers(ctx, w, r)
+					done <- true
+				}()
+				offer := <-snowflake.offerChannel
+				So(offer.sdp, ShouldResemble, []byte("fake"))
+				snowflake.answerChannel <- "fake answer"
+				<-done
+				So(w.Body.String(), ShouldEqual, `{"answer":"fake answer"}`)
+				So(w.Code, ShouldEqual, http.StatusOK)
+			})
+
+			Convey("Times out when no proxy responds.", func() {
+				if testing.Short() {
+					return
+				}
+				done := make(chan bool)
+				snowflake := ctx.AddSnowflake("fake", "", NATUnrestricted)
+				go func() {
+					clientOffers(ctx, w, r)
+					// Takes a few seconds here...
+					done <- true
+				}()
+				offer := <-snowflake.offerChannel
+				So(offer.sdp, ShouldResemble, []byte("fake"))
+				<-done
+				So(w.Code, ShouldEqual, http.StatusOK)
+				So(w.Body.String(), ShouldEqual, `{"error":"timed out waiting for answer!"}`)
+			})
+		})
+
+		Convey("Responds to legacy client offers...", func() {
+			w := httptest.NewRecorder()
+			data := bytes.NewReader([]byte("{test}"))
+			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
+			r.Header.Set("Snowflake-NAT-TYPE", "restricted")
+
 			Convey("with 503 when no snowflakes are available.", func() {
 				clientOffers(ctx, w, r)
 				So(w.Code, ShouldEqual, http.StatusServiceUnavailable)
@@ -89,8 +138,8 @@ func TestBroker(t *testing.T) {
 					done <- true
 				}()
 				offer := <-snowflake.offerChannel
-				So(offer.sdp, ShouldResemble, []byte("test"))
-				snowflake.answerChannel <- []byte("fake answer")
+				So(offer.sdp, ShouldResemble, []byte("{test}"))
+				snowflake.answerChannel <- "fake answer"
 				<-done
 				So(w.Body.String(), ShouldEqual, "fake answer")
 				So(w.Code, ShouldEqual, http.StatusOK)
@@ -108,10 +157,11 @@ func TestBroker(t *testing.T) {
 					done <- true
 				}()
 				offer := <-snowflake.offerChannel
-				So(offer.sdp, ShouldResemble, []byte("test"))
+				So(offer.sdp, ShouldResemble, []byte("{test}"))
 				<-done
 				So(w.Code, ShouldEqual, http.StatusGatewayTimeout)
 			})
+
 		})
 
 		Convey("Responds to proxy polls...", func() {
@@ -163,7 +213,7 @@ func TestBroker(t *testing.T) {
 				}(ctx)
 				answer := <-s.answerChannel
 				So(w.Code, ShouldEqual, http.StatusOK)
-				So(answer, ShouldResemble, []byte("test"))
+				So(answer, ShouldResemble, "test")
 			})
 
 			Convey("with client gone status if the proxy is not recognized", func() {
@@ -272,7 +322,8 @@ func TestBroker(t *testing.T) {
 			So(ctx.idToSnowflake["ymbcCMto7KHNGYlp"], ShouldNotBeNil)
 
 			// Client request blocks until proxy answer arrives.
-			dataC := bytes.NewReader([]byte("fake offer"))
+			dataC := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unknown\"}"))
 			wC := httptest.NewRecorder()
 			rC, err := http.NewRequest("POST", "snowflake.broker/client", dataC)
 			So(err, ShouldBeNil)
@@ -283,7 +334,7 @@ func TestBroker(t *testing.T) {
 
 			<-polled
 			So(wP.Code, ShouldEqual, http.StatusOK)
-			So(wP.Body.String(), ShouldResemble, `{"Status":"client match","Offer":"fake offer","NAT":"unknown"}`)
+			So(wP.Body.String(), ShouldResemble, `{"Status":"client match","Offer":"fake","NAT":"unknown"}`)
 			So(ctx.idToSnowflake["ymbcCMto7KHNGYlp"], ShouldNotBeNil)
 			// Follow up with the answer request afterwards
 			wA := httptest.NewRecorder()
@@ -295,7 +346,7 @@ func TestBroker(t *testing.T) {
 
 			<-done
 			So(wC.Code, ShouldEqual, http.StatusOK)
-			So(wC.Body.String(), ShouldEqual, "test")
+			So(wC.Body.String(), ShouldEqual, `{"answer":"test"}`)
 		})
 	})
 }
@@ -517,7 +568,8 @@ func TestMetrics(t *testing.T) {
 		//Test addition of client failures
 		Convey("for no proxies available", func() {
 			w := httptest.NewRecorder()
-			data := bytes.NewReader([]byte("test"))
+			data := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unknown\"}"))
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
 			So(err, ShouldBeNil)
 
@@ -535,7 +587,8 @@ func TestMetrics(t *testing.T) {
 		//Test addition of client matches
 		Convey("for client-proxy match", func() {
 			w := httptest.NewRecorder()
-			data := bytes.NewReader([]byte("test"))
+			data := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unknown\"}"))
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
 			So(err, ShouldBeNil)
 
@@ -546,8 +599,8 @@ func TestMetrics(t *testing.T) {
 				done <- true
 			}()
 			offer := <-snowflake.offerChannel
-			So(offer.sdp, ShouldResemble, []byte("test"))
-			snowflake.answerChannel <- []byte("fake answer")
+			So(offer.sdp, ShouldResemble, []byte("fake"))
+			snowflake.answerChannel <- "fake answer"
 			<-done
 
 			ctx.metrics.printMetrics()
@@ -556,22 +609,63 @@ func TestMetrics(t *testing.T) {
 		//Test rounding boundary
 		Convey("binning boundary", func() {
 			w := httptest.NewRecorder()
-			data := bytes.NewReader([]byte("test"))
+			data := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
 			So(err, ShouldBeNil)
 
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
 
 			ctx.metrics.printMetrics()
 			So(buf.String(), ShouldContainSubstring, "client-denied-count 8\nclient-restricted-denied-count 8\nclient-unrestricted-denied-count 0\n")
 
+			w = httptest.NewRecorder()
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
+			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
+			So(err, ShouldBeNil)
 			clientOffers(ctx, w, r)
 			buf.Reset()
 			ctx.metrics.printMetrics()
@@ -648,9 +742,9 @@ func TestMetrics(t *testing.T) {
 		//Test client failures by NAT type
 		Convey("client failures by NAT type", func() {
 			w := httptest.NewRecorder()
-			data := bytes.NewReader([]byte("test"))
+			data := bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"restricted\"}"))
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
-			r.Header.Set("Snowflake-NAT-TYPE", "restricted")
 			So(err, ShouldBeNil)
 
 			clientOffers(ctx, w, r)
@@ -661,8 +755,9 @@ func TestMetrics(t *testing.T) {
 			buf.Reset()
 			ctx.metrics.zeroMetrics()
 
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unrestricted\"}"))
 			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
-			r.Header.Set("Snowflake-NAT-TYPE", "unrestricted")
 			So(err, ShouldBeNil)
 
 			clientOffers(ctx, w, r)
@@ -673,8 +768,9 @@ func TestMetrics(t *testing.T) {
 			buf.Reset()
 			ctx.metrics.zeroMetrics()
 
+			data = bytes.NewReader(
+				[]byte("1.0\n{\"offer\": \"fake\", \"nat\": \"unknown\"}"))
 			r, err = http.NewRequest("POST", "snowflake.broker/client", data)
-			r.Header.Set("Snowflake-NAT-TYPE", "unknown")
 			So(err, ShouldBeNil)
 
 			clientOffers(ctx, w, r)

broker/snowflake-heap.go

@@ -13,7 +13,7 @@ type Snowflake struct {
 	proxyType     string
 	natType       string
 	offerChannel  chan *ClientOffer
-	answerChannel chan []byte
+	answerChannel chan string
 	clients       int
 	index         int
 }