ppn/free-turn-proxy
3
0
Fork 0
mirror of https://github.com/hackdiaz-dev/free-turn-proxy.git synced 2026-10-12 09:41:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(client): провайдер direct - прямо на сервер без TURN-реле

Browse Source 
Тот же DTLS с обфускацией, сервер не меняется. По умолчанию один поток.
This commit is contained in:
samosvalishe
2026-09-22 00:02:19 +03:00
parent 7df3a3fd91
commit ee30afc922
17 changed files with 244 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/config/cliflags.go
+2 -2
View File
@@ -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; должен совпадать с сервером")
internal/config/config.go
+2 -1
View File
@@ -67,7 +67,8 @@ type ProviderOpts struct {
}
const (
ProviderVK = "vk"
ProviderVK = "vk"
ProviderDirect = "direct"
)
type DNSOpts struct {
internal/config/config_test.go
+14
View File
@@ -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)
}
}
internal/config/defaults.go
+1 -1
View File
@@ -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,
internal/config/raw.go
+3
View File
@@ -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 == "" {
internal/config/validate.go
+2 -1
View File
@@ -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 {
internal/proxy/tcprelay/session_test.go
+6 -7
View File
@@ -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)
internal/proxy/tcprelay/tcprelay.go
+9 -25
View File
@@ -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
}
internal/proxy/udprelay/dial.go
+2
View File
@@ -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)
internal/proxy/udprelay/loop.go
+1 -4
View File
@@ -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)
internal/proxy/udprelay/run.go
+10 -8
View File
@@ -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)
internal/proxy/udprelay/run_test.go
+7 -14
View File
@@ -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 {
internal/session/provider.go
+35
View File
@@ -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,
internal/session/session.go
+23 -39
View File
@@ -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(),
internal/session/session_test.go
+15
View File
@@ -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)
}
}
internal/transport/turndial/direct.go
+54
View File
@@ -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)
}
internal/transport/turndial/direct_test.go
+58
View File
@@ -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)
}
}