1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-22 18:17:52 -05:00
v2fly/app/dispatcher/default.go

280 lines
6.7 KiB
Go
Raw Normal View History

package dispatcher
2016-05-18 02:05:52 -04:00
2018-09-30 17:08:41 -04:00
//go:generate errorgen
2017-04-08 19:43:25 -04:00
2016-05-18 02:05:52 -04:00
import (
2017-01-13 07:41:40 -05:00
"context"
2018-07-16 07:47:00 -04:00
"strings"
2018-07-21 06:11:49 -04:00
"sync"
2017-04-24 18:49:27 -04:00
"time"
2017-01-04 06:34:01 -05:00
"v2ray.com/core"
2016-08-20 14:55:45 -04:00
"v2ray.com/core/app/proxyman"
2017-01-06 09:32:36 -05:00
"v2ray.com/core/common"
2017-04-25 04:48:06 -04:00
"v2ray.com/core/common/buf"
"v2ray.com/core/common/net"
2018-03-30 17:48:45 -04:00
"v2ray.com/core/common/protocol"
"v2ray.com/core/common/session"
"v2ray.com/core/features/outbound"
2018-10-11 16:34:31 -04:00
"v2ray.com/core/features/policy"
"v2ray.com/core/features/routing"
2018-10-15 03:03:40 -04:00
"v2ray.com/core/features/stats"
2018-11-03 07:36:29 -04:00
"v2ray.com/core/transport"
2018-04-16 18:31:10 -04:00
"v2ray.com/core/transport/pipe"
2016-05-18 02:05:52 -04:00
)
2017-04-24 18:49:27 -04:00
var (
errSniffingTimeout = newError("timeout on sniffing")
)
2018-04-16 18:31:10 -04:00
type cachedReader struct {
2018-07-21 06:11:49 -04:00
sync.Mutex
2018-04-16 18:31:10 -04:00
reader *pipe.Reader
cache buf.MultiBuffer
}
func (r *cachedReader) Cache(b *buf.Buffer) {
mb, _ := r.reader.ReadMultiBufferTimeout(time.Millisecond * 100)
2018-07-21 06:11:49 -04:00
r.Lock()
2018-04-16 18:31:10 -04:00
if !mb.IsEmpty() {
2018-11-17 16:45:07 -05:00
r.cache, _ = buf.MergeMulti(r.cache, mb)
2018-04-16 18:31:10 -04:00
}
2018-11-02 16:34:04 -04:00
b.Clear()
rawBytes := b.Extend(buf.Size)
n := r.cache.Copy(rawBytes)
b.Resize(0, int32(n))
2018-07-21 06:11:49 -04:00
r.Unlock()
2018-04-16 18:31:10 -04:00
}
func (r *cachedReader) readInternal() buf.MultiBuffer {
2018-07-21 06:11:49 -04:00
r.Lock()
defer r.Unlock()
2018-07-23 17:39:55 -04:00
if r.cache != nil && !r.cache.IsEmpty() {
2018-04-16 18:31:10 -04:00
mb := r.cache
r.cache = nil
return mb
}
return nil
}
func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
mb := r.readInternal()
if mb != nil {
2018-04-16 18:31:10 -04:00
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)
}
2018-12-31 15:25:10 -05:00
func (r *cachedReader) Interrupt() {
2018-07-21 06:11:49 -04:00
r.Lock()
2018-07-23 17:39:55 -04:00
if r.cache != nil {
2018-11-17 16:45:07 -05:00
r.cache = buf.ReleaseMulti(r.cache)
2018-07-23 17:39:55 -04:00
}
2018-07-21 06:11:49 -04:00
r.Unlock()
2018-12-31 15:25:10 -05:00
r.reader.Interrupt()
2018-04-16 18:31:10 -04:00
}
2017-04-25 04:48:06 -04:00
// DefaultDispatcher is a default implementation of Dispatcher.
2016-05-18 02:05:52 -04:00
type DefaultDispatcher struct {
2018-10-12 17:57:56 -04:00
ohm outbound.Manager
router routing.Router
2018-10-11 16:34:31 -04:00
policy policy.Manager
2018-10-15 03:03:40 -04:00
stats stats.Manager
2016-05-18 02:05:52 -04:00
}
2018-10-22 09:58:52 -04: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) error {
return d.Init(config.(*Config), om, router, pm, sm)
}); err != nil {
return nil, err
}
return d, nil
}))
2017-01-13 07:41:40 -05:00
}
2018-10-22 05:26:22 -04:00
// Init initializes DefaultDispatcher.
2018-10-22 09:58:52 -04:00
func (d *DefaultDispatcher) Init(config *Config, om outbound.Manager, router routing.Router, pm policy.Manager, sm stats.Manager) error {
2018-10-22 05:26:22 -04:00
d.ohm = om
d.router = router
d.policy = pm
d.stats = sm
2018-10-22 09:58:52 -04:00
return nil
2018-10-22 05:26:22 -04:00
}
2018-10-15 02:51:24 -04:00
// Type implements common.HasType.
2018-10-12 17:57:56 -04:00
func (*DefaultDispatcher) Type() interface{} {
return routing.DispatcherType()
}
2018-04-03 05:11:54 -04:00
// Start implements common.Runnable.
2017-04-25 04:48:06 -04:00
func (*DefaultDispatcher) Start() error {
2017-02-01 15:35:40 -05:00
return nil
}
2018-04-03 05:11:54 -04:00
// Close implements common.Closable.
2018-02-08 09:39:46 -05:00
func (*DefaultDispatcher) Close() error { return nil }
2017-02-01 15:35:40 -05:00
2018-11-03 07:36:29 -04:00
func (d *DefaultDispatcher) getLink(ctx context.Context) (*transport.Link, *transport.Link) {
2018-05-25 06:08:28 -04:00
opt := pipe.OptionsFromContext(ctx)
uplinkReader, uplinkWriter := pipe.New(opt...)
downlinkReader, downlinkWriter := pipe.New(opt...)
2018-03-30 17:48:45 -04:00
2018-11-03 07:36:29 -04:00
inboundLink := &transport.Link{
2018-04-16 18:31:10 -04:00
Reader: downlinkReader,
Writer: uplinkWriter,
}
2018-11-03 07:36:29 -04:00
outboundLink := &transport.Link{
2018-04-16 18:31:10 -04:00
Reader: uplinkReader,
Writer: downlinkWriter,
}
2018-10-15 02:36:50 -04:00
sessionInbound := session.InboundFromContext(ctx)
var user *protocol.MemoryUser
if sessionInbound != nil {
user = sessionInbound.User
}
2018-03-30 17:48:45 -04:00
if user != nil && len(user.Email) > 0 {
2018-03-31 14:30:49 -04:00
p := d.policy.ForLevel(user.Level)
if p.Stats.UserUplink {
name := "user>>>" + user.Email + ">>>traffic>>>uplink"
2018-10-15 03:03:40 -04:00
if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
2018-11-03 08:05:23 -04:00
inboundLink.Writer = &SizeStatWriter{
2018-04-16 18:31:10 -04:00
Counter: c,
Writer: inboundLink.Writer,
}
2018-03-31 14:30:49 -04:00
}
2018-03-30 17:48:45 -04:00
}
2018-03-31 14:30:49 -04:00
if p.Stats.UserDownlink {
name := "user>>>" + user.Email + ">>>traffic>>>downlink"
2018-10-15 03:03:40 -04:00
if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
2018-11-03 08:05:23 -04:00
outboundLink.Writer = &SizeStatWriter{
2018-04-16 18:31:10 -04:00
Counter: c,
Writer: outboundLink.Writer,
}
2018-03-31 14:30:49 -04:00
}
2018-03-30 17:48:45 -04:00
}
2018-03-31 14:30:49 -04:00
}
2018-03-30 17:48:45 -04:00
2018-04-16 18:31:10 -04:00
return inboundLink, outboundLink
2018-03-31 14:30:49 -04:00
}
2018-07-16 07:47:00 -04:00
func shouldOverride(result SniffResult, domainOverride []string) bool {
for _, p := range domainOverride {
if strings.HasPrefix(result.Protocol(), p) {
return true
}
}
return false
}
// Dispatch implements routing.Dispatcher.
2018-11-03 07:36:29 -04:00
func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destination) (*transport.Link, error) {
2018-03-31 14:30:49 -04:00
if !destination.IsValid() {
panic("Dispatcher: Invalid destination.")
2018-03-30 17:48:45 -04:00
}
ob := &session.Outbound{
Target: destination,
}
ctx = session.ContextWithOutbound(ctx, ob)
2018-03-31 14:30:49 -04:00
2018-04-16 18:31:10 -04:00
inbound, outbound := d.getLink(ctx)
2018-07-16 07:47:00 -04:00
sniffingConfig := proxyman.SniffingConfigFromContext(ctx)
if destination.Network != net.Network_TCP || sniffingConfig == nil || !sniffingConfig.Enabled {
2017-04-24 18:49:27 -04:00
go d.routedDispatch(ctx, outbound, destination)
} else {
go func() {
2018-04-16 18:31:10 -04:00
cReader := &cachedReader{
reader: outbound.Reader.(*pipe.Reader),
}
outbound.Reader = cReader
2018-07-16 07:47:00 -04:00
result, err := sniffer(ctx, cReader)
2017-04-25 18:43:59 -04:00
if err == nil {
2018-07-16 07:47:00 -04:00
ctx = ContextWithSniffingResult(ctx, result)
}
2018-07-20 07:09:42 -04:00
if err == nil && shouldOverride(result, sniffingConfig.DestinationOverride) {
2018-07-16 07:47:00 -04:00
domain := result.Domain()
newError("sniffed domain: ", domain).WriteToLog(session.ExportIDToError(ctx))
2017-04-25 18:43:59 -04:00
destination.Address = net.ParseAddress(domain)
ob.Target = destination
2017-04-24 18:49:27 -04:00
}
d.routedDispatch(ctx, outbound, destination)
}()
}
2018-04-16 18:31:10 -04:00
return inbound, nil
2017-04-24 18:49:27 -04:00
}
2018-07-16 07:47:00 -04:00
func sniffer(ctx context.Context, cReader *cachedReader) (SniffResult, error) {
2017-04-25 04:48:06 -04:00
payload := buf.New()
defer payload.Release()
2018-07-16 07:47:00 -04:00
sniffer := NewSniffer()
2017-04-24 18:49:27 -04:00
totalAttempt := 0
for {
select {
case <-ctx.Done():
2018-07-16 07:47:00 -04:00
return nil, ctx.Err()
2017-04-26 15:24:58 -04:00
default:
2017-04-24 18:49:27 -04:00
totalAttempt++
2018-08-28 13:52:31 -04:00
if totalAttempt > 2 {
2018-07-16 07:47:00 -04:00
return nil, errSniffingTimeout
2017-04-24 18:49:27 -04:00
}
2018-04-16 18:31:10 -04:00
cReader.Cache(payload)
2017-04-26 15:24:58 -04:00
if !payload.IsEmpty() {
2018-07-16 07:47:00 -04:00
result, err := sniffer.Sniff(payload.Bytes())
if err != common.ErrNoClue {
2018-07-16 07:47:00 -04:00
return result, err
2017-04-24 18:49:27 -04:00
}
}
2017-04-25 04:48:06 -04:00
if payload.IsFull() {
2018-07-16 07:47:00 -04:00
return nil, errUnknownContent
2017-04-24 19:40:34 -04:00
}
2017-04-24 18:49:27 -04:00
}
}
}
2018-11-03 07:36:29 -04:00
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.Link, destination net.Destination) {
2019-01-12 03:44:59 -05:00
var handler outbound.Handler
2017-04-23 13:16:56 -04:00
if d.router != nil {
if tag, err := d.router.PickRoute(ctx); err == nil {
2019-01-12 03:44:59 -05:00
if h := d.ohm.GetHandler(tag); h != nil {
newError("taking detour [", tag, "] for [", destination, "]").WriteToLog(session.ExportIDToError(ctx))
2019-01-12 03:44:59 -05:00
handler = h
2016-05-18 02:05:52 -04:00
} else {
newError("non existing tag: ", tag).AtWarning().WriteToLog(session.ExportIDToError(ctx))
2016-05-18 02:05:52 -04:00
}
} else {
newError("default route for ", destination).WriteToLog(session.ExportIDToError(ctx))
2016-05-18 02:05:52 -04:00
}
}
2018-12-31 04:43:08 -05:00
2019-01-12 03:44:59 -05:00
if handler == nil {
handler = d.ohm.GetDefaultHandler()
}
if handler == nil {
2018-12-31 04:43:08 -05:00
newError("default outbound handler not exist").WriteToLog(session.ExportIDToError(ctx))
common.Close(link.Writer)
2018-12-31 15:25:10 -05:00
common.Interrupt(link.Reader)
2018-12-31 04:43:08 -05:00
return
}
2019-01-12 03:44:59 -05:00
handler.Dispatch(ctx, link)
2016-05-18 02:05:52 -04:00
}