@ -30,6 +30,7 @@ import (
"time"
"github.com/bschaatsbergen/dnsdialer"
"github.com/cacggghp/vk-turn-proxy/tcputil"
"github.com/cbeuw/connutil"
"github.com/google/uuid"
"github.com/gorilla/websocket"
@ -38,6 +39,7 @@ import (
"github.com/pion/logging"
"github.com/pion/transport/v4"
"github.com/pion/turn/v5"
"github.com/xtaci/smux"
)
type getCredsFunc func ( string ) ( string , string , string , error )
@ -1286,6 +1288,7 @@ func main() { //nolint:cyclop
n := flag . Int ( "n" , 0 , "connections to TURN (default 10 for VK, 1 for Yandex)" )
udp := flag . Bool ( "udp" , false , "connect to TURN with UDP" )
direct := flag . Bool ( "no-dtls" , false , "connect without obfuscation. DO NOT USE" )
tcpMode := flag . Bool ( "tcp" , false , "TCP mode: forward TCP connections (for VLESS) instead of UDP packets" )
flag . Parse ( )
if * peerAddr == "" {
log . Panicf ( "Need peer address!" )
@ -1335,6 +1338,11 @@ func main() { //nolint:cyclop
getCreds : poolCreds ( getCreds , * n ) ,
}
if * tcpMode {
runTCPMode ( ctx , params , peer , * listen , * n )
return
}
listenConnChan := make ( chan net . PacketConn )
listenConn , err := net . ListenPacket ( "udp" , * listen ) // nolint: noctx
if err != nil {
@ -1402,3 +1410,348 @@ func main() { //nolint:cyclop
wg1 . Wait ( )
}
// sessionPool manages a pool of smux sessions for round-robin TCP distribution.
type sessionPool struct {
mu sync . RWMutex
sessions [ ] * smux . Session
counter atomic . Uint64
}
func ( p * sessionPool ) add ( s * smux . Session ) {
p . mu . Lock ( )
p . sessions = append ( p . sessions , s )
p . mu . Unlock ( )
}
func ( p * sessionPool ) remove ( s * smux . Session ) {
p . mu . Lock ( )
for i , sess := range p . sessions {
if sess == s {
p . sessions = append ( p . sessions [ : i ] , p . sessions [ i + 1 : ] ... )
break
}
}
p . mu . Unlock ( )
}
func ( p * sessionPool ) pick ( ) * smux . Session {
p . mu . RLock ( )
defer p . mu . RUnlock ( )
n := len ( p . sessions )
if n == 0 {
return nil
}
idx := p . counter . Add ( 1 ) % uint64 ( n )
return p . sessions [ idx ]
}
func ( p * sessionPool ) count ( ) int {
p . mu . RLock ( )
defer p . mu . RUnlock ( )
return len ( p . sessions )
}
// runTCPMode implements TCP forwarding with round-robin across N TURN sessions.
func runTCPMode ( ctx context . Context , tp * turnParams , peer * net . UDPAddr , listenAddr string , numSessions int ) {
pool := & sessionPool { }
// Start N session maintainers with staggered startup
var wgMaint sync . WaitGroup
for i := 0 ; i < numSessions ; i ++ {
wgMaint . Add ( 1 )
go func ( id int ) {
defer wgMaint . Done ( )
select {
case <- ctx . Done ( ) :
return
case <- time . After ( time . Duration ( id ) * 300 * time . Millisecond ) :
}
maintainTCPSession ( ctx , tp , peer , id , pool )
} ( i )
}
// Wait for at least one session
log . Printf ( "TCP mode: waiting for sessions to connect (total: %d)..." , numSessions )
for {
select {
case <- ctx . Done ( ) :
wgMaint . Wait ( )
return
case <- time . After ( 100 * time . Millisecond ) :
}
if pool . count ( ) > 0 {
break
}
}
listener , err := net . Listen ( "tcp" , listenAddr )
if err != nil {
log . Panicf ( "TCP listen: %s" , err )
}
context . AfterFunc ( ctx , func ( ) { listener . Close ( ) } )
log . Printf ( "TCP mode: listening on %s (round-robin across %d sessions)" , listenAddr , numSessions )
var wgConn sync . WaitGroup
for {
tcpConn , err := listener . Accept ( )
if err != nil {
select {
case <- ctx . Done ( ) :
wgConn . Wait ( )
wgMaint . Wait ( )
return
default :
}
log . Printf ( "TCP accept error: %s" , err )
continue
}
sess := pool . pick ( )
if sess == nil || sess . IsClosed ( ) {
log . Printf ( "No active sessions, rejecting connection" )
tcpConn . Close ( )
continue
}
wgConn . Add ( 1 )
go func ( tc net . Conn , s * smux . Session ) {
defer wgConn . Done ( )
defer tc . Close ( )
stream , err := s . OpenStream ( )
if err != nil {
log . Printf ( "smux open stream error: %s" , err )
return
}
defer stream . Close ( )
pipe ( ctx , tc , stream )
} ( tcpConn , sess )
}
}
// maintainTCPSession keeps one TURN+DTLS+KCP+smux session alive, reconnecting on failure.
func maintainTCPSession ( ctx context . Context , tp * turnParams , peer * net . UDPAddr , id int , pool * sessionPool ) {
for {
select {
case <- ctx . Done ( ) :
return
default :
}
smuxSess , cleanup , err := createSmuxSession ( ctx , tp , peer )
if err != nil {
log . Printf ( "[session %d] setup error: %s, retrying..." , id , err )
select {
case <- ctx . Done ( ) :
return
case <- time . After ( 3 * time . Second ) :
}
continue
}
pool . add ( smuxSess )
log . Printf ( "[session %d] connected (active: %d)" , id , pool . count ( ) )
for ! smuxSess . IsClosed ( ) {
select {
case <- ctx . Done ( ) :
pool . remove ( smuxSess )
cleanup ( )
return
case <- time . After ( 1 * time . Second ) :
}
}
pool . remove ( smuxSess )
cleanup ( )
log . Printf ( "[session %d] disconnected (active: %d), reconnecting..." , id , pool . count ( ) )
select {
case <- ctx . Done ( ) :
return
case <- time . After ( 2 * time . Second ) :
}
}
}
// createSmuxSession establishes a full TURN+DTLS+KCP+smux pipeline and returns
// the smux session along with a cleanup function to tear down all layers.
func createSmuxSession ( ctx context . Context , tp * turnParams , peer * net . UDPAddr ) ( * smux . Session , func ( ) , error ) {
var cleanupFns [ ] func ( )
cleanup := func ( ) {
for i := len ( cleanupFns ) - 1 ; i >= 0 ; i -- {
cleanupFns [ i ] ( )
}
}
// 1. Get TURN credentials
user , pass , rawURL , err := tp . getCreds ( tp . link )
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 ( )
// 2. Connect to TURN server
var turnConn net . PacketConn
ctx1 , cancel1 := context . WithTimeout ( ctx , 5 * time . Second )
defer cancel1 ( )
if tp . udp {
conn , err := net . DialUDP ( "udp" , nil , turnServerUdpAddr )
if err != nil {
return nil , nil , fmt . Errorf ( "dial TURN (udp): %w" , err )
}
cleanupFns = append ( cleanupFns , func ( ) { conn . Close ( ) } )
turnConn = & connectedUDPConn { conn }
} else {
var d net . Dialer
conn , err := d . DialContext ( ctx1 , "tcp" , turnServerAddr )
if err != nil {
return nil , nil , fmt . Errorf ( "dial TURN (tcp): %w" , err )
}
cleanupFns = append ( cleanupFns , func ( ) { conn . Close ( ) } )
turnConn = turn . NewSTUNConn ( conn )
}
// 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 ,
Username : user ,
Password : pass ,
RequestedAddressFamily : addrFamily ,
LoggerFactory : logging . NewDefaultLoggerFactory ( ) ,
}
turnClient , err := turn . NewClient ( cfg )
if err != nil {
cleanup ( )
return nil , nil , fmt . Errorf ( "create TURN client: %w" , 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 ( ) } )
log . Printf ( "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 )
}
dtlsPC := & relayPacketConn { relay : relayConn , peer : peer }
dtlsConfig := & dtls . Config {
Certificates : [ ] tls . Certificate { certificate } ,
InsecureSkipVerify : true ,
ExtendedMasterSecret : dtls . RequireExtendedMasterSecret ,
CipherSuites : [ ] dtls . CipherSuiteID { dtls . TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256 } ,
ConnectionIDGenerator : dtls . OnlySendCIDGenerator ( ) ,
}
dtlsConn , err := dtls . Client ( dtlsPC , peer , dtlsConfig )
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 )
}
cleanupFns = append ( cleanupFns , func ( ) { dtlsConn . Close ( ) } )
log . Printf ( "DTLS connection established" )
// 5. Create KCP session over DTLS
kcpSess , err := tcputil . NewKCPOverDTLS ( dtlsConn , false )
if err != nil {
cleanup ( )
return nil , nil , fmt . Errorf ( "KCP session: %w" , err )
}
cleanupFns = append ( cleanupFns , func ( ) { kcpSess . Close ( ) } )
log . Printf ( "KCP session established" )
// 6. Create smux client session over KCP
smuxSess , err := smux . Client ( kcpSess , tcputil . DefaultSmuxConfig ( ) )
if err != nil {
cleanup ( )
return nil , nil , fmt . Errorf ( "smux client: %w" , err )
}
cleanupFns = append ( cleanupFns , func ( ) { smuxSess . Close ( ) } )
log . Printf ( "smux session established" )
return smuxSess , cleanup , nil
}
// relayPacketConn wraps a TURN relay PacketConn to direct all writes to the peer.
type relayPacketConn struct {
relay net . PacketConn
peer net . Addr
}
func ( r * relayPacketConn ) ReadFrom ( b [ ] byte ) ( int , net . Addr , error ) {
return r . relay . ReadFrom ( b )
}
func ( r * relayPacketConn ) WriteTo ( b [ ] byte , _ net . Addr ) ( int , error ) {
return r . relay . WriteTo ( b , r . peer )
}
func ( r * relayPacketConn ) Close ( ) error { return r . relay . Close ( ) }
func ( r * relayPacketConn ) LocalAddr ( ) net . Addr { return r . relay . LocalAddr ( ) }
func ( r * relayPacketConn ) SetDeadline ( t time . Time ) error { return r . relay . SetDeadline ( t ) }
func ( r * relayPacketConn ) SetReadDeadline ( t time . Time ) error { return r . relay . SetReadDeadline ( t ) }
func ( r * relayPacketConn ) SetWriteDeadline ( t time . Time ) error { return r . relay . SetWriteDeadline ( t ) }
// pipe copies data bidirectionally between two connections.
func pipe ( ctx context . Context , c1 , c2 net . Conn ) {
ctx2 , cancel := context . WithCancel ( ctx )
context . AfterFunc ( ctx2 , func ( ) {
c1 . SetDeadline ( time . Now ( ) )
c2 . SetDeadline ( time . Now ( ) )
} )
var wg sync . WaitGroup
wg . Add ( 2 )
go func ( ) {
defer wg . Done ( )
defer cancel ( )
io . Copy ( c1 , c2 )
} ( )
go func ( ) {
defer wg . Done ( )
defer cancel ( )
io . Copy ( c2 , c1 )
} ( )
wg . Wait ( )
c1 . SetDeadline ( time . Time { } )
c2 . SetDeadline ( time . Time { } )
}
Moroka8
3 months ago