diff --git a/client/main.go b/client/main.go index 511a13b..f584261 100644 --- a/client/main.go +++ b/client/main.go @@ -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) } diff --git a/internal/dtlsdial/dtlsdial.go b/internal/dtlsdial/dtlsdial.go new file mode 100644 index 0000000..a31937c --- /dev/null +++ b/internal/dtlsdial/dtlsdial.go @@ -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 +} diff --git a/internal/dtlsdial/dtlsdial_test.go b/internal/dtlsdial/dtlsdial_test.go new file mode 100644 index 0000000..ccb974d --- /dev/null +++ b/internal/dtlsdial/dtlsdial_test.go @@ -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 diff --git a/internal/turnpipe/turnpipe.go b/internal/turnpipe/turnpipe.go new file mode 100644 index 0000000..b626bf1 --- /dev/null +++ b/internal/turnpipe/turnpipe.go @@ -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 +} diff --git a/internal/turnpipe/turnpipe_test.go b/internal/turnpipe/turnpipe_test.go new file mode 100644 index 0000000..42c6519 --- /dev/null +++ b/internal/turnpipe/turnpipe_test.go @@ -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) + } +}