Xray-core/app/dispatcher/default.go

447 lines
12 KiB
Go
Raw Normal View History

2020-11-25 13:01:53 +02:00
package dispatcher
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 (
"context"
"strings"
"sync"
"time"
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/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"
2020-12-04 03:36:16 +02:00
"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"
2020-11-25 13:01:53 +02:00
)
2021-10-19 19:57:14 +03:00
var errSniffingTimeout = newError("timeout on sniffing")
2020-11-25 13:01:53 +02:00
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
hosts dns.HostsLookup
2020-11-25 13:01:53 +02:00
}
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 {
return d.Init(config.(*Config), om, router, pm, sm, dc)
2020-11-25 13:01:53 +02:00
}); 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, dc dns.Client) error {
2020-11-25 13:01:53 +02:00
d.ohm = om
d.router = router
d.policy = pm
d.stats = sm
if hosts, ok := dc.(dns.HostsLookup); ok {
d.hosts = hosts
}
2020-11-25 13:01:53 +02:00
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 shouldOverride(ctx context.Context, result SniffResult, request session.SniffingRequest, destination net.Destination) bool {
domain := result.Domain()
for _, d := range request.ExcludeForDomain {
if strings.ToLower(domain) == d {
return false
}
}
var fakeDNSEngine dns.FakeDNSEngine
core.RequireFeatures(ctx, func(fdns dns.FakeDNSEngine) {
fakeDNSEngine = fdns
})
protocolString := result.Protocol()
if resComp, ok := result.(SnifferResultComposite); ok {
protocolString = resComp.ProtocolForDomainResult()
}
for _, p := range request.OverrideDestinationForProtocol {
if strings.HasPrefix(protocolString, p) {
return true
}
if fkr0, ok := fakeDNSEngine.(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
}
2020-11-25 13:01:53 +02:00
}
}
2020-11-25 13:01:53 +02:00
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.Outbound{
Target: destination,
}
ctx = session.ContextWithOutbound(ctx, ob)
inbound, outbound := d.getLink(ctx)
content := session.ContentFromContext(ctx)
if content == nil {
content = new(session.Content)
ctx = session.ContextWithContent(ctx, content)
}
sniffingRequest := content.SniffingRequest
switch {
case !sniffingRequest.Enabled:
go d.routedDispatch(ctx, outbound, destination)
case destination.Network != net.Network_TCP:
// Only metadata sniff will be used for non tcp connection
result, err := sniffer(ctx, nil, true)
if err == nil {
content.Protocol = result.Protocol()
if shouldOverride(ctx, result, sniffingRequest, destination) {
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
destination.Address = net.ParseAddress(domain)
2021-09-16 10:05:48 +03:00
if sniffingRequest.RouteOnly && result.Protocol() != "fakedns" {
ob.RouteTarget = destination
} else {
ob.Target = destination
}
}
}
2020-11-25 13:01:53 +02:00
go d.routedDispatch(ctx, outbound, destination)
default:
2020-11-25 13:01:53 +02:00
go func() {
cReader := &cachedReader{
reader: outbound.Reader.(*pipe.Reader),
}
outbound.Reader = cReader
result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly)
2020-11-25 13:01:53 +02:00
if err == nil {
content.Protocol = result.Protocol()
}
if err == nil && shouldOverride(ctx, result, sniffingRequest, destination) {
2020-11-25 13:01:53 +02:00
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
destination.Address = net.ParseAddress(domain)
2021-09-16 10:05:48 +03:00
if sniffingRequest.RouteOnly && result.Protocol() != "fakedns" {
ob.RouteTarget = destination
} else {
ob.Target = destination
}
2020-11-25 13:01:53 +02:00
}
d.routedDispatch(ctx, outbound, destination)
}()
}
return inbound, nil
}
2021-09-28 04:42:57 +03:00
// 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.Outbound{
Target: destination,
}
ctx = session.ContextWithOutbound(ctx, ob)
content := session.ContentFromContext(ctx)
if content == nil {
content = new(session.Content)
ctx = session.ContextWithContent(ctx, content)
}
sniffingRequest := content.SniffingRequest
switch {
case !sniffingRequest.Enabled:
go d.routedDispatch(ctx, outbound, destination)
case destination.Network != net.Network_TCP:
// Only metadata sniff will be used for non tcp connection
result, err := sniffer(ctx, nil, true)
if err == nil {
content.Protocol = result.Protocol()
if shouldOverride(ctx, result, sniffingRequest, destination) {
2021-09-28 04:42:57 +03:00
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
destination.Address = net.ParseAddress(domain)
if sniffingRequest.RouteOnly && result.Protocol() != "fakedns" {
ob.RouteTarget = destination
} else {
ob.Target = destination
}
}
}
go d.routedDispatch(ctx, outbound, destination)
default:
go func() {
cReader := &cachedReader{
reader: outbound.Reader.(*pipe.Reader),
}
outbound.Reader = cReader
result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly)
if err == nil {
content.Protocol = result.Protocol()
}
if err == nil && shouldOverride(ctx, result, sniffingRequest, destination) {
2021-09-28 04:42:57 +03:00
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
destination.Address = net.ParseAddress(domain)
if sniffingRequest.RouteOnly && result.Protocol() != "fakedns" {
ob.RouteTarget = destination
} else {
ob.Target = destination
}
}
d.routedDispatch(ctx, outbound, destination)
}()
}
return nil
}
func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool) (SniffResult, error) {
2020-11-25 13:01:53 +02:00
payload := buf.New()
defer payload.Release()
sniffer := NewSniffer(ctx)
metaresult, metadataErr := sniffer.SniffMetadata(ctx)
if metadataOnly {
return metaresult, metadataErr
}
2020-11-25 13:01:53 +02:00
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())
if err != common.ErrNoClue {
return result, err
}
}
if payload.IsFull() {
return nil, errUnknownContent
2020-11-25 13:01:53 +02:00
}
}
}
}()
if contentErr != nil && metadataErr == nil {
return metaresult, nil
}
if contentErr == nil && metadataErr == nil {
return CompositeResult(metaresult, contentResult), nil
2020-11-25 13:01:53 +02:00
}
return contentResult, contentErr
2020-11-25 13:01:53 +02:00
}
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.Link, destination net.Destination) {
ob := session.OutboundFromContext(ctx)
if d.hosts != nil && destination.Address.Family().IsDomain() {
proxied := d.hosts.LookupHosts(ob.Target.String())
if proxied != nil {
ro := ob.RouteTarget == destination
destination.Address = *proxied
if ro {
ob.RouteTarget = destination
} else {
ob.Target = destination
}
}
}
2020-11-25 13:01:53 +02:00
var handler outbound.Handler
if forcedOutboundTag := session.GetForcedOutboundTagFromContext(ctx); forcedOutboundTag != "" {
ctx = session.SetForcedOutboundTagToContext(ctx, "")
if h := d.ohm.GetHandler(forcedOutboundTag); h != nil {
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 {
Merge dns (#722) * DNS: add clientip for specific nameserver * Refactoring: DNS App * DNS: add DNS over QUIC support * Feat: add disableCache option for DNS * Feat: add queryStrategy option for DNS * Feat: add disableFallback & skipFallback option for DNS * Feat: DNS hosts support multiple addresses * Feat: DNS transport over TCP * DNS: fix typo & refine code * DNS: refine code * Add disableFallbackIfMatch dns option * Feat: routing and freedom outbound ignore Fake DNS Turn off fake DNS for request sent from Routing and Freedom outbound. Fake DNS now only apply to DNS outbound. This is important for Android, where VPN service take over all system DNS traffic and pass it to core. "UseIp" option can be used in Freedom outbound to avoid getting fake IP and fail connection. * Fix test * Fix dns return * Fix local dns return empty * Apply timeout to dns outbound * Update app/dns/config.go Co-authored-by: Loyalsoldier <10487845+loyalsoldier@users.noreply.github.com> Co-authored-by: Ye Zhihao <vigilans@foxmail.com> Co-authored-by: maskedeken <52683904+maskedeken@users.noreply.github.com> Co-authored-by: V2Fly Team <51714622+vcptr@users.noreply.github.com> Co-authored-by: CalmLong <37164399+calmlong@users.noreply.github.com> Co-authored-by: Shelikhoo <xiaokangwang@outlook.com> Co-authored-by: 秋のかえで <autmaple@protonmail.com> Co-authored-by: 朱聖黎 <digglife@gmail.com> Co-authored-by: rurirei <72071920+rurirei@users.noreply.github.com> Co-authored-by: yuhan6665 <1588741+yuhan6665@users.noreply.github.com> Co-authored-by: Arthur Morgan <4637240+badO1a5A90@users.noreply.github.com>
2021-10-16 16:02:51 +03:00
if route, err := d.router.PickRoute(routing_session.AsRoutingContext(ctx)); err == nil {
tag := route.GetOutboundTag()
if h := d.ohm.GetHandler(tag); h != nil {
newError("taking detour [", tag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))
2020-11-25 13:01:53 +02:00
handler = h
} else {
Merge dns (#722) * DNS: add clientip for specific nameserver * Refactoring: DNS App * DNS: add DNS over QUIC support * Feat: add disableCache option for DNS * Feat: add queryStrategy option for DNS * Feat: add disableFallback & skipFallback option for DNS * Feat: DNS hosts support multiple addresses * Feat: DNS transport over TCP * DNS: fix typo & refine code * DNS: refine code * Add disableFallbackIfMatch dns option * Feat: routing and freedom outbound ignore Fake DNS Turn off fake DNS for request sent from Routing and Freedom outbound. Fake DNS now only apply to DNS outbound. This is important for Android, where VPN service take over all system DNS traffic and pass it to core. "UseIp" option can be used in Freedom outbound to avoid getting fake IP and fail connection. * Fix test * Fix dns return * Fix local dns return empty * Apply timeout to dns outbound * Update app/dns/config.go Co-authored-by: Loyalsoldier <10487845+loyalsoldier@users.noreply.github.com> Co-authored-by: Ye Zhihao <vigilans@foxmail.com> Co-authored-by: maskedeken <52683904+maskedeken@users.noreply.github.com> Co-authored-by: V2Fly Team <51714622+vcptr@users.noreply.github.com> Co-authored-by: CalmLong <37164399+calmlong@users.noreply.github.com> Co-authored-by: Shelikhoo <xiaokangwang@outlook.com> Co-authored-by: 秋のかえで <autmaple@protonmail.com> Co-authored-by: 朱聖黎 <digglife@gmail.com> Co-authored-by: rurirei <72071920+rurirei@users.noreply.github.com> Co-authored-by: yuhan6665 <1588741+yuhan6665@users.noreply.github.com> Co-authored-by: Arthur Morgan <4637240+badO1a5A90@users.noreply.github.com>
2021-10-16 16:02:51 +03:00
newError("non existing outTag: ", tag).AtWarning().WriteToLog(session.ExportIDToError(ctx))
2020-11-25 13:01:53 +02:00
}
} 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 != "" {
Merge dns (#722) * DNS: add clientip for specific nameserver * Refactoring: DNS App * DNS: add DNS over QUIC support * Feat: add disableCache option for DNS * Feat: add queryStrategy option for DNS * Feat: add disableFallback & skipFallback option for DNS * Feat: DNS hosts support multiple addresses * Feat: DNS transport over TCP * DNS: fix typo & refine code * DNS: refine code * Add disableFallbackIfMatch dns option * Feat: routing and freedom outbound ignore Fake DNS Turn off fake DNS for request sent from Routing and Freedom outbound. Fake DNS now only apply to DNS outbound. This is important for Android, where VPN service take over all system DNS traffic and pass it to core. "UseIp" option can be used in Freedom outbound to avoid getting fake IP and fail connection. * Fix test * Fix dns return * Fix local dns return empty * Apply timeout to dns outbound * Update app/dns/config.go Co-authored-by: Loyalsoldier <10487845+loyalsoldier@users.noreply.github.com> Co-authored-by: Ye Zhihao <vigilans@foxmail.com> Co-authored-by: maskedeken <52683904+maskedeken@users.noreply.github.com> Co-authored-by: V2Fly Team <51714622+vcptr@users.noreply.github.com> Co-authored-by: CalmLong <37164399+calmlong@users.noreply.github.com> Co-authored-by: Shelikhoo <xiaokangwang@outlook.com> Co-authored-by: 秋のかえで <autmaple@protonmail.com> Co-authored-by: 朱聖黎 <digglife@gmail.com> Co-authored-by: rurirei <72071920+rurirei@users.noreply.github.com> Co-authored-by: yuhan6665 <1588741+yuhan6665@users.noreply.github.com> Co-authored-by: Arthur Morgan <4637240+badO1a5A90@users.noreply.github.com>
2021-10-16 16:02:51 +03:00
accessMessage.Detour = tag
2020-11-25 13:01:53 +02:00
}
log.Record(accessMessage)
}
handler.Dispatch(ctx, link)
}