Xray-core/app/dispatcher/default.go

367 lines
9.6 KiB
Go
Raw Normal View History

2020-11-25 11:01:53 +00:00
package dispatcher
2020-12-04 01:36:16 +00:00
//go:generate go run github.com/xtls/xray-core/common/errors/errorgen
2020-11-25 11:01:53 +00:00
import (
"context"
"strings"
"sync"
"time"
2020-12-04 01:36:16 +00: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 01:36:16 +00: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 11:01:53 +00:00
)
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
}
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) error {
return d.Init(config.(*Config), om, router, pm, sm)
}); 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) error {
d.ohm = om
d.router = router
d.policy = pm
d.stats = sm
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 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 fakeDNSEngine != nil && protocolString != "bittorrent" && p == "fakedns" &&
destination.Address.Family().IsIP() && fakeDNSEngine.GetFakeIPRange().Contains(destination.Address.IP()) {
newError("Using sniffer ", protocolString, " since the fake DNS missed").WriteToLog(session.ExportIDToError(ctx))
2020-11-25 11:01:53 +00:00
return true
}
}
2020-11-25 11:01:53 +00: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)
ob.Target = destination
}
}
2020-11-25 11:01:53 +00:00
go d.routedDispatch(ctx, outbound, destination)
default:
2020-11-25 11:01:53 +00:00
go func() {
cReader := &cachedReader{
reader: outbound.Reader.(*pipe.Reader),
}
outbound.Reader = cReader
result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly)
2020-11-25 11:01:53 +00:00
if err == nil {
content.Protocol = result.Protocol()
}
if err == nil && shouldOverride(ctx, result, sniffingRequest, destination) {
2020-11-25 11:01:53 +00:00
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
destination.Address = net.ParseAddress(domain)
ob.Target = destination
}
d.routedDispatch(ctx, outbound, destination)
}()
}
return inbound, nil
}
func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool) (SniffResult, error) {
2020-11-25 11:01:53 +00:00
payload := buf.New()
defer payload.Release()
sniffer := NewSniffer(ctx)
metaresult, metadataErr := sniffer.SniffMetadata(ctx)
if metadataOnly {
return metaresult, metadataErr
}
2020-11-25 11:01:53 +00: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 11:01:53 +00:00
}
}
}
}()
if contentErr != nil && metadataErr == nil {
return metaresult, nil
}
if contentErr == nil && metadataErr == nil {
return CompositeResult(metaresult, contentResult), nil
2020-11-25 11:01:53 +00:00
}
return contentResult, contentErr
2020-11-25 11:01:53 +00:00
}
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.Link, destination net.Destination) {
var handler outbound.Handler
skipRoutePick := false
if content := session.ContentFromContext(ctx); content != nil {
skipRoutePick = content.SkipRoutePick
}
2021-01-14 14:45:23 +00:00
routingLink := routing_session.AsRoutingContext(ctx)
inTag := routingLink.GetInboundTag()
isPickRoute := false
2020-11-25 11:01:53 +00:00
if d.router != nil && !skipRoutePick {
2021-01-14 14:45:23 +00:00
if route, err := d.router.PickRoute(routingLink); err == nil {
outTag := route.GetOutboundTag()
isPickRoute = true
if h := d.ohm.GetHandler(outTag); h != nil {
newError("taking detour [", outTag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))
2020-11-25 11:01:53 +00:00
handler = h
} else {
2021-01-14 14:45:23 +00:00
newError("non existing outTag: ", outTag).AtWarning().WriteToLog(session.ExportIDToError(ctx))
2020-11-25 11:01:53 +00: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 != "" {
2021-01-14 14:45:23 +00:00
if isPickRoute {
if inTag != "" {
accessMessage.Detour = inTag + " -> " + tag
} else {
accessMessage.Detour = tag
}
2021-01-04 05:05:38 +00:00
} else {
2021-01-14 14:45:23 +00:00
if inTag != "" {
accessMessage.Detour = inTag + " >> " + tag
} else {
accessMessage.Detour = tag
}
2021-01-04 05:05:38 +00:00
}
2020-11-25 11:01:53 +00:00
}
log.Record(accessMessage)
}
handler.Dispatch(ctx, link)
}