mirror of
https://github.com/XTLS/Xray-core.git
synced 2024-12-23 11:53:33 +02:00
efd32b0fb2
- Add outbound name - Add outbound conn in ctx - Refactor splice: it can be turn on from all inbounds and outbounds - Refactor splice: Add splice copy to vless inbound - Fix http error test - Add freedom splice toggle via env var - Populate outbound obj in context - Use CanSpliceCopy to mark a connection - Turn off splice by default
444 lines
12 KiB
Go
444 lines
12 KiB
Go
package dispatcher
|
|
|
|
//go:generate go run github.com/xtls/xray-core/common/errors/errorgen
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/xtls/xray-core/common"
|
|
"github.com/xtls/xray-core/common/buf"
|
|
"github.com/xtls/xray-core/common/log"
|
|
"github.com/xtls/xray-core/common/net"
|
|
"github.com/xtls/xray-core/common/protocol"
|
|
"github.com/xtls/xray-core/common/session"
|
|
"github.com/xtls/xray-core/core"
|
|
"github.com/xtls/xray-core/features/dns"
|
|
"github.com/xtls/xray-core/features/outbound"
|
|
"github.com/xtls/xray-core/features/policy"
|
|
"github.com/xtls/xray-core/features/routing"
|
|
routing_session "github.com/xtls/xray-core/features/routing/session"
|
|
"github.com/xtls/xray-core/features/stats"
|
|
"github.com/xtls/xray-core/transport"
|
|
"github.com/xtls/xray-core/transport/pipe"
|
|
)
|
|
|
|
var errSniffingTimeout = newError("timeout on sniffing")
|
|
|
|
type cachedReader struct {
|
|
sync.Mutex
|
|
reader *pipe.Reader
|
|
cache buf.MultiBuffer
|
|
}
|
|
|
|
func (r *cachedReader) Cache(b *buf.Buffer) {
|
|
mb, _ := r.reader.ReadMultiBufferTimeout(time.Millisecond * 100)
|
|
r.Lock()
|
|
if !mb.IsEmpty() {
|
|
r.cache, _ = buf.MergeMulti(r.cache, mb)
|
|
}
|
|
b.Clear()
|
|
rawBytes := b.Extend(buf.Size)
|
|
n := r.cache.Copy(rawBytes)
|
|
b.Resize(0, int32(n))
|
|
r.Unlock()
|
|
}
|
|
|
|
func (r *cachedReader) readInternal() buf.MultiBuffer {
|
|
r.Lock()
|
|
defer r.Unlock()
|
|
|
|
if r.cache != nil && !r.cache.IsEmpty() {
|
|
mb := r.cache
|
|
r.cache = nil
|
|
return mb
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
|
|
mb := r.readInternal()
|
|
if mb != nil {
|
|
return mb, nil
|
|
}
|
|
|
|
return r.reader.ReadMultiBuffer()
|
|
}
|
|
|
|
func (r *cachedReader) ReadMultiBufferTimeout(timeout time.Duration) (buf.MultiBuffer, error) {
|
|
mb := r.readInternal()
|
|
if mb != nil {
|
|
return mb, nil
|
|
}
|
|
|
|
return r.reader.ReadMultiBufferTimeout(timeout)
|
|
}
|
|
|
|
func (r *cachedReader) Interrupt() {
|
|
r.Lock()
|
|
if r.cache != nil {
|
|
r.cache = buf.ReleaseMulti(r.cache)
|
|
}
|
|
r.Unlock()
|
|
r.reader.Interrupt()
|
|
}
|
|
|
|
// DefaultDispatcher is a default implementation of Dispatcher.
|
|
type DefaultDispatcher struct {
|
|
ohm outbound.Manager
|
|
router routing.Router
|
|
policy policy.Manager
|
|
stats stats.Manager
|
|
dns dns.Client
|
|
fdns dns.FakeDNSEngine
|
|
}
|
|
|
|
func init() {
|
|
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
|
|
d := new(DefaultDispatcher)
|
|
if err := core.RequireFeatures(ctx, func(om outbound.Manager, router routing.Router, pm policy.Manager, sm stats.Manager, dc dns.Client) error {
|
|
core.RequireFeatures(ctx, func(fdns dns.FakeDNSEngine) {
|
|
d.fdns = fdns
|
|
})
|
|
return d.Init(config.(*Config), om, router, pm, sm, dc)
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
return d, nil
|
|
}))
|
|
}
|
|
|
|
// Init initializes DefaultDispatcher.
|
|
func (d *DefaultDispatcher) Init(config *Config, om outbound.Manager, router routing.Router, pm policy.Manager, sm stats.Manager, dns dns.Client) error {
|
|
d.ohm = om
|
|
d.router = router
|
|
d.policy = pm
|
|
d.stats = sm
|
|
d.dns = dns
|
|
return nil
|
|
}
|
|
|
|
// Type implements common.HasType.
|
|
func (*DefaultDispatcher) Type() interface{} {
|
|
return routing.DispatcherType()
|
|
}
|
|
|
|
// Start implements common.Runnable.
|
|
func (*DefaultDispatcher) Start() error {
|
|
return nil
|
|
}
|
|
|
|
// Close implements common.Closable.
|
|
func (*DefaultDispatcher) Close() error { return nil }
|
|
|
|
func (d *DefaultDispatcher) getLink(ctx context.Context) (*transport.Link, *transport.Link) {
|
|
opt := pipe.OptionsFromContext(ctx)
|
|
uplinkReader, uplinkWriter := pipe.New(opt...)
|
|
downlinkReader, downlinkWriter := pipe.New(opt...)
|
|
|
|
inboundLink := &transport.Link{
|
|
Reader: downlinkReader,
|
|
Writer: uplinkWriter,
|
|
}
|
|
|
|
outboundLink := &transport.Link{
|
|
Reader: uplinkReader,
|
|
Writer: downlinkWriter,
|
|
}
|
|
|
|
sessionInbound := session.InboundFromContext(ctx)
|
|
var user *protocol.MemoryUser
|
|
if sessionInbound != nil {
|
|
user = sessionInbound.User
|
|
}
|
|
|
|
if user != nil && len(user.Email) > 0 {
|
|
p := d.policy.ForLevel(user.Level)
|
|
if p.Stats.UserUplink {
|
|
name := "user>>>" + user.Email + ">>>traffic>>>uplink"
|
|
if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
|
|
inboundLink.Writer = &SizeStatWriter{
|
|
Counter: c,
|
|
Writer: inboundLink.Writer,
|
|
}
|
|
}
|
|
}
|
|
if p.Stats.UserDownlink {
|
|
name := "user>>>" + user.Email + ">>>traffic>>>downlink"
|
|
if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
|
|
outboundLink.Writer = &SizeStatWriter{
|
|
Counter: c,
|
|
Writer: outboundLink.Writer,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return inboundLink, outboundLink
|
|
}
|
|
|
|
func (d *DefaultDispatcher) shouldOverride(ctx context.Context, result SniffResult, request session.SniffingRequest, destination net.Destination) bool {
|
|
domain := result.Domain()
|
|
if domain == "" {
|
|
return false
|
|
}
|
|
for _, d := range request.ExcludeForDomain {
|
|
if strings.ToLower(domain) == d {
|
|
return false
|
|
}
|
|
}
|
|
protocolString := result.Protocol()
|
|
if resComp, ok := result.(SnifferResultComposite); ok {
|
|
protocolString = resComp.ProtocolForDomainResult()
|
|
}
|
|
for _, p := range request.OverrideDestinationForProtocol {
|
|
if strings.HasPrefix(protocolString, p) || strings.HasPrefix(p, protocolString) {
|
|
return true
|
|
}
|
|
if fkr0, ok := d.fdns.(dns.FakeDNSEngineRev0); ok && protocolString != "bittorrent" && p == "fakedns" &&
|
|
destination.Address.Family().IsIP() && fkr0.IsIPInIPPool(destination.Address) {
|
|
newError("Using sniffer ", protocolString, " since the fake DNS missed").WriteToLog(session.ExportIDToError(ctx))
|
|
return true
|
|
}
|
|
if resultSubset, ok := result.(SnifferIsProtoSubsetOf); ok {
|
|
if resultSubset.IsProtoSubsetOf(p) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// Dispatch implements routing.Dispatcher.
|
|
func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destination) (*transport.Link, error) {
|
|
if !destination.IsValid() {
|
|
panic("Dispatcher: Invalid destination.")
|
|
}
|
|
ob := session.OutboundFromContext(ctx)
|
|
if ob == nil {
|
|
ob = &session.Outbound{}
|
|
ctx = session.ContextWithOutbound(ctx, ob)
|
|
}
|
|
ob.OriginalTarget = destination
|
|
ob.Target = destination
|
|
content := session.ContentFromContext(ctx)
|
|
if content == nil {
|
|
content = new(session.Content)
|
|
ctx = session.ContextWithContent(ctx, content)
|
|
}
|
|
sniffingRequest := content.SniffingRequest
|
|
inbound, outbound := d.getLink(ctx)
|
|
if !sniffingRequest.Enabled {
|
|
go d.routedDispatch(ctx, outbound, destination)
|
|
} else {
|
|
go func() {
|
|
cReader := &cachedReader{
|
|
reader: outbound.Reader.(*pipe.Reader),
|
|
}
|
|
outbound.Reader = cReader
|
|
result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly, destination.Network)
|
|
if err == nil {
|
|
content.Protocol = result.Protocol()
|
|
}
|
|
if err == nil && d.shouldOverride(ctx, result, sniffingRequest, destination) {
|
|
domain := result.Domain()
|
|
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
|
|
destination.Address = net.ParseAddress(domain)
|
|
protocol := result.Protocol()
|
|
if resComp, ok := result.(SnifferResultComposite); ok {
|
|
protocol = resComp.ProtocolForDomainResult()
|
|
}
|
|
isFakeIP := false
|
|
if fkr0, ok := d.fdns.(dns.FakeDNSEngineRev0); ok && ob.Target.Address.Family().IsIP() && fkr0.IsIPInIPPool(ob.Target.Address) {
|
|
isFakeIP = true
|
|
}
|
|
if sniffingRequest.RouteOnly && protocol != "fakedns" && protocol != "fakedns+others" && !isFakeIP {
|
|
ob.RouteTarget = destination
|
|
} else {
|
|
ob.Target = destination
|
|
}
|
|
}
|
|
d.routedDispatch(ctx, outbound, destination)
|
|
}()
|
|
}
|
|
return inbound, nil
|
|
}
|
|
|
|
// DispatchLink implements routing.Dispatcher.
|
|
func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.Destination, outbound *transport.Link) error {
|
|
if !destination.IsValid() {
|
|
return newError("Dispatcher: Invalid destination.")
|
|
}
|
|
ob := session.OutboundFromContext(ctx)
|
|
if ob == nil {
|
|
ob = &session.Outbound{}
|
|
ctx = session.ContextWithOutbound(ctx, ob)
|
|
}
|
|
ob.OriginalTarget = destination
|
|
ob.Target = destination
|
|
content := session.ContentFromContext(ctx)
|
|
if content == nil {
|
|
content = new(session.Content)
|
|
ctx = session.ContextWithContent(ctx, content)
|
|
}
|
|
sniffingRequest := content.SniffingRequest
|
|
if !sniffingRequest.Enabled {
|
|
d.routedDispatch(ctx, outbound, destination)
|
|
} else {
|
|
cReader := &cachedReader{
|
|
reader: outbound.Reader.(*pipe.Reader),
|
|
}
|
|
outbound.Reader = cReader
|
|
result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly, destination.Network)
|
|
if err == nil {
|
|
content.Protocol = result.Protocol()
|
|
}
|
|
if err == nil && d.shouldOverride(ctx, result, sniffingRequest, destination) {
|
|
domain := result.Domain()
|
|
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
|
|
destination.Address = net.ParseAddress(domain)
|
|
protocol := result.Protocol()
|
|
if resComp, ok := result.(SnifferResultComposite); ok {
|
|
protocol = resComp.ProtocolForDomainResult()
|
|
}
|
|
isFakeIP := false
|
|
if fkr0, ok := d.fdns.(dns.FakeDNSEngineRev0); ok && ob.Target.Address.Family().IsIP() && fkr0.IsIPInIPPool(ob.Target.Address) {
|
|
isFakeIP = true
|
|
}
|
|
if sniffingRequest.RouteOnly && protocol != "fakedns" && protocol != "fakedns+others" && !isFakeIP {
|
|
ob.RouteTarget = destination
|
|
} else {
|
|
ob.Target = destination
|
|
}
|
|
}
|
|
d.routedDispatch(ctx, outbound, destination)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool, network net.Network) (SniffResult, error) {
|
|
payload := buf.New()
|
|
defer payload.Release()
|
|
|
|
sniffer := NewSniffer(ctx)
|
|
|
|
metaresult, metadataErr := sniffer.SniffMetadata(ctx)
|
|
|
|
if metadataOnly {
|
|
return metaresult, metadataErr
|
|
}
|
|
|
|
contentResult, contentErr := func() (SniffResult, error) {
|
|
totalAttempt := 0
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
default:
|
|
totalAttempt++
|
|
if totalAttempt > 2 {
|
|
return nil, errSniffingTimeout
|
|
}
|
|
|
|
cReader.Cache(payload)
|
|
if !payload.IsEmpty() {
|
|
result, err := sniffer.Sniff(ctx, payload.Bytes(), network)
|
|
if err != common.ErrNoClue {
|
|
return result, err
|
|
}
|
|
}
|
|
if payload.IsFull() {
|
|
return nil, errUnknownContent
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
if contentErr != nil && metadataErr == nil {
|
|
return metaresult, nil
|
|
}
|
|
if contentErr == nil && metadataErr == nil {
|
|
return CompositeResult(metaresult, contentResult), nil
|
|
}
|
|
return contentResult, contentErr
|
|
}
|
|
|
|
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.Link, destination net.Destination) {
|
|
ob := session.OutboundFromContext(ctx)
|
|
if hosts, ok := d.dns.(dns.HostsLookup); ok && destination.Address.Family().IsDomain() {
|
|
proxied := hosts.LookupHosts(ob.Target.String())
|
|
if proxied != nil {
|
|
ro := ob.RouteTarget == destination
|
|
destination.Address = *proxied
|
|
if ro {
|
|
ob.RouteTarget = destination
|
|
} else {
|
|
ob.Target = destination
|
|
}
|
|
}
|
|
}
|
|
|
|
var handler outbound.Handler
|
|
|
|
routingLink := routing_session.AsRoutingContext(ctx)
|
|
inTag := routingLink.GetInboundTag()
|
|
isPickRoute := 0
|
|
if forcedOutboundTag := session.GetForcedOutboundTagFromContext(ctx); forcedOutboundTag != "" {
|
|
ctx = session.SetForcedOutboundTagToContext(ctx, "")
|
|
if h := d.ohm.GetHandler(forcedOutboundTag); h != nil {
|
|
isPickRoute = 1
|
|
newError("taking platform initialized detour [", forcedOutboundTag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))
|
|
handler = h
|
|
} else {
|
|
newError("non existing tag for platform initialized detour: ", forcedOutboundTag).AtError().WriteToLog(session.ExportIDToError(ctx))
|
|
common.Close(link.Writer)
|
|
common.Interrupt(link.Reader)
|
|
return
|
|
}
|
|
} else if d.router != nil {
|
|
if route, err := d.router.PickRoute(routingLink); err == nil {
|
|
outTag := route.GetOutboundTag()
|
|
if h := d.ohm.GetHandler(outTag); h != nil {
|
|
isPickRoute = 2
|
|
newError("taking detour [", outTag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))
|
|
handler = h
|
|
} else {
|
|
newError("non existing outTag: ", outTag).AtWarning().WriteToLog(session.ExportIDToError(ctx))
|
|
}
|
|
} else {
|
|
newError("default route for ", destination).WriteToLog(session.ExportIDToError(ctx))
|
|
}
|
|
}
|
|
|
|
if handler == nil {
|
|
handler = d.ohm.GetDefaultHandler()
|
|
}
|
|
|
|
if handler == nil {
|
|
newError("default outbound handler not exist").WriteToLog(session.ExportIDToError(ctx))
|
|
common.Close(link.Writer)
|
|
common.Interrupt(link.Reader)
|
|
return
|
|
}
|
|
|
|
if accessMessage := log.AccessMessageFromContext(ctx); accessMessage != nil {
|
|
if tag := handler.Tag(); tag != "" {
|
|
if inTag == "" {
|
|
accessMessage.Detour = tag
|
|
} else if isPickRoute == 1 {
|
|
accessMessage.Detour = inTag + " ==> " + tag
|
|
} else if isPickRoute == 2 {
|
|
accessMessage.Detour = inTag + " -> " + tag
|
|
} else {
|
|
accessMessage.Detour = inTag + " >> " + tag
|
|
}
|
|
}
|
|
log.Record(accessMessage)
|
|
}
|
|
|
|
handler.Dispatch(ctx, link)
|
|
}
|