Release 2022.3.1

Setting the body to nil was rendering cloudflared to crashing with
a SIGSEGV in the odd case where the hostname accessed maps to a
TCP origin (e.g. SSH/RDP/...) but the eyeball sends a plain HTTP
request that does not go through cloudflared access (thus not wrapped
in websocket as it should).

Instead, QUIC transport now sets http.noBody in that condition, which
deals with the situation gracefully.

RELEASE_NOTES

@@ -1,3 +1,9 @@
+2022.3.1
+- 2022-03-04 TUN-5837: Log panic recovery in http2 logic with debug level log
+- 2022-03-04  TUN-5696: HTTP/2 Configuration Update
+- 2022-03-04 TUN-5836: Avoid websocket#Stream function from crashing cloudflared with unexpected memory access
+- 2022-03-05 TUN-5836: QUIC transport no longer sets body to nil in any condition
+
 2022.3.0
 - 2022-03-02 TUN-5680: Adapt component tests for new service install based on token
 - 2022-02-21 TUN-5682: Remove name field from credentials

connection/connection.go

@@ -80,6 +80,7 @@ const (
 	TypeTCP
 	TypeControlStream
 	TypeHTTP
+	TypeConfiguration
 )
 
 // ShouldFlush returns whether this kind of connection should actively flush data

connection/http2.go

@@ -2,6 +2,7 @@ package connection
 
 import (
 	"context"
+	gojson "encoding/json"
 	"fmt"
 	"io"
 	"net"
@@ -23,6 +24,7 @@ const (
 	InternalTCPProxySrcHeader = "Cf-Cloudflared-Proxy-Src"
 	WebsocketUpgrade          = "websocket"
 	ControlStreamUpgrade      = "control-stream"
+	ConfigurationUpdate       = "update-configuration"
 )
 
 var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
@@ -100,7 +102,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	connType := determineHTTP2Type(r)
 	handleMissingRequestParts(connType, r)
 
-	respWriter, err := NewHTTP2RespWriter(r, w, connType)
+	respWriter, err := NewHTTP2RespWriter(r, w, connType, c.log)
 	if err != nil {
 		c.observer.log.Error().Msg(err.Error())
 		return
@@ -120,6 +122,13 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			respWriter.WriteErrorResponse()
 		}
 
+	case TypeConfiguration:
+		fmt.Println("TYPE CONFIGURATION?")
+		if err := c.handleConfigurationUpdate(respWriter, r); err != nil {
+			c.log.Error().Err(err)
+			respWriter.WriteErrorResponse()
+		}
+
 	case TypeWebsocket, TypeHTTP:
 		stripWebsocketUpgradeHeader(r)
 		if err := originProxy.ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
@@ -152,6 +161,26 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// ConfigurationUpdateBody is the representation followed by the edge to send updates to cloudflared.
+type ConfigurationUpdateBody struct {
+	Version int32             `json:"version"`
+	Config  gojson.RawMessage `json:"config"`
+}
+
+func (c *HTTP2Connection) handleConfigurationUpdate(respWriter *http2RespWriter, r *http.Request) error {
+	var configBody ConfigurationUpdateBody
+	if err := json.NewDecoder(r.Body).Decode(&configBody); err != nil {
+		return err
+	}
+	resp := c.orchestrator.UpdateConfig(configBody.Version, configBody.Config)
+	bdy, err := json.Marshal(resp)
+	if err != nil {
+		return err
+	}
+	_, err = respWriter.Write(bdy)
+	return err
+}
+
 func (c *HTTP2Connection) close() {
 	// Wait for all serve HTTP handlers to return
 	c.activeRequestsWG.Wait()
@@ -163,14 +192,16 @@ type http2RespWriter struct {
 	w           http.ResponseWriter
 	flusher     http.Flusher
 	shouldFlush bool
+	log         *zerolog.Logger
 }
 
-func NewHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (*http2RespWriter, error) {
+func NewHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type, log *zerolog.Logger) (*http2RespWriter, error) {
 	flusher, isFlusher := w.(http.Flusher)
 	if !isFlusher {
 		respWriter := &http2RespWriter{
-			r: r.Body,
-			w: w,
+			r:   r.Body,
+			w:   w,
+			log: log,
 		}
 		respWriter.WriteErrorResponse()
 		return nil, fmt.Errorf("%T doesn't implement http.Flusher", w)
@@ -181,6 +212,7 @@ func NewHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (
 		w:           w,
 		flusher:     flusher,
 		shouldFlush: connType.shouldFlush(),
+		log:         log,
 	}, nil
 }
 
@@ -239,7 +271,7 @@ func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
 		// Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
 		// Register a recover routine just in case.
 		if r := recover(); r != nil {
-			println(fmt.Sprintf("Recover from http2 response writer panic, error %s", debug.Stack()))
+			rp.log.Debug().Msgf("Recover from http2 response writer panic, error %s", debug.Stack())
 		}
 	}()
 	n, err = rp.w.Write(p)
@@ -255,6 +287,8 @@ func (rp *http2RespWriter) Close() error {
 
 func determineHTTP2Type(r *http.Request) Type {
 	switch {
+	case isConfigurationUpdate(r):
+		return TypeConfiguration
 	case isWebsocketUpgrade(r):
 		return TypeWebsocket
 	case IsTCPStream(r):
@@ -288,6 +322,10 @@ func isWebsocketUpgrade(r *http.Request) bool {
 	return r.Header.Get(InternalUpgradeHeader) == WebsocketUpgrade
 }
 
+func isConfigurationUpdate(r *http.Request) bool {
+	return r.Header.Get(InternalUpgradeHeader) == ConfigurationUpdate
+}
+
 // IsTCPStream discerns if the connection request needs a tcp stream proxy.
 func IsTCPStream(r *http.Request) bool {
 	return r.Header.Get(InternalTCPProxySrcHeader) != ""

connection/http2_test.go

@@ -1,6 +1,7 @@
 package connection
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -53,6 +54,41 @@ func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
 	), edgeConn
 }
 
+func TestHTTP2ConfigurationSet(t *testing.T) {
+	http2Conn, edgeConn := newTestHTTP2Connection()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		http2Conn.Serve(ctx)
+	}()
+
+	edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
+	require.NoError(t, err)
+
+	endpoint := fmt.Sprintf("http://localhost:8080/ok")
+	reqBody := []byte(`{
+"version": 2, 
+"config": {"warp-routing": {"enabled": true},  "originRequest" : {"connectTimeout": 10}, "ingress" : [ {"hostname": "test", "service": "https://localhost:8000" } , {"service": "http_status:404"} ]}}
+`)
+	reader := bytes.NewReader(reqBody)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, endpoint, reader)
+	require.NoError(t, err)
+	req.Header.Set(InternalUpgradeHeader, ConfigurationUpdate)
+
+	resp, err := edgeHTTP2Conn.RoundTrip(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	bdy, err := ioutil.ReadAll(resp.Body)
+	require.NoError(t, err)
+	assert.Equal(t, `{"lastAppliedVersion":2,"err":null}`, string(bdy))
+	cancel()
+	wg.Wait()
+
+}
+
 func TestServeHTTP(t *testing.T) {
 	tests := []testRequest{
 		{

connection/quic.go

@@ -342,7 +342,7 @@ func buildHTTPRequest(connectRequest *quicpogs.ConnectRequest, body io.ReadClose
 	//   * there is no transfer-encoding=chunked already set.
 	// So, if transfer cannot be chunked and content length is 0, we dont set a request body.
 	if !isWebsocket && !isTransferEncodingChunked(req) && req.ContentLength == 0 {
-		req.Body = nil
+		req.Body = http.NoBody
 	}
 	stripWebsocketUpgradeHeader(req)
 	return req, err

connection/quic_test.go

@@ -345,7 +345,7 @@ func TestBuildHTTPRequest(t *testing.T) {
 				},
 				ContentLength: 0,
 				Host:          "cf.host",
-				Body:          nil,
+				Body:          http.NoBody,
 			},
 			body: io.NopCloser(&bytes.Buffer{}),
 		},

orchestration/orchestrator_test.go

@@ -332,7 +332,8 @@ func proxyHTTP(t *testing.T, originProxy connection.OriginProxy, hostname string
 	require.NoError(t, err)
 
 	w := httptest.NewRecorder()
-	respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeHTTP)
+	log := zerolog.Nop()
+	respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeHTTP, &log)
 	require.NoError(t, err)
 
 	err = originProxy.ProxyHTTP(respWriter, req, false)
@@ -358,7 +359,8 @@ func proxyTCP(t *testing.T, originProxy connection.OriginProxy, originAddr strin
 	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", originAddr), reqBody)
 	require.NoError(t, err)
 
-	respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeTCP)
+	log := zerolog.Nop()
+	respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeTCP, &log)
 	require.NoError(t, err)
 
 	tcpReq := &connection.TCPRequest{
@@ -578,7 +580,8 @@ func TestPersistentConnection(t *testing.T) {
 		// ProxyHTTP will add Connection, Upgrade and Sec-Websocket-Version headers
 		req.Header.Add("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
 
-		respWriter, err := connection.NewHTTP2RespWriter(req, wsRespReadWriter, connection.TypeWebsocket)
+		log := zerolog.Nop()
+		respWriter, err := connection.NewHTTP2RespWriter(req, wsRespReadWriter, connection.TypeWebsocket, &log)
 		require.NoError(t, err)
 
 		err = originProxy.ProxyHTTP(respWriter, req, true)

proxy/proxy_test.go

@@ -60,6 +60,11 @@ func (w *mockHTTPRespWriter) Read(data []byte) (int, error) {
 	return 0, fmt.Errorf("mockHTTPRespWriter doesn't implement io.Reader")
 }
 
+// respHeaders is a test function to read respHeaders
+func (w *mockHTTPRespWriter) headers() http.Header {
+	return w.Header()
+}
+
 type mockWSRespWriter struct {
 	*mockHTTPRespWriter
 	writeNotification chan []byte
@@ -554,6 +559,24 @@ func TestConnections(t *testing.T) {
 				},
 			},
 		},
+		{
+			// Send (unexpected) HTTP when origin expects WS (to unwrap for raw TCP)
+			name: "http-(ws)tcp proxy",
+			args: args{
+				ingressServiceScheme:  "tcp://",
+				originService:         runEchoTCPService,
+				eyeballResponseWriter: newMockHTTPRespWriter(),
+				eyeballRequestBody:    http.NoBody,
+				connectionType:        connection.TypeHTTP,
+				requestHeaders: map[string][]string{
+					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
+				},
+			},
+			want: want{
+				message: []byte{},
+				headers: map[string][]string{},
+			},
+		},
 		{
 			name: "tcp-tcp proxy without warpRoutingService enabled",
 			args: args{
@@ -650,8 +673,8 @@ func TestConnections(t *testing.T) {
 				}()
 			}
 			if test.args.connectionType == connection.TypeTCP {
-				rws := connection.NewHTTPResponseReadWriterAcker(respWriter, req)
-				err = proxy.ProxyTCP(ctx, rws, &connection.TCPRequest{Dest: dest})
+				rwa := connection.NewHTTPResponseReadWriterAcker(respWriter, req)
+				err = proxy.ProxyTCP(ctx, rwa, &connection.TCPRequest{Dest: dest})
 			} else {
 				err = proxy.ProxyHTTP(respWriter, req, test.args.connectionType == connection.TypeWebsocket)
 			}

websocket/websocket.go

@@ -8,9 +8,11 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"runtime/debug"
 	"sync/atomic"
 	"time"
 
+	"github.com/getsentry/raven-go"
 	"github.com/gorilla/websocket"
 	"github.com/rs/zerolog"
 )
@@ -71,13 +73,25 @@ func unidirectionalStream(dst io.Writer, src io.Reader, dir string, status *bidi
 		// close. In such case, if the other direction did not stop (due to application level stopping, e.g., if a
 		// server/origin listens forever until closure), it may read/write from the underlying ReadWriter (backed by
 		// the Edge<->cloudflared transport) in an unexpected state.
-
-		if status.isAnyDone() {
-			// Because of this, we set this recover() logic, which kicks-in *only* if any stream is known to have
-			// exited. In such case, we stop a possible panic from propagating upstream.
-			if r := recover(); r != nil {
+		// Because of this, we set this recover() logic.
+		if r := recover(); r != nil {
+			if status.isAnyDone() {
 				// We handle such unexpected errors only when we detect that one side of the streaming is done.
-				log.Debug().Msgf("Handled gracefully error %v in Streaming for %s", r, dir)
+				log.Debug().Msgf("Gracefully handled error %v in Streaming for %s, error %s", r, dir, debug.Stack())
+			} else {
+				// Otherwise, this is unexpected, but we prevent the program from crashing anyway.
+				log.Warn().Msgf("Gracefully handled unexpected error %v in Streaming for %s, error %s", r, dir, debug.Stack())
+
+				tags := make(map[string]string)
+				tags["root"] = "websocket.stream"
+				tags["dir"] = dir
+				switch rval := r.(type) {
+				case error:
+					raven.CaptureError(rval, tags)
+				default:
+					rvalStr := fmt.Sprint(rval)
+					raven.CaptureMessage(rvalStr, tags)
+				}
 			}
 		}
 	}()