From ee30afc922acbfce9f2bf961f0b7e8fc05cb06c8 Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Tue, 22 Sep 2026 00:02:19 +0300 Subject: [PATCH] =?UTF-8?q?feat(client):=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B0?= =?UTF-8?q?=D0=B9=D0=B4=D0=B5=D1=80=20direct=20-=20=D0=BF=D1=80=D1=8F?= =?UTF-8?q?=D0=BC=D0=BE=20=D0=BD=D0=B0=20=D1=81=D0=B5=D1=80=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=20=D0=B1=D0=B5=D0=B7=20TURN-=D1=80=D0=B5=D0=BB=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Тот же DTLS с обфускацией, сервер не меняется. По умолчанию один поток. --- internal/config/cliflags.go | 4 +- internal/config/config.go | 3 +- internal/config/config_test.go | 14 +++++ internal/config/defaults.go | 2 +- internal/config/raw.go | 3 ++ internal/config/validate.go | 3 +- internal/proxy/tcprelay/session_test.go | 13 +++-- internal/proxy/tcprelay/tcprelay.go | 34 ++++-------- internal/proxy/udprelay/dial.go | 2 + internal/proxy/udprelay/loop.go | 5 +- internal/proxy/udprelay/run.go | 18 ++++--- internal/proxy/udprelay/run_test.go | 21 +++----- internal/session/provider.go | 35 ++++++++++++ internal/session/session.go | 62 ++++++++-------------- internal/session/session_test.go | 15 ++++++ internal/transport/turndial/direct.go | 54 +++++++++++++++++++ internal/transport/turndial/direct_test.go | 58 ++++++++++++++++++++ 17 files changed, 244 insertions(+), 102 deletions(-) create mode 100644 internal/transport/turndial/direct.go create mode 100644 internal/transport/turndial/direct_test.go diff --git a/internal/config/cliflags.go b/internal/config/cliflags.go index 518399c..53ce7b2 100644 --- a/internal/config/cliflags.go +++ b/internal/config/cliflags.go @@ -40,11 +40,11 @@ func ParseClient(args []string, errOut io.Writer) (*Client, error) { fs.StringVar(&r.Turn, "turn", r.Turn, "IP TURN-сервера; override creds провайдера") fs.StringVar(&r.Port, "port", r.Port, "порт TURN-сервера; override creds провайдера") fs.StringVar(&r.Listen, "listen", r.Listen, "локальный ip:port для WireGuard/Xray клиента") - fs.StringVar(&r.Provider, "provider", r.Provider, "источник TURN-creds: vk") + fs.StringVar(&r.Provider, "provider", r.Provider, "источник потоков: vk (TURN-реле VK Calls) | direct (напрямую на -peer)") fs.StringVar(&r.Link, "link", r.Link, "(устарел) одна ссылка VK Calls, используйте -links") fs.StringVar(&r.Links, "links", r.Links, "ссылки VK Calls через запятую: https://vk.ru/call/join/...,https://vk.ru/call/join/...") fs.StringVar(&r.Peer, "peer", r.Peer, "адрес сервера на VPS, host:port; обязательно") - fs.IntVar(&r.N, "n", r.N, "число параллельных TURN-потоков") + 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.StringVar(&r.ObfProfile, "obf-profile", r.ObfProfile, "wire-профиль обфускации: none | rtpopus | rtpopus2 | rtpopus3; должен совпадать с сервером") diff --git a/internal/config/config.go b/internal/config/config.go index 9751a36..b0683da 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -67,7 +67,8 @@ type ProviderOpts struct { } const ( - ProviderVK = "vk" + ProviderVK = "vk" + ProviderDirect = "direct" ) type DNSOpts struct { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 3b289f2..0d19570 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -407,3 +407,17 @@ func TestParseClient_KCPFlagsRequireTCPMode(t *testing.T) { t.Errorf("expected kcp/mode error, got %v", err) } } + +func TestParseClient_ProviderDirect(t *testing.T) { + c, err := ParseClient([]string{"-peer", "1.2.3.4:5000", "-provider", "direct"}, io.Discard) + if err != nil { + t.Fatalf("direct without links: %v", err) + } + if c.TURN.N != DefaultDirectStreams { + t.Errorf("TURN.N = %d, want %d", c.TURN.N, DefaultDirectStreams) + } + c, err = ParseClient([]string{"-peer", "1.2.3.4:5000", "-provider", "direct", "-n", "5"}, io.Discard) + if err != nil || c.TURN.N != 5 { + t.Fatalf("explicit -n: %v, n=%d", err, c.TURN.N) + } +} diff --git a/internal/config/defaults.go b/internal/config/defaults.go index c9fe0f1..6659ec6 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -23,6 +23,7 @@ const ( const ( DefaultListen = "127.0.0.1:9000" DefaultStreams = 12 + DefaultDirectStreams = 1 DefaultStreamsPerCred = 12 DefaultTransport = TransportTCP DefaultMode = ModeUDP @@ -38,7 +39,6 @@ func defaultRaw() raw { return raw{ Listen: DefaultListen, Provider: DefaultProvider, - N: DefaultStreams, StreamsPerCred: DefaultStreamsPerCred, Transport: DefaultTransport, Mode: DefaultMode, diff --git a/internal/config/raw.go b/internal/config/raw.go index dfaae4c..7f8bf39 100644 --- a/internal/config/raw.go +++ b/internal/config/raw.go @@ -131,6 +131,9 @@ func assemble(r raw) (*Client, error) { n := r.N if n <= 0 { n = DefaultStreams + if r.Provider == ProviderDirect { + n = DefaultDirectStreams + } } platform := Platform(r.Platform) if platform == "" { diff --git a/internal/config/validate.go b/internal/config/validate.go index 0fcfd24..1dd47a5 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -31,8 +31,9 @@ func Validate(c *Client) error { default: return fmt.Errorf("invalid -platform value %q: must be %s | %s", c.VK.Platform, PlatformDesktop, PlatformMobile) } + case ProviderDirect: default: - return fmt.Errorf("invalid -provider value %q: must be %s", c.Provider.Name, ProviderVK) + return fmt.Errorf("invalid -provider value %q: must be %s | %s", c.Provider.Name, ProviderVK, ProviderDirect) } switch c.DNS.Mode { diff --git a/internal/proxy/tcprelay/session_test.go b/internal/proxy/tcprelay/session_test.go index 5587000..cee9b9a 100644 --- a/internal/proxy/tcprelay/session_test.go +++ b/internal/proxy/tcprelay/session_test.go @@ -10,7 +10,6 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/logx" "github.com/samosvalishe/free-turn-proxy/internal/provider" - "github.com/samosvalishe/free-turn-proxy/internal/proxy/allocpace" "github.com/samosvalishe/free-turn-proxy/internal/stats" "github.com/samosvalishe/free-turn-proxy/internal/transport/turndial" ) @@ -56,8 +55,8 @@ func TestMaintainSessionReportsFatal(t *testing.T) { auth := &fakeAuth{} deps := &Deps{Log: logx.Nop(), Auth: auth} params := &Params{ - GetCreds: func(context.Context, int) (string, string, []string, error) { - return "", "", nil, provider.ErrFatalNoStreams + Dial: func(context.Context, int) (*turndial.Stream, error) { + return nil, provider.ErrFatalNoStreams }, } @@ -66,7 +65,7 @@ func TestMaintainSessionReportsFatal(t *testing.T) { go func() { defer close(done) maintainSession(ctx, deps, params, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1}, - 1, newSessionPool(nil), allocpace.New(0), func(err error) { fatalCh <- err }) + 1, newSessionPool(nil), func(err error) { fatalCh <- err }) }() select { @@ -98,8 +97,8 @@ func TestMaintainSessionHandlesAuthError(t *testing.T) { auth := &fakeAuth{authErr: authErr} deps := &Deps{Log: logx.Nop(), Auth: auth} params := &Params{ - GetCreds: func(context.Context, int) (string, string, []string, error) { - return "", "", nil, authErr + Dial: func(context.Context, int) (*turndial.Stream, error) { + return nil, authErr }, } @@ -107,7 +106,7 @@ func TestMaintainSessionHandlesAuthError(t *testing.T) { go func() { defer close(done) maintainSession(ctx, deps, params, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1}, - 1, newSessionPool(nil), allocpace.New(0), func(error) {}) + 1, newSessionPool(nil), func(error) {}) }() deadline := time.Now().Add(15 * time.Second) diff --git a/internal/proxy/tcprelay/tcprelay.go b/internal/proxy/tcprelay/tcprelay.go index 56b6e77..58e6373 100644 --- a/internal/proxy/tcprelay/tcprelay.go +++ b/internal/proxy/tcprelay/tcprelay.go @@ -14,7 +14,6 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/logx" "github.com/samosvalishe/free-turn-proxy/internal/netconn" "github.com/samosvalishe/free-turn-proxy/internal/provider" - "github.com/samosvalishe/free-turn-proxy/internal/proxy/allocpace" "github.com/samosvalishe/free-turn-proxy/internal/proxy/udprelay" "github.com/samosvalishe/free-turn-proxy/internal/randx" "github.com/samosvalishe/free-turn-proxy/internal/safego" @@ -40,20 +39,17 @@ const ( // ErrFatal возвращается при фатальных ошибках провайдера, требующих остановки клиента. var ErrFatal = errors.New("tcprelay: fatal error") -// GetCredsFunc переиспользуется из udprelay: контракт с провайдером один на оба режима. -type GetCredsFunc = udprelay.GetCredsFunc +// DialFunc переиспользуется из udprelay: подъём потока один на оба режима. +type DialFunc = udprelay.DialFunc // AuthHandler переиспользуется из udprelay: жизненный цикл реквизитов один на оба режима. type AuthHandler = udprelay.AuthHandler type Params struct { - Host string - Port string - TransportUDP bool + Dial DialFunc Profile string ObfKey []byte ObfTiming time.Duration - GetCreds GetCredsFunc KCPProfile kcpmux.Profile ClientID string TrafficStats *stats.Stats @@ -79,19 +75,11 @@ func (d *Deps) log() logx.Logger { func (d *Deps) auth() AuthHandler { if d.Auth == nil { - return nopAuth{} + return udprelay.NopAuth{} } return d.Auth } -type nopAuth struct{} - -func (nopAuth) IsAuthError(error) bool { return false } -func (nopAuth) HandleAuthError(int) bool { return false } -func (nopAuth) ResetErrors(int) {} -func (nopAuth) DropCredentials(int) {} -func (nopAuth) BackoffUntilUnix() int64 { return 0 } - // Run поднимает пул сессий и блокирует вызывающую горутину до отмены ctx. func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, listenAddr string, numSessions int) error { if numSessions <= 0 { @@ -99,7 +87,6 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis } log := deps.log() pool := newSessionPool(deps.ConnectedStreams) - pacer := allocpace.New(allocpace.DefaultInterval) runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -126,7 +113,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis for i := range numSessions { id := i + 1 wgBG.Go(func() { - _ = safego.Run(log, func() { maintainSession(runCtx, deps, params, peer, id, pool, pacer, fatal) }) + _ = safego.Run(log, func() { maintainSession(runCtx, deps, params, peer, id, pool, fatal) }) }) } if deps.Recycle != nil { @@ -279,11 +266,11 @@ type session struct { cleanup func() } -func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, id int, pool *sessionPool, pacer *allocpace.Pacer, fatal func(error)) { +func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, id int, pool *sessionPool, fatal func(error)) { log := deps.log() auth := deps.auth() for ctx.Err() == nil { - s, err := createSession(ctx, deps, params, peer, id, pacer) + s, err := createSession(ctx, deps, params, peer, id) if err != nil { if ctx.Err() != nil { return @@ -361,7 +348,7 @@ func awaitDead(ctx context.Context, log logx.Logger, s *session, id int) bool { } // createSession поднимает стек TURN -> obf -> DTLS -> KCP -> smux. -func createSession(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, id int, pacer *allocpace.Pacer) (*session, error) { +func createSession(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, id int) (*session, error) { log := deps.log() var closers []func() cleanup := func() { @@ -370,10 +357,7 @@ func createSession(ctx context.Context, deps *Deps, params *Params, peer *net.UD } } - if !pacer.Wait(ctx) { - return nil, ctx.Err() - } - stream, err := udprelay.DialTURN(ctx, params.Host, params.Port, params.TransportUDP, peer, id, params.GetCreds, log) + stream, err := params.Dial(ctx, id) if err != nil { return nil, err } diff --git a/internal/proxy/udprelay/dial.go b/internal/proxy/udprelay/dial.go index 64d8a3c..b15f5b6 100644 --- a/internal/proxy/udprelay/dial.go +++ b/internal/proxy/udprelay/dial.go @@ -10,6 +10,8 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/transport/turndial" ) +type DialFunc func(ctx context.Context, streamID int) (*turndial.Stream, error) + // GetCredsFunc разрешает TURN-реквизиты для streamID. type GetCredsFunc func(ctx context.Context, streamID int) (user, pass string, rawURLs []string, err error) diff --git a/internal/proxy/udprelay/loop.go b/internal/proxy/udprelay/loop.go index d98c5d4..1761e40 100644 --- a/internal/proxy/udprelay/loop.go +++ b/internal/proxy/udprelay/loop.go @@ -74,9 +74,6 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr case <-ctx.Done(): return case pair := <-connchan: - if !deps.allocPace.Wait(ctx) { - return - } c := make(chan error, 1) go deps.guard(func() { oneTURN(ctx, deps, params, peer, pair.pipe, streamID, c) })() @@ -234,7 +231,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, c <- err }() - stream, derr := DialTURN(ctx, params.Host, params.Port, params.TransportUDP, peer, streamID, params.GetCreds, deps.log()) + stream, derr := params.Dial(ctx, streamID) if derr != nil { if deps.Auth.IsAuthError(derr) { deps.Auth.HandleAuthError(streamID) diff --git a/internal/proxy/udprelay/run.go b/internal/proxy/udprelay/run.go index 2947088..681808c 100644 --- a/internal/proxy/udprelay/run.go +++ b/internal/proxy/udprelay/run.go @@ -11,7 +11,6 @@ import ( "time" "github.com/samosvalishe/free-turn-proxy/internal/logx" - "github.com/samosvalishe/free-turn-proxy/internal/proxy/allocpace" "github.com/samosvalishe/free-turn-proxy/internal/safego" "github.com/samosvalishe/free-turn-proxy/internal/stats" "github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial" @@ -26,21 +25,26 @@ type AuthHandler interface { BackoffUntilUnix() int64 } -// Params содержит конфигурацию подключения к TURN и параметры обфускации. +// Params содержит способ подъёма потока и параметры обфускации. type Params struct { - Host string - Port string - TransportUDP bool + Dial DialFunc Profile string ObfKey []byte ObfTiming time.Duration - GetCreds GetCredsFunc ClientID string TrafficStats *stats.Stats } const streamStartBarrier = 20 * time.Second +type NopAuth struct{} + +func (NopAuth) IsAuthError(error) bool { return false } +func (NopAuth) HandleAuthError(int) bool { return false } +func (NopAuth) ResetErrors(int) {} +func (NopAuth) DropCredentials(int) {} +func (NopAuth) BackoffUntilUnix() int64 { return 0 } + // ErrFatal возвращается при фатальных ошибках провайдера, требующих остановки клиента. var ErrFatal = errors.New("udprelay: fatal error") @@ -52,7 +56,6 @@ type Deps struct { ConnectedStreams *atomic.Int32 OnTURNServer func(ip net.IP) fatalCh chan error - allocPace *allocpace.Pacer } func (d *Deps) log() logx.Logger { @@ -93,7 +96,6 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log ConnectedStreams: connectedStreams, OnTURNServer: onTURNServer, fatalCh: fatalCh, - allocPace: allocpace.New(allocpace.DefaultInterval), } runCtx, runCancel := context.WithCancel(ctx) diff --git a/internal/proxy/udprelay/run_test.go b/internal/proxy/udprelay/run_test.go index 909f721..3c852ee 100644 --- a/internal/proxy/udprelay/run_test.go +++ b/internal/proxy/udprelay/run_test.go @@ -14,16 +14,9 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/provider" "github.com/samosvalishe/free-turn-proxy/internal/safego" "github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial" + "github.com/samosvalishe/free-turn-proxy/internal/transport/turndial" ) -type stubAuth struct{} - -func (stubAuth) IsAuthError(error) bool { return false } -func (stubAuth) HandleAuthError(int) bool { return false } -func (stubAuth) ResetErrors(int) {} -func (stubAuth) DropCredentials(int) {} -func (stubAuth) BackoffUntilUnix() int64 { return 0 } - type deadlineRecorder struct { net.PacketConn mu sync.Mutex @@ -51,8 +44,8 @@ func runFatalDeps(t *testing.T) (*dtlsdial.Dialer, *Params, *net.UDPAddr, *deadl t.Cleanup(func() { _ = pipe.Close(); _ = peerSide.Close() }) params := &Params{ - GetCreds: func(context.Context, int) (string, string, []string, error) { - return "", "", nil, provider.ErrFatalNoStreams + Dial: func(context.Context, int) (*turndial.Stream, error) { + return nil, provider.ErrFatalNoStreams }, } return &dtlsdial.Dialer{HandshakeTimeout: 100 * time.Millisecond}, @@ -70,7 +63,7 @@ func TestRunReturnsOnFatalProviderError(t *testing.T) { var connected atomic.Int32 done := make(chan error, 1) go func() { - done <- Run(context.Background(), dialer, stubAuth{}, logx.Nop(), &connected, nil, params, peer, local, 1) + done <- Run(context.Background(), dialer, NopAuth{}, logx.Nop(), &connected, nil, params, peer, local, 1) }() select { @@ -97,14 +90,14 @@ func TestRunReturnsOnFatalProviderError(t *testing.T) { func TestRunReturnsOnStreamPanic(t *testing.T) { t.Parallel() dialer, params, peer, local := runFatalDeps(t) - params.GetCreds = func(context.Context, int) (string, string, []string, error) { + params.Dial = func(context.Context, int) (*turndial.Stream, error) { panic("boom") } var connected atomic.Int32 done := make(chan error, 1) go func() { - done <- Run(context.Background(), dialer, stubAuth{}, logx.Nop(), &connected, nil, params, peer, local, 1) + done <- Run(context.Background(), dialer, NopAuth{}, logx.Nop(), &connected, nil, params, peer, local, 1) }() select { @@ -128,7 +121,7 @@ func TestRunFatalDoesNotWaitWarmupBarrier(t *testing.T) { done := make(chan error, 1) start := time.Now() go func() { - done <- Run(context.Background(), dialer, stubAuth{}, logx.Nop(), &connected, nil, params, peer, local, 4) + done <- Run(context.Background(), dialer, NopAuth{}, logx.Nop(), &connected, nil, params, peer, local, 4) }() select { diff --git a/internal/session/provider.go b/internal/session/provider.go index 0810fe3..ea53f70 100644 --- a/internal/session/provider.go +++ b/internal/session/provider.go @@ -1,6 +1,7 @@ package session import ( + "context" "fmt" "net" "sync/atomic" @@ -10,8 +11,42 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/provider" "github.com/samosvalishe/free-turn-proxy/internal/provider/multi" "github.com/samosvalishe/free-turn-proxy/internal/provider/vk" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/allocpace" + "github.com/samosvalishe/free-turn-proxy/internal/proxy/udprelay" + "github.com/samosvalishe/free-turn-proxy/internal/transport/turndial" ) +func (s *Session) link(dialer net.Dialer, peer *net.UDPAddr) (udprelay.AuthHandler, udprelay.DialFunc, error) { + log := s.deps.Logger + if s.cfg.Provider.Name == config.ProviderDirect { + log.Infof("provider=%s", config.ProviderDirect) + return udprelay.NopAuth{}, func(ctx context.Context, _ int) (*turndial.Stream, error) { + return turndial.Direct(ctx, peer) + }, nil + } + + prov, err := buildProvider(s.cfg, dialer, &s.connected, s.deps.Solver, log, s.total) + if err != nil { + return nil, nil, err + } + log.Infof("provider=%s", prov.Name()) + getCreds := func(ctx context.Context, streamID int) (string, string, []string, error) { + c, err := prov.GetCredentials(ctx, streamID) + if err != nil { + return "", "", nil, err + } + return c.User, c.Pass, c.ServerAddrs, nil + } + pacer := allocpace.New(allocpace.DefaultInterval) + turn := s.cfg.TURN + return prov, func(ctx context.Context, streamID int) (*turndial.Stream, error) { + if !pacer.Wait(ctx) { + return nil, ctx.Err() + } + return udprelay.DialTURN(ctx, turn.Host, turn.Port, turn.TransportUDP, peer, streamID, getCreds, log) + }, nil +} + // buildProvider создаёт экземпляр provider.Provider в зависимости от конфигурации. func buildProvider( cfg *config.Client, diff --git a/internal/session/session.go b/internal/session/session.go index 54e58e6..4c41b4f 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -14,7 +14,6 @@ import ( "github.com/samosvalishe/free-turn-proxy/internal/client/dnsdial" "github.com/samosvalishe/free-turn-proxy/internal/config" "github.com/samosvalishe/free-turn-proxy/internal/logx" - "github.com/samosvalishe/free-turn-proxy/internal/provider" "github.com/samosvalishe/free-turn-proxy/internal/provider/vk" "github.com/samosvalishe/free-turn-proxy/internal/proxy/tcprelay" "github.com/samosvalishe/free-turn-proxy/internal/proxy/udprelay" @@ -148,7 +147,10 @@ func New(cfg *config.Client, deps Deps) (*Session, error) { } deps.Logger = logx.WithPrefix(deps.Logger, "[session="+id.String()+"] ") - total := cfg.TURN.N * max(len(cfg.VK.Links), 1) + total := cfg.TURN.N + if cfg.Provider.Name != config.ProviderDirect { + total *= max(len(cfg.VK.Links), 1) + } s := &Session{ cfg: cfg, @@ -197,20 +199,19 @@ func (s *Session) Run(ctx context.Context) (err error) { appDialer := dnsdial.AppDialer(s.cfg.DNS.Mode) dnsdial.InstallGlobalResolver(s.cfg.DNS.Mode) - prov, err := buildProvider(s.cfg, appDialer, &s.connected, s.deps.Solver, log, s.total) - if err != nil { - return fmt.Errorf("provider init: %w", err) - } - log.Infof("provider=%s", prov.Name()) - if s.cfg.Obf.Enabled() { - log.Infof("OBF profile=%s: peer server must use matching -obf-profile and -obf-key", s.cfg.Obf.Profile) - } - peer, err := net.ResolveUDPAddr("udp", s.cfg.Proxy.Peer) if err != nil { return fmt.Errorf("resolve peer addr: %w", err) } + auth, dial, err := s.link(appDialer, peer) + if err != nil { + return fmt.Errorf("provider init: %w", err) + } + if s.cfg.Obf.Enabled() { + log.Infof("OBF profile=%s: peer server must use matching -obf-profile and -obf-key", s.cfg.Obf.Profile) + } + runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -237,7 +238,7 @@ func (s *Session) Run(ctx context.Context) (err error) { watchdogErr = s.watch(runCtx, cancel) })) - relayErr := s.runRelayLoop(runCtx, prov, peer) + relayErr := s.relayLoop(runCtx, func(ctx context.Context) error { return s.relay(ctx, auth, dial, peer) }) stopped := runCtx.Err() != nil cancel() bg.Wait() @@ -251,10 +252,6 @@ func (s *Session) Run(ctx context.Context) (err error) { return watchdogErr } -func (s *Session) runRelayLoop(ctx context.Context, prov provider.Provider, peer *net.UDPAddr) error { - return s.relayLoop(ctx, func(ctx context.Context) error { return s.relay(ctx, prov, peer) }) -} - func (s *Session) relayLoop(ctx context.Context, attempt func(context.Context) error) error { log := s.deps.Logger for generation := uint64(1); ; generation++ { @@ -424,17 +421,10 @@ func (s *Session) watch(ctx context.Context, cancel context.CancelFunc) error { } } -func (s *Session) relay(ctx context.Context, prov provider.Provider, peer *net.UDPAddr) error { +func (s *Session) relay(ctx context.Context, auth udprelay.AuthHandler, dial udprelay.DialFunc, peer *net.UDPAddr) error { log := s.deps.Logger - getCreds := udprelay.GetCredsFunc(func(ctx context.Context, streamID int) (string, string, []string, error) { - c, err := prov.GetCredentials(ctx, streamID) - if err != nil { - return "", "", nil, err - } - return c.User, c.Pass, c.ServerAddrs, nil - }) - // host-route для IP TURN-серверов в обход VPN-туннеля. + // host-route до TURN-серверов (в direct - до peer) в обход VPN-туннеля. var routeCallback func(net.IP) if s.cfg.Routes && !s.cfg.Tunnel.Enabled() { rm, rmErr := routemgr.New(log) @@ -450,12 +440,12 @@ func (s *Session) relay(ctx context.Context, prov provider.Provider, peer *net.U } if s.cfg.Proxy.Mode == config.ProxyModeTCP { - return s.relayTCP(ctx, prov, getCreds, peer, routeCallback) + return s.relayTCP(ctx, auth, dial, peer, routeCallback) } - return s.relayUDP(ctx, prov, getCreds, peer, routeCallback) + return s.relayUDP(ctx, auth, dial, peer, routeCallback) } -func (s *Session) relayUDP(ctx context.Context, prov provider.Provider, getCreds udprelay.GetCredsFunc, peer *net.UDPAddr, routeCallback func(net.IP)) error { +func (s *Session) relayUDP(ctx context.Context, auth udprelay.AuthHandler, dial udprelay.DialFunc, peer *net.UDPAddr, routeCallback func(net.IP)) error { log := s.deps.Logger local, err := s.localConn(ctx) if err != nil { @@ -470,39 +460,33 @@ func (s *Session) relayUDP(ctx context.Context, prov provider.Provider, getCreds HandshakeSem: make(chan struct{}, s.opts.HandshakeConcurrency), } params := &udprelay.Params{ - Host: s.cfg.TURN.Host, - Port: s.cfg.TURN.Port, - TransportUDP: s.cfg.TURN.TransportUDP, + Dial: dial, Profile: string(s.cfg.Obf.Profile), ObfKey: s.cfg.Obf.Key, ObfTiming: s.cfg.Obf.Timing, - GetCreds: getCreds, ClientID: s.cfg.ClientID, TrafficStats: s.trafficStats(), } - return udprelay.Run(ctx, dialer, prov, log, &s.connected, routeCallback, params, peer, local, s.total) + return udprelay.Run(ctx, dialer, auth, log, &s.connected, routeCallback, params, peer, local, s.total) } -func (s *Session) relayTCP(ctx context.Context, prov provider.Provider, getCreds udprelay.GetCredsFunc, peer *net.UDPAddr, routeCallback func(net.IP)) error { +func (s *Session) relayTCP(ctx context.Context, auth udprelay.AuthHandler, dial udprelay.DialFunc, peer *net.UDPAddr, routeCallback func(net.IP)) error { deps := &tcprelay.Deps{ DTLSDialer: &dtlsdial.Dialer{ HandshakeTimeout: s.opts.TCPHandshakeTimeout, HandshakeSem: make(chan struct{}, s.opts.HandshakeConcurrency), }, - Auth: prov, + Auth: auth, Log: s.deps.Logger, ConnectedStreams: &s.connected, OnTURNServer: routeCallback, Recycle: s.recycleCh, } params := &tcprelay.Params{ - Host: s.cfg.TURN.Host, - Port: s.cfg.TURN.Port, - TransportUDP: s.cfg.TURN.TransportUDP, + Dial: dial, Profile: string(s.cfg.Obf.Profile), ObfKey: s.cfg.Obf.Key, ObfTiming: s.cfg.Obf.Timing, - GetCreds: getCreds, KCPProfile: s.cfg.KCP.Profile, ClientID: s.cfg.ClientID, TrafficStats: s.trafficStats(), diff --git a/internal/session/session_test.go b/internal/session/session_test.go index 5a17634..5fbe480 100644 --- a/internal/session/session_test.go +++ b/internal/session/session_test.go @@ -357,3 +357,18 @@ func TestReconnectRoutesByMode(t *testing.T) { t.Error("udp mode: reconnectCh empty") } } + +func TestNewDirectIgnoresLinks(t *testing.T) { + cfg := &config.Client{ + TURN: config.TURNOpts{N: 2}, + Provider: config.ProviderOpts{Name: config.ProviderDirect}, + VK: config.VKOpts{Links: []string{"a", "b"}}, + } + s, err := New(cfg, Deps{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if got := s.Snapshot().Total; got != 2 { + t.Fatalf("Total = %d, want 2", got) + } +} diff --git a/internal/transport/turndial/direct.go b/internal/transport/turndial/direct.go new file mode 100644 index 0000000..06b5eb1 --- /dev/null +++ b/internal/transport/turndial/direct.go @@ -0,0 +1,54 @@ +package turndial + +import ( + "context" + "fmt" + "net" + "net/netip" + + "github.com/samosvalishe/free-turn-proxy/internal/netctl" +) + +func Direct(ctx context.Context, peer *net.UDPAddr) (*Stream, error) { + network := "udp4" + if peer.IP.To4() == nil { + network = "udp6" + } + pc, err := (&net.ListenConfig{Control: netctl.Apply}).ListenPacket(ctx, network, "") + if err != nil { + return nil, fmt.Errorf("direct listen: %w", err) + } + c, ok := pc.(*net.UDPConn) + if !ok { + _ = pc.Close() + return nil, fmt.Errorf("turndial: expected *net.UDPConn, got %T", pc) + } + ap := peer.AddrPort() + return &Stream{ + Relay: &peerConn{UDPConn: c, peer: peer, want: netip.AddrPortFrom(ap.Addr().Unmap(), ap.Port())}, + ServerUDPAddr: peer, + close: c.Close, + }, nil +} + +type peerConn struct { + *net.UDPConn + peer *net.UDPAddr + want netip.AddrPort +} + +func (c *peerConn) ReadFrom(p []byte) (int, net.Addr, error) { + for { + n, from, err := c.ReadFromUDPAddrPort(p) + if err != nil { + return n, nil, err + } + if from.Addr().Unmap() == c.want.Addr() && from.Port() == c.want.Port() { + return n, c.peer, nil + } + } +} + +func (c *peerConn) WriteTo(p []byte, _ net.Addr) (int, error) { + return c.WriteToUDPAddrPort(p, c.want) +} diff --git a/internal/transport/turndial/direct_test.go b/internal/transport/turndial/direct_test.go new file mode 100644 index 0000000..536a140 --- /dev/null +++ b/internal/transport/turndial/direct_test.go @@ -0,0 +1,58 @@ +package turndial + +import ( + "net" + "testing" + "time" +) + +func TestDirectAcceptsOnlyPeer(t *testing.T) { + server, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = server.Close() }() + stranger, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = stranger.Close() }() + + peer, ok := server.LocalAddr().(*net.UDPAddr) + if !ok { + t.Fatal("server addr is not UDP") + } + s, err := Direct(t.Context(), peer) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + if s.PermDead != nil { + t.Fatal("PermDead must be nil without relay") + } + + if _, err = s.Relay.WriteTo([]byte("ping"), nil); err != nil { + t.Fatal(err) + } + buf := make([]byte, 16) + _ = server.SetReadDeadline(time.Now().Add(2 * time.Second)) + n, client, err := server.ReadFromUDP(buf) + if err != nil || string(buf[:n]) != "ping" { + t.Fatalf("server read = %q, %v", buf[:n], err) + } + + if _, err = stranger.WriteToUDP([]byte("junk"), client); err != nil { + t.Fatal(err) + } + if _, err = server.WriteToUDP([]byte("pong"), client); err != nil { + t.Fatal(err) + } + _ = s.Relay.SetReadDeadline(time.Now().Add(2 * time.Second)) + n, from, err := s.Relay.ReadFrom(buf) + if err != nil || string(buf[:n]) != "pong" { + t.Fatalf("client read = %q, %v", buf[:n], err) + } + if from.String() != peer.String() { + t.Fatalf("from = %s, want %s", from, peer) + } +}