@ -3,9 +3,15 @@ package outbound
import ( import (
"context" "context"
"crypto/rand" "crypto/rand"
"errors" goerrors "errors"
"io"
"math/big"
gonet "net"
"os"
"github.com/xtls/xray-core/app/proxyman" "github.com/xtls/xray-core/app/proxyman"
"github.com/xtls/xray-core/common" "github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/errors"
"github.com/xtls/xray-core/common/buf" "github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/mux" "github.com/xtls/xray-core/common/mux"
"github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/common/net"
@ -21,10 +27,6 @@ import (
"github.com/xtls/xray-core/transport/internet/stat" "github.com/xtls/xray-core/transport/internet/stat"
"github.com/xtls/xray-core/transport/internet/tls" "github.com/xtls/xray-core/transport/internet/tls"
"github.com/xtls/xray-core/transport/pipe" "github.com/xtls/xray-core/transport/pipe"
"io"
"math/big"
gonet "net"
"os"
) )
func getStatCounter ( v * core . Instance , tag string ) ( stats . Counter , stats . Counter ) { func getStatCounter ( v * core . Instance , tag string ) ( stats . Counter , stats . Counter ) {
@ -87,11 +89,11 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
h . senderSettings = s h . senderSettings = s
mss , err := internet . ToMemoryStreamConfig ( s . StreamSettings ) mss , err := internet . ToMemoryStreamConfig ( s . StreamSettings )
if err != nil { if err != nil {
return nil , newError ( "failed to parse stream settings" ) . Base ( err ) . AtWarning ( ) return nil , errors . New ( "failed to parse stream settings" ) . Base ( err ) . AtWarning ( )
} }
h . streamSettings = mss h . streamSettings = mss
default : default :
return nil , newError ( "settings is not SenderConfig" ) return nil , errors . New ( "settings is not SenderConfig" )
} }
} }
@ -107,7 +109,7 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
proxyHandler , ok := rawProxyHandler . ( proxy . Outbound ) proxyHandler , ok := rawProxyHandler . ( proxy . Outbound )
if ! ok { if ! ok {
return nil , newError ( "not an outbound handler" ) return nil , errors . New ( "not an outbound handler" )
} }
if h . senderSettings != nil && h . senderSettings . MultiplexSettings != nil { if h . senderSettings != nil && h . senderSettings . MultiplexSettings != nil {
@ -170,7 +172,7 @@ func (h *Handler) Tag() string {
// Dispatch implements proxy.Outbound.Dispatch.
// Dispatch implements proxy.Outbound.Dispatch.
func ( h * Handler ) Dispatch ( ctx context . Context , link * transport . Link ) { func ( h * Handler ) Dispatch ( ctx context . Context , link * transport . Link ) {
outbounds := session . OutboundsFromContext ( ctx ) outbounds := session . OutboundsFromContext ( ctx )
ob := outbounds [ len ( outbounds ) - 1 ] ob := outbounds [ len ( outbounds ) - 1 ]
if ob . Target . Network == net . Network_UDP && ob . OriginalTarget . Address != nil && ob . OriginalTarget . Address != ob . Target . Address { if ob . Target . Network == net . Network_UDP && ob . OriginalTarget . Address != nil && ob . OriginalTarget . Address != ob . Target . Address {
link . Reader = & buf . EndpointOverrideReader { Reader : link . Reader , Dest : ob . Target . Address , OriginalDest : ob . OriginalTarget . Address } link . Reader = & buf . EndpointOverrideReader { Reader : link . Reader , Dest : ob . Target . Address , OriginalDest : ob . OriginalTarget . Address }
link . Writer = & buf . EndpointOverrideWriter { Writer : link . Writer , Dest : ob . Target . Address , OriginalDest : ob . OriginalTarget . Address } link . Writer = & buf . EndpointOverrideWriter { Writer : link . Writer , Dest : ob . Target . Address , OriginalDest : ob . OriginalTarget . Address }
@ -178,16 +180,16 @@ func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) {
if h . mux != nil { if h . mux != nil {
test := func ( err error ) { test := func ( err error ) {
if err != nil { if err != nil {
err := newError ( "failed to process mux outbound traffic" ) . Base ( err ) err := errors . New ( "failed to process mux outbound traffic" ) . Base ( err )
session . SubmitOutboundErrorToOriginator ( ctx , err ) session . SubmitOutboundErrorToOriginator ( ctx , err )
err . WriteToLog ( session . ExportIDToError ( ctx ) ) errors . LogInfo ( ctx , err . Error ( ) )
common . Interrupt ( link . Writer ) common . Interrupt ( link . Writer )
} }
} }
if ob . Target . Network == net . Network_UDP && ob . Target . Port == 443 { if ob . Target . Network == net . Network_UDP && ob . Target . Port == 443 {
switch h . udp443 { switch h . udp443 {
case "reject" : case "reject" :
test ( newError ( "XUDP rejected UDP/443 traffic" ) . AtInfo ( ) ) test ( errors . New ( "XUDP rejected UDP/443 traffic" ) . AtInfo ( ) )
return return
case "skip" : case "skip" :
goto out goto out
@ -208,15 +210,15 @@ func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) {
out : out :
err := h . proxy . Process ( ctx , link , h ) err := h . proxy . Process ( ctx , link , h )
if err != nil { if err != nil {
if errors . Is ( err , io . EOF ) || errors . Is ( err , io . ErrClosedPipe ) || errors . Is ( err , context . Canceled ) { if go errors. Is ( err , io . EOF ) || go errors. Is ( err , io . ErrClosedPipe ) || go errors. Is ( err , context . Canceled ) {
err = nil err = nil
} }
} }
if err != nil { if err != nil {
// Ensure outbound ray is properly closed.
// Ensure outbound ray is properly closed.
err := newError ( "failed to process outbound traffic" ) . Base ( err ) err := errors . New ( "failed to process outbound traffic" ) . Base ( err )
session . SubmitOutboundErrorToOriginator ( ctx , err ) session . SubmitOutboundErrorToOriginator ( ctx , err )
err . WriteToLog ( session . ExportIDToError ( ctx ) ) errors . LogInfo ( ctx , err . Error ( ) )
common . Interrupt ( link . Writer ) common . Interrupt ( link . Writer )
} else { } else {
common . Close ( link . Writer ) common . Close ( link . Writer )
@ -243,11 +245,11 @@ func (h *Handler) Dial(ctx context.Context, dest net.Destination) (stat.Connecti
tag := h . senderSettings . ProxySettings . Tag tag := h . senderSettings . ProxySettings . Tag
handler := h . outboundManager . GetHandler ( tag ) handler := h . outboundManager . GetHandler ( tag )
if handler != nil { if handler != nil {
newError ( "proxying to " , tag , " for dest " , dest ) . AtDebug ( ) . WriteToLog ( session . ExportIDToError ( ctx ) ) errors . LogDebug ( ctx , "proxying to " , tag , " for dest " , dest )
outbounds := session . OutboundsFromContext ( ctx ) outbounds := session . OutboundsFromContext ( ctx )
ctx = session . ContextWithOutbounds ( ctx , append ( outbounds , & session . Outbound { ctx = session . ContextWithOutbounds ( ctx , append ( outbounds , & session . Outbound {
Target : dest , Target : dest ,
Tag : tag , Tag : tag ,
} ) ) // add another outbound in session ctx
} ) ) // add another outbound in session ctx
opts := pipe . OptionsFromContext ( ctx ) opts := pipe . OptionsFromContext ( ctx )
uplinkReader , uplinkWriter := pipe . New ( opts ... ) uplinkReader , uplinkWriter := pipe . New ( opts ... )
@ -264,12 +266,12 @@ func (h *Handler) Dial(ctx context.Context, dest net.Destination) (stat.Connecti
return h . getStatCouterConnection ( conn ) , nil return h . getStatCouterConnection ( conn ) , nil
} }
newError ( "failed to get outbound handler with tag: " , tag ) . AtWarning ( ) . WriteToLog ( session . ExportIDToError ( ctx ) ) errors . LogWarning ( ctx , "failed to get outbound handler with tag: " , tag )
} }
if h . senderSettings . Via != nil { if h . senderSettings . Via != nil {
outbounds := session . OutboundsFromContext ( ctx ) outbounds := session . OutboundsFromContext ( ctx )
ob := outbounds [ len ( outbounds ) - 1 ] ob := outbounds [ len ( outbounds ) - 1 ]
if h . senderSettings . ViaCidr == "" { if h . senderSettings . ViaCidr == "" {
ob . Gateway = h . senderSettings . Via . AsAddress ( ) ob . Gateway = h . senderSettings . Via . AsAddress ( )
} else { //Get a random address.
} else { //Get a random address.
@ -285,7 +287,7 @@ func (h *Handler) Dial(ctx context.Context, dest net.Destination) (stat.Connecti
conn , err := internet . Dial ( ctx , dest , h . streamSettings ) conn , err := internet . Dial ( ctx , dest , h . streamSettings )
conn = h . getStatCouterConnection ( conn ) conn = h . getStatCouterConnection ( conn )
outbounds := session . OutboundsFromContext ( ctx ) outbounds := session . OutboundsFromContext ( ctx )
ob := outbounds [ len ( outbounds ) - 1 ] ob := outbounds [ len ( outbounds ) - 1 ]
ob . Conn = conn ob . Conn = conn
return conn , err return conn , err
} }
@ -317,7 +319,6 @@ func (h *Handler) Close() error {
return nil return nil
} }
func ParseRandomIPv6 ( address net . Address , prefix string ) net . Address { func ParseRandomIPv6 ( address net . Address , prefix string ) net . Address {
_ , network , _ := gonet . ParseCIDR ( address . IP ( ) . String ( ) + "/" + prefix ) _ , network , _ := gonet . ParseCIDR ( address . IP ( ) . String ( ) + "/" + prefix )
yuhan6665
5 months ago
committed by