Xray-core/proxy/vless/outbound/outbound.go

297 lines
9.6 KiB
Go
Raw Normal View History

2020-11-25 13:01:53 +02:00
package outbound
2020-12-04 03:36:16 +02:00
//go:generate go run github.com/xtls/xray-core/common/errors/errorgen
2020-11-25 13:01:53 +02:00
import (
"bytes"
2020-11-25 13:01:53 +02:00
"context"
gotls "crypto/tls"
"reflect"
"time"
"unsafe"
2020-11-25 13:01:53 +02:00
utls "github.com/refraction-networking/utls"
2020-12-04 03:36:16 +02:00
"github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/net"
"github.com/xtls/xray-core/common/protocol"
"github.com/xtls/xray-core/common/retry"
"github.com/xtls/xray-core/common/session"
"github.com/xtls/xray-core/common/signal"
"github.com/xtls/xray-core/common/task"
"github.com/xtls/xray-core/common/xudp"
"github.com/xtls/xray-core/core"
2020-12-04 03:36:16 +02:00
"github.com/xtls/xray-core/features/policy"
"github.com/xtls/xray-core/proxy"
2020-12-04 03:36:16 +02:00
"github.com/xtls/xray-core/proxy/vless"
"github.com/xtls/xray-core/proxy/vless/encoding"
"github.com/xtls/xray-core/transport"
"github.com/xtls/xray-core/transport/internet"
"github.com/xtls/xray-core/transport/internet/reality"
"github.com/xtls/xray-core/transport/internet/stat"
"github.com/xtls/xray-core/transport/internet/tls"
2020-11-25 13:01:53 +02:00
)
func init() {
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
return New(ctx, config.(*Config))
}))
}
// Handler is an outbound connection handler for VLess protocol.
type Handler struct {
serverList *protocol.ServerList
serverPicker protocol.ServerPicker
policyManager policy.Manager
cone bool
2020-11-25 13:01:53 +02:00
}
// New creates a new VLess outbound handler.
func New(ctx context.Context, config *Config) (*Handler, error) {
serverList := protocol.NewServerList()
for _, rec := range config.Vnext {
s, err := protocol.NewServerSpecFromPB(rec)
if err != nil {
return nil, newError("failed to parse server spec").Base(err).AtError()
}
serverList.AddServer(s)
}
v := core.MustFromContext(ctx)
handler := &Handler{
serverList: serverList,
serverPicker: protocol.NewRoundRobinServerPicker(serverList),
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
cone: ctx.Value("cone").(bool),
2020-11-25 13:01:53 +02:00
}
return handler, nil
}
// Process implements proxy.Outbound.Process().
func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer internet.Dialer) error {
outbound := session.OutboundFromContext(ctx)
if outbound == nil || !outbound.Target.IsValid() {
return newError("target not specified").AtError()
}
outbound.Name = "vless"
inbound := session.InboundFromContext(ctx)
2020-11-25 13:01:53 +02:00
var rec *protocol.ServerSpec
var conn stat.Connection
2020-11-25 13:01:53 +02:00
if err := retry.ExponentialBackoff(5, 200).On(func() error {
rec = h.serverPicker.PickServer()
var err error
conn, err = dialer.Dial(ctx, rec.Destination())
if err != nil {
return err
}
return nil
}); err != nil {
return newError("failed to find an available destination").Base(err).AtWarning()
}
defer conn.Close()
iConn := conn
if statConn, ok := iConn.(*stat.CounterConnection); ok {
2020-11-25 13:01:53 +02:00
iConn = statConn.Connection
}
target := outbound.Target
Quic related improvements (#915) * DialSystem for Quic DialSystem() is needed in case of Android client, where the raw conn is protected for vpn service * Fix client dialer log Log such as: tunneling request to tcp:www.google.com:80 via tcp:x.x.x.x:443 the second "tcp" is misleading when using mKcp or quic transport Remove the second "tcp" and add the correct logging for transport dialer: - transport/internet/tcp: dialing TCP to tcp:x.x.x.x:443 - transport/internet/quic: dialing quic to udp:x.x.x.x:443 * Quic new stream allocation mode Currently this is how Quic works: client muxing all tcp and udp traffic through a single session, when there are more than 32 running streams in the session, the next stream request will fail and open with a new session (port). Imagine lineup the session from left to right: | | | | | | As the streams finishes, we still open stream from the left, original session. So the base session will always be there and new sessions on the right come and go. However, either due to QOS or bugs in Quic implementation, the traffic "wear out" the base session. It will become slower and in the end not receiving any data from server side. I couldn't figure out a solution for this problem at the moment, as a workaround: | | | | | | | | I came up with this new stream allocation mode, that it will never open new streams in the old sessions, but only from current or new session from right. The keeplive config is turned off from server and client side. This way old sessions will natually close and new sessions keep generating. Note the frequency of new session is still controlled by the server side. Server can assign a large max stream limit. In this case the new allocation mode will be similar to the current mode.
2022-01-29 01:11:30 +02:00
newError("tunneling request to ", target, " via ", rec.Destination().NetAddr()).AtInfo().WriteToLog(session.ExportIDToError(ctx))
2020-11-25 13:01:53 +02:00
command := protocol.RequestCommandTCP
if target.Network == net.Network_UDP {
command = protocol.RequestCommandUDP
}
if target.Address.Family().IsDomain() && target.Address.Domain() == "v1.mux.cool" {
command = protocol.RequestCommandMux
}
request := &protocol.RequestHeader{
Version: encoding.Version,
User: rec.PickUser(),
Command: command,
Address: target.Address,
Port: target.Port,
}
account := request.User.Account.(*vless.MemoryAccount)
requestAddons := &encoding.Addons{
Flow: account.Flow,
}
var input *bytes.Reader
var rawInput *bytes.Buffer
2020-11-25 13:01:53 +02:00
allowUDP443 := false
switch requestAddons.Flow {
2023-03-04 12:39:26 +02:00
case vless.XRV + "-udp443":
2020-11-25 13:01:53 +02:00
allowUDP443 = true
requestAddons.Flow = requestAddons.Flow[:16]
fallthrough
2023-03-04 12:39:26 +02:00
case vless.XRV:
if inbound != nil {
inbound.SetCanSpliceCopy(2)
}
2020-11-25 13:01:53 +02:00
switch request.Command {
case protocol.RequestCommandUDP:
if !allowUDP443 && request.Port == 443 {
return newError("XTLS rejected UDP/443 traffic").AtInfo()
2020-11-25 13:01:53 +02:00
}
requestAddons.Flow = ""
case protocol.RequestCommandMux:
fallthrough // let server break Mux connections that contain TCP requests
2020-11-25 13:01:53 +02:00
case protocol.RequestCommandTCP:
2023-03-04 12:39:26 +02:00
var t reflect.Type
var p uintptr
if tlsConn, ok := iConn.(*tls.Conn); ok {
t = reflect.TypeOf(tlsConn.Conn).Elem()
p = uintptr(unsafe.Pointer(tlsConn.Conn))
} else if utlsConn, ok := iConn.(*tls.UConn); ok {
t = reflect.TypeOf(utlsConn.Conn).Elem()
p = uintptr(unsafe.Pointer(utlsConn.Conn))
} else if realityConn, ok := iConn.(*reality.UConn); ok {
t = reflect.TypeOf(realityConn.Conn).Elem()
p = uintptr(unsafe.Pointer(realityConn.Conn))
2020-11-25 13:01:53 +02:00
} else {
return newError("XTLS only supports TLS and REALITY directly for now.").AtWarning()
2020-11-25 13:01:53 +02:00
}
2023-03-04 12:39:26 +02:00
i, _ := t.FieldByName("input")
r, _ := t.FieldByName("rawInput")
input = (*bytes.Reader)(unsafe.Pointer(p + i.Offset))
rawInput = (*bytes.Buffer)(unsafe.Pointer(p + r.Offset))
2020-11-25 13:01:53 +02:00
}
default:
if inbound != nil {
inbound.SetCanSpliceCopy(3)
}
2020-11-25 13:01:53 +02:00
}
var newCtx context.Context
var newCancel context.CancelFunc
if session.TimeoutOnlyFromContext(ctx) {
newCtx, newCancel = context.WithCancel(context.Background())
}
2020-11-25 13:01:53 +02:00
sessionPolicy := h.policyManager.ForLevel(request.User.Level)
ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, func() {
cancel()
if newCancel != nil {
newCancel()
}
}, sessionPolicy.Timeouts.ConnectionIdle)
2020-11-25 13:01:53 +02:00
clientReader := link.Reader // .(*pipe.Reader)
clientWriter := link.Writer // .(*pipe.Writer)
trafficState := proxy.NewTrafficState(account.ID.Bytes())
if request.Command == protocol.RequestCommandUDP && h.cone && request.Port != 53 && request.Port != 443 {
request.Command = protocol.RequestCommandMux
request.Address = net.DomainAddress("v1.mux.cool")
request.Port = net.Port(666)
}
2020-11-25 13:01:53 +02:00
postRequest := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(conn))
if err := encoding.EncodeRequestHeader(bufferWriter, request, requestAddons); err != nil {
return newError("failed to encode request header").Base(err).AtWarning()
}
// default: serverWriter := bufferWriter
serverWriter := encoding.EncodeBodyAddons(bufferWriter, request, requestAddons, trafficState, ctx)
if request.Command == protocol.RequestCommandMux && request.Port == 666 {
serverWriter = xudp.NewPacketWriter(serverWriter, target, xudp.GetGlobalID(ctx))
}
timeoutReader, ok := clientReader.(buf.TimeoutReader)
if ok {
2022-12-26 02:37:35 +02:00
multiBuffer, err1 := timeoutReader.ReadMultiBufferTimeout(time.Millisecond * 500)
if err1 == nil {
if err := serverWriter.WriteMultiBuffer(multiBuffer); err != nil {
return err // ...
}
} else if err1 != buf.ErrReadTimeout {
return err1
} else if requestAddons.Flow == vless.XRV {
mb := make(buf.MultiBuffer, 1)
newError("Insert padding with empty content to camouflage VLESS header ", mb.Len()).WriteToLog(session.ExportIDToError(ctx))
if err := serverWriter.WriteMultiBuffer(mb); err != nil {
return err // ...
}
}
} else {
newError("Reader is not timeout reader, will send out vless header separately from first payload").AtDebug().WriteToLog(session.ExportIDToError(ctx))
2020-11-25 13:01:53 +02:00
}
// Flush; bufferWriter.WriteMultiBufer now is bufferWriter.writer.WriteMultiBuffer
if err := bufferWriter.SetBuffered(false); err != nil {
return newError("failed to write A request payload").Base(err).AtWarning()
}
var err error
2023-06-22 09:43:22 +03:00
if requestAddons.Flow == vless.XRV {
if tlsConn, ok := iConn.(*tls.Conn); ok {
if tlsConn.ConnectionState().Version != gotls.VersionTLS13 {
return newError(`failed to use `+requestAddons.Flow+`, found outer tls version `, tlsConn.ConnectionState().Version).AtWarning()
}
} else if utlsConn, ok := iConn.(*tls.UConn); ok {
if utlsConn.ConnectionState().Version != utls.VersionTLS13 {
return newError(`failed to use `+requestAddons.Flow+`, found outer tls version `, utlsConn.ConnectionState().Version).AtWarning()
}
}
ctx1 := session.ContextWithOutbound(ctx, nil) // TODO enable splice
err = encoding.XtlsWrite(clientReader, serverWriter, timer, conn, trafficState, ctx1)
} else {
// from clientReader.ReadMultiBuffer to serverWriter.WriteMultiBufer
err = buf.Copy(clientReader, serverWriter, buf.UpdateActivity(timer))
}
if err != nil {
2020-11-25 13:01:53 +02:00
return newError("failed to transfer request payload").Base(err).AtInfo()
}
// Indicates the end of request payload.
switch requestAddons.Flow {
default:
}
return nil
}
getResponse := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
responseAddons, err := encoding.DecodeResponseHeader(conn, request)
if err != nil {
return newError("failed to decode response header").Base(err).AtInfo()
}
// default: serverReader := buf.NewReader(conn)
serverReader := encoding.DecodeBodyAddons(conn, request, responseAddons)
if request.Command == protocol.RequestCommandMux && request.Port == 666 {
serverReader = xudp.NewPacketReader(conn)
}
2020-11-25 13:01:53 +02:00
2023-06-22 09:43:22 +03:00
if requestAddons.Flow == vless.XRV {
err = encoding.XtlsRead(serverReader, clientWriter, timer, conn, input, rawInput, trafficState, ctx)
2020-11-25 13:01:53 +02:00
} else {
// from serverReader.ReadMultiBuffer to clientWriter.WriteMultiBufer
err = buf.Copy(serverReader, clientWriter, buf.UpdateActivity(timer))
}
if err != nil {
return newError("failed to transfer response payload").Base(err).AtInfo()
}
return nil
}
if newCtx != nil {
ctx = newCtx
}
2020-11-25 13:01:53 +02:00
if err := task.Run(ctx, postRequest, task.OnSuccess(getResponse, task.Close(clientWriter))); err != nil {
return newError("connection ends").Base(err).AtInfo()
}
return nil
}