1
func Run(log *logrus.Logger, conf *config.Config, externGeoDataDirs []string) (err error) {
2
  // Remove AbortFile at beginning.
3
  _ = os.Remove(AbortFile)
4

5
  // New ControlPlane.
6
  c, err := newControlPlane(log, nil, nil, conf, externGeoDataDirs)
7
  if err != nil {
8
    return err
9
  }
10

11
  var pprofServer *http.Server
12
  if conf.Global.PprofPort != 0 {
13
    pprofAddr := fmt.Sprintf("localhost:%d", conf.Global.PprofPort)
14
    pprofServer = &http.Server{Addr: pprofAddr, Handler: nil}
15
    go pprofServer.ListenAndServe()
16
  }
17

18
  // Start HTTP server if configured.
19
  if conf.Global.HttpPort != 0 {
20
    go startServer(log, conf)
21
  }
22

23
  // Serve tproxy TCP/UDP server util signals.
24
  var listener *control.Listener
25
  sigs := make(chan os.Signal, 1)
26
  signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGKILL, syscall.SIGILL, syscall.SIGUSR1, syscall.SIGUSR2)
27
  go func() {
28
    readyChan := make(chan bool, 1)
29
    go func() {
30
      <-readyChan
31
      sdnotify.Ready()
32
      if !disablePidFile {
33
        _ = os.WriteFile(PidFilePath, []byte(strconv.Itoa(os.Getpid())), 0644)
34
      }
35
      _ = os.WriteFile(SignalProgressFilePath, []byte{consts.ReloadDone}, 0644)
36
    }()
37
    control.GetDaeNetns().With(func() error {
38
      if listener, err = c.ListenAndServe(readyChan, conf.Global.TproxyPort); err != nil {
39
        log.Errorln("ListenAndServe:", err)
40
      }
41
      return err
42
    })
43
    sigs <- nil
44
  }()

1
func (c *ControlPlane) handlePkt(lConn *net.UDPConn, data []byte, src, pktDst, realDst netip.AddrPort, routingResult *bpfRoutingResult, skipSniffing bool) (err error) {
2
  var realSrc netip.AddrPort
3
  var domain string
4
  realSrc = src
5
  ue, ueExists := DefaultUdpEndpointPool.Get(realSrc)
6
  if ueExists && ue.SniffedDomain != "" {
7
    // It is quic ...
8
    // Fast path.
9
    domain := ue.SniffedDomain
10
    dialTarget := realDst.String()
11

12
    if c.log.IsLevelEnabled(logrus.TraceLevel) {
13
      fields := logrus.Fields{
14
        "network":  "udp(fp)",
15
        "outbound": ue.Outbound.Name,
16
        "policy":   ue.Outbound.GetSelectionPolicy(),
17
        "dialer":   ue.Dialer.Property().Name,
18
        "sniffed":  domain,
19
        "ip":       RefineAddrPortToShow(realDst),
20
        "pid":      routingResult.Pid,
21
        "dscp":     routingResult.Dscp,
22
        "pname":    ProcessName2String(routingResult.Pname[:]),
23
        "mac":      Mac2String(routingResult.Mac[:]),
24
      }
25
      c.log.WithFields(fields).Tracef("%v <-> %v", RefineSourceToShow(realSrc, realDst.Addr()), dialTarget)
26
    }
27

28
    _, err = ue.WriteTo(data, dialTarget)
29
    if err != nil {
30
      return err
31
    }
32
    return nil
33
  }
34

35
  // To keep consistency with kernel program, we only sniff DNS request sent to 53.
36
  dnsMessage, natTimeout := ChooseNatTimeout(data, realDst.Port() == 53)
37
  // We should cache DNS records and set record TTL to 0, in order to monitor the dns req and resp in real time.
38
  isDns := dnsMessage != nil
39
  if !isDns && !skipSniffing && !ueExists {
40
    // Sniff Quic, ...
41
    key := PacketSnifferKey{
42
      LAddr: realSrc,
43
      RAddr: realDst,
44
    }
45
    _sniffer, _ := DefaultPacketSnifferSessionMgr.GetOrCreate(key, nil)
46
    _sniffer.Mu.Lock()
47
    // Re-get sniffer from pool to confirm the transaction is not done.
48
    sniffer := DefaultPacketSnifferSessionMgr.Get(key)
49
    if _sniffer == sniffer {
50
      sniffer.AppendData(data)
51
      domain, err = sniffer.SniffUdp()
52
      if err != nil && !sniffing.IsSniffingError(err) {
53
        sniffer.Mu.Unlock()
54
        return err
55
      }
56
      if sniffer.NeedMore() {
57
        sniffer.Mu.Unlock()
58
        return nil
59
      }
60
      if err != nil {
61
        logrus.WithError(err).
62
          WithField("from", realSrc).
63
          WithField("to", realDst).
64
          Trace("sniffUdp")
65
      }
66
      defer DefaultPacketSnifferSessionMgr.Remove(key, sniffer)
67
      // Re-handlePkt after self func.
68
      toRehandle := sniffer.Data()[1 : len(sniffer.Data())-1] // Skip the first empty and the last (self).
69
      sniffer.Mu.Unlock()
70
      if len(toRehandle) > 0 {
71
        defer func() {
72
          if err == nil {
73
            for _, d := range toRehandle {
74
              dCopy := pool.Get(len(d))
75
              copy(dCopy, d)
76
              go c.handlePkt(lConn, dCopy, src, pktDst, realDst, routingResult, true)
77
            }
78
          }
79
        }()
80
      }
81
    } else {
82
      _sniffer.Mu.Unlock()
83
      // sniffer may be nil.
84
    }
85
  }
86
  if routingResult.Must > 0 {
87
    isDns = false // Regard as plain traffic.
88
  }
89
  if routingResult.Mark == 0 {
90
    routingResult.Mark = c.soMarkFromDae
91
  }
92
  if isDns {
93
    return c.dnsController.Handle_(dnsMessage, &udpRequest{
94
      realSrc:       realSrc,
95
      realDst:       realDst,
96
      src:           src,
97
      lConn:         lConn,
98
      routingResult: routingResult,
99
    })
100
  }

1
func (c *DnsController) handleWithResponseWriter_(
2
  dnsMessage *dnsmessage.Msg,
3
  req *udpRequest,
4
  needResp bool,
5
  responseWriter dnsmessage.ResponseWriter,
6
) (err error) {
7
  // Prepare qname, qtype.
8
  var qname string
9
  var qtype uint16
10
  if len(dnsMessage.Question) != 0 {
11
    q := dnsMessage.Question[0]
12
    qname = q.Name
13
    qtype = q.Qtype
14
  }
15

16
  // Route request.
17
  upstreamIndex, upstream, err := c.routing.RequestSelect(qname, qtype)
18
  if err != nil {
19
    return err
20
  }
21

22
  cacheKey := c.cacheKey(qname, qtype, req.src)
23
  if c.log.IsLevelEnabled(logrus.DebugLevel) {
24
    c.log.WithFields(logrus.Fields{
25
      "qname":    qname,
26
      "qtype":    QtypeToString(qtype),
27
      "src":      req.src.String(),
28
      "realSrc":  req.realSrc.String(),
29
      "cacheKey": cacheKey,
30
    }).Debugf("DNS request - looking up cache")
31
  }
32

33
  if upstreamIndex == consts.DnsRequestOutboundIndex_Reject {
34
    // Reject with empty answer.
35
    c.RemoveDnsRespCache(cacheKey)
36
    return c.sendRejectWithResponseWriter_(dnsMessage, req, responseWriter)
37
  }
38

39
  // No parallel for the same lookup.
40
  handlingState_, _ := c.handling.LoadOrStore(cacheKey, new(handlingState))
41
  handlingState := handlingState_.(*handlingState)
42
  atomic.AddUint32(&handlingState.ref, 1)
43
  handlingState.mu.Lock()
44
  defer func() {
45
    handlingState.mu.Unlock()
46
    atomic.AddUint32(&handlingState.ref, ^uint32(0))
47
    if atomic.LoadUint32(&handlingState.ref) == 0 {
48
      c.handling.Delete(cacheKey)
49
    }
50
  }()
51

52
  if resp := c.LookupDnsRespCache_(dnsMessage, cacheKey, false); resp != nil {
53
    // Send cache to client directly.
54
    if needResp {
55
      if responseWriter != nil {
56
        var respMsg dnsmessage.Msg
57
        if err = respMsg.Unpack(resp); err != nil {
58
          return fmt.Errorf("failed to unpack DNS response: %w", err)
59
        }
60
        return responseWriter.WriteMsg(&respMsg)
61
      }
62
      if err = sendPkt(c.log, resp, req.realDst, req.realSrc, req.src, req.lConn); err != nil {
63
        return fmt.Errorf("failed to write cached DNS resp: %w", err)
64
      }
65
    }
66
    if c.log.IsLevelEnabled(logrus.DebugLevel) && len(dnsMessage.Question) > 0 {
67
      q := dnsMessage.Question[0]
68
      c.log.WithFields(logrus.Fields{
69
        "cacheKey": cacheKey,
70
        "qname":    strings.ToLower(q.Name),
71
        "qtype":    QtypeToString(q.Qtype),
72
        "src":      req.src.String(),
73
        "realSrc":  req.realSrc.String(),
74
      }).Debugf("UDP(DNS) %v <-> Cache: %v %v",
75
        RefineSourceToShow(req.realSrc, req.realDst.Addr()), strings.ToLower(q.Name), QtypeToString(q.Qtype),
76
      )
77
    }
78
    return nil
79
  }
80

81
  if c.log.IsLevelEnabled(logrus.DebugLevel) {
82
    upstreamName := upstreamIndex.String()
83
    if upstream != nil {
84
      upstreamName = upstream.String()
85
    }
86
    c.log.WithFields(logrus.Fields{
87
      "cacheKey": cacheKey,
88
      "qname":    qname,
89
      "qtype":    QtypeToString(qtype),
90
      "src":      req.src.String(),
91
      "realSrc":  req.realSrc.String(),
92
      "upstream": upstreamName,
93
    }).Debugf("DNS cache miss - requesting from upstream")
94
  }
95
  if c.log.IsLevelEnabled(logrus.TraceLevel) {
96
    upstreamName := upstreamIndex.String()
97
    if upstream != nil {
98
      upstreamName = upstream.String()
99
    }
100
    c.log.WithFields(logrus.Fields{
101
      "question": dnsMessage.Question,
102
      "upstream": upstreamName,
103
    }).Traceln("Request to DNS upstream")
104
  }
105

106
  // Re-pack DNS packet.
107
  data, err := dnsMessage.Pack()
108
  if err != nil {
109
    return fmt.Errorf("pack DNS packet: %w", err)
110
  }
111
  return c.dialSend(0, req, data, dnsMessage.Id, upstream, needResp)
112
}

1
func (c *DnsController) dialSend(invokingDepth int, req *udpRequest, data []byte, id uint16, upstream *dns.Upstream, needResp bool) (err error) {
2
  if invokingDepth >= MaxDnsLookupDepth {
3
    return fmt.Errorf("too deep DNS lookup invoking (depth: %v); there may be infinite loop in your DNS response routing", MaxDnsLookupDepth)
4
  }
5

6
  upstreamName := "asis"
7
  if upstream == nil {
8
    // As-is.
9

10
    // As-is should not be valid in response routing, thus using connection realDest is reasonable.
11
    var ip46 netutils.Ip46
12
    if req.realDst.Addr().Is4() {
13
      ip46.Ip4 = req.realDst.Addr()
14
    } else {
15
      ip46.Ip6 = req.realDst.Addr()
16
    }
17
    upstream = &dns.Upstream{
18
      Scheme:   "udp",
19
      Hostname: req.realDst.Addr().String(),
20
      Port:     req.realDst.Port(),
21
      Ip46:     &ip46,
22
    }
23
  } else {
24
    upstreamName = upstream.String()
25
  }
26

27
  // Select best dial arguments (outbound, dialer, l4proto, ipversion, etc.)
28
  dialArgument, err := c.bestDialerChooser(req, upstream)
29
  if err != nil {
30
    return err
31
  }
32

33
  c.log.Infof("*************************************************")
34
  c.log.Infof("dialArgument: %v", dialArgument)
35
  c.log.Infof("*************************************************")
36

37
  networkType := &dialer.NetworkType{
38
    L4Proto:   dialArgument.l4proto,
39
    IpVersion: dialArgument.ipversion,
40
    IsDns:     true,
41
  }
42

43
  // Dial and send.
44
  var respMsg *dnsmessage.Msg
45
  // defer in a recursive call will delay Close(), thus we Close() before
46
  // the next recursive call. However, a connection cannot be closed twice.
47
  // We should set a connClosed flag to avoid it.
48
  var connClosed bool
49

50
  ctxDial, cancel := context.WithTimeout(context.TODO(), consts.DefaultDialTimeout)
51
  defer cancel()
52

53
  // get forwarder from cache
54
  c.dnsForwarderCacheMu.Lock()
55
  forwarder, ok := c.dnsForwarderCache[dnsForwarderKey{upstream: upstream.String(), dialArgument: *dialArgument}]
56
  if !ok {
57
    forwarder, err = newDnsForwarder(upstream, *dialArgument)
58
    if err != nil {
59
      c.dnsForwarderCacheMu.Unlock()
60
      return err
61
    }
62
    c.dnsForwarderCache[dnsForwarderKey{upstream: upstream.String(), dialArgument: *dialArgument}] = forwarder
63
  }
64
  c.dnsForwarderCacheMu.Unlock()
65

66
  defer func() {
67
    if !connClosed {
68
      forwarder.Close()
69
    }
70
  }()
71

72
  if err != nil {
73
    return err
74
  }
75

76
  c.log.Infof("*************************************************")
77
  c.log.Infof("ForwardDNS request: %v", data)
78
  c.log.Infof("*************************************************")
79

80
  respMsg, err = forwarder.ForwardDNS(ctxDial, data)
81
  c.log.Infof("*************************************************")
82
  c.log.Infof("ForwardDNS response: %v", respMsg)
83
  c.log.Infof("ForwardDNS error: %v", err)
84
  c.log.Infof("*************************************************")
85
  if err != nil {
86
    return err
87
  }

1
func (d *DoUDP) ForwardDNS(ctx context.Context, data []byte) (*dnsmessage.Msg, error) {
2
  conn, err := d.dialArgument.bestDialer.DialContext(
3
    ctx,
4
    common.MagicNetwork("udp", d.dialArgument.mark, d.dialArgument.mptcp),
5
    d.dialArgument.bestTarget.String(),
6
  )
7
  if err != nil {
8
    return nil, err
9
  }
10
  defer conn.Close()
11

12
  timeout := 10 * time.Second
13
  targetAddr := d.dialArgument.bestTarget.String()
14

15
  // Check if connection is PacketConn (like in udp_endpoint_pool.go)
16
  // PacketConn uses ReadFrom/WriteTo, while connected Conn uses Read/Write
17
  var pktConn netproxy.PacketConn
18
  var usePacketConn bool
19
  if pc, ok := conn.(netproxy.PacketConn); ok {
20
    pktConn = pc
21
    usePacketConn = true
22
    _ = pktConn.SetDeadline(time.Now().Add(timeout))
23
    fmt.Println("*************************************************")
24
    fmt.Println("DoUDP using PacketConn (ReadFrom/WriteTo), timeout: %v", timeout)
25
    fmt.Println("DoUDP target: %v", targetAddr)
26
    fmt.Println("*************************************************")
27
  } else {
28
    // Fallback to Conn (Read/Write)
29
    _ = conn.SetDeadline(time.Now().Add(timeout))
30
    fmt.Println("*************************************************")
31
    fmt.Println("DoUDP using Conn (Read/Write), timeout: %v", timeout)
32
    fmt.Println("DoUDP target: %v", targetAddr)
33
    fmt.Println("*************************************************")
34
  }
35

36
  dnsReqCtx, cancelDnsReqCtx := context.WithTimeout(context.TODO(), timeout)
37
  defer cancelDnsReqCtx()
38

39
  go func() {
40
    // Send DNS request every seconds.
41
    writeCount := 0
42
    for {
43
      var err error
44
      var n int
45
      if usePacketConn {
46
        n, err = pktConn.WriteTo(data, targetAddr)
47
        writeCount++
48
        if writeCount == 1 {
49
          fmt.Println("*************************************************")
50
          fmt.Println("DoUDP WriteTo bytes: %v, target: %v, error: %v", n, targetAddr, err)
51
          fmt.Println("*************************************************")
52
        }
53
      } else {
54
        n, err = conn.Write(data)
55
        writeCount++
56
        if writeCount == 1 {
57
          fmt.Println("*************************************************")
58
          fmt.Println("DoUDP Write bytes: %v, error: %v", n, err)
59
          fmt.Println("*************************************************")
60
        }
61
      }
62
      if err != nil {
63
        fmt.Println("*************************************************")
64
        fmt.Println("DoUDP Write error: %v", err)
65
        fmt.Println("*************************************************")
66
        return
67
      }
68
      select {
69
      case <-dnsReqCtx.Done():
70
        return
71
      case <-time.After(1 * time.Second):
72
      }
73
    }
74
  }()
75

76
  // We can block here because we are in a coroutine.
77
  respBuf := pool.GetFullCap(consts.EthernetMtu)
78
  fmt.Println("*************************************************")
79
  fmt.Println("DoUDP respBuf cap: %v", cap(respBuf))
80
  fmt.Println("DoUDP respBuf len: %v", len(respBuf))
81
  fmt.Println("*************************************************")
82
  defer pool.Put(respBuf)
83

84
  // Wait for response
85
  var n int
86
  readStartTime := time.Now()
87
  if usePacketConn {
88
    // Use ReadFrom for PacketConn (unconnected socket)
89
    var from netip.AddrPort
90
    n, from, err = pktConn.ReadFrom(respBuf)
91
    readDuration := time.Since(readStartTime)
92
    fmt.Println("*************************************************")
93
    fmt.Println("DoUDP ReadFrom n: %v", n)
94
    if from.IsValid() {
95
      fmt.Println("DoUDP ReadFrom from: %v", from.String())
96
    }
97
    fmt.Println("DoUDP ReadFrom error: %v", err)
98
    fmt.Println("DoUDP ReadFrom duration: %v", readDuration)
99
    fmt.Println("DoUDP expected from: %v", targetAddr)
100
    fmt.Println("*************************************************")
101
  } else {
102
    // Use Read for Conn (connected socket)
103
    n, err = conn.Read(respBuf)
104
    readDuration := time.Since(readStartTime)
105
    fmt.Println("*************************************************")
106
    fmt.Println("DoUDP Read n: %v", n)
107
    fmt.Println("DoUDP Read error: %v", err)
108
    fmt.Println("DoUDP Read duration: %v", readDuration)
109
    fmt.Println("*************************************************")
110
  }
111

112
  if err != nil {
113
    return nil, err
114
  }
115
  var msg dnsmessage.Msg
116
  if err = msg.Unpack(respBuf[:n]); err != nil {
117
    return nil, err
118
  }
119
  return &msg, nil
120
}