ccm,ocm: fix connector-side bufio data loss in reverse proxy

connectorConnect() creates a bufio.NewReader to read the HTTP 101
upgrade response, but then passes the raw conn to yamux.Server().
If TCP coalesces the 101 response with initial yamux frames, the
bufio reader over-reads into its buffer and those bytes are lost
to yamux, causing session failure.

Wrap the bufio.Reader and raw conn into a bufferedConn so yamux
reads through the buffer first.

service/ccm/reverse.go

@@ -27,6 +27,15 @@ func reverseYamuxConfig() *yamux.Config {
 	return config
 }
 
+type bufferedConn struct {
+	reader *bufio.Reader
+	net.Conn
+}
+
+func (c *bufferedConn) Read(p []byte) (int, error) {
+	return c.reader.Read(p)
+}
+
 type yamuxNetListener struct {
 	session *yamux.Session
 }
@@ -219,7 +228,7 @@ func (c *externalCredential) connectorConnect(ctx context.Context) (time.Duratio
 		}
 	}
 
-	session, err := yamux.Server(conn, reverseYamuxConfig())
+	session, err := yamux.Server(&bufferedConn{reader: reader, Conn: conn}, reverseYamuxConfig())
 	if err != nil {
 		conn.Close()
 		return 0, E.Cause(err, "create yamux server")

service/ocm/reverse.go

@@ -27,6 +27,15 @@ func reverseYamuxConfig() *yamux.Config {
 	return config
 }
 
+type bufferedConn struct {
+	reader *bufio.Reader
+	net.Conn
+}
+
+func (c *bufferedConn) Read(p []byte) (int, error) {
+	return c.reader.Read(p)
+}
+
 type yamuxNetListener struct {
 	session *yamux.Session
 }
@@ -219,7 +228,7 @@ func (c *externalCredential) connectorConnect(ctx context.Context) (time.Duratio
 		}
 	}
 
-	session, err := yamux.Server(conn, reverseYamuxConfig())
+	session, err := yamux.Server(&bufferedConn{reader: reader, Conn: conn}, reverseYamuxConfig())
 	if err != nil {
 		conn.Close()
 		return 0, E.Cause(err, "create yamux server")