Merge branch 'dev' into no-ioutil

.github/workflows/tests.yaml

@@ -18,16 +18,18 @@ jobs:
         working-directory: ${{ env.GOPATH }}/src/github.com/${{ github.repository }}
     strategy:
       matrix:
-        go: ["1.21.x"]
+        go: ["stable", "oldstable"]
         include:
-        - go: 1.21.x
+        - go: "stable"
           latest: true
           COVERAGE: "yes"
           LINT: "yes"
+        - go: "oldstable"
+          LINT: "yes"
 
     steps:
     - name: Setup Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v5
       with:
         go-version: ${{ matrix.go }}
 

CHANGELOG.md

@@ -1,6 +1,25 @@
 Changelog
 =========
 
+## [1.34.4] - 2024-06-26
+### Fixed
+
+ * fix getSysConn to work with TLS (#918)
+
+### Changed
+
+ * Switch to aliases for Go versions in CI (#919)
+
+## [1.34.3] - 2024-04-23
+### Fixed
+
+ * Fix a DoS vulnerability of the vendored apache-thrift library (#915, #916)
+
+## [1.34.2] - 2024-02-16
+### Added
+
+ * Expose `inbound.cancels.{requested,honored}` metrics (#912)
+
 ## [1.34.1] - 2023-12-11
 ### Fixed
 
@@ -382,6 +401,9 @@ Changelog
 * Thrift support, including includes.
 
 [//]: # (Version Links)
+[1.34.4]: https://github.com/uber/tchannel-go/compare/v1.34.3...v1.34.4
+[1.34.3]: https://github.com/uber/tchannel-go/compare/v1.34.2...v1.34.3
+[1.34.2]: https://github.com/uber/tchannel-go/compare/v1.34.1...v1.34.2
 [1.34.1]: https://github.com/uber/tchannel-go/compare/v1.34.0...v1.34.1
 [1.34.0]: https://github.com/uber/tchannel-go/compare/v1.33.0...v1.34.0
 [1.33.0]: https://github.com/uber/tchannel-go/compare/v1.32.1...v1.33.0

connection.go

@@ -21,6 +21,7 @@
 package tchannel
 
 import (
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
@@ -971,7 +972,18 @@ func (c *Connection) getLastActivityWriteTime() time.Time {
 }
 
 func getSysConn(conn net.Conn, log Logger) syscall.RawConn {
-	connSyscall, ok := conn.(syscall.Conn)
+	var (
+		connSyscall syscall.Conn
+		ok          bool
+	)
+	switch v := conn.(type) {
+	case syscall.Conn:
+		connSyscall = v
+		ok = true
+	case *tls.Conn:
+		connSyscall, ok = v.NetConn().(syscall.Conn)
+	}
+
 	if !ok {
 		log.WithFields(LogField{"connectionType", fmt.Sprintf("%T", conn)}).
 			Error("Connection does not implement SyscallConn.")

connection_internal_test.go

@@ -22,7 +22,9 @@ package tchannel
 
 import (
 	"bytes"
+	"crypto/tls"
 	"net"
+	"net/http/httptest"
 	"syscall"
 	"testing"
 
@@ -79,4 +81,45 @@ func TestGetSysConn(t *testing.T) {
 		require.NotNil(t, sysConn)
 		assert.Empty(t, loggerBuf.String(), "expected no logs on success")
 	})
+
+	t.Run("SyscallConn is successful with TLS", func(t *testing.T) {
+		var (
+			loggerBuf = &bytes.Buffer{}
+			logger    = NewLogger(loggerBuf)
+			server    = httptest.NewTLSServer(nil)
+		)
+		defer server.Close()
+
+		conn, err := tls.Dial("tcp", server.Listener.Addr().String(), &tls.Config{InsecureSkipVerify: true})
+		require.NoError(t, err, "failed to dial")
+		defer conn.Close()
+
+		sysConn := getSysConn(conn, logger)
+		require.NotNil(t, sysConn)
+		assert.Empty(t, loggerBuf.String(), "expected no logs on success")
+	})
+
+	t.Run("no SyscallConn - nil net.Conn", func(t *testing.T) {
+		var (
+			loggerBuf   = &bytes.Buffer{}
+			logger      = NewLogger(loggerBuf)
+			syscallConn = getSysConn(nil /* conn */, logger)
+		)
+
+		require.Nil(t, syscallConn, "expected no syscall.RawConn to be returned")
+		assert.Contains(t, loggerBuf.String(), "Connection does not implement SyscallConn", "missing log")
+		assert.Contains(t, loggerBuf.String(), "{connectionType <nil>}", "missing type in log")
+	})
+
+	t.Run("no SyscallConn - TLS with no net.Conn", func(t *testing.T) {
+		var (
+			loggerBuf   = &bytes.Buffer{}
+			logger      = NewLogger(loggerBuf)
+			syscallConn = getSysConn(&tls.Conn{}, logger)
+		)
+
+		require.Nil(t, syscallConn, "expected no syscall.RawConn to be returned")
+		assert.Contains(t, loggerBuf.String(), "Connection does not implement SyscallConn", "missing log")
+		assert.Contains(t, loggerBuf.String(), "{connectionType *tls.Conn}", "missing type in log")
+	})
 }

connection_test.go

@@ -496,6 +496,9 @@ func TestServerClientCancellation(t *testing.T) {
 	opts.DefaultConnectionOptions.SendCancelOnContextCanceled = true
 	opts.DefaultConnectionOptions.PropagateCancel = true
 
+	serverStats := newRecordingStatsReporter()
+	opts.StatsReporter = serverStats
+
 	testutils.WithTestServer(t, opts, func(t testing.TB, ts *testutils.TestServer) {
 		callReceived := make(chan struct{})
 		testutils.RegisterFunc(ts.Server(), "ctxWait", func(ctx context.Context, args *raw.Args) (*raw.Res, error) {
@@ -519,16 +522,23 @@ func TestServerClientCancellation(t *testing.T) {
 		_, _, _, err := raw.Call(ctx, ts.Server(), ts.HostPort(), ts.ServiceName(), "ctxWait", nil, nil)
 		assert.Equal(t, ErrRequestCancelled, err, "client call result")
 
+		statsTags := ts.Server().StatsTags()
+		serverStats.Expected.IncCounter("inbound.cancels.requested", statsTags, 1)
+		serverStats.Expected.IncCounter("inbound.cancels.honored", statsTags, 1)
+
 		calls := relaytest.NewMockStats()
 		calls.Add(ts.ServiceName(), ts.ServiceName(), "ctxWait").Failed("canceled").End()
 		ts.AssertRelayStats(calls)
 	})
+
+	serverStats.ValidateExpected(t)
 }
 
 func TestCancelWithoutSendCancelOnContextCanceled(t *testing.T) {
 	tests := []struct {
 		msg                         string
 		sendCancelOnContextCanceled bool
+		wantCancelRequested         bool
 	}{
 		{
 			msg:                         "no send or process cancel",
@@ -537,6 +547,7 @@ func TestCancelWithoutSendCancelOnContextCanceled(t *testing.T) {
 		{
 			msg:                         "only enable cancels on outbounds",
 			sendCancelOnContextCanceled: true,
+			wantCancelRequested:         true,
 		},
 	}
 
@@ -545,7 +556,12 @@ func TestCancelWithoutSendCancelOnContextCanceled(t *testing.T) {
 			opts := testutils.NewOpts()
 			opts.DefaultConnectionOptions.SendCancelOnContextCanceled = tt.sendCancelOnContextCanceled
 
+			serverStats := newRecordingStatsReporter()
+			opts.StatsReporter = serverStats
+
 			testutils.WithTestServer(t, opts, func(t testing.TB, ts *testutils.TestServer) {
+				serverStats.Reset()
+
 				callReceived := make(chan struct{})
 				testutils.RegisterFunc(ts.Server(), "ctxWait", func(ctx context.Context, args *raw.Args) (*raw.Res, error) {
 					require.NoError(t, ctx.Err(), "context valid before cancellation")
@@ -571,6 +587,17 @@ func TestCancelWithoutSendCancelOnContextCanceled(t *testing.T) {
 				calls := relaytest.NewMockStats()
 				calls.Add(ts.ServiceName(), ts.ServiceName(), "ctxWait").Failed("timeout").End()
 				ts.AssertRelayStats(calls)
+
+				ts.AddPostFn(func() {
+					// Validating these at the end of the test, when server has fully processed the cancellation.
+					if tt.wantCancelRequested && !ts.HasRelay() {
+						serverStats.Expected.IncCounter("inbound.cancels.requested", ts.Server().StatsTags(), 1)
+						serverStats.ValidateExpected(t)
+					} else {
+						serverStats.EnsureNotPresent(t, "inbound.cancels.requested")
+					}
+					serverStats.EnsureNotPresent(t, "inbound.cancels.honored")
+				})
 			})
 		})
 	}

inbound.go

@@ -144,13 +144,17 @@ func (c *Connection) handleCallReqContinue(frame *Frame) bool {
 }
 
 func (c *Connection) handleCancel(frame *Frame) bool {
+	c.statsReporter.IncCounter("inbound.cancels.requested", c.commonStatsTags, 1)
+
 	if !c.opts.PropagateCancel {
 		if c.log.Enabled(LogLevelDebug) {
 			c.log.Debugf("Ignoring cancel for %v", frame.Header.ID)
 		}
 		return true
 	}
 
+	c.statsReporter.IncCounter("inbound.cancels.honored", c.commonStatsTags, 1)
+
 	c.inbound.handleCancel(frame)
 
 	// Free the frame, as it's consumed immediately.

stats_utils_test.go

@@ -125,9 +125,28 @@ func (r *recordingStatsReporter) Validate(t *testing.T) {
 
 	assert.Equal(t, keysMap(r.Expected.Values), keysMap(r.Values),
 		"Metric keys are different")
-	for counterKey, counter := range r.Values {
-		expectedCounter, ok := r.Expected.Values[counterKey]
-		if !ok {
+	r.validateExpectedLocked(t)
+}
+
+// ValidateExpected only validates metrics added to expected rather than all recorded metrics.
+func (r *recordingStatsReporter) ValidateExpected(t testing.TB) {
+	r.Lock()
+	defer r.Unlock()
+
+	r.validateExpectedLocked(t)
+}
+
+func (r *recordingStatsReporter) EnsureNotPresent(t testing.TB, counter string) {
+	r.Lock()
+	defer r.Unlock()
+
+	assert.NotContains(t, r.Values, counter, "metric should not be present")
+}
+
+func (r *recordingStatsReporter) validateExpectedLocked(t testing.TB) {
+	for counterKey, expectedCounter := range r.Expected.Values {
+		counter, ok := r.Values[counterKey]
+		if !assert.True(t, ok, "expected %v not found", counterKey) {
 			continue
 		}
 

testutils/test_server.go

@@ -400,6 +400,11 @@ func (ts *TestServer) verify(ch *tchannel.Channel) {
 	assert.NoError(ts, errs, "Verification failed. Channel state:\n%v", IntrospectJSON(ch, nil /* opts */))
 }
 
+// AddPostFn registers a function that will be executed after channels are closed.
+func (ts *TestServer) AddPostFn(fn func()) {
+	ts.postFns = append(ts.postFns, fn)
+}
+
 func (ts *TestServer) post() {
 	if !ts.Failed() {
 		for _, ch := range ts.channels {

thirdparty/github.com/apache/thrift/lib/go/thrift/protocol.go

@@ -21,6 +21,7 @@ package thrift
 
 import (
 	"errors"
+	"fmt"
 )
 
 const (
@@ -88,7 +89,7 @@ func SkipDefaultDepth(prot TProtocol, typeId TType) (err error) {
 
 // Skips over the next data element from the provided input TProtocol object.
 func Skip(self TProtocol, fieldType TType, maxDepth int) (err error) {
-	
+
     if maxDepth <= 0 {
 		return NewTProtocolExceptionWithType( DEPTH_LIMIT, errors.New("Depth limit exceeded"))
 	}
@@ -143,7 +144,10 @@ func Skip(self TProtocol, fieldType TType, maxDepth int) (err error) {
 			if err != nil {
 				return err
 			}
-			self.Skip(valueType)
+			err = Skip(self, valueType, maxDepth-1)
+			if err != nil {
+				return err
+			}
 		}
 		return self.ReadMapEnd()
 	case SET:
@@ -170,6 +174,8 @@ func Skip(self TProtocol, fieldType TType, maxDepth int) (err error) {
 			}
 		}
 		return self.ReadListEnd()
+	default:
+		return NewTProtocolExceptionWithType(INVALID_DATA, fmt.Errorf("Unknown data type %d", fieldType))
 	}
 	return nil
 }