mirror of
https://github.com/wyx2685/V2bX.git
synced 2026-02-04 04:30:08 +00:00
fix: 设备数限制某些情况误拒绝旧IP链接BUG
This commit is contained in:
@@ -39,23 +39,21 @@ type cachedReader struct {
|
||||
cache buf.MultiBuffer
|
||||
}
|
||||
|
||||
func (r *cachedReader) Cache(b *buf.Buffer) {
|
||||
mb, _ := r.reader.ReadMultiBufferTimeout(time.Millisecond * 100)
|
||||
func (r *cachedReader) Cache(b *buf.Buffer, deadline time.Duration) error {
|
||||
mb, err := r.reader.ReadMultiBufferTimeout(deadline)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
r.Lock()
|
||||
if !mb.IsEmpty() {
|
||||
r.cache, _ = buf.MergeMulti(r.cache, mb)
|
||||
}
|
||||
cacheLen := r.cache.Len()
|
||||
if cacheLen <= b.Cap() {
|
||||
b.Clear()
|
||||
} else {
|
||||
b.Release()
|
||||
*b = *buf.NewWithSize(cacheLen)
|
||||
}
|
||||
rawBytes := b.Extend(cacheLen)
|
||||
b.Clear()
|
||||
rawBytes := b.Extend(min(r.cache.Len(), b.Cap()))
|
||||
n := r.cache.Copy(rawBytes)
|
||||
b.Resize(0, int32(n))
|
||||
r.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *cachedReader) readInternal() buf.MultiBuffer {
|
||||
@@ -115,7 +113,7 @@ func init() {
|
||||
core.OptionalFeatures(ctx, func(fdns dns.FakeDNSEngine) {
|
||||
d.fdns = fdns
|
||||
})
|
||||
return d.Init(config.(*Config), om, router, pm, sm, dc)
|
||||
return d.Init(config.(*Config), om, router, pm, sm)
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -124,12 +122,11 @@ func init() {
|
||||
}
|
||||
|
||||
// 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 {
|
||||
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
|
||||
d.dns = dns
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -364,7 +361,7 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
|
||||
protocol = resComp.ProtocolForDomainResult()
|
||||
}
|
||||
isFakeIP := false
|
||||
if fkr0, ok := d.fdns.(dns.FakeDNSEngineRev0); ok && ob.Target.Address.Family().IsIP() && fkr0.IsIPInIPPool(ob.Target.Address) {
|
||||
if fkr0, ok := d.fdns.(dns.FakeDNSEngineRev0); ok && fkr0.IsIPInIPPool(ob.Target.Address) {
|
||||
isFakeIP = true
|
||||
}
|
||||
if sniffingRequest.RouteOnly && protocol != "fakedns" && protocol != "fakedns+others" && !isFakeIP {
|
||||
@@ -372,7 +369,6 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
|
||||
} else {
|
||||
ob.Target = destination
|
||||
}
|
||||
destination.Address.Family()
|
||||
}
|
||||
d.routedDispatch(ctx, outbound, destination, nil, content.Protocol)
|
||||
}
|
||||
@@ -381,7 +377,7 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
|
||||
}
|
||||
|
||||
func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool, network net.Network) (SniffResult, error) {
|
||||
payload := buf.New()
|
||||
payload := buf.NewWithSize(32767)
|
||||
defer payload.Release()
|
||||
|
||||
sniffer := NewSniffer(ctx)
|
||||
@@ -393,26 +389,36 @@ func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool, netw
|
||||
}
|
||||
|
||||
contentResult, contentErr := func() (SniffResult, error) {
|
||||
cacheDeadline := 200 * time.Millisecond
|
||||
totalAttempt := 0
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
default:
|
||||
totalAttempt++
|
||||
if totalAttempt > 2 {
|
||||
return nil, errSniffingTimeout
|
||||
cachingStartingTimeStamp := time.Now()
|
||||
err := cReader.Cache(payload, cacheDeadline)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cachingTimeElapsed := time.Since(cachingStartingTimeStamp)
|
||||
cacheDeadline -= cachingTimeElapsed
|
||||
|
||||
cReader.Cache(payload)
|
||||
if !payload.IsEmpty() {
|
||||
result, err := sniffer.Sniff(ctx, payload.Bytes(), network)
|
||||
if err != common.ErrNoClue {
|
||||
switch err {
|
||||
case common.ErrNoClue: // No Clue: protocol not matches, and sniffer cannot determine whether there will be a match or not
|
||||
totalAttempt++
|
||||
case protocol.ErrProtoNeedMoreData: // Protocol Need More Data: protocol matches, but need more data to complete sniffing
|
||||
// in this case, do not add totalAttempt(allow to read until timeout)
|
||||
default:
|
||||
return result, err
|
||||
}
|
||||
} else {
|
||||
totalAttempt++
|
||||
}
|
||||
if payload.IsFull() {
|
||||
return nil, errUnknownContent
|
||||
if totalAttempt >= 2 || cacheDeadline <= 0 {
|
||||
return nil, errSniffingTimeout
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -429,29 +435,10 @@ func sniffer(ctx context.Context, cReader *cachedReader, metadataOnly bool, netw
|
||||
func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.Link, destination net.Destination, l *limiter.Limiter, protocol string) {
|
||||
outbounds := session.OutboundsFromContext(ctx)
|
||||
ob := outbounds[len(outbounds)-1]
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sessionInbound := session.InboundFromContext(ctx)
|
||||
if sessionInbound.User != nil {
|
||||
if l != nil {
|
||||
// del connect count
|
||||
if destination.Network == net.Network_TCP {
|
||||
defer func() {
|
||||
l.ConnLimiter.DelConnCount(sessionInbound.User.Email, sessionInbound.Source.Address.IP().String())
|
||||
}()
|
||||
}
|
||||
} else {
|
||||
if l == nil {
|
||||
var err error
|
||||
l, err = limiter.GetLimiter(sessionInbound.Tag)
|
||||
if err != nil {
|
||||
@@ -510,7 +497,11 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.
|
||||
outTag := route.GetOutboundTag()
|
||||
if h := d.ohm.GetHandler(outTag); h != nil {
|
||||
isPickRoute = 2
|
||||
errors.LogInfo(ctx, "taking detour [", outTag, "] for [", destination, "]")
|
||||
if route.GetRuleTag() == "" {
|
||||
errors.LogInfo(ctx, "taking detour [", outTag, "] for [", destination, "]")
|
||||
} else {
|
||||
errors.LogInfo(ctx, "Hit route rule: [", route.GetRuleTag(), "] so taking detour [", outTag, "] for [", destination, "]")
|
||||
}
|
||||
handler = h
|
||||
} else {
|
||||
errors.LogWarning(ctx, "non existing outTag: ", outTag)
|
||||
@@ -520,10 +511,6 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.
|
||||
}
|
||||
}
|
||||
|
||||
if handler == nil {
|
||||
handler = d.ohm.GetHandler(inTag)
|
||||
}
|
||||
|
||||
if handler == nil {
|
||||
handler = d.ohm.GetDefaultHandler()
|
||||
}
|
||||
@@ -535,6 +522,7 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, link *transport.
|
||||
return
|
||||
}
|
||||
|
||||
ob.Tag = handler.Tag()
|
||||
if accessMessage := log.AccessMessageFromContext(ctx); accessMessage != nil {
|
||||
if tag := handler.Tag(); tag != "" {
|
||||
if inTag == "" {
|
||||
|
||||
Reference in New Issue
Block a user