diff --git a/carrier/websocket.go b/carrier/websocket.go index ae752d19..36cd08e7 100644 --- a/carrier/websocket.go +++ b/carrier/websocket.go @@ -37,7 +37,7 @@ func (ws *Websocket) ServeStream(options *StartOptions, conn io.ReadWriter) erro } defer func() { _ = wsConn.Close() }() - stream.Pipe(wsConn, conn, stream.DefaultTimeoutAfterFirstClose, ws.log) + stream.Pipe(wsConn, conn, ws.log) return nil } diff --git a/carrier/websocket_test.go b/carrier/websocket_test.go index 94d93ec5..c784832f 100644 --- a/carrier/websocket_test.go +++ b/carrier/websocket_test.go @@ -5,12 +5,8 @@ import ( crand "crypto/rand" "crypto/tls" "crypto/x509" - "errors" "fmt" - "io" "math/big" - "net/http" - "net/http/httptest" "testing" "time" @@ -21,7 +17,6 @@ import ( "golang.org/x/net/websocket" "github.com/cloudflare/cloudflared/hello" - "github.com/cloudflare/cloudflared/stream" "github.com/cloudflare/cloudflared/tlsconfig" cfwebsocket "github.com/cloudflare/cloudflared/websocket" ) @@ -132,137 +127,3 @@ func TestWebsocketWrapper(t *testing.T) { require.Equal(t, 2, n) require.Equal(t, "bc", string(buf[:n])) } - -// halfClosePipeEnd is a bidirectional pipe built from two independent io.Pipe -// pairs. Unlike net.Pipe, closing one direction does not affect the other, which -// lets us simulate a half-close: the local client can signal EOF on its write -// side while still reading a delayed response from the remote side. -type halfClosePipeEnd struct { - r *io.PipeReader // data flowing into this end - w *io.PipeWriter // data flowing out of this end -} - -func newHalfClosePipe() (local, remote *halfClosePipeEnd) { - // Pipe A carries data from local → remote. - ar, aw := io.Pipe() - // Pipe B carries data from remote → local. - br, bw := io.Pipe() - - local = &halfClosePipeEnd{r: br, w: aw} - remote = &halfClosePipeEnd{r: ar, w: bw} - return local, remote -} - -func (p *halfClosePipeEnd) Read(b []byte) (int, error) { return p.r.Read(b) } -func (p *halfClosePipeEnd) Write(b []byte) (int, error) { return p.w.Write(b) } - -// CloseWrite signals EOF to the remote reader without closing the read side. -func (p *halfClosePipeEnd) CloseWrite() error { return p.w.Close() } - -// Close shuts down both directions. -func (p *halfClosePipeEnd) Close() error { - _ = p.r.Close() - return p.w.Close() -} - -// TestServeStreamWaitsForResponseAfterLocalClose exercises the websocket path -// It verifies that the pipe does not tear down immediately after the client closes -// its write side. -// -// The setup mirrors the cloudflared access tcp path: -// -// local app -> halfClosePipe -> ServeStream -> mock WS echo server -// -// Sequence: -// 1. Write payload to the mock server through ServeStream. -// 2. Half-close the write side (CloseWrite) — signals EOF upstream. -// 3. Sleep for halfCloseWait to make the race window explicit: a buggy -// (timeout=0) pipe would already have torn down the connection here. -// 4. Read the echo — must succeed because the patch keeps the pipe alive. -func TestServeStreamWaitsForResponseAfterLocalClose(t *testing.T) { - t.Parallel() - - const ( - payload = "half-close-test" - // halfCloseWait makes the race window visible: if ServeStream tears - // down the connection on CloseWrite the read that follows will fail - // immediately, mirroring the 3-second sleep in the cftunnel reference - // test. It must be shorter than DefaultTimeoutAfterFirstClose (10s). - halfCloseWait = 3 * time.Second - // testTimeout is an upper bound for the whole test. - testTimeout = stream.DefaultTimeoutAfterFirstClose + 5*time.Second - ) - - server := websocketServer() - defer server.Close() - - localEnd, remoteEnd := newHalfClosePipe() - - log := zerolog.Nop() - wsConn := NewWSConnection(&log) - options := &StartOptions{ - OriginURL: "ws://" + server.Listener.Addr().String(), - } - - serveErrCh := make(chan error, 1) - go func() { - serveErrCh <- wsConn.ServeStream(options, remoteEnd) - }() - - ctx, cancel := context.WithTimeout(t.Context(), testTimeout) - defer cancel() - - // 1. Write the payload. ServeStream forwards it as a WS binary frame. - _, err := localEnd.Write([]byte(payload)) - require.NoError(t, err) - - // 2. Half-close the write side - require.NoError(t, localEnd.CloseWrite()) - - // 3. Wait to make the race window explicit. - time.Sleep(halfCloseWait) - - // 4. Read the echo. - got := make([]byte, len(payload)) - _, err = io.ReadFull(localEnd, got) - require.NoError(t, err, "read after half-close failed: pipe was torn down too early") - require.Equal(t, payload, string(got)) - - // Drain ServeStream. - _ = localEnd.Close() - _ = remoteEnd.Close() - - select { - case err := <-serveErrCh: - if err != nil && err != io.EOF && !errors.Is(err, io.ErrClosedPipe) { - require.NoError(t, err) - } - case <-ctx.Done(): - t.Fatal("ServeStream did not return in time") - } -} - -func websocketServer() *httptest.Server { - upgrader := gws.Upgrader{ - ReadBufferSize: 1024, - WriteBufferSize: 1024, - CheckOrigin: func(*http.Request) bool { return true }, - } - - return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - conn, err := upgrader.Upgrade(w, r, nil) - if err != nil { - return - } - defer func() { _ = conn.Close() }() - - _, msg, err := conn.ReadMessage() - if err != nil { - return - } - - if err := conn.WriteMessage(gws.BinaryMessage, msg); err != nil { - return - } - })) -} diff --git a/cmd/cloudflared/updater/update.go b/cmd/cloudflared/updater/update.go index b1c69862..d98da821 100644 --- a/cmd/cloudflared/updater/update.go +++ b/cmd/cloudflared/updater/update.go @@ -299,5 +299,5 @@ func wasInstalledFromPackageManager() bool { } func isRunningFromTerminal() bool { - return term.IsTerminal(int(os.Stdout.Fd())) //nolint:gosec + return term.IsTerminal(int(os.Stdout.Fd())) // nolint:gosec } diff --git a/connection/connection_test.go b/connection/connection_test.go index 2aa30ae6..a03f53b2 100644 --- a/connection/connection_test.go +++ b/connection/connection_test.go @@ -151,7 +151,7 @@ func wsEchoEndpoint(w ResponseWriter, r *http.Request) error { }() originConn := &echoPipe{reader: readPipe, writer: writePipe} - stream.Pipe(wsConn, originConn, 0, &log) + stream.Pipe(wsConn, originConn, &log) cancel() wsConn.Close() return nil @@ -190,7 +190,7 @@ func wsFlakyEndpoint(w ResponseWriter, r *http.Request) error { rInt, _ := rand.Int(rand.Reader, big.NewInt(50)) closedAfter := time.Millisecond * time.Duration(rInt.Int64()) originConn := &flakyConn{closeAt: time.Now().Add(closedAfter)} - stream.Pipe(wsConn, originConn, 0, &log) + stream.Pipe(wsConn, originConn, &log) cancel() wsConn.Close() return nil diff --git a/ingress/origin_connection.go b/ingress/origin_connection.go index d4d2644e..22013013 100644 --- a/ingress/origin_connection.go +++ b/ingress/origin_connection.go @@ -25,9 +25,9 @@ type OriginConnection interface { type streamHandlerFunc func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger) // DefaultStreamHandler is an implementation of streamHandlerFunc that -// performs a two-way io.Copy between originConn and remoteConn. +// performs a two way io.Copy between originConn and remoteConn. func DefaultStreamHandler(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger) { - stream.Pipe(originConn, remoteConn, stream.DefaultTimeoutAfterFirstClose, log) + stream.Pipe(originConn, remoteConn, log) } // tcpConnection is an OriginConnection that directly streams to raw TCP. @@ -38,7 +38,7 @@ type tcpConnection struct { } func (tc *tcpConnection) Stream(_ context.Context, tunnelConn io.ReadWriter, _ *zerolog.Logger) { - stream.Pipe(tunnelConn, tc, stream.DefaultTimeoutAfterFirstClose, tc.logger) + stream.Pipe(tunnelConn, tc, tc.logger) } func (tc *tcpConnection) Write(b []byte) (int, error) { diff --git a/ingress/origin_connection_test.go b/ingress/origin_connection_test.go index 52a121a4..b031c011 100644 --- a/ingress/origin_connection_test.go +++ b/ingress/origin_connection_test.go @@ -38,7 +38,6 @@ func TestStreamTCPConnection(t *testing.T) { tcpConn := tcpConnection{ Conn: cfdConn, writeTimeout: 30 * time.Second, - logger: TestLogger, } eyeballConn, edgeConn := net.Pipe() @@ -158,7 +157,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) { require.NoError(t, err) defer func() { _ = wsForwarderInConn.Close() }() - stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, stream.DefaultTimeoutAfterFirstClose, TestLogger) + stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, TestLogger) return nil }) diff --git a/ingress/origin_service.go b/ingress/origin_service.go index 65a71dbb..e13204c5 100644 --- a/ingress/origin_service.go +++ b/ingress/origin_service.go @@ -99,6 +99,7 @@ type rawTCPService struct { name string dialer net.Dialer writeTimeout time.Duration + logger *zerolog.Logger } func (o *rawTCPService) String() string { @@ -232,12 +233,10 @@ func (o *helloWorld) start( if err != nil { return errors.Wrap(err, "Cannot start Hello World Server") } - go func() { - _ = hello.StartHelloWorldServer(log, helloListener, shutdownC) - }() + go hello.StartHelloWorldServer(log, helloListener, shutdownC) o.server = helloListener - o.url = &url.URL{ + o.httpService.url = &url.URL{ Scheme: "https", Host: o.server.Addr().String(), } @@ -357,7 +356,7 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol IdleConnTimeout: cfg.KeepAliveTimeout.Duration, TLSHandshakeTimeout: cfg.TLSTimeout.Duration, ExpectContinueTimeout: 1 * time.Second, - TLSClientConfig: &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify}, //nolint: gosec + TLSClientConfig: &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify}, ForceAttemptHTTP2: cfg.Http2Origin, } if _, isHelloWorld := service.(*helloWorld); !isHelloWorld && cfg.OriginServerName != "" { @@ -375,6 +374,7 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol // DialContext depends on which kind of origin is being used. dialContext := dialer.DialContext switch service := service.(type) { + // If this origin is a unix socket, enforce network type "unix". case *unixSocketPath: httpTransport.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) { diff --git a/proxy/proxy.go b/proxy/proxy.go index 4ec5487b..8ae27adb 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -256,7 +256,7 @@ func (p *Proxy) proxyHTTPRequest( reader: tr.Body, } - stream.Pipe(eyeballStream, rwc, time.Second*0, logger) + stream.Pipe(eyeballStream, rwc, logger) return nil } @@ -311,7 +311,7 @@ func (p *Proxy) proxyStream( // proxyTCPStream proxies private network type TCP connections as a stream towards an available origin. // -// This is different from proxyStream because it's not leveraged ingress rule services and uses the +// This is different than proxyStream because it's not leveraged ingress rule services and uses the // originDialer from OriginDialerService. func (p *Proxy) proxyTCPStream( tr *tracing.TracedContext, @@ -344,8 +344,7 @@ func (p *Proxy) proxyTCPStream( connectLatency.Observe(float64(time.Since(start).Milliseconds())) logger.Debug().Msg("proxy stream acknowledged") - stream.Pipe(tunnelConn, originConn, stream.DefaultTimeoutAfterFirstClose, logger) - + stream.Pipe(tunnelConn, originConn, logger) return nil } diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index 6392d797..f5038122 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -223,9 +223,9 @@ func testProxyWebsocket(proxy connection.OriginProxy) func(t *testing.T) { } if ctx.Err() == context.DeadlineExceeded { t.Errorf("Test timed out") - _ = readPipe.Close() - _ = writePipe.Close() - _ = responseWriter.Close() + readPipe.Close() + writePipe.Close() + responseWriter.Close() } return nil }) @@ -647,7 +647,7 @@ func TestConnections(t *testing.T) { ingressServiceScheme: "tcp://", originService: func(t *testing.T, ln net.Listener) { // closing the listener created by the test. - _ = ln.Close() + ln.Close() }, eyeballResponseWriter: newTCPRespWriter(replayer), eyeballRequestBody: newTCPRequestBody([]byte("test2")), @@ -756,8 +756,6 @@ func newTCPRequestBody(data []byte) *requestBody { pr, pw := io.Pipe() go func() { _, _ = pw.Write(data) - // Close the write side once the payload has been sent. - _ = pw.Close() }() return &requestBody{ pr: pr, @@ -803,8 +801,8 @@ func (p *pipedRequestBody) roundtrip(addr string) []byte { if err != nil { panic(err) } - defer func() { _ = conn.Close() }() - defer func() { _ = resp.Body.Close() }() + defer conn.Close() + defer resp.Body.Close() if resp.StatusCode != http.StatusSwitchingProtocols { panic(fmt.Errorf("resp returned status code: %d", resp.StatusCode)) @@ -951,7 +949,7 @@ func runEchoTCPService(t *testing.T, l net.Listener) { if err != nil { panic(err) } - defer func() { _ = conn.Close() }() + defer conn.Close() for { buf := make([]byte, 1024) @@ -989,7 +987,7 @@ func runEchoWSService(t *testing.T, l net.Listener) { t.Log(err) return } - defer func() { _ = conn.Close() }() + defer conn.Close() for { messageType, p, err := conn.ReadMessage() diff --git a/stream/stream.go b/stream/stream.go index 557cbfcf..3b623241 100644 --- a/stream/stream.go +++ b/stream/stream.go @@ -2,7 +2,6 @@ package stream import ( "encoding/hex" - "errors" "fmt" "io" "runtime/debug" @@ -10,17 +9,12 @@ import ( "time" "github.com/getsentry/sentry-go" + "github.com/pkg/errors" "github.com/rs/zerolog" "github.com/cloudflare/cloudflared/cfio" ) -// DefaultTimeoutAfterFirstClose controls the upper bound of how long we wait for the second stream to finish. -// We use bidirectional streams for our communication. Since the read and write sides can be closed independently, -// we must have a way to close the second stream once the first one finishes. We don't want to wait indefinitely, -// since we want to prevent misbehaving clients from blocking cloudflared. -const DefaultTimeoutAfterFirstClose = time.Second * 10 - type Stream interface { Reader WriterCloser @@ -43,7 +37,7 @@ type nopCloseWriterAdapter struct { io.ReadWriter } -func noopCloseWriter(stream io.ReadWriter) *nopCloseWriterAdapter { +func NopCloseWriterAdapter(stream io.ReadWriter) *nopCloseWriterAdapter { return &nopCloseWriterAdapter{stream} } @@ -78,7 +72,7 @@ func (s *bidirectionalStreamStatus) wait(maxWaitForSecondStream time.Duration) e select { case <-timer.C: - return fmt.Errorf("timeout waiting for second stream to finish %s", maxWaitForSecondStream) + return fmt.Errorf("timeout waiting for second stream to finish") case <-s.doneChan: return nil } @@ -91,19 +85,15 @@ func (s *bidirectionalStreamStatus) isAnyDone() bool { } // Pipe copies copy data to & from provided io.ReadWriters. -func Pipe(tunnelConn, originConn io.ReadWriter, timeoutAfterFirstClose time.Duration, log *zerolog.Logger) { - if err := PipeBidirectional(noopCloseWriter(tunnelConn), noopCloseWriter(originConn), timeoutAfterFirstClose, log); err != nil { - log.Warn().Err(err).Msg("Failed to pipe bidirectional stream") - } +func Pipe(tunnelConn, originConn io.ReadWriter, log *zerolog.Logger) { + _ = PipeBidirectional(NopCloseWriterAdapter(tunnelConn), NopCloseWriterAdapter(originConn), 0, log) } -// PipeBidirectional copies data between two unidirectional streams. It is a special case of Pipe that accepts streams -// whose read and write sides can be closed independently. The main difference is that when piping data from a reader -// to a writer, if EOF is read, this implementation propagates the EOF signal to the destination by closing the write -// side of the bidirectional stream. -// Finally, once EOF is received from one of the provided streams, the other direction has a configured grace period to -// finish; otherwise, the method returns a timeout error. It is, however, the responsibility of the caller to close -// the associated streams at both ends in order to free all resources and goroutines. +// PipeBidirectional copies data to two unidirectional streams. It is a special case of Pipe where it receives a concept that allows for Read and Write side to be closed independently. +// The main difference is that when piping data from a reader to a writer, if EOF is read, then this implementation propagates the EOF signal to the destination/writer by closing the write side of the +// Bidirectional Stream. +// Finally, depending on once EOF is ready from one of the provided streams, the other direction of streaming data will have a configured time period to also finish, otherwise, +// the method will return immediately with a timeout error. It is however, the responsibility of the caller to close the associated streams in both ends in order to free all the resources/go-routines. func PipeBidirectional(downstream, upstream Stream, maxWaitForSecondStream time.Duration, log *zerolog.Logger) error { status := newBiStreamStatus() @@ -111,7 +101,7 @@ func PipeBidirectional(downstream, upstream Stream, maxWaitForSecondStream time. go unidirectionalStream(upstream, downstream, "downstream->upstream", status, log) if err := status.wait(maxWaitForSecondStream); err != nil { - return fmt.Errorf("unable to wait for both streams while proxying: %w", err) + return errors.Wrap(err, "unable to wait for both streams while proxying") } return nil diff --git a/stream/stream_test.go b/stream/stream_test.go index 8c99e72d..6db372b3 100644 --- a/stream/stream_test.go +++ b/stream/stream_test.go @@ -30,8 +30,8 @@ func TestPipeBidirectionalFinishOneSideTimeout(t *testing.T) { func TestPipeBidirectionalClosingWriteBothSidesAlsoExists(t *testing.T) { fun := func(upstream, downstream *mockedStream) { - _ = downstream.CloseWrite() - _ = upstream.CloseWrite() + downstream.CloseWrite() + upstream.CloseWrite() downstream.writeToReader("abc") upstream.writeToReader("abc") @@ -42,7 +42,7 @@ func TestPipeBidirectionalClosingWriteBothSidesAlsoExists(t *testing.T) { func TestPipeBidirectionalClosingWriteSingleSideAlsoExists(t *testing.T) { fun := func(upstream, downstream *mockedStream) { - _ = downstream.CloseWrite() + downstream.CloseWrite() downstream.writeToReader("abc") upstream.writeToReader("abc") @@ -51,46 +51,6 @@ func TestPipeBidirectionalClosingWriteSingleSideAlsoExists(t *testing.T) { testPipeBidirectionalUnblocking(t, fun, time.Millisecond*200, true) } -// TestPipeBidirectionalReturnsWhenBothSidesFinish verifies that -// PipeBidirectional returns as soon as both stream directions finish, without -// waiting for the full timeout grace period to expire. This guards against a -// regression where the second-stream wait would block for the whole timeout -// even when the result is already available. -func TestPipeBidirectionalReturnsWhenBothSidesFinish(t *testing.T) { - t.Parallel() - - const ( - timeout = time.Second * 5 - maxWallTime = time.Millisecond * 500 - ) - - logger := zerolog.Nop() - downstream := newMockedStream() - upstream := newMockedStream() - - resultCh := make(chan error, 1) - go func() { - resultCh <- PipeBidirectional(downstream, upstream, timeout, &logger) - }() - - // Close both reader sides so both stream directions reach EOF promptly. - downstream.closeReader() - upstream.closeReader() - - start := time.Now() - select { - case err := <-resultCh: - elapsed := time.Since(start) - require.NoError(t, err) - require.Less(t, elapsed, maxWallTime, - "PipeBidirectional should return as soon as both streams finish, not after the full %s timeout (took %s)", - timeout, elapsed, - ) - case <-time.After(timeout): - require.Fail(t, "PipeBidirectional did not return before the timeout expired") - } -} - func testPipeBidirectionalUnblocking(t *testing.T, afterFun func(*mockedStream, *mockedStream), timeout time.Duration, expectTimeout bool) { logger := zerolog.Nop() @@ -107,9 +67,9 @@ func testPipeBidirectionalUnblocking(t *testing.T, afterFun func(*mockedStream, select { case err := <-resultCh: if expectTimeout { - require.Error(t, err) + require.NotNil(t, err) } else { - require.NoError(t, err) + require.Nil(t, err) } case <-time.After(timeout * 2):