mirror of
https://github.com/hackdiaz-dev/free-turn-proxy.git
synced 2026-10-12 09:41:53 +02:00
refactor: вынести turnpipe и dtlsdial в internal/
Переносит общий пайплайн TURN dial+allocate в internal/turnpipe и DTLS-настройку клиента в internal/dtlsdial. И oneTurnConnection, и createSmuxSession теперь используют одни строительные блоки, убирая ~140 строк дублированной настройки. Глобал handshakeSem убран, заменён двумя сконфигурированными dtlsdial.Dialer: udpDtlsDialer (timeout 20s, sem cap=3) и vlessDtlsDialer (timeout 30s, без gate). Поведение идентично прежним путям. client/main.go: 1380 → 1195. Импорты netadapt, pion/dtls, selfsign, pion/logging и pion/turn вынесены из client/main.go в новые пакеты.
This commit is contained in:
samosvalishe
+30
-215
@@ -24,15 +24,12 @@ import (
|
||||
"github.com/cacggghp/vk-turn-proxy/client/internal/dnsdial"
|
||||
"github.com/cacggghp/vk-turn-proxy/client/internal/vkauth"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/bond"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/netadapt"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/dtlsdial"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/stats"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/turnpipe"
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/wrap"
|
||||
"github.com/cacggghp/vk-turn-proxy/tcputil"
|
||||
"github.com/cbeuw/connutil"
|
||||
"github.com/pion/dtls/v3"
|
||||
"github.com/pion/dtls/v3/pkg/crypto/selfsign"
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/turn/v5"
|
||||
"github.com/xtaci/smux"
|
||||
)
|
||||
|
||||
@@ -43,8 +40,12 @@ var (
|
||||
activeLocalPeer atomic.Value
|
||||
connectedStreams atomic.Int32
|
||||
globalAppCancel context.CancelFunc
|
||||
handshakeSem = make(chan struct{}, 3)
|
||||
isDebug bool
|
||||
udpDtlsDialer = &dtlsdial.Dialer{
|
||||
HandshakeTimeout: 20 * time.Second,
|
||||
HandshakeSem: make(chan struct{}, 3),
|
||||
}
|
||||
vlessDtlsDialer = &dtlsdial.Dialer{HandshakeTimeout: 30 * time.Second}
|
||||
isDebug bool
|
||||
)
|
||||
|
||||
var appDialer net.Dialer
|
||||
@@ -82,37 +83,7 @@ func manualCaptchaSolver(_ context.Context, e *captcha.Error, d net.Dialer) (str
|
||||
}
|
||||
|
||||
func dtlsFunc(ctx context.Context, conn net.PacketConn, peer *net.UDPAddr) (net.Conn, error) {
|
||||
certificate, err := selfsign.GenerateSelfSigned()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
select {
|
||||
case handshakeSem <- struct{}{}:
|
||||
defer func() { <-handshakeSem }()
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
ctx1, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||||
defer cancel()
|
||||
dtlsConn, err := dtls.ClientWithOptions(
|
||||
conn,
|
||||
peer,
|
||||
dtls.WithCertificates(certificate),
|
||||
dtls.WithInsecureSkipVerify(true),
|
||||
dtls.WithExtendedMasterSecret(dtls.RequireExtendedMasterSecret),
|
||||
dtls.WithCipherSuites(dtls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256),
|
||||
dtls.WithConnectionIDGenerator(dtls.OnlySendCIDGenerator()),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := dtlsConn.HandshakeContext(ctx1); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return dtlsConn, nil
|
||||
return udpDtlsDialer.Dial(ctx, conn, peer)
|
||||
}
|
||||
|
||||
func oneDtlsConnection(ctx context.Context, peer *net.UDPAddr, listenConn net.PacketConn, inboundChan <-chan *UDPPacket, connchan chan<- net.PacketConn, okchan chan<- struct{}, streamID int) error {
|
||||
@@ -217,98 +188,20 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD
|
||||
err = fmt.Errorf("failed to get TURN credentials: %s", err1)
|
||||
return
|
||||
}
|
||||
urlhost, urlport, err1 := net.SplitHostPort(urlTarget)
|
||||
if err1 != nil {
|
||||
err = fmt.Errorf("failed to parse TURN server address: %s", err1)
|
||||
return
|
||||
}
|
||||
if turnParams.host != "" {
|
||||
urlhost = turnParams.host
|
||||
}
|
||||
if turnParams.port != "" {
|
||||
urlport = turnParams.port
|
||||
}
|
||||
var turnServerAddr string
|
||||
turnServerAddr = net.JoinHostPort(urlhost, urlport)
|
||||
turnServerUDPAddr, err1 := net.ResolveUDPAddr("udp", turnServerAddr)
|
||||
if err1 != nil {
|
||||
err = fmt.Errorf("failed to resolve TURN server address: %s", err1)
|
||||
return
|
||||
}
|
||||
turnServerAddr = turnServerUDPAddr.String()
|
||||
debugf("[STREAM %d] TURN server IP: %s", streamID, turnServerUDPAddr.IP)
|
||||
var cfg *turn.ClientConfig
|
||||
var turnConn net.PacketConn
|
||||
var d net.Dialer
|
||||
ctx1, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
if turnParams.udp {
|
||||
conn, err2 := net.DialUDP("udp", nil, turnServerUDPAddr) // nolint: noctx
|
||||
if err2 != nil {
|
||||
err = fmt.Errorf("failed to connect to TURN server: %s", err2)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err1 = conn.Close(); err1 != nil {
|
||||
err = fmt.Errorf("failed to close TURN server connection: %s", err1)
|
||||
return
|
||||
}
|
||||
}()
|
||||
turnConn = &netadapt.ConnectedUDPConn{UDPConn: conn}
|
||||
} else {
|
||||
conn, err2 := d.DialContext(ctx1, "tcp", turnServerAddr)
|
||||
if err2 != nil {
|
||||
err = fmt.Errorf("failed to connect to TURN server: %s", err2)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err1 = conn.Close(); err1 != nil {
|
||||
err = fmt.Errorf("failed to close TURN server connection: %s", err1)
|
||||
return
|
||||
}
|
||||
}()
|
||||
wrappedConn := &netadapt.SplitFirstWriteConn{Conn: conn, SplitAt: 6, Delay: 20 * time.Millisecond}
|
||||
turnConn = turn.NewSTUNConn(wrappedConn)
|
||||
}
|
||||
var addrFamily turn.RequestedAddressFamily
|
||||
if peer.IP.To4() != nil {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv4
|
||||
} else {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv6
|
||||
}
|
||||
|
||||
cfg = &turn.ClientConfig{
|
||||
STUNServerAddr: turnServerAddr,
|
||||
TURNServerAddr: turnServerAddr,
|
||||
Conn: turnConn,
|
||||
Net: netadapt.New(),
|
||||
Username: user,
|
||||
Password: pass,
|
||||
RequestedAddressFamily: addrFamily,
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
}
|
||||
|
||||
client, err1 := turn.NewClient(cfg)
|
||||
if err1 != nil {
|
||||
err = fmt.Errorf("failed to create TURN client: %s", err1)
|
||||
return
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
err1 = client.Listen()
|
||||
if err1 != nil {
|
||||
err = fmt.Errorf("failed to listen: %s", err1)
|
||||
return
|
||||
}
|
||||
|
||||
relayConn, err1 := client.Allocate()
|
||||
stream, err1 := turnpipe.Open(ctx, turnpipe.Config{
|
||||
HostOverride: turnParams.host,
|
||||
PortOverride: turnParams.port,
|
||||
UDP: turnParams.udp,
|
||||
}, peer, user, pass, urlTarget)
|
||||
if err1 != nil {
|
||||
if vkauth.IsAuthError(err1) {
|
||||
vkAuth.HandleAuthError(streamID)
|
||||
}
|
||||
err = fmt.Errorf("failed to allocate: %s", err1)
|
||||
err = err1
|
||||
return
|
||||
}
|
||||
relayConn := stream.Relay
|
||||
debugf("[STREAM %d] TURN server IP: %s", streamID, stream.ServerUDPAddr.IP)
|
||||
|
||||
// Reset error count on successful allocation
|
||||
vkAuth.ResetErrors(streamID)
|
||||
@@ -317,8 +210,8 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD
|
||||
connectedStreams.Add(1)
|
||||
defer func() {
|
||||
connectedStreams.Add(-1)
|
||||
if err1 := relayConn.Close(); err1 != nil {
|
||||
err = fmt.Errorf("failed to close TURN allocated connection: %s", err1)
|
||||
if cerr := stream.Close(); cerr != nil {
|
||||
err = fmt.Errorf("failed to close TURN stream: %s", cerr)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -1192,87 +1085,22 @@ func createSmuxSession(ctx context.Context, tp *turnParams, peer *net.UDPAddr, i
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("get TURN creds: %w", err)
|
||||
}
|
||||
urlhost, urlport, err := net.SplitHostPort(rawURL)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("parse TURN addr: %w", err)
|
||||
}
|
||||
if tp.host != "" {
|
||||
urlhost = tp.host
|
||||
}
|
||||
if tp.port != "" {
|
||||
urlport = tp.port
|
||||
}
|
||||
turnServerAddr := net.JoinHostPort(urlhost, urlport)
|
||||
turnServerUDPAddr, err := net.ResolveUDPAddr("udp", turnServerAddr)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("resolve TURN addr: %w", err)
|
||||
}
|
||||
turnServerAddr = turnServerUDPAddr.String()
|
||||
debugf("[session %d] TURN server IP: %s", id, turnServerUDPAddr.IP)
|
||||
|
||||
// 2. Connect to TURN server
|
||||
var turnConn net.PacketConn
|
||||
ctx1, cancel1 := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel1()
|
||||
if tp.udp {
|
||||
c, err1 := net.DialUDP("udp", nil, turnServerUDPAddr)
|
||||
if err1 != nil {
|
||||
return nil, nil, fmt.Errorf("dial TURN (udp): %w", err1)
|
||||
}
|
||||
cleanupFns = append(cleanupFns, func() { _ = c.Close() })
|
||||
turnConn = &netadapt.ConnectedUDPConn{UDPConn: c}
|
||||
} else {
|
||||
var d net.Dialer
|
||||
c, err1 := d.DialContext(ctx1, "tcp", turnServerAddr)
|
||||
if err1 != nil {
|
||||
return nil, nil, fmt.Errorf("dial TURN (tcp): %w", err1)
|
||||
}
|
||||
cleanupFns = append(cleanupFns, func() { _ = c.Close() })
|
||||
wrappedC := &netadapt.SplitFirstWriteConn{Conn: c, SplitAt: 6, Delay: 20 * time.Millisecond}
|
||||
turnConn = turn.NewSTUNConn(wrappedC)
|
||||
}
|
||||
|
||||
// 3. Create TURN client and allocate relay
|
||||
var addrFamily turn.RequestedAddressFamily
|
||||
if peer.IP.To4() != nil {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv4
|
||||
} else {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv6
|
||||
}
|
||||
cfg := &turn.ClientConfig{
|
||||
STUNServerAddr: turnServerAddr,
|
||||
TURNServerAddr: turnServerAddr,
|
||||
Conn: turnConn,
|
||||
Net: netadapt.New(),
|
||||
Username: user,
|
||||
Password: pass,
|
||||
RequestedAddressFamily: addrFamily,
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
}
|
||||
turnClient, err := turn.NewClient(cfg)
|
||||
// 2-3. Dial TURN and allocate relay.
|
||||
stream, err := turnpipe.Open(ctx, turnpipe.Config{
|
||||
HostOverride: tp.host,
|
||||
PortOverride: tp.port,
|
||||
UDP: tp.udp,
|
||||
}, peer, user, pass, rawURL)
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("create TURN client: %w", err)
|
||||
return nil, nil, err
|
||||
}
|
||||
cleanupFns = append(cleanupFns, func() { turnClient.Close() })
|
||||
if err = turnClient.Listen(); err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("TURN listen: %w", err)
|
||||
}
|
||||
relayConn, err := turnClient.Allocate()
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("TURN allocate: %w", err)
|
||||
}
|
||||
cleanupFns = append(cleanupFns, func() { _ = relayConn.Close() })
|
||||
cleanupFns = append(cleanupFns, func() { _ = stream.Close() })
|
||||
relayConn := stream.Relay
|
||||
debugf("[session %d] TURN server IP: %s", id, stream.ServerUDPAddr.IP)
|
||||
debugf("relayed-address=%s", relayConn.LocalAddr().String())
|
||||
|
||||
// 4. Establish DTLS over TURN relay
|
||||
certificate, err := selfsign.GenerateSelfSigned()
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("generate cert: %w", err)
|
||||
}
|
||||
var relayWC *wrap.Conn
|
||||
if len(tp.wrapKey) == wrap.KeyLen {
|
||||
relayWC, err = wrap.NewConn(tp.wrapKey, false)
|
||||
@@ -1282,21 +1110,8 @@ func createSmuxSession(ctx context.Context, tp *turnParams, peer *net.UDPAddr, i
|
||||
}
|
||||
}
|
||||
dtlsPC := &wrap.RelayPacketConn{Relay: relayConn, Peer: peer, Conn: relayWC}
|
||||
dtlsConn, err := dtls.ClientWithOptions(dtlsPC, peer,
|
||||
dtls.WithCertificates(certificate),
|
||||
dtls.WithInsecureSkipVerify(true),
|
||||
dtls.WithExtendedMasterSecret(dtls.RequireExtendedMasterSecret),
|
||||
dtls.WithCipherSuites(dtls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256),
|
||||
dtls.WithConnectionIDGenerator(dtls.OnlySendCIDGenerator()),
|
||||
)
|
||||
dtlsConn, err := vlessDtlsDialer.Dial(ctx, dtlsPC, peer)
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("DTLS client create: %w", err)
|
||||
}
|
||||
ctx2, cancel2 := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel2()
|
||||
if err = dtlsConn.HandshakeContext(ctx2); err != nil {
|
||||
_ = dtlsConn.Close()
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("DTLS handshake: %w", err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
// Package dtlsdial wraps pion-dtls client setup (self-signed cert, EMS,
|
||||
// AES-128-GCM, send-only CID) plus an optional concurrency gate around the
|
||||
// handshake. Used by the client UDP and VLESS pipelines.
|
||||
package dtlsdial
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/pion/dtls/v3"
|
||||
"github.com/pion/dtls/v3/pkg/crypto/selfsign"
|
||||
)
|
||||
|
||||
// Dialer configures the DTLS client handshake.
|
||||
type Dialer struct {
|
||||
// HandshakeTimeout caps the handshake context. Zero means no timeout.
|
||||
HandshakeTimeout time.Duration
|
||||
// HandshakeSem, if non-nil, gates concurrent handshakes (Dial blocks
|
||||
// until a slot is available or ctx fires).
|
||||
HandshakeSem chan struct{}
|
||||
}
|
||||
|
||||
// Dial generates a fresh self-signed cert, acquires the optional handshake
|
||||
// slot, and performs a DTLS client handshake over pc to peer. On success
|
||||
// returns the connected *dtls.Conn. Caller closes it.
|
||||
func (d *Dialer) Dial(ctx context.Context, pc net.PacketConn, peer *net.UDPAddr) (*dtls.Conn, error) {
|
||||
certificate, err := selfsign.GenerateSelfSigned()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if d.HandshakeSem != nil {
|
||||
select {
|
||||
case d.HandshakeSem <- struct{}{}:
|
||||
defer func() { <-d.HandshakeSem }()
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
hsCtx := ctx
|
||||
if d.HandshakeTimeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
hsCtx, cancel = context.WithTimeout(ctx, d.HandshakeTimeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
dtlsConn, err := dtls.ClientWithOptions(
|
||||
pc,
|
||||
peer,
|
||||
dtls.WithCertificates(certificate),
|
||||
dtls.WithInsecureSkipVerify(true),
|
||||
dtls.WithExtendedMasterSecret(dtls.RequireExtendedMasterSecret),
|
||||
dtls.WithCipherSuites(dtls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256),
|
||||
dtls.WithConnectionIDGenerator(dtls.OnlySendCIDGenerator()),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := dtlsConn.HandshakeContext(hsCtx); err != nil {
|
||||
_ = dtlsConn.Close()
|
||||
return nil, err
|
||||
}
|
||||
return dtlsConn, nil
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package dtlsdial
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fakePC blocks forever on ReadFrom and accepts writes — enough to keep the
|
||||
// DTLS handshake stuck so we can exercise the gate/cancel paths.
|
||||
type fakePC struct{}
|
||||
|
||||
func (fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
|
||||
select {} // block forever
|
||||
}
|
||||
func (fakePC) WriteTo(b []byte, _ net.Addr) (int, error) { return len(b), nil }
|
||||
func (fakePC) Close() error { return nil }
|
||||
func (fakePC) LocalAddr() net.Addr { return &net.UDPAddr{IP: net.IPv4zero} }
|
||||
func (fakePC) SetDeadline(time.Time) error { return nil }
|
||||
func (fakePC) SetReadDeadline(time.Time) error { return nil }
|
||||
func (fakePC) SetWriteDeadline(time.Time) error { return nil }
|
||||
|
||||
// When HandshakeSem is full and ctx fires, Dial must return ctx.Err()
|
||||
// without ever touching the underlying PacketConn.
|
||||
func TestDial_SemBlocksUntilCtx(t *testing.T) {
|
||||
sem := make(chan struct{}, 1)
|
||||
sem <- struct{}{} // pre-fill: no slots free
|
||||
d := &Dialer{HandshakeSem: sem, HandshakeTimeout: 5 * time.Second}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
peer := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1}
|
||||
_, err := d.Dial(ctx, fakePC{}, peer)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("expected DeadlineExceeded, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
var _ = errors.New
|
||||
@@ -0,0 +1,157 @@
|
||||
// Package turnpipe centralizes the TURN dial+allocate pipeline shared by
|
||||
// the client UDP (oneTurnConnection) and VLESS (createSmuxSession) modes.
|
||||
//
|
||||
// One call to Open performs: parse target, apply host/port overrides,
|
||||
// resolve UDP addr, dial UDP-or-TCP (with SplitFirstWriteConn over TCP),
|
||||
// turn.NewClient, Listen, Allocate. Returns the relay PacketConn plus a
|
||||
// Close that tears down the allocation, TURN client, and underlying conn.
|
||||
package turnpipe
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/cacggghp/vk-turn-proxy/internal/netadapt"
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/turn/v5"
|
||||
)
|
||||
|
||||
// Config configures one Open call.
|
||||
type Config struct {
|
||||
// HostOverride, if non-empty, replaces the host part returned by the
|
||||
// credentials lookup.
|
||||
HostOverride string
|
||||
// PortOverride, if non-empty, replaces the port part returned by the
|
||||
// credentials lookup.
|
||||
PortOverride string
|
||||
// UDP=true dials TURN over UDP; otherwise over TCP via STUNConn.
|
||||
UDP bool
|
||||
// DialTimeout caps the TCP dial. Zero defaults to 5s.
|
||||
DialTimeout time.Duration
|
||||
}
|
||||
|
||||
// Stream is a live TURN allocation with its dependencies. Close tears
|
||||
// everything down in reverse order.
|
||||
type Stream struct {
|
||||
// Relay is the allocated relay PacketConn returned by turn.Client.Allocate.
|
||||
Relay net.PacketConn
|
||||
// ServerUDPAddr is the resolved TURN server UDP address (host:port).
|
||||
ServerUDPAddr *net.UDPAddr
|
||||
close func() error
|
||||
}
|
||||
|
||||
// Close releases the allocation, TURN client and underlying transport.
|
||||
// Safe to call once. Returns the first non-nil error (if any).
|
||||
func (s *Stream) Close() error {
|
||||
if s == nil || s.close == nil {
|
||||
return nil
|
||||
}
|
||||
return s.close()
|
||||
}
|
||||
|
||||
// Open dials TURN, creates a turn.Client and allocates a relay. rawAddr is
|
||||
// the host:port returned by the credentials lookup; user/pass are the TURN
|
||||
// long-term credentials.
|
||||
func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAddr string) (*Stream, error) {
|
||||
urlhost, urlport, err := net.SplitHostPort(rawAddr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse TURN addr: %w", err)
|
||||
}
|
||||
if cfg.HostOverride != "" {
|
||||
urlhost = cfg.HostOverride
|
||||
}
|
||||
if cfg.PortOverride != "" {
|
||||
urlport = cfg.PortOverride
|
||||
}
|
||||
turnServerAddr := net.JoinHostPort(urlhost, urlport)
|
||||
turnServerUDPAddr, err := net.ResolveUDPAddr("udp", turnServerAddr)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve TURN addr: %w", err)
|
||||
}
|
||||
turnServerAddr = turnServerUDPAddr.String()
|
||||
|
||||
dialTimeout := cfg.DialTimeout
|
||||
if dialTimeout == 0 {
|
||||
dialTimeout = 5 * time.Second
|
||||
}
|
||||
|
||||
var (
|
||||
turnConn net.PacketConn
|
||||
closeConn func() error
|
||||
)
|
||||
if cfg.UDP {
|
||||
c, derr := net.DialUDP("udp", nil, turnServerUDPAddr) //nolint:noctx
|
||||
if derr != nil {
|
||||
return nil, fmt.Errorf("dial TURN (udp): %w", derr)
|
||||
}
|
||||
turnConn = &netadapt.ConnectedUDPConn{UDPConn: c}
|
||||
closeConn = c.Close
|
||||
} else {
|
||||
dctx, cancel := context.WithTimeout(ctx, dialTimeout)
|
||||
defer cancel()
|
||||
var d net.Dialer
|
||||
c, derr := d.DialContext(dctx, "tcp", turnServerAddr)
|
||||
if derr != nil {
|
||||
return nil, fmt.Errorf("dial TURN (tcp): %w", derr)
|
||||
}
|
||||
wrapped := &netadapt.SplitFirstWriteConn{Conn: c, SplitAt: 6, Delay: 20 * time.Millisecond}
|
||||
turnConn = turn.NewSTUNConn(wrapped)
|
||||
closeConn = c.Close
|
||||
}
|
||||
|
||||
var addrFamily turn.RequestedAddressFamily
|
||||
if peer.IP.To4() != nil {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv4
|
||||
} else {
|
||||
addrFamily = turn.RequestedAddressFamilyIPv6
|
||||
}
|
||||
client, err := turn.NewClient(&turn.ClientConfig{
|
||||
STUNServerAddr: turnServerAddr,
|
||||
TURNServerAddr: turnServerAddr,
|
||||
Conn: turnConn,
|
||||
Net: netadapt.New(),
|
||||
Username: user,
|
||||
Password: pass,
|
||||
RequestedAddressFamily: addrFamily,
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
if cerr := closeConn(); cerr != nil {
|
||||
err = fmt.Errorf("%w (close: %v)", err, cerr)
|
||||
}
|
||||
return nil, fmt.Errorf("create TURN client: %w", err)
|
||||
}
|
||||
if err = client.Listen(); err != nil {
|
||||
client.Close()
|
||||
if cerr := closeConn(); cerr != nil {
|
||||
err = fmt.Errorf("%w (close: %v)", err, cerr)
|
||||
}
|
||||
return nil, fmt.Errorf("TURN listen: %w", err)
|
||||
}
|
||||
relay, err := client.Allocate()
|
||||
if err != nil {
|
||||
client.Close()
|
||||
if cerr := closeConn(); cerr != nil {
|
||||
err = fmt.Errorf("%w (close: %v)", err, cerr)
|
||||
}
|
||||
return nil, fmt.Errorf("TURN allocate: %w", err)
|
||||
}
|
||||
|
||||
return &Stream{
|
||||
Relay: relay,
|
||||
ServerUDPAddr: turnServerUDPAddr,
|
||||
close: func() error {
|
||||
var firstErr error
|
||||
if cerr := relay.Close(); cerr != nil {
|
||||
firstErr = cerr
|
||||
}
|
||||
client.Close()
|
||||
if cerr := closeConn(); cerr != nil && firstErr == nil {
|
||||
firstErr = cerr
|
||||
}
|
||||
return firstErr
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package turnpipe
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Open dialing a bogus address must surface a wrapped resolve/dial error.
|
||||
func TestOpen_BadAddress(t *testing.T) {
|
||||
peer := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 1}
|
||||
_, err := Open(context.Background(), Config{}, peer, "u", "p", "not-a-host-port")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for malformed addr")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "parse TURN addr") {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Open with overrides resolves the override host:port (still fails because
|
||||
// there's no TURN server, but the dial error must mention the override).
|
||||
func TestOpen_HostOverrideApplied(t *testing.T) {
|
||||
peer := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 1}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
_, err := Open(ctx, Config{HostOverride: "127.0.0.1", PortOverride: "1", UDP: false, DialTimeout: 200 * time.Millisecond}, peer, "u", "p", "8.8.8.8:443")
|
||||
if err == nil {
|
||||
t.Fatal("expected dial error against unreachable :1")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "dial TURN") {
|
||||
t.Fatalf("expected dial error, got: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user