1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-06-09 17:30:44 +00:00
v2fly/app/dispatcher/default.go

209 lines
5.4 KiB
Go
Raw Normal View History

package dispatcher
2016-05-18 06:05:52 +00:00
2017-12-03 00:04:57 +00:00
//go:generate go run $GOPATH/src/v2ray.com/core/common/errors/errorgen/main.go -pkg impl -path App,Dispatcher,Default
2017-04-08 23:43:25 +00:00
2016-05-18 06:05:52 +00:00
import (
2017-01-13 12:41:40 +00:00
"context"
2017-04-24 22:49:27 +00:00
"time"
2017-01-04 11:34:01 +00:00
"v2ray.com/core"
2016-08-20 18:55:45 +00:00
"v2ray.com/core/app/proxyman"
2017-01-06 14:32:36 +00:00
"v2ray.com/core/common"
2017-04-25 08:48:06 +00:00
"v2ray.com/core/common/buf"
"v2ray.com/core/common/net"
2018-03-30 21:48:45 +00:00
"v2ray.com/core/common/protocol"
2018-04-16 22:31:10 +00:00
"v2ray.com/core/common/stats"
2016-08-20 18:55:45 +00:00
"v2ray.com/core/proxy"
2018-04-16 22:31:10 +00:00
"v2ray.com/core/transport/pipe"
2016-05-18 06:05:52 +00:00
)
2017-04-24 22:49:27 +00:00
var (
errSniffingTimeout = newError("timeout on sniffing")
)
2018-04-16 22:31:10 +00:00
type cachedReader struct {
reader *pipe.Reader
cache buf.MultiBuffer
}
func (r *cachedReader) Cache(b *buf.Buffer) {
mb, _ := r.reader.ReadMultiBufferWithTimeout(time.Millisecond * 100)
if !mb.IsEmpty() {
2018-04-19 19:33:18 +00:00
common.Must(r.cache.WriteMultiBuffer(mb))
2018-04-16 22:31:10 +00:00
}
common.Must(b.Reset(func(x []byte) (int, error) {
return r.cache.Copy(x), nil
}))
}
func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
if !r.cache.IsEmpty() {
mb := r.cache
r.cache = nil
return mb, nil
}
return r.reader.ReadMultiBuffer()
}
func (r *cachedReader) CloseError() {
r.cache.Release()
r.reader.CloseError()
}
2017-04-25 08:48:06 +00:00
// DefaultDispatcher is a default implementation of Dispatcher.
2016-05-18 06:05:52 +00:00
type DefaultDispatcher struct {
ohm core.OutboundHandlerManager
router core.Router
2018-03-30 21:48:45 +00:00
policy core.PolicyManager
stats core.StatManager
2016-05-18 06:05:52 +00:00
}
2017-04-25 08:48:06 +00:00
// NewDefaultDispatcher create a new DefaultDispatcher.
func NewDefaultDispatcher(ctx context.Context, config *Config) (*DefaultDispatcher, error) {
2018-02-21 16:05:29 +00:00
v := core.MustFromContext(ctx)
d := &DefaultDispatcher{
ohm: v.OutboundHandlerManager(),
router: v.Router(),
2018-03-30 21:48:45 +00:00
policy: v.PolicyManager(),
stats: v.Stats(),
}
if err := v.RegisterFeature((*core.Dispatcher)(nil), d); err != nil {
2018-04-11 18:17:08 +00:00
return nil, newError("unable to register Dispatcher").Base(err)
2017-01-13 12:41:40 +00:00
}
return d, nil
}
2018-04-03 09:11:54 +00:00
// Start implements common.Runnable.
2017-04-25 08:48:06 +00:00
func (*DefaultDispatcher) Start() error {
2017-02-01 20:35:40 +00:00
return nil
}
2018-04-03 09:11:54 +00:00
// Close implements common.Closable.
2018-02-08 14:39:46 +00:00
func (*DefaultDispatcher) Close() error { return nil }
2017-02-01 20:35:40 +00:00
2018-04-16 22:31:10 +00:00
func (d *DefaultDispatcher) getLink(ctx context.Context) (*core.Link, *core.Link) {
2018-05-25 10:08:28 +00:00
opt := pipe.OptionsFromContext(ctx)
uplinkReader, uplinkWriter := pipe.New(opt...)
downlinkReader, downlinkWriter := pipe.New(opt...)
2018-03-30 21:48:45 +00:00
2018-04-16 22:31:10 +00:00
inboundLink := &core.Link{
Reader: downlinkReader,
Writer: uplinkWriter,
}
outboundLink := &core.Link{
Reader: uplinkReader,
Writer: downlinkWriter,
}
user := protocol.UserFromContext(ctx)
2018-03-30 21:48:45 +00:00
if user != nil && len(user.Email) > 0 {
2018-03-31 18:30:49 +00:00
p := d.policy.ForLevel(user.Level)
if p.Stats.UserUplink {
name := "user>>>" + user.Email + ">>>traffic>>>uplink"
2018-04-11 22:10:14 +00:00
if c, _ := core.GetOrRegisterStatCounter(d.stats, name); c != nil {
2018-04-16 22:31:10 +00:00
inboundLink.Writer = &stats.SizeStatWriter{
Counter: c,
Writer: inboundLink.Writer,
}
2018-03-31 18:30:49 +00:00
}
2018-03-30 21:48:45 +00:00
}
2018-03-31 18:30:49 +00:00
if p.Stats.UserDownlink {
name := "user>>>" + user.Email + ">>>traffic>>>downlink"
2018-04-11 22:10:14 +00:00
if c, _ := core.GetOrRegisterStatCounter(d.stats, name); c != nil {
2018-04-16 22:31:10 +00:00
outboundLink.Writer = &stats.SizeStatWriter{
Counter: c,
Writer: outboundLink.Writer,
}
2018-03-31 18:30:49 +00:00
}
2018-03-30 21:48:45 +00:00
}
2018-03-31 18:30:49 +00:00
}
2018-03-30 21:48:45 +00:00
2018-04-16 22:31:10 +00:00
return inboundLink, outboundLink
2018-03-31 18:30:49 +00:00
}
// Dispatch implements core.Dispatcher.
2018-04-16 22:31:10 +00:00
func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destination) (*core.Link, error) {
2018-03-31 18:30:49 +00:00
if !destination.IsValid() {
panic("Dispatcher: Invalid destination.")
2018-03-30 21:48:45 +00:00
}
2018-03-31 18:30:49 +00:00
ctx = proxy.ContextWithTarget(ctx, destination)
2018-04-16 22:31:10 +00:00
inbound, outbound := d.getLink(ctx)
2018-03-15 02:32:10 +00:00
snifferList := proxyman.ProtocolSniffersFromContext(ctx)
2018-05-20 19:29:40 +00:00
if destination.Address.Family().IsDomain() || destination.Network != net.Network_TCP || len(snifferList) == 0 {
2017-04-24 22:49:27 +00:00
go d.routedDispatch(ctx, outbound, destination)
} else {
go func() {
2018-04-16 22:31:10 +00:00
cReader := &cachedReader{
reader: outbound.Reader.(*pipe.Reader),
}
outbound.Reader = cReader
domain, err := sniffer(ctx, snifferList, cReader)
2017-04-25 22:43:59 +00:00
if err == nil {
2018-02-22 14:26:00 +00:00
newError("sniffed domain: ", domain).WithContext(ctx).WriteToLog()
2017-04-25 22:43:59 +00:00
destination.Address = net.ParseAddress(domain)
ctx = proxy.ContextWithTarget(ctx, destination)
2017-04-24 22:49:27 +00:00
}
d.routedDispatch(ctx, outbound, destination)
}()
}
2018-04-16 22:31:10 +00:00
return inbound, nil
2017-04-24 22:49:27 +00:00
}
2018-04-16 22:31:10 +00:00
func sniffer(ctx context.Context, snifferList []proxyman.KnownProtocols, cReader *cachedReader) (string, error) {
2017-04-25 08:48:06 +00:00
payload := buf.New()
defer payload.Release()
2018-02-26 01:06:27 +00:00
sniffer := NewSniffer(snifferList)
2017-04-24 22:49:27 +00:00
totalAttempt := 0
for {
select {
case <-ctx.Done():
2017-04-25 08:48:06 +00:00
return "", ctx.Err()
2017-04-26 19:24:58 +00:00
default:
2017-04-24 22:49:27 +00:00
totalAttempt++
if totalAttempt > 5 {
2017-04-25 08:48:06 +00:00
return "", errSniffingTimeout
2017-04-24 22:49:27 +00:00
}
2018-04-16 22:31:10 +00:00
cReader.Cache(payload)
2017-04-26 19:24:58 +00:00
if !payload.IsEmpty() {
2017-06-05 21:27:50 +00:00
domain, err := sniffer.Sniff(payload.Bytes())
2017-04-24 22:49:27 +00:00
if err != ErrMoreData {
2017-04-25 08:48:06 +00:00
return domain, err
2017-04-24 22:49:27 +00:00
}
}
2017-04-25 08:48:06 +00:00
if payload.IsFull() {
return "", ErrInvalidData
2017-04-24 23:40:34 +00:00
}
2017-04-26 19:24:58 +00:00
time.Sleep(time.Millisecond * 100)
2017-04-24 22:49:27 +00:00
}
}
}
2018-04-16 22:31:10 +00:00
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *core.Link, destination net.Destination) {
2017-04-24 22:49:27 +00:00
dispatcher := d.ohm.GetDefaultHandler()
2017-04-23 17:16:56 +00:00
if d.router != nil {
if tag, err := d.router.PickRoute(ctx); err == nil {
2017-04-23 17:16:56 +00:00
if handler := d.ohm.GetHandler(tag); handler != nil {
2018-02-22 14:26:00 +00:00
newError("taking detour [", tag, "] for [", destination, "]").WithContext(ctx).WriteToLog()
2016-05-18 06:05:52 +00:00
dispatcher = handler
} else {
2018-03-15 02:32:10 +00:00
newError("non existing tag: ", tag).AtWarning().WithContext(ctx).WriteToLog()
2016-05-18 06:05:52 +00:00
}
} else {
2018-02-22 14:26:00 +00:00
newError("default route for ", destination).WithContext(ctx).WriteToLog()
2016-05-18 06:05:52 +00:00
}
}
2018-04-16 22:31:10 +00:00
dispatcher.Dispatch(ctx, link)
2016-05-18 06:05:52 +00:00
}
2016-12-16 14:39:47 +00:00
func init() {
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
return NewDefaultDispatcher(ctx, config.(*Config))
2017-01-13 12:41:40 +00:00
}))
2016-12-16 14:39:47 +00:00
}