From 6f84a18b43da17a03fb5fdab2c0d235569a50f9e Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Tue, 22 Sep 2026 22:33:56 +0300 Subject: [PATCH] =?UTF-8?q?feat(tcp):=20=D0=B2=D0=B5=D1=80=D0=BD=D1=83?= =?UTF-8?q?=D1=82=D1=8C=20=D0=BF=D0=BE=D0=B4=D0=B4=D0=B5=D1=80=D0=B6=D0=BA?= =?UTF-8?q?=D1=83=20bond?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/server/bond_test.go | 21 ++ cmd/server/main.go | 21 +- docs/flags.md | 1 + docs/modes.md | 12 ++ docs/uri.md | 1 + internal/clientsdb/db.go | 7 +- internal/config/args.go | 1 + internal/config/bond_test.go | 54 +++++ internal/config/cliflags.go | 1 + internal/config/config.go | 1 + internal/config/json.go | 2 + internal/config/raw.go | 3 + internal/config/validate.go | 20 ++ internal/proxy/bond/bond_test.go | 300 ++++++++++++++++++++++++++ internal/proxy/bond/copy.go | 205 ++++++++++++++++++ internal/proxy/bond/frame.go | 99 +++++++++ internal/proxy/bond/server.go | 112 ++++++++++ internal/proxy/tcprelay/bond.go | 79 +++++++ internal/proxy/tcprelay/bond_test.go | 125 +++++++++++ internal/proxy/tcprelay/pool.go | 1 + internal/proxy/tcprelay/tcprelay.go | 16 +- internal/proxy/tcpserver/tcpserver.go | 20 +- internal/session/session.go | 1 + internal/uri/uri.go | 4 + mobile/README.md | 2 + 25 files changed, 1099 insertions(+), 10 deletions(-) create mode 100644 cmd/server/bond_test.go create mode 100644 internal/config/bond_test.go create mode 100644 internal/proxy/bond/bond_test.go create mode 100644 internal/proxy/bond/copy.go create mode 100644 internal/proxy/bond/frame.go create mode 100644 internal/proxy/bond/server.go create mode 100644 internal/proxy/tcprelay/bond.go create mode 100644 internal/proxy/tcprelay/bond_test.go diff --git a/cmd/server/bond_test.go b/cmd/server/bond_test.go new file mode 100644 index 0000000..62ac677 --- /dev/null +++ b/cmd/server/bond_test.go @@ -0,0 +1,21 @@ +package main + +import ( + "testing" + + "github.com/samosvalishe/free-turn-proxy/internal/clientsdb" + "github.com/samosvalishe/free-turn-proxy/internal/config" +) + +func TestAdmitTCPAndBond(t *testing.T) { + for _, mode := range []config.ProxyMode{config.ProxyModeTCP, config.ProxyModeUDP} { + cfg := &config.Server{Proxy: config.ProxyOpts{Mode: mode}} + for _, tag := range []byte{clientsdb.ModeUnset, clientsdb.ModeUDP, clientsdb.ModeTCP, clientsdb.ModeTCPBond, 255} { + err := admitClient(cfg, nil, "test-client", tag) + want := (mode == config.ProxyModeTCP && (tag == clientsdb.ModeTCP || tag == clientsdb.ModeTCPBond)) || (mode == config.ProxyModeUDP && (tag == clientsdb.ModeUnset || tag == clientsdb.ModeUDP)) + if (err == nil) != want { + t.Fatalf("mode=%s tag=%d err=%v", mode, tag, err) + } + } + } +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 83fe01e..3dd7af7 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -15,6 +15,7 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/clientsdb" "github.com/samosvalishe/free-turn-proxy/internal/config" "github.com/samosvalishe/free-turn-proxy/internal/logx" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/bond" "github.com/samosvalishe/free-turn-proxy/internal/proxy/tcpserver" "github.com/samosvalishe/free-turn-proxy/internal/proxy/udpserver" "github.com/samosvalishe/free-turn-proxy/internal/safego" @@ -124,6 +125,7 @@ func listen(cfg *config.Server, addr *net.UDPAddr, logger logx.Logger) (net.List } func serve(ctx context.Context, logger logx.Logger, listener net.Listener, db *clientsdb.DB, cfg *config.Server) { + bonds := &bond.Server{} var wg sync.WaitGroup var backoff time.Duration for { @@ -152,7 +154,7 @@ func serve(ctx context.Context, logger logx.Logger, listener net.Listener, db *c } backoff = 0 wg.Go(func() { - _ = safego.Run(logger, func() { handleAccepted(ctx, logger, db, conn, cfg) }) + _ = safego.Run(logger, func() { handleAccepted(ctx, logger, db, conn, cfg, bonds) }) }) } } @@ -180,13 +182,16 @@ func wireMode(m config.ProxyMode) byte { } func modeName(b byte) string { + if b == clientsdb.ModeTCPBond { + return "tcp-bond" + } if b == clientsdb.ModeTCP { return string(config.ProxyModeTCP) } return string(config.ProxyModeUDP) } -func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, conn net.Conn, cfg *config.Server) { +func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, conn net.Conn, cfg *config.Server, bonds *bond.Server) { defer func() { if closeErr := conn.Close(); closeErr != nil { logger.Warnf("failed to close incoming connection: %s", closeErr) @@ -211,7 +216,9 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c } logger.Debugf("Handshake done") + var clientMode byte clientID, data, err := clientsdb.AcceptClientID(dtlsConn, func(id string, mode byte) error { + clientMode = mode return admitClient(cfg, db, id, mode) }) if err != nil { @@ -220,9 +227,12 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c } logger.Infof("Session up: client=%s from=%s", clientID, conn.RemoteAddr()) - if cfg.Proxy.Mode == config.ProxyModeTCP { + switch { + case clientMode == clientsdb.ModeTCPBond: + tcpserver.HandleBond(ctx, logger, data, cfg.Proxy.Connect, cfg.KCP.Profile, bonds, clientID) + case cfg.Proxy.Mode == config.ProxyModeTCP: tcpserver.Handle(ctx, logger, data, cfg.Proxy.Connect, cfg.KCP.Profile) - } else { + default: udpserver.Handle(ctx, logger, data, cfg.Proxy.Connect) } logger.Infof("Session down: client=%s from=%s", clientID, conn.RemoteAddr()) @@ -230,6 +240,9 @@ func handleAccepted(ctx context.Context, logger logx.Logger, db *clientsdb.DB, c // admitClient: режим клиента обязан совпасть с сервером, ID - быть в allowlist (если он есть). func admitClient(cfg *config.Server, db *clientsdb.DB, id string, mode byte) error { + if mode == clientsdb.ModeTCPBond { + mode = clientsdb.ModeTCP + } if want := wireMode(cfg.Proxy.Mode); mode != clientsdb.ModeUnset && mode != want { return fmt.Errorf("mode mismatch: клиент %s, сервер %s - приведите -mode к одному значению", modeName(mode), cfg.Proxy.Mode) diff --git a/docs/flags.md b/docs/flags.md index 68d7f4f..dd50f06 100644 --- a/docs/flags.md +++ b/docs/flags.md @@ -12,6 +12,7 @@ | `-n` | `12` (`direct`: `1`) | параллельных потоков; в `-mode tcp` это число сессий пула, по которым round-robin раскладываются локальные TCP-соединения | | `-transport` | `tcp` | транспорт до TURN-реле: `tcp` (TCP/TLS) \| `udp` | | `-mode` | `udp` | режим туннеля: `udp` (UDP-релей для WireGuard) \| `tcp` (TCP-форвардер для Xray/sing-box) | +| `-bond` | `false` | объединять сессии для одного TCP-соединения; только `-mode tcp`, до 256 сессий суммарно; нужен обновлённый сервер | | `-turn` | из creds | переопределить IP TURN-сервера | | `-port` | из creds | переопределить порт TURN-сервера | | `-obf-profile` | `none` | wire-профиль обфускации payload: `none` \| `rtpopus` (RTP/opus + ChaCha20-Poly1305 AEAD) \| `rtpopus2` (+ RTP header extension, ближе к WebRTC) \| `rtpopus3` (+ abs-send-time, VAD, имитация потерь, вариативный timestamp); должен совпадать с сервером | diff --git a/docs/modes.md b/docs/modes.md index 3e8ac29..2104385 100644 --- a/docs/modes.md +++ b/docs/modes.md @@ -32,6 +32,18 @@ Relayed-данные TURN - всегда датаграммы, даже при ` `-n` в этом режиме задаёт число сессий пула: каждое локальное TCP-соединение уходит целиком в одну из них (round-robin). +### Bond для одного TCP-соединения + +Добавьте клиенту `-bond`: данные каждого TCP-соединения будут распределяться между доступными сессиями. На сервере достаточно `-mode tcp`, он одновременно принимает обычные и bond-клиенты. По умолчанию bond выключен. + +```bash +./client -listen 127.0.0.1:9000 -peer :56000 -links "" -mode tcp -bond -n 4 +``` + +Состав каналов фиксируется при открытии TCP-соединения; с одной доступной сессией bond тоже работает. Поддерживается до 256 сессий суммарно по всем ссылкам. Новые сессии используются следующими соединениями. При отказе участвующей сессии текущее соединение закрывается, а Xray может открыть новое через оставшиеся. Бесшовного переноса уже переданных данных между каналами нет. Медленный канал задерживает сборку потока; прирост скорости зависит от сети. + +Для bond обновите и клиент, и сервер: сервер без его поддержки отклонит такой клиент при проверке режима. Обычный TCP сохраняет прежний протокол. В JSON клиента настройка называется `proxy.bond`, в `freeturn://` - `bond`. + Встроенный WireGuard (`-tunnel`) с `-mode tcp` не сочетается: он гонит датаграммы. Режим должен совпадать на обеих сторонах: клиент помечает им первую запись, и сервер с другим `-mode` закрывает соединение с явной ошибкой вместо молчаливого зависания. diff --git a/docs/uri.md b/docs/uri.md index efd8d1a..4b1237c 100644 --- a/docs/uri.md +++ b/docs/uri.md @@ -26,6 +26,7 @@ Payload - JSON-объект, закодированный `base64url` (без pa | `peer` | `-peer` | адрес сервера на VPS (`ip:port`). | | `transport` | `-transport` | транспорт до TURN-реле: `tcp` \| `udp`. | | `mode` | `-mode` | режим туннеля: `udp` \| `tcp`; отсутствует - `udp`. | +| `bond` | `-bond` | объединять сессии для одного TCP-соединения (`true`); только `mode: tcp`, по умолчанию выключено. | | `obf` | `-obf-profile` | профиль обфускации (`rtpopus` \| `rtpopus2` \| `rtpopus3`); `none` опускается. | | `key` | `-obf-key` | ключ обфускации (hex), только при заданном `obf`. | | `n` | `-n` | число TURN-потоков; отсутствует - дефолт клиента. | diff --git a/internal/clientsdb/db.go b/internal/clientsdb/db.go index 0c5174d..2a860bf 100644 --- a/internal/clientsdb/db.go +++ b/internal/clientsdb/db.go @@ -190,9 +190,10 @@ func syncDir(dir string) { } const ( - ModeUnset byte = 0 - ModeUDP byte = 1 - ModeTCP byte = 2 + ModeUnset byte = 0 + ModeUDP byte = 1 + ModeTCP byte = 2 + ModeTCPBond byte = 3 // idVersionAck - клиент ждёт подтверждения ID. idVersionAck byte = 2 diff --git a/internal/config/args.go b/internal/config/args.go index afd2c37..4ac35d7 100644 --- a/internal/config/args.go +++ b/internal/config/args.go @@ -30,6 +30,7 @@ func ClientArgs(c *Client) []string { addIf(c.VK.StreamsPerCred != def.VK.StreamsPerCred, "-streams-per-cred", strconv.Itoa(c.VK.StreamsPerCred)) addIf(c.TURN.TransportUDP != def.TURN.TransportUDP, "-transport", TransportUDP) addIf(c.Proxy.Mode != def.Proxy.Mode, "-mode", string(c.Proxy.Mode)) + addIf(c.Proxy.Bond, "-bond") args = append(args, kcpArgs(c.KCP.Profile, def.KCP.Profile)...) addIf(c.Obf.Enabled(), "-obf-profile", string(c.Obf.Profile)) addIf(c.Obf.Enabled(), "-obf-key", hex.EncodeToString(c.Obf.Key)) diff --git a/internal/config/bond_test.go b/internal/config/bond_test.go new file mode 100644 index 0000000..dffc2b8 --- /dev/null +++ b/internal/config/bond_test.go @@ -0,0 +1,54 @@ +package config + +import ( + "errors" + "io" + "strings" + "testing" + + "github.com/samosvalishe/free-turn-proxy/internal/uri" +) + +func TestBondConfigRoundTrip(t *testing.T) { + args := []string{"-provider", "direct", "-peer", "127.0.0.1:56000", "-mode", "tcp", "-bond", "-n", "3"} + c, err := ParseClient(args, io.Discard) + if err != nil { + t.Fatal(err) + } + round, err := ParseClient(ClientArgs(c), io.Discard) + if err != nil || !round.Proxy.Bond { + t.Fatalf("args round trip: %v", err) + } + link := (&uri.Config{Provider: "direct", Peer: "127.0.0.1:56000", Mode: "tcp", Bond: true, N: 3}).String() + parsed, err := uri.Parse(link) + if err != nil || !parsed.Bond { + t.Fatalf("URI round trip: %v", err) + } + c, err = ParseClient([]string{link}, io.Discard) + if err != nil || !c.Proxy.Bond { + t.Fatalf("URI to CLI: %v", err) + } + c, err = ParseClientJSON([]byte(`{"provider":"direct","peer":"127.0.0.1:56000","proxy":{"mode":"tcp","bond":true}}`), "") + if err != nil || !c.Proxy.Bond { + t.Fatalf("JSON: %v", err) + } + c, err = ParseClientJSON([]byte(`{}`), link) + if err != nil || !c.Proxy.Bond { + t.Fatalf("URI overlay: %v", err) + } + if strings.Contains(DefaultClientJSON(), `"bond"`) { + t.Fatal("default JSON changed") + } +} + +func TestBondValidation(t *testing.T) { + for _, extra := range [][]string{{}, {"-mode", "tcp", "-n", "257"}} { + args := append([]string{"-provider", "direct", "-peer", "127.0.0.1:56000", "-bond"}, extra...) + if _, err := ParseClient(args, io.Discard); !errors.Is(err, ErrBondConfig) { + t.Fatalf("accepted %v: %v", args, err) + } + } + if _, err := ParseClientJSON([]byte(`{"provider":"direct","peer":"127.0.0.1:1","proxy":{"mode":"tcp","bonds":true}}`), ""); err == nil { + t.Fatal("unknown JSON field accepted") + } +} diff --git a/internal/config/cliflags.go b/internal/config/cliflags.go index 53ce7b2..ae95e6b 100644 --- a/internal/config/cliflags.go +++ b/internal/config/cliflags.go @@ -47,6 +47,7 @@ func ParseClient(args []string, errOut io.Writer) (*Client, error) { fs.IntVar(&r.N, "n", r.N, "число параллельных потоков; 0 - по провайдеру (vk 12, direct 1)") fs.StringVar(&r.Transport, "transport", r.Transport, "транспорт до TURN-реле: tcp | udp") fs.StringVar(&r.Mode, "mode", r.Mode, "режим туннеля: udp (WireGuard) | tcp (Xray/sing-box)") + fs.BoolVar(&r.Bond, "bond", r.Bond, "объединять сессии для одного TCP-соединения; только -mode tcp") fs.StringVar(&r.ObfProfile, "obf-profile", r.ObfProfile, "wire-профиль обфускации: none | rtpopus | rtpopus2 | rtpopus3; должен совпадать с сервером") fs.StringVar(&r.ObfKey, "obf-key", r.ObfKey, "ключ для -obf-profile != none: 32 байта hex (64 символа)") fs.BoolVar(&r.GenObfKey, "gen-obf-key", r.GenObfKey, "напечатать новый -obf-key и выйти") diff --git a/internal/config/config.go b/internal/config/config.go index b0683da..b8ba665 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -42,6 +42,7 @@ const ( ) type ProxyOpts struct { + Bond bool Mode ProxyMode Listen string Connect string diff --git a/internal/config/json.go b/internal/config/json.go index 54cc717..0ac59ef 100644 --- a/internal/config/json.go +++ b/internal/config/json.go @@ -43,6 +43,7 @@ type turnJSON struct { } type proxyJSON struct { + Bond bool `json:"bond,omitempty"` Mode string `json:"mode"` Listen string `json:"listen"` } @@ -167,6 +168,7 @@ func (j ClientJSON) toRaw() raw { StreamsPerCred: j.VK.StreamsPerCred, Transport: j.TURN.Transport, Mode: j.Proxy.Mode, + Bond: j.Proxy.Bond, ObfProfile: j.Obf.Profile, ObfKey: j.Obf.Key, diff --git a/internal/config/raw.go b/internal/config/raw.go index 70600a7..99b86aa 100644 --- a/internal/config/raw.go +++ b/internal/config/raw.go @@ -26,6 +26,7 @@ type raw struct { StreamsPerCred int Transport string Mode string + Bond bool ObfProfile string ObfKey string @@ -55,6 +56,7 @@ func (r *raw) applyURI(u *uri.Config) { override(&r.Provider, u.Provider) override(&r.Transport, u.Transport) override(&r.Mode, u.Mode) + r.Bond = r.Bond || u.Bond if u.KCP != nil { r.KCP = kcpmux.Profile{ NoDelay: u.KCP.NoDelay, @@ -142,6 +144,7 @@ func assemble(r raw) (*Client, error) { Timing: r.ObfTiming, }, Proxy: ProxyOpts{ + Bond: r.Bond, Mode: ProxyMode(r.Mode), Listen: r.Listen, Peer: r.Peer, diff --git a/internal/config/validate.go b/internal/config/validate.go index 6dc1cf8..e3db2bc 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/bond" "github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux" "github.com/samosvalishe/free-turn-proxy/internal/tunnel" ) @@ -30,6 +31,9 @@ func Validate(c *Client) error { if err := validateProxyMode(c.Proxy.Mode); err != nil { return err } + if err := validateBond(c); err != nil { + return err + } if err := validateTunnel(c.Tunnel, c.Proxy.Mode); err != nil { return err } @@ -39,6 +43,22 @@ func Validate(c *Client) error { return validateObfTiming(c.Obf) } +var ErrBondConfig = errors.New("invalid bond configuration") + +func validateBond(c *Client) error { + if !c.Proxy.Bond { + return nil + } + groups := 1 + if c.Provider.Name == ProviderVK { + groups = max(len(c.VK.Links), 1) + } + if c.Proxy.Mode != ProxyModeTCP || c.TURN.N > bond.MaxLanes/groups { + return fmt.Errorf("%w: -bond requires -mode tcp and at most %d sessions in total", ErrBondConfig, bond.MaxLanes) + } + return nil +} + func validateProvider(c *Client) error { switch c.Provider.Name { case ProviderDirect: diff --git a/internal/proxy/bond/bond_test.go b/internal/proxy/bond/bond_test.go new file mode 100644 index 0000000..86cdb3e --- /dev/null +++ b/internal/proxy/bond/bond_test.go @@ -0,0 +1,300 @@ +package bond + +import ( + "bytes" + "context" + "encoding/hex" + "errors" + "io" + "net" + "strconv" + "sync" + "testing" + "time" + + "github.com/samosvalishe/free-turn-proxy/internal/stats" + "github.com/xtaci/smux" +) + +func tcpPair(t *testing.T) (*net.TCPConn, net.Conn) { + t.Helper() + ln, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer func() { _ = ln.Close() }() + a, err := (&net.Dialer{}).DialContext(t.Context(), "tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + b, err := ln.Accept() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = a.Close(); _ = b.Close() }) + if err := a.SetDeadline(time.Now().Add(15 * time.Second)); err != nil { + t.Fatal(err) + } + tcp, ok := a.(*net.TCPConn) + if !ok { + t.Fatal("expected TCP") + } + return tcp, b +} + +func TestSessionFailureAfterFINStopsCopy(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + a, b := net.Pipe() + clientSession, err := smux.Client(a, nil) + if err != nil { + t.Fatal(err) + } + defer func() { _ = clientSession.Close() }() + serverSession, err := smux.Server(b, nil) + if err != nil { + t.Fatal(err) + } + defer func() { _ = serverSession.Close() }() + lane, err := clientSession.OpenStream() + if err != nil { + t.Fatal(err) + } + remote, err := serverSession.AcceptStream() + if err != nil { + t.Fatal(err) + } + client, local := tcpPair(t) + done := make(chan error, 1) + go func() { done <- Copy(ctx, local, []net.Conn{lane}, nil) }() + if err = writeFrame(remote, frameFIN, 0, nil); err != nil { + t.Fatal(err) + } + if _, err = client.Read(make([]byte, 1)); !errors.Is(err, io.EOF) { + t.Fatalf("half-close: %v", err) + } + _ = serverSession.Close() + select { + case <-done: + case <-time.After(time.Second): + cancel() + <-done + t.Fatal("session failure after FIN left local read blocked") + } +} + +type delayedRead struct { + net.Conn + once sync.Once +} + +type blockedWriter struct { + net.Conn + writing chan struct{} +} + +func (w *blockedWriter) Write(b []byte) (int, error) { + close(w.writing) + return w.Conn.Write(b) +} + +func TestLaneFailureUnblocksLocalWrite(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + a, b := net.Pipe() + defer func() { _ = b.Close() }() + local := &blockedWriter{Conn: a, writing: make(chan struct{})} + lane, remote := net.Pipe() + defer func() { _ = remote.Close() }() + done := make(chan error, 1) + go func() { done <- Copy(ctx, local, []net.Conn{lane}, nil) }() + if err := writeFrame(remote, frameData, 0, []byte("blocked")); err != nil { + t.Fatal(err) + } + <-local.writing + _ = remote.Close() + select { + case <-done: + case <-time.After(time.Second): + cancel() + <-done + t.Fatal("lane failure left local write blocked") + } +} + +func (d *delayedRead) Read(p []byte) (int, error) { + d.once.Do(func() { time.Sleep(30 * time.Millisecond) }) + return d.Conn.Read(p) +} + +func TestCopyReordersAndPreservesHalfClose(t *testing.T) { + for _, count := range []int{1, 3} { + t.Run(strconv.Itoa(count), func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) + defer cancel() + client, clientLocal := tcpPair(t) + backend, serverLocal := tcpPair(t) + left, right := make([]net.Conn, count), make([]net.Conn, count) + for i := range left { + left[i], right[i] = net.Pipe() + } + right[0] = &delayedRead{Conn: right[0]} + traffic := stats.New(true) + done := make(chan error, 2) + go func() { done <- Copy(ctx, clientLocal, left, traffic) }() + go func() { done <- Copy(ctx, serverLocal, right, nil) }() + request := bytes.Repeat([]byte("request contents"), 128*1024) + response := bytes.Repeat([]byte("response after EOF"), 128*1024) + writer := make(chan error, 1) + go func() { + _, err := client.Write(request) + writer <- errors.Join(err, client.CloseWrite()) + }() + got, err := io.ReadAll(backend) + if err != nil || !bytes.Equal(got, request) { + t.Fatalf("request: len=%d err=%v", len(got), err) + } + if err = <-writer; err != nil { + t.Fatal(err) + } + go func() { + _, werr := backend.Write(response) + writer <- errors.Join(werr, backend.CloseWrite()) + }() + got, err = io.ReadAll(client) + if err != nil || !bytes.Equal(got, response) { + t.Fatalf("response: len=%d err=%v", len(got), err) + } + if err := <-writer; err != nil { + t.Fatal(err) + } + for range 2 { + if err := <-done; err != nil { + t.Fatal(err) + } + } + tx, rx := traffic.Counters() + if tx != uint64(len(request)) || rx != uint64(len(response)) { + t.Fatalf("traffic=%d/%d", tx, rx) + } + }) + } +} + +func TestCopyLaneFailureAndCancel(t *testing.T) { + for _, failure := range []bool{false, true} { + t.Run(map[bool]string{false: "cancel", true: "lane failure"}[failure], func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + local, peer := net.Pipe() + defer func() { _ = peer.Close() }() + lanes, remote := make([]net.Conn, 3), make([]net.Conn, 3) + for i := range lanes { + lanes[i], remote[i] = net.Pipe() + defer func() { _ = remote[i].Close() }() + } + done := make(chan error, 1) + go func() { done <- Copy(ctx, local, lanes, nil) }() + if failure { + _ = remote[1].Close() + } else { + cancel() + } + select { + case err := <-done: + if err == nil { + t.Fatal("expected failure") + } + case <-time.After(3 * time.Second): + t.Fatal("copy did not stop") + } + if _, err := peer.Read(make([]byte, 1)); !errors.Is(err, io.EOF) { + t.Fatalf("local not closed: %v", err) + } + }) + } +} + +func TestHelloWireAndValidation(t *testing.T) { + var b bytes.Buffer + h := Hello{ID: 0x0102030405060708, Index: 1, Count: 3} + if err := WriteHello(&b, h); err != nil { + t.Fatal(err) + } + if got := hex.EncodeToString(b.Bytes()); got != "564c423101010203040506070800010003" { + t.Fatal(got) + } + got, err := ReadHello(&b) + if err != nil || got != h { + t.Fatalf("hello=%+v, %v", got, err) + } + for _, invalid := range []Hello{{Count: 0}, {Count: MaxLanes + 1}, {Count: 2, Index: 2}} { + b.Reset() + if err := WriteHello(&b, invalid); err != nil { + t.Fatal(err) + } + if _, err := ReadHello(&b); !errors.Is(err, ErrProtocol) { + t.Fatalf("accepted %+v: %v", invalid, err) + } + } +} + +func TestReorderRejectsInvalidSequences(t *testing.T) { + for name, frames := range map[string][]frame{ + "gap": {{typ: frameFIN, seq: 1}}, + "duplicate": {{typ: frameData, seq: 1}, {typ: frameData, seq: 1}}, + "overflow": {{typ: frameData, seq: pendingCap + 1}}, + "conflicting FIN": {{typ: frameData, seq: 1}, {typ: frameFIN, seq: 0}}, + } { + t.Run(name, func(t *testing.T) { + ch := make(chan received, len(frames)) + for _, f := range frames { + ch <- received{frame: f} + } + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + local, peer := net.Pipe() + defer func() { _ = local.Close(); _ = peer.Close() }() + if err := reorder(ctx, local, ch, 1, nil); !errors.Is(err, ErrProtocol) { + t.Fatalf("error=%v", err) + } + }) + } +} + +func TestServerSeparatesClientsAndRejectsDuplicateLane(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + s := &Server{} + laneA, remoteA := net.Pipe() + laneB, remoteB := net.Pipe() + defer func() { _ = remoteA.Close(); _ = remoteB.Close() }() + h := Hello{ID: 7, Index: 0, Count: 2} + key := connectionKey{client: "a", id: h.ID} + a, err := s.attach(ctx, key, h, laneA, "127.0.0.1:1") + if err != nil { + t.Fatal(err) + } + b, err := s.attach(ctx, connectionKey{client: "b", id: h.ID}, h, laneB, "127.0.0.1:1") + if err != nil { + t.Fatal(err) + } + if a == b { + t.Fatal("different clients share group") + } + if _, err := s.attach(ctx, key, h, laneA, "127.0.0.1:1"); !errors.Is(err, ErrProtocol) { + t.Fatal("duplicate accepted") + } + cancel() + for _, g := range []*group{a, b} { + select { + case <-g.done: + case <-time.After(3 * time.Second): + t.Fatal("group did not stop") + } + } + s.mu.Lock() + defer s.mu.Unlock() + if len(s.groups) != 0 { + t.Fatal("groups leaked") + } +} diff --git a/internal/proxy/bond/copy.go b/internal/proxy/bond/copy.go new file mode 100644 index 0000000..7b92f2a --- /dev/null +++ b/internal/proxy/bond/copy.go @@ -0,0 +1,205 @@ +package bond + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "sync" + + "github.com/samosvalishe/free-turn-proxy/internal/stats" +) + +// Copy владеет local и lanes; ошибка любого направления закрывает всё соединение. +func Copy(ctx context.Context, local net.Conn, lanes []net.Conn, traffic *stats.Stats) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + stop := context.AfterFunc(ctx, func() { + _ = local.Close() + for _, lane := range lanes { + _ = lane.Close() + } + }) + defer stop() + defer func() { + _ = local.Close() + for _, lane := range lanes { + _ = lane.Close() + } + }() + if len(lanes) == 0 || len(lanes) > MaxLanes { + return fmt.Errorf("%w: lane count", ErrProtocol) + } + results := make(chan error, 2) + go func() { results <- send(local, lanes, traffic) }() + var receiver sync.WaitGroup + receiver.Go(func() { + err := receive(ctx, local, lanes, traffic, cancel) + results <- err + if err == nil { + watchClosure(lanes, cancel) + } + }) + defer func() { cancel(); receiver.Wait() }() + err := <-results + if err != nil { + cancel() + } + return errors.Join(err, <-results) +} + +// После FIN чтение EOF/ошибки всё ещё должно прерывать заблокированный local.Read. +func watchClosure(lanes []net.Conn, cancel context.CancelFunc) { + var wg sync.WaitGroup + for _, lane := range lanes { + wg.Go(func() { + var b [1]byte + _, _ = lane.Read(b[:]) + cancel() + }) + } + wg.Wait() +} + +func send(local net.Conn, lanes []net.Conn, traffic *stats.Stats) error { + buf := make([]byte, maxChunk) + var seq uint64 + for { + n, err := local.Read(buf) + if n > 0 { + if werr := writeFrame(lanes[seq%uint64(len(lanes))], frameData, seq, buf[:n]); werr != nil { + return werr + } + seq++ + if traffic != nil { + traffic.AddTx(n) + } + } + if errors.Is(err, io.EOF) { + for _, lane := range lanes { + if werr := writeFrame(lane, frameFIN, seq, nil); werr != nil { + return werr + } + } + return nil + } + if err != nil { + return fmt.Errorf("bond local read: %w", err) + } + } +} + +type received struct { + frame + err error +} + +func receive(ctx context.Context, local net.Conn, lanes []net.Conn, traffic *stats.Stats, abort context.CancelFunc) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + frames := make(chan received, len(lanes)) + var readers sync.WaitGroup + for _, lane := range lanes { + readers.Go(func() { readLane(ctx, lane, frames, abort) }) + } + err := reorder(ctx, local, frames, len(lanes), traffic) + cancel() + // При ошибке чтения другие lanes могут ждать данных бесконечно. + if err != nil { + for _, lane := range lanes { + _ = lane.Close() + } + } + readers.Wait() + return err +} + +func readLane(ctx context.Context, lane net.Conn, frames chan<- received, abort context.CancelFunc) { + for { + f, err := readFrame(lane) + if err != nil { + // Получатель может стоять в local.Write, не читая frames. + abort() + } + select { + case frames <- received{frame: f, err: err}: + case <-ctx.Done(): + return + } + if err != nil || f.typ == frameFIN { + return + } + } +} + +func reorder(ctx context.Context, local net.Conn, frames <-chan received, count int, traffic *stats.Stats) error { + pending := make(map[uint64][]byte) + var expect, end uint64 + var fins int + for { + select { + case <-ctx.Done(): + return fmt.Errorf("bond receive: %w", ctx.Err()) + case f := <-frames: + if f.err != nil { + return f.err + } + if f.typ == frameFIN { + if f.seq < expect || (fins > 0 && end != f.seq) { + return fmt.Errorf("%w: conflicting FIN", ErrProtocol) + } + end = f.seq + fins++ + } else { + if err := queueFrame(pending, f.frame, expect, end, fins); err != nil { + return err + } + } + if err := drain(local, pending, &expect, traffic); err != nil { + return err + } + if fins == count { + if expect != end || len(pending) != 0 { + return fmt.Errorf("%w: missing data before FIN", ErrProtocol) + } + return closeWrite(local) + } + } + } +} + +func closeWrite(local net.Conn) error { + if cw, ok := local.(interface{ CloseWrite() error }); ok { + if err := cw.CloseWrite(); err != nil { + return fmt.Errorf("bond half-close: %w", err) + } + } + return nil +} + +func queueFrame(pending map[uint64][]byte, f frame, expect, end uint64, fins int) error { + _, duplicate := pending[f.seq] + if f.seq < expect || duplicate || f.seq-expect > pendingCap || (fins > 0 && f.seq >= end) { + return fmt.Errorf("%w: data sequence", ErrProtocol) + } + pending[f.seq] = f.data + return nil +} + +func drain(local net.Conn, pending map[uint64][]byte, expect *uint64, traffic *stats.Stats) error { + for { + data, ok := pending[*expect] + if !ok { + return nil + } + if err := writeFull(local, data); err != nil { + return err + } + if traffic != nil { + traffic.AddRx(len(data)) + } + delete(pending, *expect) + *expect++ + } +} diff --git a/internal/proxy/bond/frame.go b/internal/proxy/bond/frame.go new file mode 100644 index 0000000..ae701c3 --- /dev/null +++ b/internal/proxy/bond/frame.go @@ -0,0 +1,99 @@ +// Package bond объединяет smux-потоки разных сессий в одно TCP-соединение. +package bond + +import ( + "encoding/binary" + "errors" + "fmt" + "io" + "time" +) + +const ( + MaxLanes = 256 + SetupTimeout = 10 * time.Second + maxChunk = 16 * 1024 + pendingCap = 256 + frameData byte = 1 + frameFIN byte = 2 +) + +var ErrProtocol = errors.New("bond: invalid protocol") + +// Сохранён формат VLB1: magic(4), version(1), connection(8), index(2), count(2); big endian. +type Hello struct { + ID uint64 + Index uint16 + Count uint16 +} + +func WriteHello(w io.Writer, h Hello) error { + var b [17]byte + copy(b[:], "VLB1") + b[4] = 1 + binary.BigEndian.PutUint64(b[5:13], h.ID) + binary.BigEndian.PutUint16(b[13:15], h.Index) + binary.BigEndian.PutUint16(b[15:17], h.Count) + return writeFull(w, b[:]) +} + +func ReadHello(r io.Reader) (Hello, error) { + var b [17]byte + if _, err := io.ReadFull(r, b[:]); err != nil { + return Hello{}, fmt.Errorf("bond hello: %w", err) + } + h := Hello{ID: binary.BigEndian.Uint64(b[5:13]), Index: binary.BigEndian.Uint16(b[13:15]), Count: binary.BigEndian.Uint16(b[15:17])} + if string(b[:4]) != "VLB1" || b[4] != 1 || h.Count == 0 || h.Count > MaxLanes || h.Index >= h.Count { + return Hello{}, fmt.Errorf("%w: hello", ErrProtocol) + } + return h, nil +} + +// Кадр VLB1: type(1), sequence(8), size(4), payload(size); FIN обозначает следующий sequence. +type frame struct { + typ byte + seq uint64 + data []byte +} + +func writeFrame(w io.Writer, typ byte, seq uint64, data []byte) error { + var b [13]byte + b[0] = typ + binary.BigEndian.PutUint64(b[1:9], seq) + binary.BigEndian.PutUint32(b[9:13], uint32(len(data))) //nolint:gosec // payload ограничен maxChunk + if err := writeFull(w, b[:]); err != nil { + return err + } + return writeFull(w, data) +} + +func readFrame(r io.Reader) (frame, error) { + var b [13]byte + if _, err := io.ReadFull(r, b[:]); err != nil { + return frame{}, fmt.Errorf("bond frame: %w", err) + } + size := binary.BigEndian.Uint32(b[9:13]) + f := frame{typ: b[0], seq: binary.BigEndian.Uint64(b[1:9])} + if (f.typ != frameData && f.typ != frameFIN) || size > maxChunk || (f.typ == frameFIN && size != 0) { + return frame{}, fmt.Errorf("%w: frame", ErrProtocol) + } + f.data = make([]byte, size) + if _, err := io.ReadFull(r, f.data); err != nil { + return frame{}, fmt.Errorf("bond payload: %w", err) + } + return f, nil +} + +func writeFull(w io.Writer, b []byte) error { + if len(b) == 0 { + return nil + } + n, err := w.Write(b) + if err != nil { + return fmt.Errorf("bond write: %w", err) + } + if n != len(b) { + return fmt.Errorf("bond write: %w", io.ErrShortWrite) + } + return nil +} diff --git a/internal/proxy/bond/server.go b/internal/proxy/bond/server.go new file mode 100644 index 0000000..8a7b371 --- /dev/null +++ b/internal/proxy/bond/server.go @@ -0,0 +1,112 @@ +package bond + +import ( + "context" + "fmt" + "net" + "sync" + "time" +) + +type connectionKey struct { + client string + id uint64 +} + +type group struct { + lanes []net.Conn + attached int + ready chan struct{} + done chan struct{} + cancel context.CancelFunc + err error +} + +// Server общий для всех DTLS-сессий одного listener; Client ID изолирует группы клиентов. +type Server struct { + mu sync.Mutex + groups map[connectionKey]*group +} + +func (s *Server) Handle(ctx context.Context, lane net.Conn, client, backend string) error { + defer func() { _ = lane.Close() }() + stop := context.AfterFunc(ctx, func() { _ = lane.Close() }) + defer stop() + if err := lane.SetReadDeadline(time.Now().Add(SetupTimeout)); err != nil { + return fmt.Errorf("bond hello deadline: %w", err) + } + h, err := ReadHello(lane) + if err != nil { + return err + } + if err = lane.SetReadDeadline(time.Time{}); err != nil { + return fmt.Errorf("bond hello deadline: %w", err) + } + g, err := s.attach(ctx, connectionKey{client: client, id: h.ID}, h, lane, backend) + if err != nil { + return err + } + select { + case <-ctx.Done(): + g.cancel() + <-g.done + case <-g.done: + } + return g.err +} + +func (s *Server) attach(ctx context.Context, key connectionKey, h Hello, lane net.Conn, backend string) (*group, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.groups == nil { + s.groups = make(map[connectionKey]*group) + } + g := s.groups[key] + if g == nil { + groupCtx, cancel := context.WithCancel(ctx) + g = &group{lanes: make([]net.Conn, h.Count), ready: make(chan struct{}), done: make(chan struct{}), cancel: cancel} + s.groups[key] = g + go s.run(groupCtx, key, g, backend) + } + if len(g.lanes) != int(h.Count) || g.lanes[h.Index] != nil { + return nil, fmt.Errorf("%w: duplicate lane or count mismatch", ErrProtocol) + } + g.lanes[h.Index] = lane + g.attached++ + if g.attached == len(g.lanes) { + close(g.ready) + } + return g, nil +} + +func (s *Server) run(ctx context.Context, key connectionKey, g *group, backend string) { + g.err = serveGroup(ctx, g, backend) + g.cancel() + s.mu.Lock() + delete(s.groups, key) + lanes := append([]net.Conn(nil), g.lanes...) + s.mu.Unlock() + for _, lane := range lanes { + if lane != nil { + _ = lane.Close() + } + } + close(g.done) +} + +func serveGroup(ctx context.Context, g *group, backend string) error { + timer := time.NewTimer(SetupTimeout) + defer timer.Stop() + select { + case <-g.ready: + case <-ctx.Done(): + return fmt.Errorf("bond attach: %w", ctx.Err()) + case <-timer.C: + return fmt.Errorf("bond attach: %w", context.DeadlineExceeded) + } + conn, err := (&net.Dialer{Timeout: SetupTimeout}).DialContext(ctx, "tcp", backend) + if err != nil { + return fmt.Errorf("bond backend: %w", err) + } + return Copy(ctx, conn, g.lanes, nil) +} diff --git a/internal/proxy/tcprelay/bond.go b/internal/proxy/tcprelay/bond.go new file mode 100644 index 0000000..adf5307 --- /dev/null +++ b/internal/proxy/tcprelay/bond.go @@ -0,0 +1,79 @@ +package tcprelay + +import ( + "context" + "crypto/rand" + "encoding/binary" + "net" + "time" + + "github.com/samosvalishe/free-turn-proxy/internal/logx" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/bond" + "github.com/samosvalishe/free-turn-proxy/internal/stats" +) + +func proxyBond(ctx context.Context, log logx.Logger, conn net.Conn, pool *sessionPool) { + defer func() { _ = conn.Close() }() + lanes, traffic := openBondLanes(pool) + defer func() { + for _, lane := range lanes { + _ = lane.Close() + } + }() + if len(lanes) == 0 { + return + } + stop := context.AfterFunc(ctx, func() { + for _, lane := range lanes { + _ = lane.Close() + } + }) + defer stop() + var id [8]byte + _, _ = rand.Read(id[:]) + deadline := time.Now().Add(bond.SetupTimeout) + for i, lane := range lanes { + h := bond.Hello{ID: binary.BigEndian.Uint64(id[:]), Index: uint16(i), Count: uint16(len(lanes))} //nolint:gosec // len(lanes) <= MaxLanes + if err := helloBondLane(lane, h, deadline); err != nil { + log.Debugf("TCP bond setup: %v", err) + return + } + } + log.Debugf("TCP bond connected: lanes=%d", len(lanes)) + if err := bond.Copy(ctx, conn, lanes, traffic); err != nil && ctx.Err() == nil { + log.Debugf("TCP bond closed: %v", err) + } +} + +func helloBondLane(lane net.Conn, h bond.Hello, deadline time.Time) error { + if err := lane.SetWriteDeadline(deadline); err != nil { + return err + } + if err := bond.WriteHello(lane, h); err != nil { + return err + } + return lane.SetWriteDeadline(time.Time{}) +} + +func openBondLanes(pool *sessionPool) ([]net.Conn, *stats.Stats) { + pool.mu.RLock() + candidates := append([]*pooledSession(nil), pool.sessions...) + pool.mu.RUnlock() + lanes := make([]net.Conn, 0, len(candidates)) + var traffic *stats.Stats + for _, ps := range candidates { + if len(lanes) == bond.MaxLanes { + break + } + if ps.sess.IsClosed() { + continue + } + stream, err := ps.sess.OpenStream() + if err != nil { + continue + } + lanes = append(lanes, stream) + traffic = ps.traffic + } + return lanes, traffic +} diff --git a/internal/proxy/tcprelay/bond_test.go b/internal/proxy/tcprelay/bond_test.go new file mode 100644 index 0000000..6618847 --- /dev/null +++ b/internal/proxy/tcprelay/bond_test.go @@ -0,0 +1,125 @@ +package tcprelay + +import ( + "bytes" + "context" + "fmt" + "io" + "net" + "testing" + "time" + + "github.com/samosvalishe/free-turn-proxy/internal/logx" + "github.com/samosvalishe/free-turn-proxy/internal/netconn" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/bond" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/tcpserver" + "github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux" + "github.com/xtaci/smux" +) + +func pairedBondSession(t *testing.T, ctx context.Context, server *bond.Server, backend string) *smux.Session { + t.Helper() + a, b := netconn.DatagramPipe(2048, 1024) + t.Cleanup(func() { _ = a.Close(); _ = b.Close() }) + done := make(chan struct{}) + go func() { + defer close(done) + tcpserver.HandleBond(ctx, logx.Nop(), b, backend, kcpmux.DefaultProfile(), server, "test-client") + }() + kcp, err := kcpmux.Dial(a, kcpmux.DefaultProfile()) + if err != nil { + t.Fatal(err) + } + sess, err := smux.Client(kcp, kcpmux.SmuxConfig()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = sess.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("bond server leaked") + } + }) + return sess +} + +func TestBondAcrossSessionsAndAfterSessionFailure(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + pool := newSessionPool(nil) + pool.bond = true + server := &bond.Server{} + backend := echoBackend(t) + for i := range 3 { + pool.Add(i+1, pairedBondSession(t, ctx, server, backend), nil) + } + ln, err := (&net.ListenConfig{}).Listen(ctx, "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer func() { _ = ln.Close() }() + stop := context.AfterFunc(ctx, func() { _ = ln.Close() }) + defer stop() + loopDone := make(chan struct{}) + go func() { defer close(loopDone); acceptLoop(ctx, &Deps{Log: logx.Nop()}, ln, pool) }() + conn := dialBondTest(t, ctx, ln.Addr().String()) + payload := bytes.Repeat([]byte("bond payload"), 128*1024) + checkBondEcho(t, conn, payload) + for _, ps := range pool.sessions { + if ps.sess.NumStreams() != 1 { + t.Fatalf("session %d: expected one bond lane", ps.id) + } + } + _ = pool.sessions[1].sess.Close() + if _, err := conn.Read(make([]byte, 1)); err == nil { + t.Fatal("broken bond still open") + } + _ = conn.Close() + // Следующее соединение использует оставшиеся сессии без перезапуска listener. + conn = dialBondTest(t, ctx, ln.Addr().String()) + checkBondEcho(t, conn, payload) + _ = conn.Close() + cancel() + select { + case <-loopDone: + case <-time.After(5 * time.Second): + t.Fatal("accept loop leaked") + } +} + +func dialBondTest(t *testing.T, ctx context.Context, addr string) net.Conn { + t.Helper() + conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + t.Fatal(err) + } + return conn +} + +func checkBondEcho(t *testing.T, conn net.Conn, payload []byte) { + t.Helper() + done := make(chan error, 1) + go func() { + n, err := conn.Write(payload) + if err == nil && n != len(payload) { + err = fmt.Errorf("short write %d", n) + } + done <- err + }() + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil { + t.Fatal(err) + } + if err := <-done; err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, payload) { + t.Fatal("corrupted echo") + } +} diff --git a/internal/proxy/tcprelay/pool.go b/internal/proxy/tcprelay/pool.go index 92aaa81..b17eecc 100644 --- a/internal/proxy/tcprelay/pool.go +++ b/internal/proxy/tcprelay/pool.go @@ -21,6 +21,7 @@ type pooledSession struct { // sessionPool - конкурентно-безопасный round-robin пул живых сессий. type sessionPool struct { + bond bool mu sync.RWMutex sessions []*pooledSession counter atomic.Uint64 diff --git a/internal/proxy/tcprelay/tcprelay.go b/internal/proxy/tcprelay/tcprelay.go index 7867203..3b15c14 100644 --- a/internal/proxy/tcprelay/tcprelay.go +++ b/internal/proxy/tcprelay/tcprelay.go @@ -45,6 +45,7 @@ type DialFunc = udprelay.DialFunc type AuthHandler = udprelay.AuthHandler type Params struct { + Bond bool Dial DialFunc Profile string ObfKey []byte @@ -86,6 +87,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis } log := deps.log() pool := newSessionPool(deps.ConnectedStreams) + pool.bond = params.Bond runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -98,7 +100,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis } stopCloser := context.AfterFunc(runCtx, func() { _ = listener.Close() }) defer stopCloser() - log.Infof("TCP mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions) + log.Infof("TCP mode: listening on %s (sessions=%d bond=%t)", listenAddr, numSessions, params.Bond) fatalCh := make(chan error, 1) fatal := func(err error) { @@ -201,6 +203,12 @@ func acceptLoop(ctx context.Context, deps *Deps, listener net.Listener, pool *se continue } backoff = 0 + if pool.bond { + wg.Go(func() { + _ = safego.Run(log, func() { proxyBond(ctx, log, conn, pool) }) + }) + continue + } ps := pool.Pick() if ps == nil { @@ -387,7 +395,11 @@ func createSession(ctx context.Context, deps *Deps, params *Params, peer *net.UD closers = append(closers, func() { _ = dtlsConn.Close() }) // Wire-контракт: Client ID первой app-record, до KCP. - if err = clientsdb.WriteClientID(ctx, dtlsConn, params.ClientID, clientsdb.ModeTCP); err != nil { + mode := clientsdb.ModeTCP + if params.Bond { + mode = clientsdb.ModeTCPBond + } + if err = clientsdb.WriteClientID(ctx, dtlsConn, params.ClientID, mode); err != nil { cleanup() return nil, fmt.Errorf("send client ID: %w", err) } diff --git a/internal/proxy/tcpserver/tcpserver.go b/internal/proxy/tcpserver/tcpserver.go index 3574fe1..3e9cae6 100644 --- a/internal/proxy/tcpserver/tcpserver.go +++ b/internal/proxy/tcpserver/tcpserver.go @@ -11,6 +11,7 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/logx" "github.com/samosvalishe/free-turn-proxy/internal/netconn" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/bond" "github.com/samosvalishe/free-turn-proxy/internal/safego" "github.com/samosvalishe/free-turn-proxy/internal/transport/kcpmux" "github.com/xtaci/smux" @@ -20,6 +21,22 @@ const backendDialTimeout = 10 * time.Second // Handle блокирует вызывающую горутину до закрытия сессии клиентом или ctx. func Handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectAddr string, profile kcpmux.Profile) { + handle(ctx, logger, dtlsConn, profile, func(ctx context.Context, stream *smux.Stream) { + handleStream(ctx, logger, stream, connectAddr) + }) +} + +func HandleBond(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectAddr string, profile kcpmux.Profile, server *bond.Server, clientID string) { + handle(ctx, logger, dtlsConn, profile, func(ctx context.Context, stream *smux.Stream) { + if err := server.Handle(ctx, stream, clientID, connectAddr); err != nil && ctx.Err() == nil { + logger.Debugf("tcpserver bond: %v", err) + } + }) +} + +func handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, profile kcpmux.Profile, streamHandler func(context.Context, *smux.Stream)) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() kcpSess, err := kcpmux.Accept(dtlsConn, profile) if err != nil { logger.Errorf("tcpserver: %s", err) @@ -57,9 +74,10 @@ func Handle(ctx context.Context, logger logx.Logger, dtlsConn net.Conn, connectA break } wg.Go(func() { - _ = safego.Run(logger, func() { handleStream(ctx, logger, stream, connectAddr) }) + _ = safego.Run(logger, func() { streamHandler(ctx, stream) }) }) } + cancel() wg.Wait() } diff --git a/internal/session/session.go b/internal/session/session.go index 4c41b4f..87d5dc3 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -483,6 +483,7 @@ func (s *Session) relayTCP(ctx context.Context, auth udprelay.AuthHandler, dial Recycle: s.recycleCh, } params := &tcprelay.Params{ + Bond: s.cfg.Proxy.Bond, Dial: dial, Profile: string(s.cfg.Obf.Profile), ObfKey: s.cfg.Obf.Key, diff --git a/internal/uri/uri.go b/internal/uri/uri.go index 9ae384e..39f7600 100644 --- a/internal/uri/uri.go +++ b/internal/uri/uri.go @@ -18,6 +18,7 @@ type Config struct { Peer string Transport string Mode string + Bond bool ObfProfile string ObfKey string ObfTimingMs int @@ -41,6 +42,7 @@ type wire struct { Peer string `json:"peer"` Transport string `json:"transport,omitempty"` Mode string `json:"mode,omitempty"` + Bond bool `json:"bond,omitempty"` Obf string `json:"obf,omitempty"` Key string `json:"key,omitempty"` TimingMs int `json:"timing,omitempty"` @@ -92,6 +94,7 @@ func Parse(s string) (*Config, error) { Peer: w.Peer, Transport: w.Transport, Mode: w.Mode, + Bond: w.Bond, ObfProfile: w.Obf, ObfKey: w.Key, ObfTimingMs: w.TimingMs, @@ -117,6 +120,7 @@ func (c *Config) String() string { Peer: c.Peer, Transport: c.Transport, Mode: c.Mode, + Bond: c.Bond, N: c.N, StreamsPerCred: c.StreamsPerCred, ClientID: c.ClientID, diff --git a/mobile/README.md b/mobile/README.md index bf260b8..25ba77a 100644 --- a/mobile/README.md +++ b/mobile/README.md @@ -69,6 +69,8 @@ func Version() string `proxy.mode`: `udp` (по умолчанию, UDP-релей для WireGuard) или `tcp` (TCP-форвардер для Xray/sing-box). +`proxy.bond: true` включает объединение сессий для каждого TCP-соединения; допускается только с `proxy.mode: "tcp"`, по умолчанию выключено. Нужны обновлённые сервер и AAR. При отказе одной участвующей сессии соединение закрывается; новое использует оставшиеся сессии. Публичные методы gomobile не меняются. + В `tcp` ядро слушает `proxy.listen` как TCP-порт, и VLESS-клиент ходит туда своим outbound. Своего tun при этом нет: * запускать через `Start(configJSON)`, не через `StartTunnel`; `VpnService.prepare`/`establish` звать не нужно - иначе tun отбирается у чужого VPN;