mirror of
https://github.com/XTLS/Xray-core.git
synced 2024-09-06 16:29:19 +03:00
* 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.
258 lines
8.0 KiB
Go
258 lines
8.0 KiB
Go
package outbound
|
|
|
|
//go:generate go run github.com/xtls/xray-core/common/errors/errorgen
|
|
|
|
import (
|
|
"context"
|
|
"syscall"
|
|
"time"
|
|
|
|
"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/platform"
|
|
"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"
|
|
core "github.com/xtls/xray-core/core"
|
|
"github.com/xtls/xray-core/features/policy"
|
|
"github.com/xtls/xray-core/features/stats"
|
|
"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/stat"
|
|
"github.com/xtls/xray-core/transport/internet/xtls"
|
|
)
|
|
|
|
var xtls_show = false
|
|
|
|
func init() {
|
|
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
|
|
return New(ctx, config.(*Config))
|
|
}))
|
|
|
|
const defaultFlagValue = "NOT_DEFINED_AT_ALL"
|
|
|
|
xtlsShow := platform.NewEnvFlag("xray.vless.xtls.show").GetValue(func() string { return defaultFlagValue })
|
|
if xtlsShow == "true" {
|
|
xtls_show = true
|
|
}
|
|
}
|
|
|
|
// Handler is an outbound connection handler for VLess protocol.
|
|
type Handler struct {
|
|
serverList *protocol.ServerList
|
|
serverPicker protocol.ServerPicker
|
|
policyManager policy.Manager
|
|
cone bool
|
|
}
|
|
|
|
// 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),
|
|
}
|
|
|
|
return handler, nil
|
|
}
|
|
|
|
// Process implements proxy.Outbound.Process().
|
|
func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer internet.Dialer) error {
|
|
var rec *protocol.ServerSpec
|
|
var conn stat.Connection
|
|
|
|
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
|
|
statConn, ok := iConn.(*stat.CounterConnection)
|
|
if ok {
|
|
iConn = statConn.Connection
|
|
}
|
|
|
|
outbound := session.OutboundFromContext(ctx)
|
|
if outbound == nil || !outbound.Target.IsValid() {
|
|
return newError("target not specified").AtError()
|
|
}
|
|
|
|
target := outbound.Target
|
|
newError("tunneling request to ", target, " via ", rec.Destination().NetAddr()).AtInfo().WriteToLog(session.ExportIDToError(ctx))
|
|
|
|
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 rawConn syscall.RawConn
|
|
var sctx context.Context
|
|
|
|
allowUDP443 := false
|
|
switch requestAddons.Flow {
|
|
case vless.XRO + "-udp443", vless.XRD + "-udp443", vless.XRS + "-udp443":
|
|
allowUDP443 = true
|
|
requestAddons.Flow = requestAddons.Flow[:16]
|
|
fallthrough
|
|
case vless.XRO, vless.XRD, vless.XRS:
|
|
switch request.Command {
|
|
case protocol.RequestCommandMux:
|
|
return newError(requestAddons.Flow + " doesn't support Mux").AtWarning()
|
|
case protocol.RequestCommandUDP:
|
|
if !allowUDP443 && request.Port == 443 {
|
|
return newError(requestAddons.Flow + " stopped UDP/443").AtInfo()
|
|
}
|
|
requestAddons.Flow = ""
|
|
case protocol.RequestCommandTCP:
|
|
if xtlsConn, ok := iConn.(*xtls.Conn); ok {
|
|
xtlsConn.RPRX = true
|
|
xtlsConn.SHOW = xtls_show
|
|
xtlsConn.MARK = "XTLS"
|
|
if requestAddons.Flow == vless.XRS {
|
|
sctx = ctx
|
|
requestAddons.Flow = vless.XRD
|
|
}
|
|
if requestAddons.Flow == vless.XRD {
|
|
xtlsConn.DirectMode = true
|
|
if sc, ok := xtlsConn.Connection.(syscall.Conn); ok {
|
|
rawConn, _ = sc.SyscallConn()
|
|
}
|
|
}
|
|
} else {
|
|
return newError(`failed to use ` + requestAddons.Flow + `, maybe "security" is not "xtls"`).AtWarning()
|
|
}
|
|
}
|
|
default:
|
|
if _, ok := iConn.(*xtls.Conn); ok {
|
|
panic(`To avoid misunderstanding, you must fill in VLESS "flow" when using XTLS.`)
|
|
}
|
|
}
|
|
|
|
sessionPolicy := h.policyManager.ForLevel(request.User.Level)
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
|
|
|
|
clientReader := link.Reader // .(*pipe.Reader)
|
|
clientWriter := link.Writer // .(*pipe.Writer)
|
|
|
|
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)
|
|
}
|
|
|
|
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)
|
|
if request.Command == protocol.RequestCommandMux && request.Port == 666 {
|
|
serverWriter = xudp.NewPacketWriter(serverWriter, target)
|
|
}
|
|
if err := buf.CopyOnceTimeout(clientReader, serverWriter, time.Millisecond*100); err != nil && err != buf.ErrNotTimeoutReader && err != buf.ErrReadTimeout {
|
|
return err // ...
|
|
}
|
|
|
|
// 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()
|
|
}
|
|
|
|
// from clientReader.ReadMultiBuffer to serverWriter.WriteMultiBufer
|
|
if err := buf.Copy(clientReader, serverWriter, buf.UpdateActivity(timer)); err != nil {
|
|
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)
|
|
}
|
|
|
|
if rawConn != nil {
|
|
var counter stats.Counter
|
|
if statConn != nil {
|
|
counter = statConn.ReadCounter
|
|
}
|
|
err = encoding.ReadV(serverReader, clientWriter, timer, iConn.(*xtls.Conn), rawConn, counter, sctx)
|
|
} 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 err := task.Run(ctx, postRequest, task.OnSuccess(getResponse, task.Close(clientWriter))); err != nil {
|
|
return newError("connection ends").Base(err).AtInfo()
|
|
}
|
|
|
|
return nil
|
|
}
|