Merge remote-tracking branch 'origin/stable' into v2

Refactor TLS fragmenting

.github/workflows/ci.yaml

@@ -122,7 +122,7 @@ jobs:
   artifacts:
     name: Build release artifacts
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 20
     steps:
       - name: Checkout
         uses: actions/checkout@v6

Dockerfile

@@ -33,7 +33,7 @@ RUN go mod download
 COPY . /app
 
 RUN set -x \
-  && version="$(git describe --exact-match HEAD 2>/dev/null || git describe --tags --always)" \
+  && version="$(git describe --exact-match HEAD 2>/dev/null || git describe --tags --always 2>/dev/null || echo dev)" \
   && go build \
       -trimpath \
       -mod=readonly \

README.md

@@ -29,6 +29,8 @@ are the most notable:
 * [Official](https://github.com/TelegramMessenger/MTProxy)
 * [Python](https://github.com/alexbers/mtprotoproxy)
 * [Erlang](https://github.com/seriyps/mtproto_proxy)
+* [Teleproxy (C)](https://github.com/teleproxy/teleproxy)
+* [mtproto.zig (Zig)](https://github.com/sleep3r/mtproto.zig)
 * [Telemt (Rust)](https://github.com/telemt/telemt)
 
 You can use any of these. They work great and all implementations have

antireplay/stable_bloom_filter_test.go

@@ -12,7 +12,7 @@ type StableBloomFilterTestSuite struct {
 }
 
 func (suite *StableBloomFilterTestSuite) TestOp() {
-	filter := antireplay.NewStableBloomFilter(500, 0.001)
+	filter := antireplay.NewStableBloomFilter(100000, 0.001)
 
 	suite.False(filter.SeenBefore([]byte{1, 2, 3}))
 	suite.False(filter.SeenBefore([]byte{4, 5, 6}))

mise.lock

@@ -82,40 +82,40 @@ checksum = "sha256:4932cfca5e75bf60fe1c576edf459e5e809e6644664a068185d64b84af3fa
 url = "https://github.com/golangci/golangci-lint/releases/download/v2.11.4/golangci-lint-2.11.4-windows-amd64.zip"
 
 [[tools.goreleaser]]
-version = "2.14.3"
+version = "2.15.2"
 backend = "aqua:goreleaser/goreleaser"
 
 [tools.goreleaser."platforms.linux-arm64"]
-checksum = "sha256:581a10e53c1176b3e81ee45cf531e02dbf899db0bc7b795669347df4276ce948"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Linux_arm64.tar.gz"
+checksum = "sha256:5db66761a98f6693161e49e1a95d28d2673a892ba60cb4a5e16736cafd41c4c9"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Linux_arm64.tar.gz"
 provenance = "cosign"
 
 [tools.goreleaser."platforms.linux-arm64-musl"]
-checksum = "sha256:581a10e53c1176b3e81ee45cf531e02dbf899db0bc7b795669347df4276ce948"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Linux_arm64.tar.gz"
+checksum = "sha256:5db66761a98f6693161e49e1a95d28d2673a892ba60cb4a5e16736cafd41c4c9"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Linux_arm64.tar.gz"
 provenance = "cosign"
 
 [tools.goreleaser."platforms.linux-x64"]
-checksum = "sha256:dc7faeeeb6da8bdfda788626263a4ae725892a8c7504b975c3234127d4a44579"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Linux_x86_64.tar.gz"
+checksum = "sha256:0ebdbf0353aba566b969dde746cc4e4806f96c27aa2f3971b229a9df7611fedc"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Linux_x86_64.tar.gz"
 provenance = "cosign"
 
 [tools.goreleaser."platforms.linux-x64-musl"]
-checksum = "sha256:dc7faeeeb6da8bdfda788626263a4ae725892a8c7504b975c3234127d4a44579"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Linux_x86_64.tar.gz"
+checksum = "sha256:0ebdbf0353aba566b969dde746cc4e4806f96c27aa2f3971b229a9df7611fedc"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Linux_x86_64.tar.gz"
 provenance = "cosign"
 
 [tools.goreleaser."platforms.macos-arm64"]
-checksum = "sha256:3507798489e107a78aff36b169de48148a335ac26eb3161608d905f3f3a957bd"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Darwin_all.tar.gz"
-provenance = "cosign"
+checksum = "sha256:0e6bd67688ac949780bf1166813a91f89856898ef4c40d7d46c2c74ebaa4b9ee"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Darwin_all.tar.gz"
+provenance = "github-attestations"
 
 [tools.goreleaser."platforms.macos-x64"]
-checksum = "sha256:3507798489e107a78aff36b169de48148a335ac26eb3161608d905f3f3a957bd"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Darwin_all.tar.gz"
+checksum = "sha256:0e6bd67688ac949780bf1166813a91f89856898ef4c40d7d46c2c74ebaa4b9ee"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Darwin_all.tar.gz"
 provenance = "cosign"
 
 [tools.goreleaser."platforms.windows-x64"]
-checksum = "sha256:3deea8ff471aa258a2d99f3e5302971d7028647ae8ddaf103257a8113e485a31"
-url = "https://github.com/goreleaser/goreleaser/releases/download/v2.14.3/goreleaser_Windows_x86_64.zip"
+checksum = "sha256:7459832946dbe122c144f8d7f87484d8572ca005b779310aa6bb03346e8de17a"
+url = "https://github.com/goreleaser/goreleaser/releases/download/v2.15.2/goreleaser_Windows_x86_64.zip"
 provenance = "cosign"

mtglib/conns.go

@@ -3,9 +3,11 @@ package mtglib
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net"
+	"sync/atomic"
 	"time"
 
 	"github.com/9seconds/mtg/v2/essentials"
@@ -97,20 +99,63 @@ func newConnProxyProtocol(source, target essentials.Conn) *connProxyProtocol {
 	}
 }
 
-type connIdleTimeout struct {
-	essentials.Conn
-
+// idleTracker is a shared idle tracker for a pair of relay connections.
+// Both directions update the same timestamp so that activity in one direction
+// prevents the other (idle) direction from timing out.
+type idleTracker struct {
+	lastActive atomic.Pointer[time.Time]
 	timeout    time.Duration
 }
 
-func (c connIdleTimeout) Read(b []byte) (int, error) {
-	c.SetReadDeadline(time.Now().Add(c.timeout)) //nolint: errcheck
+func newIdleTracker(timeout time.Duration) *idleTracker {
+	t := &idleTracker{timeout: timeout}
+	t.touch()
 
-	return c.Conn.Read(b) //nolint: wrapcheck
+	return t
+}
+
+func (t *idleTracker) touch() {
+	stamp := time.Now()
+	t.lastActive.Store(&stamp)
+}
+
+func (t *idleTracker) isIdle() bool {
+	return time.Since(*t.lastActive.Load()) >= t.timeout
+}
+
+type connIdleTimeout struct {
+	essentials.Conn
+
+	tracker *idleTracker
+}
+
+func (c connIdleTimeout) Read(b []byte) (int, error) {
+	var netErr net.Error
+
+	for {
+		c.SetReadDeadline(time.Now().Add(c.tracker.timeout)) //nolint: errcheck
+
+		n, err := c.Conn.Read(b)
+
+		switch {
+		case err == nil:
+			c.tracker.touch()
+			return n, nil
+		case errors.As(err, &netErr) && netErr.Timeout() && !c.tracker.isIdle():
+			continue
+		}
+
+		return n, err
+	}
 }
 
 func (c connIdleTimeout) Write(b []byte) (int, error) {
-	c.SetWriteDeadline(time.Now().Add(c.timeout)) //nolint: errcheck
+	c.SetWriteDeadline(time.Now().Add(c.tracker.timeout)) //nolint: errcheck
 
-	return c.Conn.Write(b) //nolint: wrapcheck
+	n, err := c.Conn.Write(b)
+	if n > 0 {
+		c.tracker.touch()
+	}
+
+	return n, err //nolint: wrapcheck
 }

mtglib/conns_internal_test.go

@@ -16,6 +16,12 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
+type netTimeoutError struct{}
+
+func (e netTimeoutError) Error() string   { return "i/o timeout" }
+func (e netTimeoutError) Timeout() bool   { return true }
+func (e netTimeoutError) Temporary() bool { return true }
+
 type ConnRewindBaseConn struct {
 	testlib.EssentialsConnMock
 
@@ -291,6 +297,141 @@ func (suite *ConnProxyProtocolTestSuite) TearDownTest() {
 	suite.targetConnMock.AssertExpectations(suite.T())
 }
 
+type IdleTrackerTestSuite struct {
+	suite.Suite
+}
+
+func (suite *IdleTrackerTestSuite) TestNewNotIdle() {
+	tracker := newIdleTracker(time.Second)
+	suite.False(tracker.isIdle())
+}
+
+func (suite *IdleTrackerTestSuite) TestIdleAfterTimeout() {
+	tracker := newIdleTracker(10 * time.Millisecond)
+	time.Sleep(20 * time.Millisecond)
+
+	suite.True(tracker.isIdle())
+}
+
+func (suite *IdleTrackerTestSuite) TestTouchResetsIdle() {
+	tracker := newIdleTracker(50 * time.Millisecond)
+	time.Sleep(30 * time.Millisecond)
+
+	tracker.touch()
+
+	suite.False(tracker.isIdle())
+}
+
+type ConnIdleTimeoutTestSuite struct {
+	suite.Suite
+
+	connMock *testlib.EssentialsConnMock
+	tracker  *idleTracker
+	conn     connIdleTimeout
+}
+
+func (suite *ConnIdleTimeoutTestSuite) SetupTest() {
+	suite.connMock = &testlib.EssentialsConnMock{}
+	suite.tracker = newIdleTracker(time.Second)
+	suite.conn = connIdleTimeout{
+		Conn:    suite.connMock,
+		tracker: suite.tracker,
+	}
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TearDownTest() {
+	suite.connMock.AssertExpectations(suite.T())
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestReadOk() {
+	suite.connMock.On("SetReadDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Read", mock.Anything).Once().Return(5, nil)
+
+	n, err := suite.conn.Read(make([]byte, 10))
+	suite.NoError(err)
+	suite.Equal(5, n)
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestReadNonTimeoutErr() {
+	suite.connMock.On("SetReadDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Read", mock.Anything).Once().Return(0, io.EOF)
+
+	n, err := suite.conn.Read(make([]byte, 10))
+	suite.True(errors.Is(err, io.EOF))
+	suite.Equal(0, n)
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestReadTimeoutRetriesWhenNotIdle() {
+	suite.connMock.On("SetReadDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Read", mock.Anything).Once().Return(0, netTimeoutError{})
+	suite.connMock.On("Read", mock.Anything).Once().Return(5, nil)
+
+	n, err := suite.conn.Read(make([]byte, 10))
+	suite.NoError(err)
+	suite.Equal(5, n)
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestReadTimeoutClosesWhenIdle() {
+	suite.tracker = newIdleTracker(time.Millisecond)
+	suite.conn = connIdleTimeout{
+		Conn:    suite.connMock,
+		tracker: suite.tracker,
+	}
+
+	time.Sleep(5 * time.Millisecond)
+
+	suite.connMock.On("SetReadDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Read", mock.Anything).Once().Return(0, netTimeoutError{})
+
+	n, err := suite.conn.Read(make([]byte, 10))
+	suite.Equal(0, n)
+
+	netErr, ok := err.(net.Error) //nolint: errorlint
+	suite.True(ok)
+	suite.True(netErr.Timeout())
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestSharedTrackerPreventsFalseTimeout() {
+	connMock2 := &testlib.EssentialsConnMock{}
+	conn2 := connIdleTimeout{
+		Conn:    connMock2,
+		tracker: suite.tracker,
+	}
+
+	connMock2.On("SetWriteDeadline", mock.Anything).Return(nil)
+	connMock2.On("Write", mock.Anything).Once().Return(5, nil)
+
+	_, _ = conn2.Write(make([]byte, 5))
+
+	suite.connMock.On("SetReadDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Read", mock.Anything).Once().Return(0, netTimeoutError{})
+	suite.connMock.On("Read", mock.Anything).Once().Return(3, nil)
+
+	n, err := suite.conn.Read(make([]byte, 10))
+	suite.NoError(err)
+	suite.Equal(3, n)
+
+	connMock2.AssertExpectations(suite.T())
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestWriteOk() {
+	suite.connMock.On("SetWriteDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Write", mock.Anything).Once().Return(5, nil)
+
+	n, err := suite.conn.Write(make([]byte, 5))
+	suite.NoError(err)
+	suite.Equal(5, n)
+}
+
+func (suite *ConnIdleTimeoutTestSuite) TestWriteErr() {
+	suite.connMock.On("SetWriteDeadline", mock.Anything).Return(nil)
+	suite.connMock.On("Write", mock.Anything).Once().Return(0, io.EOF)
+
+	n, err := suite.conn.Write(make([]byte, 5))
+	suite.True(errors.Is(err, io.EOF))
+	suite.Equal(0, n)
+}
+
 func TestConnTraffic(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ConnTrafficTestSuite{})
@@ -305,3 +446,13 @@ func TestConnProxyProtocol(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ConnProxyProtocolTestSuite{})
 }
+
+func TestIdleTracker(t *testing.T) {
+	t.Parallel()
+	suite.Run(t, &IdleTrackerTestSuite{})
+}
+
+func TestConnIdleTimeout(t *testing.T) {
+	t.Parallel()
+	suite.Run(t, &ConnIdleTimeoutTestSuite{})
+}

mtglib/internal/dc/view.go

@@ -5,15 +5,19 @@ type dcView struct {
 }
 
 func (d dcView) getV4(dc int) []Addr {
-	addrs := d.publicConfigs.getV4(dc)
+	var addrs []Addr
+
 	addrs = append(addrs, defaultDCAddrSet.getV4(dc)...)
+	addrs = append(addrs, d.publicConfigs.getV4(dc)...)
 
 	return addrs
 }
 
 func (d dcView) getV6(dc int) []Addr {
-	addrs := d.publicConfigs.getV6(dc)
+	var addrs []Addr
+
 	addrs = append(addrs, defaultDCAddrSet.getV6(dc)...)
+	addrs = append(addrs, d.publicConfigs.getV6(dc)...)
 
 	return addrs
 }

mtglib/internal/doppel/scout.go

@@ -61,23 +61,29 @@ func (s Scout) learn(ctx context.Context, url string) (ScoutResult, error) {
 		client.CloseIdleConnections()
 	}
 
-	if err != nil || len(results.data) == 0 {
+	if err != nil {
 		return ScoutResult{}, err
 	}
 
+	data, writeIndex := results.Snapshot()
+
+	if len(data) == 0 {
+		return ScoutResult{}, nil
+	}
+
 	var result ScoutResult
 
 	// Compute inter-record durations (existing logic).
 	lastTimestamp := time.Time{}
 
-	for i, v := range results.data {
+	for i, v := range data {
 		if v.recordType != tls.TypeApplicationData {
 			continue
 		}
 
 		if lastTimestamp.IsZero() {
 			if i > 0 {
-				lastTimestamp = results.data[i-1].timestamp
+				lastTimestamp = data[i-1].timestamp
 			} else {
 				lastTimestamp = v.timestamp
 			}
@@ -90,12 +96,12 @@ func (s Scout) learn(ctx context.Context, url string) (ScoutResult, error) {
 	// Compute cert size: sum of ApplicationData payload between CCS and
 	// the first client Write (which marks the end of server handshake).
 	seenCCS := false
-	boundary := results.writeIndex
+	boundary := writeIndex
 	if boundary < 0 {
-		boundary = len(results.data)
+		boundary = len(data)
 	}
 
-	for i, v := range results.data {
+	for i, v := range data {
 		if i >= boundary {
 			break
 		}

mtglib/internal/doppel/scout_conn_collected.go

@@ -1,6 +1,10 @@
 package doppel
 
-import "time"
+import (
+	"slices"
+	"sync"
+	"time"
+)
 
 const (
 	ScoutConnCollectedPreallocSize = 100
@@ -13,23 +17,38 @@ type ScoutConnResult struct {
 }
 
 type ScoutConnCollected struct {
+	mu         sync.Mutex
 	data       []ScoutConnResult
 	writeIndex int // index at which client first wrote post-handshake data; -1 if not set
 }
 
 func (s *ScoutConnCollected) Add(record byte, payloadLen int) {
+	s.mu.Lock()
 	s.data = append(s.data, ScoutConnResult{
 		timestamp:  time.Now(),
 		recordType: record,
 		payloadLen: payloadLen,
 	})
+	s.mu.Unlock()
 }
 
 // MarkWrite records the current data length as the handshake boundary.
 func (s *ScoutConnCollected) MarkWrite() {
+	s.mu.Lock()
 	if s.writeIndex < 0 {
 		s.writeIndex = len(s.data)
 	}
+	s.mu.Unlock()
+}
+
+// Snapshot returns a copy of the collected data and the write index.
+func (s *ScoutConnCollected) Snapshot() ([]ScoutConnResult, int) {
+	s.mu.Lock()
+	snapshot := slices.Clone(s.data)
+	writeIndex := s.writeIndex
+	s.mu.Unlock()
+
+	return snapshot, writeIndex
 }
 
 func NewScoutConnCollected() *ScoutConnCollected {

mtglib/internal/doppel/scout_conn_collected_test.go

@@ -1,6 +1,7 @@
 package doppel
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -16,8 +17,10 @@ func (suite *ScoutConnCollectedTestSuite) TestAddSingle() {
 	collected := NewScoutConnCollected()
 	collected.Add(tls.TypeApplicationData, 100)
 
-	suite.Len(collected.data, 1)
-	suite.Equal(byte(tls.TypeApplicationData), collected.data[0].recordType)
+	data, _ := collected.Snapshot()
+
+	suite.Len(data, 1)
+	suite.Equal(byte(tls.TypeApplicationData), data[0].recordType)
 }
 
 func (suite *ScoutConnCollectedTestSuite) TestAddTimestampsAreMonotonic() {
@@ -31,11 +34,52 @@ func (suite *ScoutConnCollectedTestSuite) TestAddTimestampsAreMonotonic() {
 	time.Sleep(time.Microsecond)
 	collected.Add(tls.TypeApplicationData, 100)
 
-	for i := 1; i < len(collected.data); i++ {
-		suite.True(collected.data[i].timestamp.After(collected.data[i-1].timestamp))
+	data, _ := collected.Snapshot()
+
+	for i := 1; i < len(data); i++ {
+		suite.True(data[i].timestamp.After(data[i-1].timestamp))
 	}
 }
 
+func (suite *ScoutConnCollectedTestSuite) TestConcurrentAddSnapshot() {
+	collected := NewScoutConnCollected()
+
+	var wg sync.WaitGroup
+
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 1000; i++ {
+			collected.Add(tls.TypeApplicationData, i)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 100; i++ {
+			collected.MarkWrite()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 1000; i++ {
+			// call Snapshot concurrently to exercise the lock under -race
+			collected.Snapshot() //nolint:errcheck
+		}
+	}()
+
+	wg.Wait()
+
+	data, writeIndex := collected.Snapshot()
+	suite.Len(data, 1000)
+	suite.GreaterOrEqual(writeIndex, 0)
+}
+
 func TestScoutConnCollected(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ScoutConnCollectedTestSuite{})

mtglib/internal/tls/fake/client_side.go

@@ -11,8 +11,6 @@ import (
 	"net"
 	"slices"
 	"time"
-
-	"github.com/9seconds/mtg/v2/mtglib/internal/tls"
 )
 
 const (
@@ -56,25 +54,17 @@ func ReadClientHello(
 	//  4. New digest should be all 0 except of last 4 bytes
 	//  5. Last 4 bytes are little endian uint32 of UNIX timestamp when
 	//     this message was created.
-	handshakeCopyBuf := &bytes.Buffer{}
-	reader := io.TeeReader(conn, handshakeCopyBuf)
-
-	reader, err := parseTLSHeader(reader)
+	clientHelloCopy, handshakeReader, err := parseClientHello(conn)
 	if err != nil {
-		return nil, fmt.Errorf("cannot parse tls header: %w", err)
+		return nil, fmt.Errorf("cannot read client hello: %w", err)
 	}
 
-	reader, err = parseHandshakeHeader(reader)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse handshake header: %w", err)
-	}
-
-	hello, err := parseHandshake(reader)
+	hello, err := parseHandshake(handshakeReader)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse handshake: %w", err)
 	}
 
-	sniHostnames, err := parseSNI(reader)
+	sniHostnames, err := parseSNI(handshakeReader)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse SNI: %w", err)
 	}
@@ -85,10 +75,10 @@ func ReadClientHello(
 
 	digest := hmac.New(sha256.New, secret)
 	// we write a copy of the handshake with client random all nullified.
-	digest.Write(handshakeCopyBuf.Next(RandomOffset))
-	handshakeCopyBuf.Next(RandomLen)
+	digest.Write(clientHelloCopy.Next(RandomOffset))
+	clientHelloCopy.Next(RandomLen)
 	digest.Write(emptyRandom[:])
-	digest.Write(handshakeCopyBuf.Bytes())
+	digest.Write(clientHelloCopy.Bytes())
 
 	computed := digest.Sum(nil)
 
@@ -110,58 +100,6 @@ func ReadClientHello(
 	return hello, nil
 }
 
-func parseTLSHeader(r io.Reader) (io.Reader, error) {
-	// record_type(1) + version(2) + size(2)
-	//   16 - type is 0x16 (handshake record)
-	//   03 01 - protocol version is "3,1" (also known as TLS 1.0)
-	//   00 f8 - 0xF8 (248) bytes of handshake message follows
-	header := [1 + 2 + 2]byte{}
-
-	if _, err := io.ReadFull(r, header[:]); err != nil {
-		return nil, fmt.Errorf("cannot read record header: %w", err)
-	}
-
-	if header[0] != tls.TypeHandshake {
-		return nil, fmt.Errorf("unexpected record type %#x", header[0])
-	}
-
-	if header[1] != 3 || header[2] != 1 {
-		return nil, fmt.Errorf("unexpected protocol version %#x %#x", header[1], header[2])
-	}
-
-	length := int64(binary.BigEndian.Uint16(header[3:]))
-	buf := &bytes.Buffer{}
-
-	_, err := io.CopyN(buf, r, length)
-
-	return buf, err
-}
-
-func parseHandshakeHeader(r io.Reader) (io.Reader, error) {
-	// type(1) + size(3 / uint24)
-	// 01 - handshake message type 0x01 (client hello)
-	// 00 00 f4 - 0xF4 (244) bytes of client hello data follows
-	header := [1 + 3]byte{}
-
-	if _, err := io.ReadFull(r, header[:]); err != nil {
-		return nil, fmt.Errorf("cannot read handshake header: %w", err)
-	}
-
-	if header[0] != TypeHandshakeClient {
-		return nil, fmt.Errorf("incorrect handshake type: %#x", header[0])
-	}
-
-	// unfortunately there is not uint24 in golang, so we just reust header
-	header[0] = 0
-
-	length := int64(binary.BigEndian.Uint32(header[:]))
-	buf := &bytes.Buffer{}
-
-	_, err := io.CopyN(buf, r, length)
-
-	return buf, err
-}
-
 func parseHandshake(r io.Reader) (*ClientHello, error) {
 	//  A protocol version of "3,3" (meaning TLS 1.2) is given.
 	header := [2]byte{}

mtglib/internal/tls/fake/client_side_test.go

@@ -3,8 +3,10 @@ package fake_test
 import (
 	"bytes"
 	"encoding/binary"
+	"encoding/json"
 	"errors"
 	"io"
+	"os"
 	"testing"
 	"time"
 
@@ -393,3 +395,273 @@ func TestParseClientHelloSNI(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ParseClientHelloSNITestSuite{})
 }
+
+// fragmentTLSRecord splits a single TLS record into n TLS records by
+// dividing the payload into roughly equal parts. Each part gets its own
+// TLS record header with the same record type and version.
+func fragmentTLSRecord(t testing.TB, full []byte, n int) []byte {
+	t.Helper()
+
+	recordType := full[0]
+	version := full[1:3]
+	payload := full[tls.SizeHeader:]
+
+	chunkSize := len(payload) / n
+	result := &bytes.Buffer{}
+
+	for i := 0; i < n; i++ {
+		start := i * chunkSize
+		end := start + chunkSize
+
+		if i == n-1 {
+			end = len(payload)
+		}
+
+		chunk := payload[start:end]
+		result.WriteByte(recordType)
+		result.Write(version)
+		require.NoError(t, binary.Write(result, binary.BigEndian, uint16(len(chunk))))
+		result.Write(chunk)
+	}
+
+	return result.Bytes()
+}
+
+// splitPayloadAt creates two TLS records from a single record by splitting
+// the payload at the given byte position.
+func splitPayloadAt(t testing.TB, full []byte, pos int) []byte {
+	t.Helper()
+
+	payload := full[tls.SizeHeader:]
+	buf := &bytes.Buffer{}
+
+	buf.WriteByte(tls.TypeHandshake)
+	buf.Write(full[1:3])
+	require.NoError(t, binary.Write(buf, binary.BigEndian, uint16(pos)))
+	buf.Write(payload[:pos])
+
+	buf.WriteByte(tls.TypeHandshake)
+	buf.Write(full[1:3])
+	require.NoError(t, binary.Write(buf, binary.BigEndian, uint16(len(payload)-pos)))
+	buf.Write(payload[pos:])
+
+	return buf.Bytes()
+}
+
+type ParseClientHelloFragmentedTestSuite struct {
+	suite.Suite
+
+	secret   mtglib.Secret
+	snapshot *clientHelloSnapshot
+}
+
+func (s *ParseClientHelloFragmentedTestSuite) SetupSuite() {
+	parsed, err := mtglib.ParseSecret(
+		"ee367a189aee18fa31c190054efd4a8e9573746f726167652e676f6f676c65617069732e636f6d",
+	)
+	require.NoError(s.T(), err)
+
+	s.secret = parsed
+
+	fileData, err := os.ReadFile("testdata/client-hello-ok-19dfe38384b9884b.json")
+	require.NoError(s.T(), err)
+
+	s.snapshot = &clientHelloSnapshot{}
+	require.NoError(s.T(), json.Unmarshal(fileData, s.snapshot))
+}
+
+func (s *ParseClientHelloFragmentedTestSuite) makeConn(data []byte) *parseClientHelloConnMock {
+	readBuf := &bytes.Buffer{}
+	readBuf.Write(data)
+
+	connMock := &parseClientHelloConnMock{
+		readBuf: readBuf,
+	}
+
+	connMock.
+		On("SetReadDeadline", mock.AnythingOfType("time.Time")).
+		Twice().
+		Return(nil)
+
+	return connMock
+}
+
+func (s *ParseClientHelloFragmentedTestSuite) TestReassemblySuccess() {
+	full := s.snapshot.GetFull()
+
+	tests := []struct {
+		name string
+		data []byte
+	}{
+		{"two equal fragments", fragmentTLSRecord(s.T(), full, 2)},
+		{"three equal fragments", fragmentTLSRecord(s.T(), full, 3)},
+		{"single byte first fragment", splitPayloadAt(s.T(), full, 1)},
+		{"three byte first fragment", splitPayloadAt(s.T(), full, 3)},
+	}
+
+	for _, tt := range tests {
+		s.Run(tt.name, func() {
+			connMock := s.makeConn(tt.data)
+			defer connMock.AssertExpectations(s.T())
+
+			hello, err := fake.ReadClientHello(
+				connMock,
+				s.secret.Key[:],
+				s.secret.Host,
+				TolerateTime,
+			)
+			s.Require().NoError(err)
+
+			s.Equal(s.snapshot.GetRandom(), hello.Random[:])
+			s.Equal(s.snapshot.GetSessionID(), hello.SessionID)
+			s.Equal(uint16(s.snapshot.CipherSuite), hello.CipherSuite)
+		})
+	}
+}
+
+func (s *ParseClientHelloFragmentedTestSuite) TestReassemblyErrors() {
+	full := s.snapshot.GetFull()
+	payload := full[tls.SizeHeader:]
+
+	tests := []struct {
+		name      string
+		buildData func() []byte
+		errMsg    string
+	}{
+		{
+			name: "wrong continuation record type",
+			buildData: func() []byte {
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(10)))
+				buf.Write(payload[:10])
+				// Wrong type: application data instead of handshake
+				buf.WriteByte(tls.TypeApplicationData)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(len(payload)-10)))
+				buf.Write(payload[10:])
+				return buf.Bytes()
+			},
+			errMsg: "unexpected record type",
+		},
+		{
+			name: "too many continuation records",
+			buildData: func() []byte {
+				// Handshake header claiming 256 bytes, but we only send 1 byte per continuation
+				handshakePayload := []byte{0x01, 0x00, 0x01, 0x00}
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write([]byte{3, 1})
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(len(handshakePayload))))
+				buf.Write(handshakePayload)
+				for range 11 {
+					buf.WriteByte(tls.TypeHandshake)
+					buf.Write([]byte{3, 1})
+					require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(1)))
+					buf.WriteByte(0xAB)
+				}
+				return buf.Bytes()
+			},
+			errMsg: "too many fragments",
+		},
+		{
+			name: "zero-length continuation record",
+			buildData: func() []byte {
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(10)))
+				buf.Write(payload[:10])
+				// Valid header but zero-length payload
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(0)))
+				return buf.Bytes()
+			},
+			errMsg: "cannot read record header",
+		},
+		{
+			name: "wrong continuation record version",
+			buildData: func() []byte {
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(10)))
+				buf.Write(payload[:10])
+				// Wrong version: 3.3 instead of 3.1
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write([]byte{3, 3})
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(len(payload)-10)))
+				buf.Write(payload[10:])
+				return buf.Bytes()
+			},
+			errMsg: "unexpected protocol version",
+		},
+		{
+			name: "handshake message too large",
+			buildData: func() []byte {
+				// Handshake header claiming 0x010000 (65536) bytes — exceeds 0xFFFF limit
+				handshakePayload := []byte{0x01, 0x01, 0x00, 0x00}
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write([]byte{3, 1})
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(len(handshakePayload))))
+				buf.Write(handshakePayload)
+				return buf.Bytes()
+			},
+			errMsg: "cannot read record header",
+		},
+		{
+			name: "truncated continuation record header",
+			buildData: func() []byte {
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(10)))
+				buf.Write(payload[:10])
+				// Connection ends mid-header (only 2 bytes)
+				buf.WriteByte(tls.TypeHandshake)
+				buf.WriteByte(3)
+				return buf.Bytes()
+			},
+			errMsg: "cannot read record header",
+		},
+		{
+			name: "truncated continuation record payload",
+			buildData: func() []byte {
+				buf := &bytes.Buffer{}
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(10)))
+				buf.Write(payload[:10])
+				// Claims 100 bytes but no payload follows
+				buf.WriteByte(tls.TypeHandshake)
+				buf.Write(full[1:3])
+				require.NoError(s.T(), binary.Write(buf, binary.BigEndian, uint16(100)))
+				return buf.Bytes()
+			},
+			errMsg: "EOF",
+		},
+	}
+
+	for _, tt := range tests {
+		s.Run(tt.name, func() {
+			connMock := s.makeConn(tt.buildData())
+			defer connMock.AssertExpectations(s.T())
+
+			_, err := fake.ReadClientHello(
+				connMock,
+				s.secret.Key[:],
+				s.secret.Host,
+				TolerateTime,
+			)
+			s.ErrorContains(err, tt.errMsg)
+		})
+	}
+}
+
+func TestParseClientHelloFragmented(t *testing.T) {
+	t.Parallel()
+	suite.Run(t, &ParseClientHelloFragmentedTestSuite{})
+}

mtglib/internal/tls/fake/utils.go

@@ -0,0 +1,158 @@
+package fake
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/9seconds/mtg/v2/mtglib/internal/tls"
+)
+
+const (
+	maxFragmentsCount = 10
+)
+
+var ErrTooManyFragments = errors.New("too many fragments")
+
+// https://datatracker.ietf.org/doc/html/rfc5246#section-6.2.1
+// client hello can be fragmented in a series of packets:
+//
+//	Bytes on the wire:
+//
+// 16 03 01 00 F8 01 00 00 F4 03 03 [32 bytes random] [session_id] [ciphers] [SNI...]
+// ├─────────────┤├──────────────────────────────────────────────────────────────────┤
+//
+//	TLS record       Payload (248 bytes)
+//	header (5B)
+//
+//	16    = Handshake
+//	03 01 = TLS 1.0 (record layer version)
+//	00 F8 = 248 bytes follow
+//
+//	01       = ClientHello (handshake type)
+//	00 00 F4 = 244 bytes of handshake body
+//	03 03    = TLS 1.2 (actual protocol version)
+//	...rest of ClientHello...
+//
+// Fragmented record look like:
+//
+//	Record 1:
+//
+// 16 03 01 00 03 01 00 00
+// ├─────────────┤├──────┤
+//
+//	TLS header    3 bytes of payload
+//
+//	16    = Handshake
+//	03 01 = TLS 1.0
+//	00 03 = only 3 bytes follow
+//
+//	01       = ClientHello type
+//	00 00    = first 2 bytes of the uint24 length (INCOMPLETE!)
+//
+// Record 2:
+// 16 03 01 00 F5 F4 03 03 [32 bytes random] [session_id] [ciphers] [SNI...]
+// ├─────────────┤├────────────────────────────────────────────────────────────┤
+//
+//	TLS header    remaining 245 bytes of payload
+//
+//	16    = Handshake
+//	03 01 = TLS 1.0
+//	00 F5 = 245 bytes follow
+//
+//	F4    = last byte of uint24 length (now complete: 00 00 F4 = 244)
+//	03 03 = TLS 1.2
+//	...rest of ClientHello continues...
+//
+// So it means that there could be a series of handshake packets of different
+// lengths. The goal of this function is to concatenate these fragments.
+type fragmentedHandshakeReader struct {
+	r             io.Reader
+	buf           bytes.Buffer
+	readFragments int
+}
+
+func (f *fragmentedHandshakeReader) Read(p []byte) (int, error) {
+	if n, err := f.buf.Read(p); err == nil {
+		return n, nil
+	}
+
+	f.buf.Reset()
+
+	for f.buf.Len() == 0 {
+		if f.readFragments > maxFragmentsCount {
+			return 0, ErrTooManyFragments
+		}
+
+		if err := f.parseNextFragment(); err != nil {
+			return 0, err
+		}
+
+		f.readFragments++
+	}
+
+	return f.buf.Read(p)
+}
+
+func (f *fragmentedHandshakeReader) parseNextFragment() error {
+	// record_type(1) + version(2) + size(2)
+	//   16 - type is 0x16 (handshake record)
+	//   03 01 - protocol version is "3,1" (also known as TLS 1.0)
+	//   00 f8 - 0xF8 (248) bytes of handshake message follows
+	header := [1 + 2 + 2]byte{}
+
+	if _, err := io.ReadFull(f.r, header[:]); err != nil {
+		return fmt.Errorf("cannot read record header: %w", err)
+	}
+
+	if header[0] != tls.TypeHandshake {
+		return fmt.Errorf("unexpected record type %#x", header[0])
+	}
+
+	if header[1] != 3 || header[2] != 1 {
+		return fmt.Errorf("unexpected protocol version %#x %#x", header[1], header[2])
+	}
+
+	length := int64(binary.BigEndian.Uint16(header[3:]))
+	_, err := io.CopyN(&f.buf, f.r, length)
+
+	return err
+}
+
+func parseClientHello(r io.Reader) (*bytes.Buffer, *bytes.Buffer, error) {
+	r = &fragmentedHandshakeReader{r: r}
+	header := [1 + 3]byte{}
+
+	if _, err := io.ReadFull(r, header[:]); err != nil {
+		return nil, nil, fmt.Errorf("cannot read handshake header: %w", err)
+	}
+
+	if header[0] != TypeHandshakeClient {
+		return nil, nil, fmt.Errorf("incorrect handshake type: %#x", header[0])
+	}
+
+	// unfortunately there is not uint24 in golang, so we just reuse header
+	header[0] = 0
+	length := int64(binary.BigEndian.Uint32(header[:]))
+
+	clientHelloCopy := &bytes.Buffer{}
+	clientHelloCopy.Write([]byte{tls.TypeHandshake, 3, 1})
+	binary.Write( //nolint: errcheck
+		clientHelloCopy,
+		binary.BigEndian,
+		// 1 for handshake type
+		// 3 for handshake length
+		uint16(1+3+length),
+	)
+	clientHelloCopy.WriteByte(TypeHandshakeClient)
+	clientHelloCopy.Write(header[1:])
+
+	handshakeCopy := &bytes.Buffer{}
+	writer := io.MultiWriter(clientHelloCopy, handshakeCopy)
+
+	_, err := io.CopyN(writer, r, length)
+
+	return clientHelloCopy, handshakeCopy, err
+}

mtglib/proxy.go

@@ -102,11 +102,13 @@ func (p *Proxy) ServeConn(conn essentials.Conn) {
 		return
 	}
 
+	tracker := newIdleTracker(p.idleTimeout)
+
 	relay.Relay(
 		ctx,
 		ctx.logger.Named("relay"),
-		connIdleTimeout{Conn: ctx.telegramConn, timeout: p.idleTimeout},
-		connIdleTimeout{Conn: ctx.clientConn, timeout: p.idleTimeout},
+		connIdleTimeout{Conn: ctx.telegramConn, tracker: tracker},
+		connIdleTimeout{Conn: ctx.clientConn, tracker: tracker},
 	)
 }
 
@@ -305,11 +307,13 @@ func (p *Proxy) doDomainFronting(ctx *streamContext, conn *connRewind) {
 		stream:   p.eventStream,
 	}
 
+	tracker := newIdleTracker(p.idleTimeout)
+
 	relay.Relay(
 		ctx,
 		ctx.logger.Named("domain-fronting"),
-		connIdleTimeout{Conn: frontConn, timeout: p.idleTimeout},
-		connIdleTimeout{Conn: conn, timeout: p.idleTimeout},
+		connIdleTimeout{Conn: frontConn, tracker: tracker},
+		connIdleTimeout{Conn: conn, tracker: tracker},
 	)
 }
 

mtglib/proxy_opts.go

@@ -160,7 +160,6 @@ type ProxyOpts struct {
 
 	// DoppelGangerDRS defines if TLS Dynamic Record Sizing is active.
 	DoppelGangerDRS bool
-
 }
 
 func (p ProxyOpts) valid() error {

mtglib/proxy_test.go

@@ -175,7 +175,7 @@ func (suite *ProxyTestSuite) TestHTTPSRequest() {
 	addr := fmt.Sprintf("https://%s/headers", suite.ProxyAddress())
 
 	resp, err := client.Get(addr) //nolint: noctx
-	suite.NoError(err)
+	suite.Require().NoError(err)
 
 	defer resp.Body.Close() //nolint: errcheck