@ -33,6 +33,7 @@ import (
"github.com/bogdanfinn/tls-client/profiles"
"github.com/bschaatsbergen/dnsdialer"
"github.com/cacggghp/vk-turn-proxy/tcputil"
"github.com/cbeuw/connutil"
"github.com/google/uuid"
"github.com/gorilla/websocket"
@ -41,6 +42,7 @@ import (
"github.com/pion/logging"
"github.com/pion/transport/v4"
"github.com/pion/turn/v5"
"github.com/xtaci/smux"
)
type getCredsFunc func ( ctx context . Context , link string , streamID int ) ( string , string , string , error )
@ -277,7 +279,7 @@ type VkCaptchaError struct {
ErrorMsg string
CaptchaSid string
CaptchaImg string
RedirectUri string
RedirectURI string
IsSoundCaptchaAvailable bool
SessionToken string
CaptchaTs string
@ -285,29 +287,65 @@ type VkCaptchaError struct {
}
func ParseVkCaptchaError ( errData map [ string ] interface { } ) * VkCaptchaError {
codeFloat , _ := errData [ "error_code" ] . ( float64 )
// Extract error_code
codeFloat , ok := errData [ "error_code" ] . ( float64 )
if ! ok {
log . Printf ( "missing error_code in captcha error data" )
return nil
}
code := int ( codeFloat )
redirectUri , _ := errData [ "redirect_uri" ] . ( string )
captchaSid , _ := errData [ "captcha_sid" ] . ( string )
if captchaSid == "" {
if sidNum , ok := errData [ "captcha_sid" ] . ( float64 ) ; ok {
// Extract redirect_uri
RedirectURI , ok := errData [ "redirect_uri" ] . ( string )
if ! ok {
log . Printf ( "missing redirect_uri in captcha error data" )
return nil
}
// Extract captcha_sid
captchaSid , ok := errData [ "captcha_sid" ] . ( string )
if ! ok {
// try numeric
if sidNum , ok2 := errData [ "captcha_sid" ] . ( float64 ) ; ok2 {
captchaSid = fmt . Sprintf ( "%.0f" , sidNum )
} else {
log . Printf ( "missing captcha_sid in captcha error data" )
return nil
}
}
captchaImg , _ := errData [ "captcha_img" ] . ( string )
errorMsg , _ := errData [ "error_msg" ] . ( string )
// Extract captcha_img
captchaImg , ok := errData [ "captcha_img" ] . ( string )
if ! ok {
log . Printf ( "missing captcha_img in captcha error data" )
return nil
}
// Extract error_msg
errorMsg , ok := errData [ "error_msg" ] . ( string )
if ! ok {
log . Printf ( "missing error_msg in captcha error data" )
return nil
}
// Extract session token if redirect_uri present
var sessionToken string
if redirectUri != "" {
if parsed , err := neturl . Parse ( redirectUri ) ; err == nil {
if RedirectURI != "" {
if parsed , err := neturl . Parse ( RedirectURI ) ; err == nil {
sessionToken = parsed . Query ( ) . Get ( "session_token" )
} else {
log . Printf ( "failed to parse redirect_uri: %v" , err )
return nil
}
}
isSound , _ := errData [ "is_sound_captcha_available" ] . ( bool )
// Extract is_sound_captcha_available
isSound , ok := errData [ "is_sound_captcha_available" ] . ( bool )
if ! ok {
isSound = false
}
// Extract captcha_ts
var captchaTs string
if tsFloat , ok := errData [ "captcha_ts" ] . ( float64 ) ; ok {
captchaTs = fmt . Sprintf ( "%.0f" , tsFloat )
@ -315,6 +353,7 @@ func ParseVkCaptchaError(errData map[string]interface{}) *VkCaptchaError {
captchaTs = tsStr
}
// Extract captcha_attempt
var captchaAttempt string
if attFloat , ok := errData [ "captcha_attempt" ] . ( float64 ) ; ok {
captchaAttempt = fmt . Sprintf ( "%.0f" , attFloat )
@ -322,12 +361,13 @@ func ParseVkCaptchaError(errData map[string]interface{}) *VkCaptchaError {
captchaAttempt = attStr
}
// Build VkCaptchaError
return & VkCaptchaError {
ErrorCode : code ,
ErrorMsg : errorMsg ,
CaptchaSid : captchaSid ,
CaptchaImg : captchaImg ,
RedirectUri : redirectUri ,
RedirectURI : RedirectURI ,
IsSoundCaptchaAvailable : isSound ,
SessionToken : sessionToken ,
CaptchaTs : captchaTs ,
@ -336,7 +376,7 @@ func ParseVkCaptchaError(errData map[string]interface{}) *VkCaptchaError {
}
func ( e * VkCaptchaError ) IsCaptchaError ( ) bool {
return e . ErrorCode == 14 && e . RedirectUri != "" && e . SessionToken != ""
return e . ErrorCode == 14 && e . RedirectURI != "" && e . SessionToken != ""
}
func solveVkCaptcha ( ctx context . Context , captchaErr * VkCaptchaError , streamID int , client tlsclient . HttpClient , profile Profile , useSliderPOC bool ) ( string , error ) {
@ -349,11 +389,11 @@ func solveVkCaptcha(ctx context.Context, captchaErr *VkCaptchaError, streamID in
if captchaErr . SessionToken == "" {
return "" , fmt . Errorf ( "no session_token in redirect_uri for auto-solve" )
}
if captchaErr . RedirectUri == "" {
if captchaErr . RedirectURI == "" {
return "" , fmt . Errorf ( "no redirect_uri for auto-solve" )
}
bootstrap , err := fetchCaptchaBootstrap ( ctx , captchaErr . RedirectUri , client , profile )
bootstrap , err := fetchCaptchaBootstrap ( ctx , captchaErr . RedirectURI , client , profile )
if err != nil {
return "" , fmt . Errorf ( "failed to fetch captcha bootstrap: %w" , err )
}
@ -385,14 +425,14 @@ func solveVkCaptcha(ctx context.Context, captchaErr *VkCaptchaError, streamID in
return successToken , nil
}
func fetchCaptchaBootstrap ( ctx context . Context , redirectUri string , client tlsclient . HttpClient , profile Profile ) ( * captchaBootstrap , error ) {
parsedURL , err := neturl . Parse ( redirectUri )
func fetchCaptchaBootstrap ( ctx context . Context , redirectURI string , client tlsclient . HttpClient , profile Profile ) ( * captchaBootstrap , error ) {
parsedURL , err := neturl . Parse ( redirectURI )
if err != nil {
return nil , err
}
domain := parsedURL . Hostname ( )
req , err := fhttp . NewRequestWithContext ( ctx , "GET" , redirectUri , nil )
req , err := fhttp . NewRequestWithContext ( ctx , "GET" , redirectURI , nil )
if err != nil {
return nil , err
}
@ -435,7 +475,10 @@ func solvePoW(powInput string, difficulty int) string {
func callCaptchaNotRobot ( ctx context . Context , sessionToken , hash string , streamID int , client tlsclient . HttpClient , profile Profile ) ( string , error ) {
vkReq := func ( method string , postData string ) ( map [ string ] interface { } , error ) {
reqURL := "https://api.vk.ru/method/" + method + "?v=5.131"
parsedURL , _ := neturl . Parse ( reqURL )
parsedURL , err := neturl . Parse ( reqURL )
if err != nil {
return nil , fmt . Errorf ( "parse request URL: %w" , err )
}
domain := parsedURL . Hostname ( )
req , err := fhttp . NewRequestWithContext ( ctx , "POST" , reqURL , strings . NewReader ( postData ) )
@ -522,8 +565,8 @@ func callCaptchaNotRobot(ctx context.Context, sessionToken, hash string, streamI
if ! ok {
return "" , fmt . Errorf ( "invalid check response: %v" , checkResp )
}
status , _ := respObj [ "status" ] . ( string )
if status != "OK" {
status , ok := respObj [ "status" ] . ( string )
if ! ok || status != "OK" {
return "" , fmt . Errorf ( "check status: %s" , status )
}
successToken , ok := respObj [ "success_token" ] . ( string )
@ -789,7 +832,10 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
log . Printf ( "[STREAM %d] [VK Auth] Connecting Identity - Name: %s | User-Agent: %s" , streamID , name , profile . UserAgent )
doRequest := func ( data string , url string ) ( resp map [ string ] interface { } , err error ) {
parsedURL , _ := neturl . Parse ( url )
parsedURL , err := neturl . Parse ( url )
if err != nil {
return nil , fmt . Errorf ( "parse request URL: %w" , err )
}
domain := parsedURL . Hostname ( )
req , err := fhttp . NewRequestWithContext ( ctx , "POST" , url , bytes . NewBuffer ( [ ] byte ( data ) ) )
@ -849,7 +895,10 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
// Token 1 -> getCallPreview
data = fmt . Sprintf ( "vk_join_link=https://vk.com/call/join/%s&fields=photo_200&access_token=%s" , link , token1 )
_ , _ = doRequest ( data , "https://api.vk.ru/method/calls.getCallPreview?v=5.275&client_id=" + creds . ClientID )
_ , err = doRequest ( data , "https://api.vk.ru/method/calls.getCallPreview?v=5.275&client_id=" + creds . ClientID )
if err != nil {
log . Printf ( "[STREAM %d] [VK Auth] Warning: getCallPreview failed: %v" , streamID , err )
}
vkDelayRandom ( 200 , 400 )
@ -878,7 +927,7 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
switch solveMode {
case captchaSolveModeAuto :
if captchaErr . SessionToken != "" && captchaErr . RedirectUri != "" {
if captchaErr . SessionToken != "" && captchaErr . RedirectURI != "" {
successToken , solveErr = solveVkCaptcha ( ctx , captchaErr , streamID , client , profile , false )
if solveErr != nil {
log . Printf ( "[STREAM %d] [Captcha] Auto captcha failed: %v" , streamID , solveErr )
@ -887,7 +936,7 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
solveErr = fmt . Errorf ( "missing fields for auto solve" )
}
case captchaSolveModeSliderPOC :
if captchaErr . SessionToken != "" && captchaErr . RedirectUri != "" {
if captchaErr . SessionToken != "" && captchaErr . RedirectURI != "" {
successToken , solveErr = solveVkCaptcha ( ctx , captchaErr , streamID , client , profile , true )
if solveErr != nil {
log . Printf ( "[STREAM %d] [Captcha] Auto captcha slider POC failed: %v" , streamID , solveErr )
@ -909,8 +958,8 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
go func ( ) {
var t , k string
var e error
if captchaErr . RedirectUri != "" {
t , e = solveCaptchaViaProxy ( captchaErr . RedirectUri , dialer )
if captchaErr . RedirectURI != "" {
t , e = solveCaptchaViaProxy ( captchaErr . RedirectURI , dialer )
} else if captchaErr . CaptchaImg != "" {
k , e = solveCaptchaViaHTTP ( captchaErr . CaptchaImg )
} else {
@ -968,12 +1017,12 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
return "" , "" , "" , fmt . Errorf ( "VK API error: %v" , errObj )
}
respMap , ok := resp [ "response" ] . ( map [ string ] interface { } )
if ! ok {
respMap , okLoop := resp [ "response" ] . ( map [ string ] interface { } )
if ! okLoop {
return "" , "" , "" , fmt . Errorf ( "unexpected getAnonymousToken response: %v" , resp )
}
token2 , ok = respMap [ "token" ] . ( string )
if ! ok {
token2 , okLoop = respMap [ "token" ] . ( string )
if ! okLoop {
return "" , "" , "" , fmt . Errorf ( "missing token in response: %v" , resp )
}
break
@ -988,7 +1037,10 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
if err != nil {
return "" , "" , "" , err
}
token3 := resp [ "session_key" ] . ( string )
token3 , ok := resp [ "session_key" ] . ( string )
if ! ok {
return "" , "" , "" , fmt . Errorf ( "missing session_key in response: %v" , resp )
}
vkDelayRandom ( 100 , 150 )
@ -999,11 +1051,26 @@ func getTokenChain(ctx context.Context, link string, streamID int, creds VKCrede
return "" , "" , "" , err
}
ts := resp [ "turn_server" ] . ( map [ string ] interface { } )
user := ts [ "username" ] . ( string )
pass := ts [ "credential" ] . ( string )
urls := ts [ "urls" ] . ( [ ] interface { } )
urlStr := urls [ 0 ] . ( string )
tsRaw , ok := resp [ "turn_server" ] . ( map [ string ] interface { } )
if ! ok {
return "" , "" , "" , fmt . Errorf ( "missing turn_server in response: %v" , resp )
}
user , ok := tsRaw [ "username" ] . ( string )
if ! ok {
return "" , "" , "" , fmt . Errorf ( "missing username in turn_server" )
}
pass , ok := tsRaw [ "credential" ] . ( string )
if ! ok {
return "" , "" , "" , fmt . Errorf ( "missing credential in turn_server" )
}
urlsRaw , ok := tsRaw [ "urls" ] . ( [ ] interface { } )
if ! ok || len ( urlsRaw ) == 0 {
return "" , "" , "" , fmt . Errorf ( "missing or empty urls in turn_server" )
}
urlStr , ok := urlsRaw [ 0 ] . ( string )
if ! ok {
return "" , "" , "" , fmt . Errorf ( "turn server url is not a string" )
}
clean := strings . Split ( urlStr , "?" ) [ 0 ]
address := strings . TrimPrefix ( strings . TrimPrefix ( clean , "turn:" ) , "turns:" )
@ -1116,7 +1183,7 @@ func getYandexCreds(link string) (string, string, string, error) {
}
type WSSAck struct {
Uid string ` json:"uid" `
UID string ` json:"uid" `
Ack struct {
Status struct {
Code string ` json:"code" `
@ -1125,8 +1192,8 @@ func getYandexCreds(link string) (string, string, string, error) {
}
type WSSData struct {
ParticipantId string
RoomId string
ParticipantID string
RoomID string
Credentials string
Wss string
}
@ -1163,8 +1230,11 @@ func getYandexCreds(link string) (string, string, string, error) {
}
} ( )
if resp . StatusCode != http . StatusOK {
body , _ := io . ReadAll ( resp . Body )
return "" , "" , "" , fmt . Errorf ( "GetConference: status=%s body=%s" , resp . Status , string ( body ) )
readBody , err2 := io . ReadAll ( resp . Body )
if err2 != nil {
return "" , "" , "" , fmt . Errorf ( "GetConference: status=%s (failed to read body: %v)" , resp . Status , err2 )
}
return "" , "" , "" , fmt . Errorf ( "GetConference: status=%s body=%s" , resp . Status , string ( readBody ) )
}
var result ConferenceResponse
@ -1172,8 +1242,8 @@ func getYandexCreds(link string) (string, string, string, error) {
return "" , "" , "" , fmt . Errorf ( "decode conf: %v" , err )
}
data := WSSData {
ParticipantId : result . PeerID ,
RoomId : result . RoomID ,
ParticipantID : result . PeerID ,
RoomID : result . RoomID ,
Credentials : result . Credentials ,
Wss : result . ClientConfiguration . MediaServerURL ,
}
@ -1185,10 +1255,17 @@ func getYandexCreds(link string) (string, string, string, error) {
defer cancel ( )
dialer := websocket . Dialer { }
conn , _ , err := dialer . DialContext ( ctx , data . Wss , h )
var conn * websocket . Conn
conn , resp , err = dialer . DialContext ( ctx , data . Wss , h )
if err != nil {
if resp != nil && resp . Body != nil {
_ = resp . Body . Close ( )
}
return "" , "" , "" , fmt . Errorf ( "ws dial: %w" , err )
}
if resp != nil && resp . Body != nil {
defer func ( ) { _ = resp . Body . Close ( ) } ( )
}
defer func ( ) {
if closeErr := conn . Close ( ) ; closeErr != nil {
log . Printf ( "close websocket: %s" , closeErr )
@ -1214,8 +1291,8 @@ func getYandexCreds(link string) (string, string, string, error) {
SendVideo : false ,
SendSharing : false ,
ParticipantID : data . ParticipantId ,
RoomID : data . RoomId ,
ParticipantID : data . ParticipantID ,
RoomID : data . RoomID ,
ServiceName : "telemost" ,
Credentials : data . Credentials ,
SdkInfo : SdkInfo {
@ -1344,6 +1421,7 @@ func dtlsFunc(ctx context.Context, conn net.PacketConn, peer *net.UDPAddr) (net.
func oneDtlsConnection ( ctx context . Context , peer * net . UDPAddr , listenConn net . PacketConn , inboundChan <- chan * UDPPacket , connchan chan <- net . PacketConn , okchan chan <- struct { } , streamID int ) error {
time . Sleep ( time . Duration ( rand . Intn ( 400 ) + 100 ) * time . Millisecond )
dtlsctx , dtlscancel := context . WithCancel ( ctx )
defer dtlscancel ( )
@ -1381,7 +1459,9 @@ func oneDtlsConnection(ctx context.Context, peer *net.UDPAddr, listenConn net.Pa
wg := sync . WaitGroup { }
wg . Add ( 1 )
context . AfterFunc ( dtlsctx , func ( ) {
_ = dtlsConn . SetDeadline ( time . Now ( ) )
if err := dtlsConn . SetDeadline ( time . Now ( ) ) ; err != nil {
log . Printf ( "[STREAM %d] Warning: SetDeadline failed: %v" , streamID , err )
}
} )
go func ( ) {
@ -1409,7 +1489,11 @@ func oneDtlsConnection(ctx context.Context, peer *net.UDPAddr, listenConn net.Pa
// Send back to the active WG client
if peerAddr := activeLocalPeer . Load ( ) ; peerAddr != nil {
_ , _ = listenConn . WriteTo ( buf [ : n ] , peerAddr . ( net . Addr ) )
if addr , ok := peerAddr . ( net . Addr ) ; ok {
if _ , err := listenConn . WriteTo ( buf [ : n ] , addr ) ; err != nil {
log . Printf ( "[STREAM %d] failed to forward packet to local peer: %v" , streamID , err )
}
}
}
}
} ( )
@ -1439,7 +1523,7 @@ type turnParams struct {
func oneTurnConnection ( ctx context . Context , turnParams * turnParams , peer * net . UDPAddr , conn2 net . PacketConn , streamID int , c chan <- error ) {
time . Sleep ( time . Duration ( rand . Intn ( 400 ) + 100 ) * time . Millisecond )
var err error = nil
var err error
defer func ( ) { c <- err } ( )
user , pass , urlTarget , err1 := turnParams . getCreds ( ctx , turnParams . link , streamID )
if err1 != nil {
@ -1459,20 +1543,19 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD
}
var turnServerAddr string
turnServerAddr = net . JoinHostPort ( urlhost , urlport )
turnServerUdp Addr , err1 := net . ResolveUDPAddr ( "udp" , turnServerAddr )
turnServerUDP Addr , err1 := net . ResolveUDPAddr ( "udp" , turnServerAddr )
if err1 != nil {
err = fmt . Errorf ( "failed to resolve TURN server address: %s" , err1 )
return
}
turnServerAddr = turnServerUdpAddr . String ( )
turnServerAddr = turnServerUDPAddr . String ( )
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 )
conn , err2 := net . DialUDP ( "udp" , nil , turnServerUDPAddr ) // nolint: noctx
if err2 != nil {
err = fmt . Errorf ( "failed to connect to TURN server: %s" , err2 )
return
@ -1603,9 +1686,10 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD
continue
}
_ , err1 = conn2 . WriteTo ( buf [ : n ] , addr1 . ( net . Addr ) )
if err1 != nil {
return
if addr , ok := addr1 . ( net . Addr ) ; ok {
if _ , err := conn2 . WriteTo ( buf [ : n ] , addr ) ; err != nil {
return
}
}
}
} ( )
@ -1714,6 +1798,7 @@ func main() {
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" )
vlessMode := flag . Bool ( "vless" , false , "VLESS mode: forward TCP connections (for VLESS) instead of UDP packets" )
debugFlag := flag . Bool ( "debug" , false , "enable debug logging" )
manualCaptchaFlag := flag . Bool ( "manual-captcha" , false , "skip auto captcha solving, use manual mode immediately" )
flag . Parse ( )
@ -1772,6 +1857,11 @@ func main() {
getCreds : getCreds ,
}
if * vlessMode {
runVLESSMode ( ctx , params , peer , * listen , * n )
return
}
listenConn , err := net . ListenPacket ( "udp" , * listen )
if err != nil {
log . Panicf ( "Failed to listen: %s" , err )
@ -1792,7 +1882,12 @@ func main() {
go func ( ) {
for {
pkt := packetPool . Get ( ) . ( * UDPPacket )
pktIface := packetPool . Get ( )
pkt , ok := pktIface . ( * UDPPacket )
if ! ok {
log . Printf ( "packetPool returned unexpected type: %T" , pktIface )
continue
}
nRead , addr , err := listenConn . ReadFrom ( pkt . Data )
if err != nil {
return
@ -1800,7 +1895,13 @@ func main() {
// Save the local WireGuard peer address
current := activeLocalPeer . Load ( )
if current == nil || current . ( net . Addr ) . String ( ) != addr . String ( ) {
if current == nil {
activeLocalPeer . Store ( addr )
} else if addrStr , ok := current . ( net . Addr ) ; ok {
if addrStr . String ( ) != addr . String ( ) {
activeLocalPeer . Store ( addr )
}
} else {
activeLocalPeer . Store ( addr )
}
@ -1856,3 +1957,359 @@ func main() {
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 )
}
// runVLESSMode implements TCP forwarding with round-robin across N TURN sessions.
func runVLESSMode ( 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 ) :
}
maintainVLESSSession ( ctx , tp , peer , id , pool )
} ( i )
}
// Wait for at least one session
log . Printf ( "VLESS 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 ( "VLESS 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 func ( ) { _ = tc . Close ( ) } ( )
stream , err := s . OpenStream ( )
if err != nil {
log . Printf ( "smux open stream error: %s" , err )
return
}
defer func ( ) { _ = stream . Close ( ) } ( )
pipe ( ctx , tc , stream )
} ( tcpConn , sess )
}
}
// maintainVLESSSession keeps one TURN+DTLS+KCP+smux session alive, reconnecting on failure.
func maintainVLESSSession ( 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 , id )
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 , id int ) ( * 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 ( ctx , tp . link , id )
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 {
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 = & connectedUDPConn { 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 ( ) } )
turnConn = turn . NewSTUNConn ( c )
}
// 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 }
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 ( ) ) ,
)
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 ( ) {
if err := c1 . SetDeadline ( time . Now ( ) ) ; err != nil {
log . Printf ( "pipe: failed to set deadline c1: %v" , err )
}
if err := c2 . SetDeadline ( time . Now ( ) ) ; err != nil {
log . Printf ( "pipe: failed to set deadline c2: %v" , err )
}
} )
var wg sync . WaitGroup
wg . Add ( 2 )
go func ( ) {
defer wg . Done ( )
defer cancel ( )
if _ , err := io . Copy ( c1 , c2 ) ; err != nil {
log . Printf ( "pipe: c1<-c2 copy error: %v" , err )
}
} ( )
go func ( ) {
defer wg . Done ( )
defer cancel ( )
if _ , err := io . Copy ( c2 , c1 ) ; err != nil {
log . Printf ( "pipe: c2<-c1 copy error: %v" , err )
}
} ( )
wg . Wait ( )
if err := c1 . SetDeadline ( time . Time { } ) ; err != nil {
log . Printf ( "pipe: failed to reset deadline c1: %v" , err )
}
if err := c2 . SetDeadline ( time . Time { } ) ; err != nil {
log . Printf ( "pipe: failed to reset deadline c2: %v" , err )
}
}
cacggghp
2 months ago
GitHub