Feature: trace adapters when dialing (#170)

This commit is contained in:
Yunhao Zhang 2019-08-09 01:28:37 +08:00 committed by Dreamacro
parent 5829c3d5be
commit b926f4cf09
13 changed files with 166 additions and 56 deletions

View file

@ -30,7 +30,7 @@ func (b *Base) Type() C.AdapterType {
return b.tp return b.tp
} }
func (b *Base) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (b *Base) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
return nil, nil, errors.New("no support") return nil, nil, errors.New("no support")
} }
@ -46,6 +46,40 @@ func (b *Base) MarshalJSON() ([]byte, error) {
}) })
} }
type conn struct {
net.Conn
chain C.Chain
}
func (c *conn) Chains() C.Chain {
return c.chain
}
func (c *conn) AppendToChains(a C.ProxyAdapter) {
c.chain = append(c.chain, a.Name())
}
func newConn(c net.Conn, a C.ProxyAdapter) C.Conn {
return &conn{c, []string{a.Name()}}
}
type packetConn struct {
net.PacketConn
chain C.Chain
}
func (c *packetConn) Chains() C.Chain {
return c.chain
}
func (c *packetConn) AppendToChains(a C.ProxyAdapter) {
c.chain = append(c.chain, a.Name())
}
func newPacketConn(c net.PacketConn, a C.ProxyAdapter) C.PacketConn {
return &packetConn{c, []string{a.Name()}}
}
type Proxy struct { type Proxy struct {
C.ProxyAdapter C.ProxyAdapter
history *queue.Queue history *queue.Queue
@ -56,7 +90,7 @@ func (p *Proxy) Alive() bool {
return p.alive return p.alive
} }
func (p *Proxy) Dial(metadata *C.Metadata) (net.Conn, error) { func (p *Proxy) Dial(metadata *C.Metadata) (C.Conn, error) {
conn, err := p.ProxyAdapter.Dial(metadata) conn, err := p.ProxyAdapter.Dial(metadata)
if err != nil { if err != nil {
p.alive = false p.alive = false

View file

@ -10,7 +10,7 @@ type Direct struct {
*Base *Base
} }
func (d *Direct) Dial(metadata *C.Metadata) (net.Conn, error) { func (d *Direct) Dial(metadata *C.Metadata) (C.Conn, error) {
address := net.JoinHostPort(metadata.Host, metadata.DstPort) address := net.JoinHostPort(metadata.Host, metadata.DstPort)
if metadata.DstIP != nil { if metadata.DstIP != nil {
address = net.JoinHostPort(metadata.DstIP.String(), metadata.DstPort) address = net.JoinHostPort(metadata.DstIP.String(), metadata.DstPort)
@ -21,10 +21,10 @@ func (d *Direct) Dial(metadata *C.Metadata) (net.Conn, error) {
return nil, err return nil, err
} }
tcpKeepAlive(c) tcpKeepAlive(c)
return c, nil return newConn(c, d), nil
} }
func (d *Direct) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (d *Direct) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
pc, err := net.ListenPacket("udp", "") pc, err := net.ListenPacket("udp", "")
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
@ -34,7 +34,7 @@ func (d *Direct) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
return pc, addr, nil return newPacketConn(pc, d), addr, nil
} }
func NewDirect() *Direct { func NewDirect() *Direct {

View file

@ -31,14 +31,22 @@ func (f *Fallback) Now() string {
return proxy.Name() return proxy.Name()
} }
func (f *Fallback) Dial(metadata *C.Metadata) (net.Conn, error) { func (f *Fallback) Dial(metadata *C.Metadata) (C.Conn, error) {
proxy := f.findAliveProxy() proxy := f.findAliveProxy()
return proxy.Dial(metadata) c, err := proxy.Dial(metadata)
if err == nil {
c.AppendToChains(f)
}
return c, err
} }
func (f *Fallback) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (f *Fallback) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
proxy := f.findAliveProxy() proxy := f.findAliveProxy()
return proxy.DialUDP(metadata) pc, addr, err := proxy.DialUDP(metadata)
if err == nil {
pc.AppendToChains(f)
}
return pc, addr, err
} }
func (f *Fallback) SupportUDP() bool { func (f *Fallback) SupportUDP() bool {

View file

@ -35,7 +35,7 @@ type HttpOption struct {
SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"` SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"`
} }
func (h *Http) Dial(metadata *C.Metadata) (net.Conn, error) { func (h *Http) Dial(metadata *C.Metadata) (C.Conn, error) {
c, err := dialTimeout("tcp", h.addr, tcpTimeout) c, err := dialTimeout("tcp", h.addr, tcpTimeout)
if err == nil && h.tls { if err == nil && h.tls {
cc := tls.Client(c, h.tlsConfig) cc := tls.Client(c, h.tlsConfig)
@ -51,7 +51,7 @@ func (h *Http) Dial(metadata *C.Metadata) (net.Conn, error) {
return nil, err return nil, err
} }
return c, nil return newConn(c, h), nil
} }
func (h *Http) shakeHand(metadata *C.Metadata, rw io.ReadWriter) error { func (h *Http) shakeHand(metadata *C.Metadata, rw io.ReadWriter) error {

View file

@ -54,21 +54,34 @@ func jumpHash(key uint64, buckets int32) int32 {
return int32(b) return int32(b)
} }
func (lb *LoadBalance) Dial(metadata *C.Metadata) (net.Conn, error) { func (lb *LoadBalance) Dial(metadata *C.Metadata) (c C.Conn, err error) {
defer func() {
if err == nil {
c.AppendToChains(lb)
}
}()
key := uint64(murmur3.Sum32([]byte(getKey(metadata)))) key := uint64(murmur3.Sum32([]byte(getKey(metadata))))
buckets := int32(len(lb.proxies)) buckets := int32(len(lb.proxies))
for i := 0; i < lb.maxRetry; i, key = i+1, key+1 { for i := 0; i < lb.maxRetry; i, key = i+1, key+1 {
idx := jumpHash(key, buckets) idx := jumpHash(key, buckets)
proxy := lb.proxies[idx] proxy := lb.proxies[idx]
if proxy.Alive() { if proxy.Alive() {
return proxy.Dial(metadata) c, err = proxy.Dial(metadata)
return
} }
} }
c, err = lb.proxies[0].Dial(metadata)
return lb.proxies[0].Dial(metadata) return
} }
func (lb *LoadBalance) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (lb *LoadBalance) DialUDP(metadata *C.Metadata) (pc C.PacketConn, addr net.Addr, err error) {
defer func() {
if err == nil {
pc.AppendToChains(lb)
}
}()
key := uint64(murmur3.Sum32([]byte(getKey(metadata)))) key := uint64(murmur3.Sum32([]byte(getKey(metadata))))
buckets := int32(len(lb.proxies)) buckets := int32(len(lb.proxies))
for i := 0; i < lb.maxRetry; i, key = i+1, key+1 { for i := 0; i < lb.maxRetry; i, key = i+1, key+1 {

View file

@ -12,8 +12,8 @@ type Reject struct {
*Base *Base
} }
func (r *Reject) Dial(metadata *C.Metadata) (net.Conn, error) { func (r *Reject) Dial(metadata *C.Metadata) (C.Conn, error) {
return &NopConn{}, nil return newConn(&NopConn{}, r), nil
} }
func NewReject() *Reject { func NewReject() *Reject {

View file

@ -20,12 +20,20 @@ type SelectorOption struct {
Proxies []string `proxy:"proxies"` Proxies []string `proxy:"proxies"`
} }
func (s *Selector) Dial(metadata *C.Metadata) (net.Conn, error) { func (s *Selector) Dial(metadata *C.Metadata) (C.Conn, error) {
return s.selected.Dial(metadata) c, err := s.selected.Dial(metadata)
if err == nil {
c.AppendToChains(s)
}
return c, err
} }
func (s *Selector) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (s *Selector) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
return s.selected.DialUDP(metadata) pc, addr, err := s.selected.DialUDP(metadata)
if err == nil {
pc.AppendToChains(s)
}
return pc, addr, err
} }
func (s *Selector) SupportUDP() bool { func (s *Selector) SupportUDP() bool {

View file

@ -57,7 +57,7 @@ type v2rayObfsOption struct {
SkipCertVerify bool `obfs:"skip-cert-verify,omitempty"` SkipCertVerify bool `obfs:"skip-cert-verify,omitempty"`
} }
func (ss *ShadowSocks) Dial(metadata *C.Metadata) (net.Conn, error) { func (ss *ShadowSocks) Dial(metadata *C.Metadata) (C.Conn, error) {
c, err := dialTimeout("tcp", ss.server, tcpTimeout) c, err := dialTimeout("tcp", ss.server, tcpTimeout)
if err != nil { if err != nil {
return nil, fmt.Errorf("%s connect error: %s", ss.server, err.Error()) return nil, fmt.Errorf("%s connect error: %s", ss.server, err.Error())
@ -78,10 +78,10 @@ func (ss *ShadowSocks) Dial(metadata *C.Metadata) (net.Conn, error) {
} }
c = ss.cipher.StreamConn(c) c = ss.cipher.StreamConn(c)
_, err = c.Write(serializesSocksAddr(metadata)) _, err = c.Write(serializesSocksAddr(metadata))
return c, err return newConn(c, ss), err
} }
func (ss *ShadowSocks) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (ss *ShadowSocks) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
pc, err := net.ListenPacket("udp", "") pc, err := net.ListenPacket("udp", "")
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
@ -98,7 +98,7 @@ func (ss *ShadowSocks) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr,
} }
pc = ss.cipher.PacketConn(pc) pc = ss.cipher.PacketConn(pc)
return &ssUDPConn{PacketConn: pc, rAddr: targetAddr}, addr, nil return newPacketConn(&ssUDPConn{PacketConn: pc, rAddr: targetAddr}, ss), addr, nil
} }
func (ss *ShadowSocks) MarshalJSON() ([]byte, error) { func (ss *ShadowSocks) MarshalJSON() ([]byte, error) {

View file

@ -33,7 +33,7 @@ type Socks5Option struct {
SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"` SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"`
} }
func (ss *Socks5) Dial(metadata *C.Metadata) (net.Conn, error) { func (ss *Socks5) Dial(metadata *C.Metadata) (C.Conn, error) {
c, err := dialTimeout("tcp", ss.addr, tcpTimeout) c, err := dialTimeout("tcp", ss.addr, tcpTimeout)
if err == nil && ss.tls { if err == nil && ss.tls {
@ -56,10 +56,10 @@ func (ss *Socks5) Dial(metadata *C.Metadata) (net.Conn, error) {
if _, err := socks5.ClientHandshake(c, serializesSocksAddr(metadata), socks5.CmdConnect, user); err != nil { if _, err := socks5.ClientHandshake(c, serializesSocksAddr(metadata), socks5.CmdConnect, user); err != nil {
return nil, err return nil, err
} }
return c, nil return newConn(c, ss), nil
} }
func (ss *Socks5) DialUDP(metadata *C.Metadata) (_ net.PacketConn, _ net.Addr, err error) { func (ss *Socks5) DialUDP(metadata *C.Metadata) (_ C.PacketConn, _ net.Addr, err error) {
c, err := dialTimeout("tcp", ss.addr, tcpTimeout) c, err := dialTimeout("tcp", ss.addr, tcpTimeout)
if err != nil { if err != nil {
err = fmt.Errorf("%s connect error", ss.addr) err = fmt.Errorf("%s connect error", ss.addr)
@ -116,7 +116,7 @@ func (ss *Socks5) DialUDP(metadata *C.Metadata) (_ net.PacketConn, _ net.Addr, e
pc.Close() pc.Close()
}() }()
return &socksUDPConn{PacketConn: pc, rAddr: targetAddr}, addr, nil return newPacketConn(&socksUDPConn{PacketConn: pc, rAddr: targetAddr}, ss), addr, nil
} }
func NewSocks5(option Socks5Option) *Socks5 { func NewSocks5(option Socks5Option) *Socks5 {

View file

@ -33,16 +33,22 @@ func (u *URLTest) Now() string {
return u.fast.Name() return u.fast.Name()
} }
func (u *URLTest) Dial(metadata *C.Metadata) (net.Conn, error) { func (u *URLTest) Dial(metadata *C.Metadata) (C.Conn, error) {
a, err := u.fast.Dial(metadata) a, err := u.fast.Dial(metadata)
if err != nil { if err != nil {
u.fallback() u.fallback()
} else {
a.AppendToChains(u)
} }
return a, err return a, err
} }
func (u *URLTest) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (u *URLTest) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
return u.fast.DialUDP(metadata) pc, addr, err := u.fast.DialUDP(metadata)
if err == nil {
pc.AppendToChains(u)
}
return pc, addr, err
} }
func (u *URLTest) SupportUDP() bool { func (u *URLTest) SupportUDP() bool {

View file

@ -31,17 +31,17 @@ type VmessOption struct {
SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"` SkipCertVerify bool `proxy:"skip-cert-verify,omitempty"`
} }
func (v *Vmess) Dial(metadata *C.Metadata) (net.Conn, error) { func (v *Vmess) Dial(metadata *C.Metadata) (C.Conn, error) {
c, err := dialTimeout("tcp", v.server, tcpTimeout) c, err := dialTimeout("tcp", v.server, tcpTimeout)
if err != nil { if err != nil {
return nil, fmt.Errorf("%s connect error", v.server) return nil, fmt.Errorf("%s connect error", v.server)
} }
tcpKeepAlive(c) tcpKeepAlive(c)
c, err = v.client.New(c, parseVmessAddr(metadata)) c, err = v.client.New(c, parseVmessAddr(metadata))
return c, err return newConn(c, v), err
} }
func (v *Vmess) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error) { func (v *Vmess) DialUDP(metadata *C.Metadata) (C.PacketConn, net.Addr, error) {
c, err := dialTimeout("tcp", v.server, tcpTimeout) c, err := dialTimeout("tcp", v.server, tcpTimeout)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("%s connect error", v.server) return nil, nil, fmt.Errorf("%s connect error", v.server)
@ -51,7 +51,7 @@ func (v *Vmess) DialUDP(metadata *C.Metadata) (net.PacketConn, net.Addr, error)
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("new vmess client error: %v", err) return nil, nil, fmt.Errorf("new vmess client error: %v", err)
} }
return &fakeUDPConn{Conn: c}, c.RemoteAddr(), nil return newPacketConn(&fakeUDPConn{Conn: c}, v), c.RemoteAddr(), nil
} }
func NewVmess(option VmessOption) (*Vmess, error) { func NewVmess(option VmessOption) (*Vmess, error) {

View file

@ -2,6 +2,7 @@ package constant
import ( import (
"context" "context"
"fmt"
"net" "net"
"time" "time"
) )
@ -25,11 +26,39 @@ type ServerAdapter interface {
Metadata() *Metadata Metadata() *Metadata
} }
type Connection interface {
Chains() Chain
AppendToChains(adapter ProxyAdapter)
}
type Chain []string
func (c Chain) String() string {
switch len(c) {
case 0:
return ""
case 1:
return c[0]
default:
return fmt.Sprintf("%s[%s]", c[len(c)-1], c[0])
}
}
type Conn interface {
net.Conn
Connection
}
type PacketConn interface {
net.PacketConn
Connection
}
type ProxyAdapter interface { type ProxyAdapter interface {
Name() string Name() string
Type() AdapterType Type() AdapterType
Dial(metadata *Metadata) (net.Conn, error) Dial(metadata *Metadata) (Conn, error)
DialUDP(metadata *Metadata) (net.PacketConn, net.Addr, error) DialUDP(metadata *Metadata) (PacketConn, net.Addr, error)
SupportUDP() bool SupportUDP() bool
Destroy() Destroy()
MarshalJSON() ([]byte, error) MarshalJSON() ([]byte, error)

View file

@ -135,6 +135,7 @@ func (t *Tunnel) handleConn(localConn C.ServerAdapter) {
} }
var proxy C.Proxy var proxy C.Proxy
var rule C.Rule
switch t.mode { switch t.mode {
case Direct: case Direct:
proxy = t.proxies["DIRECT"] proxy = t.proxies["DIRECT"]
@ -143,7 +144,7 @@ func (t *Tunnel) handleConn(localConn C.ServerAdapter) {
// Rule // Rule
default: default:
var err error var err error
proxy, err = t.match(metadata) proxy, rule, err = t.match(metadata)
if err != nil { if err != nil {
return return
} }
@ -151,22 +152,29 @@ func (t *Tunnel) handleConn(localConn C.ServerAdapter) {
switch metadata.NetWork { switch metadata.NetWork {
case C.TCP: case C.TCP:
t.handleTCPConn(localConn, metadata, proxy) t.handleTCPConn(localConn, metadata, proxy, rule)
case C.UDP: case C.UDP:
t.handleUDPConn(localConn, metadata, proxy) t.handleUDPConn(localConn, metadata, proxy, rule)
} }
} }
func (t *Tunnel) handleUDPConn(localConn C.ServerAdapter, metadata *C.Metadata, proxy C.Proxy) { func (t *Tunnel) handleUDPConn(localConn C.ServerAdapter, metadata *C.Metadata, proxy C.Proxy, rule C.Rule) {
pc, addr := natTable.Get(localConn.RemoteAddr()) pc, addr := natTable.Get(localConn.RemoteAddr())
if pc == nil { if pc == nil {
var err error rawpc, naddr, err := proxy.DialUDP(metadata)
pc, addr, err = proxy.DialUDP(metadata) addr = naddr
pc = rawpc
if err != nil { if err != nil {
log.Warnln("Proxy[%s] connect [%s --> %s] error: %s", proxy.Name(), metadata.SrcIP.String(), metadata.String(), err.Error()) log.Warnln("%s --> %v match %s using %s error: %s", metadata.SrcIP.String(), metadata.String(), rule.RuleType().String(), rule.Adapter(), err.Error())
return return
} }
if rule != nil {
log.Infoln("%s --> %v match %s using %s", metadata.SrcIP.String(), metadata.String(), rule.RuleType().String(), rawpc.Chains().String())
} else {
log.Infoln("%s --> %v doesn't match any rule using DIRECT", metadata.SrcIP.String(), metadata.String())
}
natTable.Set(localConn.RemoteAddr(), pc, addr) natTable.Set(localConn.RemoteAddr(), pc, addr)
go t.handleUDPToLocal(localConn, pc) go t.handleUDPToLocal(localConn, pc)
} }
@ -174,14 +182,21 @@ func (t *Tunnel) handleUDPConn(localConn C.ServerAdapter, metadata *C.Metadata,
t.handleUDPToRemote(localConn, pc, addr) t.handleUDPToRemote(localConn, pc, addr)
} }
func (t *Tunnel) handleTCPConn(localConn C.ServerAdapter, metadata *C.Metadata, proxy C.Proxy) { func (t *Tunnel) handleTCPConn(localConn C.ServerAdapter, metadata *C.Metadata, proxy C.Proxy, rule C.Rule) {
remoConn, err := proxy.Dial(metadata) remoConn, err := proxy.Dial(metadata)
if err != nil { if err != nil {
log.Warnln("Proxy[%s] connect [%s --> %s] error: %s", proxy.Name(), metadata.SrcIP.String(), metadata.String(), err.Error()) log.Warnln("%s --> %v match %s using %s error: %s", metadata.SrcIP.String(), metadata.String(), rule.RuleType().String(), rule.Adapter(), err.Error())
return return
} }
defer remoConn.Close() defer remoConn.Close()
if rule != nil {
log.Infoln("%s --> %v match %s using %s", metadata.SrcIP.String(), metadata.String(), rule.RuleType().String(), remoConn.Chains().String())
} else {
log.Infoln("%s --> %v doesn't match any rule using DIRECT", metadata.SrcIP.String(), metadata.String())
}
switch adapter := localConn.(type) { switch adapter := localConn.(type) {
case *InboundAdapter.HTTPAdapter: case *InboundAdapter.HTTPAdapter:
t.handleHTTP(adapter, remoConn) t.handleHTTP(adapter, remoConn)
@ -194,7 +209,7 @@ func (t *Tunnel) shouldResolveIP(rule C.Rule, metadata *C.Metadata) bool {
return (rule.RuleType() == C.GEOIP || rule.RuleType() == C.IPCIDR) && metadata.Host != "" && metadata.DstIP == nil return (rule.RuleType() == C.GEOIP || rule.RuleType() == C.IPCIDR) && metadata.Host != "" && metadata.DstIP == nil
} }
func (t *Tunnel) match(metadata *C.Metadata) (C.Proxy, error) { func (t *Tunnel) match(metadata *C.Metadata) (C.Proxy, C.Rule, error) {
t.configMux.RLock() t.configMux.RLock()
defer t.configMux.RUnlock() defer t.configMux.RUnlock()
@ -204,7 +219,7 @@ func (t *Tunnel) match(metadata *C.Metadata) (C.Proxy, error) {
ip, err := t.resolveIP(metadata.Host) ip, err := t.resolveIP(metadata.Host)
if err != nil { if err != nil {
if !t.ignoreResolveFail { if !t.ignoreResolveFail {
return nil, fmt.Errorf("[DNS] resolve %s error: %s", metadata.Host, err.Error()) return nil, nil, fmt.Errorf("[DNS] resolve %s error: %s", metadata.Host, err.Error())
} }
log.Debugln("[DNS] resolve %s error: %s", metadata.Host, err.Error()) log.Debugln("[DNS] resolve %s error: %s", metadata.Host, err.Error())
} else { } else {
@ -224,13 +239,10 @@ func (t *Tunnel) match(metadata *C.Metadata) (C.Proxy, error) {
log.Debugln("%v UDP is not supported", adapter.Name()) log.Debugln("%v UDP is not supported", adapter.Name())
continue continue
} }
return adapter, rule, nil
log.Infoln("%s --> %v match %s using %s", metadata.SrcIP.String(), metadata.String(), rule.RuleType().String(), rule.Adapter())
return adapter, nil
} }
} }
log.Infoln("%s --> %v doesn't match any rule using DIRECT", metadata.SrcIP.String(), metadata.String()) return t.proxies["DIRECT"], nil, nil
return t.proxies["DIRECT"], nil
} }
func newTunnel() *Tunnel { func newTunnel() *Tunnel {