// Package session - рантайм одной клиентской сессии: провайдер TURN-реквизитов, // DNS, DTLS-диалеры и выбранный режим релея (udprelay / tcpfwd). // // Общий для cmd/client, пакета mobile и любого будущего потребителя: хост даёт // конфиг, логгер и контекст, а взамен получает блокирующий Run и Snapshot для // UI. Пакет не знает ни про CLI, ни про gomobile - ничего из os.Exit, флагов и // платформенных типов здесь быть не должно. // // Одна активная сессия на процесс: dnsdial.InstallGlobalResolver и // netctl.SetControl - process-global, две параллельные сессии затирали бы // настройки друг друга. package session import ( "context" "errors" "fmt" "net" "sync" "sync/atomic" "time" "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/bondclient" "github.com/samosvalishe/free-turn-proxy/internal/proxy/tcpfwd" "github.com/samosvalishe/free-turn-proxy/internal/proxy/udprelay" "github.com/samosvalishe/free-turn-proxy/internal/routemgr" "github.com/samosvalishe/free-turn-proxy/internal/stats" "github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial" ) // Phase - стадия подключения сессии. type Phase string const ( PhaseIdle Phase = "idle" PhaseConnecting Phase = "connecting" PhaseConnected Phase = "connected" // PhaseCaptcha - пользователь решает captcha вручную. Отдельная стадия, // чтобы UI не показывал ошибку, а watchdog не считал это зависанием. PhaseCaptcha Phase = "captcha" // PhaseError выставляет хост по ошибке из Run: сама сессия к этому моменту // уже завершена. PhaseError Phase = "error" ) var ( ErrAlreadyRun = errors.New("session: already run") // ErrConnectTimeout - ни один поток не поднялся за ConnectTimeout. ErrConnectTimeout = errors.New("session: connect timeout: no stream connected within the deadline") ) const ( defaultStatusInterval = 500 * time.Millisecond defaultRateInterval = time.Second defaultUDPHandshakeTimeout = 20 * time.Second defaultTCPHandshakeTimeout = 30 * time.Second defaultHandshakeConcurrency = 3 ) // Options - тайминги и переключатели рантайма. Нулевое значение поля означает // "дефолт пакета"; ConnectTimeout=0 - особый случай, см. поле. type Options struct { // ConnectTimeout ограничивает ожидание первого поднявшегося стрима: если за // это время ни один не поднялся, Run возвращает ошибку вместо вечного // connecting. Падение отдельного стрима не считается - таймаут снимается, // пока жив хотя бы один. 0 отключает watchdog (поведение CLI: клиент ждёт // столько, сколько попросили). ConnectTimeout time.Duration StatusInterval time.Duration RateInterval time.Duration UDPHandshakeTimeout time.Duration TCPHandshakeTimeout time.Duration HandshakeConcurrency int // Traffic включает подсчёт байт и скорости. Нужен UI; CLI без него не // платит за атомики на пути пакета. Traffic bool } func (o Options) withDefaults() Options { if o.StatusInterval <= 0 { o.StatusInterval = defaultStatusInterval } if o.RateInterval <= 0 { o.RateInterval = defaultRateInterval } if o.UDPHandshakeTimeout <= 0 { o.UDPHandshakeTimeout = defaultUDPHandshakeTimeout } if o.TCPHandshakeTimeout <= 0 { o.TCPHandshakeTimeout = defaultTCPHandshakeTimeout } if o.HandshakeConcurrency <= 0 { o.HandshakeConcurrency = defaultHandshakeConcurrency } return o } // Deps - зависимости хоста. Всё, кроме Logger, опционально. type Deps struct { Logger logx.Logger // Observer получает переходы состояния. nil - хост опрашивает Snapshot. Observer Observer // Solver - ручной решатель captcha. nil отключает ручной fallback. Solver vk.ManualSolverFunc // CaptchaActive сообщает, что прямо сейчас идёт ручное решение captcha: // на это время watchdog приостанавливает отсчёт ConnectTimeout, а Snapshot // отдаёт PhaseCaptcha. nil - captcha никогда не активна. CaptchaActive func() bool // LocalPipe подменяет локальный UDP-сокет каналом в памяти: так туннель, // поднятый внутри процесса, соединяется с релеем напрямую. nil - обычный // bind на cfg.Proxy.Listen. Только для udp-режима; сессия закрывает его сама. LocalPipe net.PacketConn Options Options } // Snapshot - консистентный срез состояния сессии для UI: и стадия подключения, // и статистика трафика. Один вызов на тик вместо нескольких геттеров - хост не // ловит рассогласование от порядка чтения. type Snapshot struct { Phase Phase Streams int Total int Err string TxTotal uint64 RxTotal uint64 TxRate int64 RxRate int64 } type statusInfo struct { phase Phase streams int total int err string } // Session - одна клиентская сессия. Создаётся New, отрабатывает один Run, // повторно не используется. type Session struct { cfg *config.Client deps Deps opts Options total int traffic *traffic connected atomic.Int32 status atomic.Pointer[statusInfo] started atomic.Bool } // Сеть не трогает: блокирующие вызовы происходят в Run. func New(cfg *config.Client, deps Deps) (*Session, error) { if cfg == nil { return nil, errors.New("session: nil config") } if deps.Logger == nil { deps.Logger = logx.Nop() } // Несколько ссылок расширяют пул: каждая даёт cfg.TURN.N стримов, все // объединяются в общий пул (больше параллельных TURN-аллокаций). total := cfg.TURN.N * max(len(cfg.VK.Links), 1) s := &Session{ cfg: cfg, deps: deps, opts: deps.Options.withDefaults(), total: total, } if s.opts.Traffic { s.traffic = newTraffic() } // Созданная сессия уже "подключается": хост читает Snapshot сразу после New, // и показывать ему idle до первой строки Run было бы враньём. Кладётся // молча - наблюдателя на этот момент ещё нет, первое событие даст Run. s.status.Store(&statusInfo{phase: PhaseConnecting, total: total}) return s, nil } // Блокирует до отмены ctx (возвращает nil) или фатальной ошибки. func (s *Session) Run(ctx context.Context) (err error) { if !s.started.CompareAndSwap(false, true) { return ErrAlreadyRun } // Первое событие сессии приходит всегда, даже если состояние совпало с // заготовленным в New: для наблюдателя это начало жизни, а не повтор. s.publish(&statusInfo{phase: PhaseConnecting, total: s.total}, true) defer func() { if err != nil { s.setStatus(PhaseError, 0, err.Error()) return } s.setStatus(PhaseIdle, 0, "") }() log := s.deps.Logger dnsdial.SetLogger(log) if s.cfg.DNS.Servers != nil { dnsdial.SetUDPDNSServers(s.cfg.DNS.Servers) log.Infof("[DNS] using custom UDP servers: %v", s.cfg.DNS.Servers) } 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) } runCtx, cancel := context.WithCancel(ctx) defer cancel() var bg sync.WaitGroup if s.traffic != nil { bg.Add(1) go func() { defer bg.Done() s.traffic.rateMeter(runCtx, s.opts.RateInterval) }() } var watchdogErr error bg.Add(1) go func() { defer bg.Done() watchdogErr = s.watch(runCtx, cancel) }() relayErr := s.relay(runCtx, prov, peer) // Причину смотрим до cancel: собственный defer сделал бы любой выход // "отменённым" и проглотил бы настоящую ошибку релея. stopped := runCtx.Err() != nil cancel() bg.Wait() if relayErr != nil && !stopped { return relayErr } // Watchdog отменяет контекст сам, поэтому его ошибка приходит именно так. return watchdogErr } // Snapshot - текущее состояние сессии. Безопасен из любой горутины. func (s *Session) Snapshot() Snapshot { st := s.status.Load() if st == nil { st = &statusInfo{phase: PhaseIdle} } snap := Snapshot{Phase: st.phase, Streams: st.streams, Total: st.total, Err: st.err} if s.traffic != nil { snap.TxTotal, snap.RxTotal = s.traffic.stats.Counters() snap.TxRate = s.traffic.txRate.Load() snap.RxRate = s.traffic.rxRate.Load() } return snap } // setStatus публикует состояние и, если оно изменилось, дёргает Observer. func (s *Session) setStatus(phase Phase, streams int, errMsg string) { s.publish(&statusInfo{phase: phase, streams: streams, total: s.total, err: errMsg}, false) } // publish кладёт состояние и уведомляет наблюдателя. force шлёт событие даже // при совпадении с предыдущим. // // Вызывается из одной горутины за раз: старт и терминальный статус - из Run, // промежуточные - из watch, который к моменту терминального уже завершён. func (s *Session) publish(next *statusInfo, force bool) { prev := s.status.Swap(next) if !force && prev != nil && *prev == *next { return } if s.deps.Observer != nil { s.deps.Observer.OnPhase(next.phase, next.streams, next.total, next.err) } } func (s *Session) captchaActive() bool { return s.deps.CaptchaActive != nil && s.deps.CaptchaActive() } // watch публикует стадию подключения и следит за ConnectTimeout. При срабатывании // отменяет сессию через cancel и возвращает ошибку - её Run отдаёт наружу. func (s *Session) watch(ctx context.Context, cancel context.CancelFunc) error { tick := time.NewTicker(s.opts.StatusInterval) defer tick.Stop() deadline := time.Now().Add(s.opts.ConnectTimeout) everConnected := false for { select { case <-ctx.Done(): return nil case <-tick.C: n := int(s.connected.Load()) if n > 0 { everConnected = true } if s.captchaActive() { deadline = time.Now().Add(s.opts.ConnectTimeout) s.setStatus(PhaseCaptcha, n, "") continue } phase := PhaseConnecting if n > 0 { phase = PhaseConnected } s.setStatus(phase, n, "") if s.opts.ConnectTimeout > 0 && !everConnected && time.Now().After(deadline) { cancel() return fmt.Errorf("%w (timeout=%s)", ErrConnectTimeout, s.opts.ConnectTimeout) } } } } func (s *Session) relay(ctx context.Context, prov provider.Provider, peer *net.UDPAddr) error { log := s.deps.Logger 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 } // Управление маршрутами: создаём host-route для IP TURN-серверов через // реальный шлюз, чтобы VPN не перехватывал TURN-трафик. var routeCallback func(net.IP) if s.cfg.Routes && !s.cfg.Tunnel.Enabled() { rm, rmErr := routemgr.New(log) if rmErr != nil { log.Warnf("route manager disabled: %v", rmErr) } else if rm != nil { defer func() { _ = rm.Close() }() routeCallback = rm.Callback() log.Infof("route manager: gateway=%s", rm.Gateway()) } } if s.cfg.Proxy.Mode != config.ProxyModeUDP { dialer := &dtlsdial.Dialer{ HandshakeTimeout: s.opts.TCPHandshakeTimeout, HandshakeSem: make(chan struct{}, s.opts.HandshakeConcurrency), } bond := &bondclient.Handler{Deps: bondclient.Deps{Log: log}} deps := &tcpfwd.Deps{ DTLSDialer: dialer, Log: log, BondHandler: bond.Handle, ConnectedStreams: &s.connected, OnTURNServer: routeCallback, } params := &tcpfwd.Params{ Host: s.cfg.TURN.Host, Port: s.cfg.TURN.Port, TransportUDP: s.cfg.TURN.TransportUDP, Profile: string(s.cfg.Obf.Profile), ObfKey: s.cfg.Obf.Key, GetCreds: tcpfwd.GetCredsFunc(getCreds), KCPProfile: s.cfg.KCP.Profile, KCPFEC: s.cfg.KCP.FEC, ClientID: s.cfg.ClientID, TrafficStats: s.trafficStats(), } return tcpfwd.Run(ctx, deps, params, peer, s.cfg.Proxy.Listen, s.total, s.cfg.Proxy.Mode == config.ProxyModeTCPFwdBond) } local, err := s.localConn(ctx) if err != nil { return err } dialer := &dtlsdial.Dialer{ HandshakeTimeout: s.opts.UDPHandshakeTimeout, 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, Profile: string(s.cfg.Obf.Profile), ObfKey: s.cfg.Obf.Key, ObfTiming: s.cfg.Obf.Timing, GetCreds: udprelay.GetCredsFunc(getCreds), ClientID: s.cfg.ClientID, TrafficStats: s.trafficStats(), } return udprelay.Run(ctx, dialer, prov, log, &s.connected, routeCallback, params, peer, local, s.total) } // localConn открывает канал до локального пира. Обычно это UDP-сокет на // cfg.Proxy.Listen, куда ходит внешний WireGuard. Если хост дал LocalPipe - // туннель живёт в этом же процессе, и петля через 127.0.0.1 не нужна. func (s *Session) localConn(ctx context.Context) (net.PacketConn, error) { if s.deps.LocalPipe != nil { s.deps.Logger.Infof("local peer: in-process pipe (no loopback socket)") return s.deps.LocalPipe, nil } conn, err := (&net.ListenConfig{}).ListenPacket(ctx, "udp", s.cfg.Proxy.Listen) if err != nil { return nil, fmt.Errorf("udprelay listen %s: %w", s.cfg.Proxy.Listen, err) } return conn, nil } func (s *Session) trafficStats() *stats.Stats { if s.traffic == nil { return nil } return s.traffic.stats }