123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715 |
- package tunnel
- import (
- "context"
- "errors"
- "fmt"
- "github.com/metacubex/mihomo/constant/features"
- "net"
- "net/netip"
- "path/filepath"
- "runtime"
- "strings"
- "sync"
- "time"
- N "github.com/metacubex/mihomo/common/net"
- "github.com/metacubex/mihomo/common/utils"
- "github.com/metacubex/mihomo/component/loopback"
- "github.com/metacubex/mihomo/component/nat"
- P "github.com/metacubex/mihomo/component/process"
- "github.com/metacubex/mihomo/component/resolver"
- "github.com/metacubex/mihomo/component/slowdown"
- "github.com/metacubex/mihomo/component/sniffer"
- C "github.com/metacubex/mihomo/constant"
- "github.com/metacubex/mihomo/constant/provider"
- icontext "github.com/metacubex/mihomo/context"
- "github.com/metacubex/mihomo/log"
- "github.com/metacubex/mihomo/tunnel/statistic"
- )
- var (
- status = newAtomicStatus(Suspend)
- tcpQueue = make(chan C.ConnContext, 200)
- udpQueue = make(chan C.PacketAdapter, 200)
- natTable = nat.New()
- rules []C.Rule
- listeners = make(map[string]C.InboundListener)
- subRules map[string][]C.Rule
- proxies = make(map[string]C.Proxy)
- providers map[string]provider.ProxyProvider
- ruleProviders map[string]provider.RuleProvider
- sniffingEnable = false
- configMux sync.RWMutex
- // Outbound Rule
- mode = Rule
- // default timeout for UDP session
- udpTimeout = 60 * time.Second
- findProcessMode P.FindProcessMode
- fakeIPRange netip.Prefix
- ruleUpdateCallback = utils.NewCallback[provider.RuleProvider]()
- )
- type tunnel struct{}
- var Tunnel = tunnel{}
- var _ C.Tunnel = Tunnel
- var _ provider.Tunnel = Tunnel
- func (t tunnel) HandleTCPConn(conn net.Conn, metadata *C.Metadata) {
- connCtx := icontext.NewConnContext(conn, metadata)
- handleTCPConn(connCtx)
- }
- func (t tunnel) HandleUDPPacket(packet C.UDPPacket, metadata *C.Metadata) {
- packetAdapter := C.NewPacketAdapter(packet, metadata)
- select {
- case udpQueue <- packetAdapter:
- default:
- }
- }
- func (t tunnel) NatTable() C.NatTable {
- return natTable
- }
- func (t tunnel) Providers() map[string]provider.ProxyProvider {
- return providers
- }
- func (t tunnel) RuleProviders() map[string]provider.RuleProvider {
- return ruleProviders
- }
- func (t tunnel) RuleUpdateCallback() *utils.Callback[provider.RuleProvider] {
- return ruleUpdateCallback
- }
- func OnSuspend() {
- status.Store(Suspend)
- }
- func OnInnerLoading() {
- status.Store(Inner)
- }
- func OnRunning() {
- status.Store(Running)
- }
- func Status() TunnelStatus {
- return status.Load()
- }
- func SetFakeIPRange(p netip.Prefix) {
- fakeIPRange = p
- }
- func FakeIPRange() netip.Prefix {
- return fakeIPRange
- }
- func SetSniffing(b bool) {
- if sniffer.Dispatcher.Enable() {
- configMux.Lock()
- sniffingEnable = b
- configMux.Unlock()
- }
- }
- func IsSniffing() bool {
- return sniffingEnable
- }
- func init() {
- go process()
- }
- // TCPIn return fan-in queue
- // Deprecated: using Tunnel instead
- func TCPIn() chan<- C.ConnContext {
- return tcpQueue
- }
- // UDPIn return fan-in udp queue
- // Deprecated: using Tunnel instead
- func UDPIn() chan<- C.PacketAdapter {
- return udpQueue
- }
- // NatTable return nat table
- func NatTable() C.NatTable {
- return natTable
- }
- // Rules return all rules
- func Rules() []C.Rule {
- return rules
- }
- func Listeners() map[string]C.InboundListener {
- return listeners
- }
- // UpdateRules handle update rules
- func UpdateRules(newRules []C.Rule, newSubRule map[string][]C.Rule, rp map[string]provider.RuleProvider) {
- configMux.Lock()
- rules = newRules
- ruleProviders = rp
- subRules = newSubRule
- configMux.Unlock()
- }
- // Proxies return all proxies
- func Proxies() map[string]C.Proxy {
- return proxies
- }
- func ProxiesWithProviders() map[string]C.Proxy {
- allProxies := make(map[string]C.Proxy)
- for name, proxy := range proxies {
- allProxies[name] = proxy
- }
- for _, p := range providers {
- for _, proxy := range p.Proxies() {
- name := proxy.Name()
- allProxies[name] = proxy
- }
- }
- return allProxies
- }
- // Providers return all compatible providers
- func Providers() map[string]provider.ProxyProvider {
- return providers
- }
- // RuleProviders return all loaded rule providers
- func RuleProviders() map[string]provider.RuleProvider {
- return ruleProviders
- }
- // UpdateProxies handle update proxies
- func UpdateProxies(newProxies map[string]C.Proxy, newProviders map[string]provider.ProxyProvider) {
- configMux.Lock()
- proxies = newProxies
- providers = newProviders
- configMux.Unlock()
- }
- func UpdateListeners(newListeners map[string]C.InboundListener) {
- configMux.Lock()
- defer configMux.Unlock()
- listeners = newListeners
- }
- func UpdateSniffer(dispatcher *sniffer.SnifferDispatcher) {
- configMux.Lock()
- sniffer.Dispatcher = dispatcher
- sniffingEnable = dispatcher.Enable()
- configMux.Unlock()
- }
- // Mode return current mode
- func Mode() TunnelMode {
- return mode
- }
- // SetMode change the mode of tunnel
- func SetMode(m TunnelMode) {
- mode = m
- }
- // SetFindProcessMode replace SetAlwaysFindProcess
- // always find process info if legacyAlways = true or mode.Always() = true, may be increase many memory
- func SetFindProcessMode(mode P.FindProcessMode) {
- findProcessMode = mode
- }
- func isHandle(t C.Type) bool {
- status := status.Load()
- return status == Running || (status == Inner && t == C.INNER)
- }
- // processUDP starts a loop to handle udp packet
- func processUDP() {
- queue := udpQueue
- for conn := range queue {
- handleUDPConn(conn)
- }
- }
- func process() {
- numUDPWorkers := 4
- if num := runtime.GOMAXPROCS(0); num > numUDPWorkers {
- numUDPWorkers = num
- }
- for i := 0; i < numUDPWorkers; i++ {
- go processUDP()
- }
- queue := tcpQueue
- for conn := range queue {
- go handleTCPConn(conn)
- }
- }
- func needLookupIP(metadata *C.Metadata) bool {
- return resolver.MappingEnabled() && metadata.Host == "" && metadata.DstIP.IsValid()
- }
- func preHandleMetadata(metadata *C.Metadata) error {
- // handle IP string on host
- if ip, err := netip.ParseAddr(metadata.Host); err == nil {
- metadata.DstIP = ip
- metadata.Host = ""
- }
- // preprocess enhanced-mode metadata
- if needLookupIP(metadata) {
- host, exist := resolver.FindHostByIP(metadata.DstIP)
- if exist {
- metadata.Host = host
- metadata.DNSMode = C.DNSMapping
- if resolver.FakeIPEnabled() {
- metadata.DstIP = netip.Addr{}
- metadata.DNSMode = C.DNSFakeIP
- } else if node, ok := resolver.DefaultHosts.Search(host, false); ok {
- // redir-host should lookup the hosts
- metadata.DstIP, _ = node.RandIP()
- } else if node != nil && node.IsDomain {
- metadata.Host = node.Domain
- }
- } else if resolver.IsFakeIP(metadata.DstIP) {
- return fmt.Errorf("fake DNS record %s missing", metadata.DstIP)
- }
- } else if node, ok := resolver.DefaultHosts.Search(metadata.Host, true); ok {
- // try use domain mapping
- metadata.Host = node.Domain
- }
- return nil
- }
- func resolveMetadata(metadata *C.Metadata) (proxy C.Proxy, rule C.Rule, err error) {
- if metadata.SpecialProxy != "" {
- var exist bool
- proxy, exist = proxies[metadata.SpecialProxy]
- if !exist {
- err = fmt.Errorf("proxy %s not found", metadata.SpecialProxy)
- }
- return
- }
- if findProcessMode.Always() {
- findPackageName(metadata)
- }
- switch mode {
- case Direct:
- proxy = proxies["DIRECT"]
- case Global:
- proxy = proxies["GLOBAL"]
- // Rule
- default:
- proxy, rule, err = match(metadata)
- }
- return
- }
- func findPackageName(metadata *C.Metadata) {
- if !features.Android {
- uid, path, err := P.FindProcessName(metadata.NetWork.String(), metadata.SrcIP, int(metadata.SrcPort))
- if err != nil {
- log.Debugln("[Process] find process %s error: %v", metadata.String(), err)
- } else {
- metadata.Process = filepath.Base(path)
- metadata.ProcessPath = path
- metadata.Uid = uid
- }
- } else {
- pkg, err := P.FindPackageName(metadata)
- if err != nil {
- log.Debugln("[Process] find process %s error: %v", metadata.String(), err)
- } else {
- metadata.Process = pkg
- }
- }
- }
- func handleUDPConn(packet C.PacketAdapter) {
- if !isHandle(packet.Metadata().Type) {
- packet.Drop()
- return
- }
- metadata := packet.Metadata()
- if !metadata.Valid() {
- packet.Drop()
- log.Warnln("[Metadata] not valid: %#v", metadata)
- return
- }
- // make a fAddr if request ip is fakeip
- var fAddr netip.Addr
- if resolver.IsExistFakeIP(metadata.DstIP) {
- fAddr = metadata.DstIP
- }
- if err := preHandleMetadata(metadata); err != nil {
- packet.Drop()
- log.Debugln("[Metadata PreHandle] error: %s", err)
- return
- }
- if sniffer.Dispatcher.Enable() && sniffingEnable {
- sniffer.Dispatcher.UDPSniff(packet)
- }
- // local resolve UDP dns
- if !metadata.Resolved() {
- ip, err := resolver.ResolveIP(context.Background(), metadata.Host)
- if err != nil {
- return
- }
- metadata.DstIP = ip
- }
- key := packet.LocalAddr().String()
- handle := func() bool {
- pc, proxy := natTable.Get(key)
- if pc != nil {
- if proxy != nil {
- proxy.UpdateWriteBack(packet)
- }
- _ = handleUDPToRemote(packet, pc, metadata)
- return true
- }
- return false
- }
- if handle() {
- packet.Drop()
- return
- }
- cond, loaded := natTable.GetOrCreateLock(key)
- go func() {
- defer packet.Drop()
- if loaded {
- cond.L.Lock()
- cond.Wait()
- handle()
- cond.L.Unlock()
- return
- }
- defer func() {
- natTable.DeleteLock(key)
- cond.Broadcast()
- }()
- proxy, rule, err := resolveMetadata(metadata)
- if err != nil {
- log.Warnln("[UDP] Parse metadata failed: %s", err.Error())
- return
- }
- ctx, cancel := context.WithTimeout(context.Background(), C.DefaultUDPTimeout)
- defer cancel()
- rawPc, err := retry(ctx, func(ctx context.Context) (C.PacketConn, error) {
- return proxy.ListenPacketContext(ctx, metadata.Pure())
- }, func(err error) {
- logMetadataErr(metadata, rule, proxy, err)
- })
- if err != nil {
- return
- }
- logMetadata(metadata, rule, rawPc)
- pc := statistic.NewUDPTracker(rawPc, statistic.DefaultManager, metadata, rule, 0, 0, true)
- if rawPc.Chains().Last() == "REJECT-DROP" {
- pc.Close()
- return
- }
- oAddrPort := metadata.AddrPort()
- writeBackProxy := nat.NewWriteBackProxy(packet)
- natTable.Set(key, pc, writeBackProxy)
- go handleUDPToLocal(writeBackProxy, pc, key, oAddrPort, fAddr)
- handle()
- }()
- }
- func handleTCPConn(connCtx C.ConnContext) {
- if !isHandle(connCtx.Metadata().Type) {
- _ = connCtx.Conn().Close()
- return
- }
- defer func(conn net.Conn) {
- _ = conn.Close()
- }(connCtx.Conn())
- metadata := connCtx.Metadata()
- if !metadata.Valid() {
- log.Warnln("[Metadata] not valid: %#v", metadata)
- return
- }
- preHandleFailed := false
- if err := preHandleMetadata(metadata); err != nil {
- log.Debugln("[Metadata PreHandle] error: %s", err)
- preHandleFailed = true
- }
- conn := connCtx.Conn()
- conn.ResetPeeked() // reset before sniffer
- if sniffer.Dispatcher.Enable() && sniffingEnable {
- // Try to sniff a domain when `preHandleMetadata` failed, this is usually
- // caused by a "Fake DNS record missing" error when enhanced-mode is fake-ip.
- if sniffer.Dispatcher.TCPSniff(conn, metadata) {
- // we now have a domain name
- preHandleFailed = false
- }
- }
- // If both trials have failed, we can do nothing but give up
- if preHandleFailed {
- log.Debugln("[Metadata PreHandle] failed to sniff a domain for connection %s --> %s, give up",
- metadata.SourceDetail(), metadata.RemoteAddress())
- return
- }
- peekMutex := sync.Mutex{}
- if !conn.Peeked() {
- peekMutex.Lock()
- go func() {
- defer peekMutex.Unlock()
- _ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
- _, _ = conn.Peek(1)
- _ = conn.SetReadDeadline(time.Time{})
- }()
- }
- proxy, rule, err := resolveMetadata(metadata)
- if err != nil {
- log.Warnln("[Metadata] parse failed: %s", err.Error())
- return
- }
- dialMetadata := metadata
- if len(metadata.Host) > 0 {
- if node, ok := resolver.DefaultHosts.Search(metadata.Host, false); ok {
- if dstIp, _ := node.RandIP(); !FakeIPRange().Contains(dstIp) {
- dialMetadata.DstIP = dstIp
- dialMetadata.DNSMode = C.DNSHosts
- dialMetadata = dialMetadata.Pure()
- }
- }
- }
- var peekBytes []byte
- var peekLen int
- ctx, cancel := context.WithTimeout(context.Background(), C.DefaultTCPTimeout)
- defer cancel()
- remoteConn, err := retry(ctx, func(ctx context.Context) (remoteConn C.Conn, err error) {
- remoteConn, err = proxy.DialContext(ctx, dialMetadata)
- if err != nil {
- return
- }
- if N.NeedHandshake(remoteConn) {
- defer func() {
- for _, chain := range remoteConn.Chains() {
- if chain == "REJECT" {
- err = nil
- return
- }
- }
- if err != nil {
- remoteConn = nil
- }
- }()
- peekMutex.Lock()
- defer peekMutex.Unlock()
- peekBytes, _ = conn.Peek(conn.Buffered())
- _, err = remoteConn.Write(peekBytes)
- if err != nil {
- return
- }
- if peekLen = len(peekBytes); peekLen > 0 {
- _, _ = conn.Discard(peekLen)
- }
- }
- return
- }, func(err error) {
- logMetadataErr(metadata, rule, proxy, err)
- })
- if err != nil {
- return
- }
- logMetadata(metadata, rule, remoteConn)
- remoteConn = statistic.NewTCPTracker(remoteConn, statistic.DefaultManager, metadata, rule, 0, int64(peekLen), true)
- defer func(remoteConn C.Conn) {
- _ = remoteConn.Close()
- }(remoteConn)
- _ = conn.SetReadDeadline(time.Now()) // stop unfinished peek
- peekMutex.Lock()
- defer peekMutex.Unlock()
- _ = conn.SetReadDeadline(time.Time{}) // reset
- handleSocket(conn, remoteConn)
- }
- func logMetadataErr(metadata *C.Metadata, rule C.Rule, proxy C.ProxyAdapter, err error) {
- if rule == nil {
- log.Warnln("[%s] dial %s %s --> %s error: %s", strings.ToUpper(metadata.NetWork.String()), proxy.Name(), metadata.SourceDetail(), metadata.RemoteAddress(), err.Error())
- } else {
- log.Warnln("[%s] dial %s (match %s/%s) %s --> %s error: %s", strings.ToUpper(metadata.NetWork.String()), proxy.Name(), rule.RuleType().String(), rule.Payload(), metadata.SourceDetail(), metadata.RemoteAddress(), err.Error())
- }
- }
- func logMetadata(metadata *C.Metadata, rule C.Rule, remoteConn C.Connection) {
- switch {
- case metadata.SpecialProxy != "":
- log.Infoln("[%s] %s --> %s using %s", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress(), metadata.SpecialProxy)
- case rule != nil:
- if rule.Payload() != "" {
- log.Infoln("[%s] %s --> %s match %s using %s", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress(), fmt.Sprintf("%s(%s)", rule.RuleType().String(), rule.Payload()), remoteConn.Chains().String())
- } else {
- log.Infoln("[%s] %s --> %s match %s using %s", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress(), rule.RuleType().String(), remoteConn.Chains().String())
- }
- case mode == Global:
- log.Infoln("[%s] %s --> %s using GLOBAL", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress())
- case mode == Direct:
- log.Infoln("[%s] %s --> %s using DIRECT", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress())
- default:
- log.Infoln("[%s] %s --> %s doesn't match any rule using %s", strings.ToUpper(metadata.NetWork.String()), metadata.SourceDetail(), metadata.RemoteAddress(), remoteConn.Chains().Last())
- }
- }
- func shouldResolveIP(rule C.Rule, metadata *C.Metadata) bool {
- return rule.ShouldResolveIP() && metadata.Host != "" && !metadata.DstIP.IsValid()
- }
- func match(metadata *C.Metadata) (C.Proxy, C.Rule, error) {
- configMux.RLock()
- defer configMux.RUnlock()
- var (
- resolved bool
- )
- if node, ok := resolver.DefaultHosts.Search(metadata.Host, false); ok {
- metadata.DstIP, _ = node.RandIP()
- resolved = true
- }
- for _, rule := range getRules(metadata) {
- if !resolved && shouldResolveIP(rule, metadata) {
- func() {
- ctx, cancel := context.WithTimeout(context.Background(), resolver.DefaultDNSTimeout)
- defer cancel()
- ip, err := resolver.ResolveIP(ctx, metadata.Host)
- if err != nil {
- log.Debugln("[DNS] resolve %s error: %s", metadata.Host, err.Error())
- } else {
- log.Debugln("[DNS] %s --> %s", metadata.Host, ip.String())
- metadata.DstIP = ip
- }
- resolved = true
- }()
- }
- if matched, ada := rule.Match(metadata); matched {
- adapter, ok := proxies[ada]
- if !ok {
- continue
- }
- // parse multi-layer nesting
- passed := false
- for adapter := adapter; adapter != nil; adapter = adapter.Unwrap(metadata, false) {
- if adapter.Type() == C.Pass {
- passed = true
- break
- }
- }
- if passed {
- log.Debugln("%s match Pass rule", adapter.Name())
- continue
- }
- if metadata.NetWork == C.UDP && !adapter.SupportUDP() {
- log.Debugln("%s UDP is not supported", adapter.Name())
- continue
- }
- return adapter, rule, nil
- }
- }
- return proxies["DIRECT"], nil, nil
- }
- func getRules(metadata *C.Metadata) []C.Rule {
- if sr, ok := subRules[metadata.SpecialRules]; ok {
- log.Debugln("[Rule] use %s rules", metadata.SpecialRules)
- return sr
- } else {
- log.Debugln("[Rule] use default rules")
- return rules
- }
- }
- func shouldStopRetry(err error) bool {
- if errors.Is(err, resolver.ErrIPNotFound) {
- return true
- }
- if errors.Is(err, resolver.ErrIPVersion) {
- return true
- }
- if errors.Is(err, resolver.ErrIPv6Disabled) {
- return true
- }
- if errors.Is(err, loopback.ErrReject) {
- return true
- }
- return false
- }
- func retry[T any](ctx context.Context, ft func(context.Context) (T, error), fe func(err error)) (t T, err error) {
- s := slowdown.New()
- for i := 0; i < 10; i++ {
- t, err = ft(ctx)
- if err != nil {
- if fe != nil {
- fe(err)
- }
- if shouldStopRetry(err) {
- return
- }
- if s.Wait(ctx) == nil {
- continue
- } else {
- return
- }
- } else {
- break
- }
- }
- return
- }
|