add drop-only option to dae trace

.gitignore

@@ -4,6 +4,8 @@
 *.tmp
 bpf_bpfeb*.go
 bpf_bpfel*.go
+bpf_*_bpfeb*.go
+bpf_*_bpfel*.go
 dae
 outline.json
 go-mod/

CHANGELOGS.md

@@ -14,7 +14,8 @@ curl --silent "https://api.github.com/repos/daeuniverse/dae/releases" | jq -r '.
 
 <!-- BEGIN NEW TOC ENTRY -->
 
-- [v0.7.1 (Latest)](#v071-latest)
+- [v0.7.2 (Latest)](#v072-latest)
+- [v0.7.1](#v071)
 - [v0.8.0rc1 (Pre-release)](#v080rc1-pre-release)
 - [v0.7.0](#v070)
 - [v0.6.0](#v060)
@@ -42,7 +43,16 @@ curl --silent "https://api.github.com/repos/daeuniverse/dae/releases" | jq -r '.
 - [v0.1.0](#v010)
 <!-- BEGIN NEW CHANGELOGS -->
 
-### v0.7.1 (Latest)
+### v0.7.2 (Latest)
+
+> Release date: 2024/09/27
+
+#### Bug Fixes
+
+- fix: connection leaks (#624)
+- fix: crash on openwrt (#640)
+
+### v0.7.1
 
 > Release date: 2024/08/21
 

cmd/run.go

@@ -105,8 +105,9 @@ var (
 					Compress:   true,
 				}
 			}
-			log := logger.NewLogger(conf.Global.LogLevel, disableTimestamp, logOpts)
-			logrus.SetLevel(log.Level)
+			log := logrus.New()
+			logger.SetLogger(log, conf.Global.LogLevel, disableTimestamp, logOpts)
+			logger.SetLogger(logrus.StandardLogger(), conf.Global.LogLevel, disableTimestamp, logOpts)
 
 			log.Infof("Include config files: [%v]", strings.Join(includes, ", "))
 			if err := Run(log, conf, []string{filepath.Dir(cfgFile)}); err != nil {
@@ -238,9 +239,11 @@ loop:
 			}
 			// New logger.
 			oldLogOutput := log.Out
-			log = logger.NewLogger(newConf.Global.LogLevel, disableTimestamp, nil)
+			log = logrus.New()
+			logger.SetLogger(log, newConf.Global.LogLevel, disableTimestamp, nil)
+			logger.SetLogger(logrus.StandardLogger(), newConf.Global.LogLevel, disableTimestamp, nil)
 			log.SetOutput(oldLogOutput) // FIXME: THIS IS A HACK.
-			logrus.SetLevel(log.Level)
+			logrus.SetOutput(oldLogOutput)
 
 			// New control plane.
 			obj := c.EjectBpf()
@@ -330,8 +333,7 @@ func newControlPlane(log *logrus.Logger, bpf interface{}, dnsCache map[string]*c
 		client := http.Client{
 			Transport: &http.Transport{
 				DialContext: func(ctx context.Context, network, addr string) (c net.Conn, err error) {
-					cd := netproxy.ContextDialerConverter{Dialer: direct.SymmetricDirect}
-					conn, err := cd.DialContext(ctx, common.MagicNetwork("tcp", conf.Global.SoMarkFromDae, conf.Global.Mptcp), addr)
+					conn, err := direct.SymmetricDirect.DialContext(ctx, common.MagicNetwork("tcp", conf.Global.SoMarkFromDae, conf.Global.Mptcp), addr)
 					if err != nil {
 						return nil, err
 					}
@@ -372,8 +374,7 @@ func newControlPlane(log *logrus.Logger, bpf interface{}, dnsCache map[string]*c
 	client := http.Client{
 		Transport: &http.Transport{
 			DialContext: func(ctx context.Context, network, addr string) (c net.Conn, err error) {
-				cd := netproxy.ContextDialerConverter{Dialer: direct.SymmetricDirect}
-				conn, err := cd.DialContext(ctx, common.MagicNetwork("tcp", conf.Global.SoMarkFromDae, conf.Global.Mptcp), addr)
+				conn, err := direct.SymmetricDirect.DialContext(ctx, common.MagicNetwork("tcp", conf.Global.SoMarkFromDae, conf.Global.Mptcp), addr)
 				if err != nil {
 					return nil, err
 				}

cmd/trace.go

@@ -24,6 +24,7 @@ var (
 	L4Proto    string
 	Port       int
 	OutputFile string
+	DropOnly   bool
 )
 
 func init() {
@@ -56,7 +57,7 @@ func init() {
 
 			ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 			defer cancel()
-			if err := trace.StartTrace(ctx, IPVersion, L4ProtoNo, Port, OutputFile); err != nil {
+			if err := trace.StartTrace(ctx, IPVersion, L4ProtoNo, Port, DropOnly, OutputFile); err != nil {
 				logrus.Fatalln(err)
 			}
 		},
@@ -66,6 +67,7 @@ func init() {
 	traceCmd.PersistentFlags().BoolVarP(&IPv6, "ipv6", "6", false, "Capture IPv6 traffic")
 	traceCmd.PersistentFlags().StringVarP(&L4Proto, "l4-proto", "p", "tcp", "Layer 4 protocol")
 	traceCmd.PersistentFlags().IntVarP(&Port, "port", "P", 80, "Port")
+	traceCmd.PersistentFlags().BoolVarP(&DropOnly, "drop-only", "", false, "only trace the dropped package")
 	traceCmd.PersistentFlags().StringVarP(&OutputFile, "output", "o", "/dev/stdout", "Output file")
 
 	rootCmd.AddCommand(traceCmd)

common/netutils/dns.go

@@ -207,8 +207,7 @@ func resolve(ctx context.Context, d netproxy.Dialer, dns netip.AddrPort, host st
 	}
 
 	// Dial and write.
-	cd := &netproxy.ContextDialerConverter{Dialer: d}
-	c, err := cd.DialContext(ctx, network, dns.String())
+	c, err := d.DialContext(ctx, network, dns.String())
 	if err != nil {
 		return nil, err
 	}

component/outbound/dialer/connectivity_check.go

@@ -600,12 +600,11 @@ func (d *Dialer) HttpCheck(ctx context.Context, u *netutils.URL, ip netip.Addr,
 	if method == "" {
 		method = http.MethodGet
 	}
-	cd := &netproxy.ContextDialerConverter{Dialer: d.Dialer}
 	cli := http.Client{
 		Transport: &http.Transport{
 			DialContext: func(ctx context.Context, network, addr string) (c net.Conn, err error) {
 				// Force to dial "ip".
-				conn, err := cd.DialContext(ctx, common.MagicNetwork("tcp", soMark, mptcp), net.JoinHostPort(ip.String(), u.Port()))
+				conn, err := d.Dialer.DialContext(ctx, common.MagicNetwork("tcp", soMark, mptcp), net.JoinHostPort(ip.String(), u.Port()))
 				if err != nil {
 					return nil, err
 				}

component/outbound/dialer/dialer.go

@@ -68,7 +68,9 @@ func NewGlobalOption(global *config.Global, log *logrus.Logger) *GlobalOption {
 		ExtraOption: D.ExtraOption{
 			AllowInsecure:     global.AllowInsecure,
 			TlsImplementation: global.TlsImplementation,
-			UtlsImitate:       global.UtlsImitate},
+			UtlsImitate:       global.UtlsImitate,
+			BandwidthMaxTx:    global.BandwidthMaxTx,
+			BandwidthMaxRx:    global.BandwidthMaxRx},
 		Log:               log,
 		TcpCheckOptionRaw: TcpCheckOptionRaw{Raw: global.TcpCheckUrl, Log: log, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Method: global.TcpCheckHttpMethod},
 		CheckDnsOptionRaw: CheckDnsOptionRaw{Raw: global.UdpCheckDns, ResolverNetwork: common.MagicNetwork("udp", global.SoMarkFromDae, global.Mptcp), Somark: global.SoMarkFromDae},

component/sniffing/sniffer.go

@@ -107,13 +107,10 @@ func (s *Sniffer) SniffTcp() (d string, err error) {
 		if s.stream {
 			go func() {
 				// Read once.
-				n, err := s.buf.ReadFromOnce(s.r)
+				_, err = s.buf.ReadFromOnce(s.r)
 				if err != nil {
 					s.dataError = err
 				}
-				if n == 0 {
-					s.dataError = io.EOF
-				}
 				close(s.dataReady)
 			}()
 

config/config.go

@@ -43,6 +43,9 @@ type Global struct {
 	UtlsImitate            string        `mapstructure:"utls_imitate" default:"chrome_auto"`
 	PprofPort              uint16        `mapstructure:"pprof_port" default:"0"`
 	Mptcp                  bool          `mapstructure:"mptcp" default:"false"`
+	// TODO: support input in human-readable format (e.g., 100Mbps, 1Gbps)
+	BandwidthMaxTx uint64 `mapstructure:"bandwidth_max_tx" default:"0"`
+	BandwidthMaxRx uint64 `mapstructure:"bandwidth_max_rx" default:"0"`
 }
 
 type Utls struct {

control/dns_control.go

@@ -166,6 +166,7 @@ func (c *DnsController) LookupDnsRespCache_(msg *dnsmessage.Msg, cacheKey string
 	cache := c.LookupDnsRespCache(cacheKey, ignoreFixedTtl)
 	if cache != nil {
 		cache.FillInto(msg)
+		msg.Compress = true
 		b, err := msg.Pack()
 		if err != nil {
 			c.log.Warnf("failed to pack: %v", err)
@@ -497,6 +498,7 @@ func (c *DnsController) sendReject_(dnsMessage *dnsmessage.Msg, req *udpRequest)
 	dnsMessage.Response = true
 	dnsMessage.RecursionAvailable = true
 	dnsMessage.Truncated = false
+	dnsMessage.Compress = true
 	if c.log.IsLevelEnabled(logrus.TraceLevel) {
 		c.log.WithFields(logrus.Fields{
 			"question": dnsMessage.Question,
@@ -560,16 +562,13 @@ func (c *DnsController) dialSend(invokingDepth int, req *udpRequest, data []byte
 
 	ctxDial, cancel := context.WithTimeout(context.TODO(), consts.DefaultDialTimeout)
 	defer cancel()
-	bestContextDialer := netproxy.ContextDialerConverter{
-		Dialer: dialArgument.bestDialer,
-	}
 
 	switch dialArgument.l4proto {
 	case consts.L4ProtoStr_UDP:
 		// Get udp endpoint.
 
 		// TODO: connection pool.
-		conn, err = bestContextDialer.DialContext(
+		conn, err = dialArgument.bestDialer.DialContext(
 			ctxDial,
 			common.MagicNetwork("udp", dialArgument.mark, dialArgument.mptcp),
 			dialArgument.bestTarget.String(),
@@ -634,7 +633,7 @@ func (c *DnsController) dialSend(invokingDepth int, req *udpRequest, data []byte
 	case consts.L4ProtoStr_TCP:
 		// We can block here because we are in a coroutine.
 
-		conn, err = bestContextDialer.DialContext(ctxDial, common.MagicNetwork("tcp", dialArgument.mark, dialArgument.mptcp), dialArgument.bestTarget.String())
+		conn, err = dialArgument.bestDialer.DialContext(ctxDial, common.MagicNetwork("tcp", dialArgument.mark, dialArgument.mptcp), dialArgument.bestTarget.String())
 		if err != nil {
 			return fmt.Errorf("failed to dial proxy to tcp: %w", err)
 		}
@@ -756,6 +755,7 @@ func (c *DnsController) dialSend(invokingDepth int, req *udpRequest, data []byte
 	if needResp {
 		// Keep the id the same with request.
 		respMsg.Id = id
+		respMsg.Compress = true
 		data, err = respMsg.Pack()
 		if err != nil {
 			return err

control/kern/tproxy.c

@@ -302,14 +302,14 @@ enum __attribute__((packed)) MatchType {
 
 enum L4ProtoType {
 	L4ProtoType_TCP = 1,
-	L4ProtoType_UDP = 2,
-	L4ProtoType_X = 3,
+	L4ProtoType_UDP,
+	L4ProtoType_X,
 };
 
 enum IpVersionType {
 	IpVersionType_4 = 1,
-	IpVersionType_6 = 2,
-	IpVersionType_X = 3,
+	IpVersionType_6,
+	IpVersionType_X,
 };
 
 struct port_range {
@@ -408,7 +408,7 @@ static __always_inline __u8 ipv4_get_dscp(const struct iphdr *iph)
 
 static __always_inline __u8 ipv6_get_dscp(const struct ipv6hdr *ipv6h)
 {
-	return (ipv6h->priority << 2) + (ipv6h->flow_lbl[0] >> 6);
+	return (ipv6h->priority << 2) | (ipv6h->flow_lbl[0] >> 6);
 }
 
 static __always_inline void
@@ -1112,9 +1112,9 @@ int tproxy_lan_ingress(struct __sk_buff *skb)
 	__be32 mac[4] = {
 		0,
 		0,
-		bpf_htonl((ethh.h_source[0] << 8) + (ethh.h_source[1])),
-		bpf_htonl((ethh.h_source[2] << 24) + (ethh.h_source[3] << 16) +
-			  (ethh.h_source[4] << 8) + (ethh.h_source[5])),
+		bpf_htonl((ethh.h_source[0] << 8) | (ethh.h_source[1])),
+		bpf_htonl((ethh.h_source[2] << 24) | (ethh.h_source[3] << 16) |
+			  (ethh.h_source[4] << 8) | (ethh.h_source[5])),
 	};
 	__s64 s64_ret;
 
@@ -1288,15 +1288,15 @@ refresh_udp_conn_state_timer(struct tuples_key *key, bool is_egress)
 	if (unlikely(!value))
 		return NULL;
 
-	if ((ret = bpf_timer_init(&value->timer, &udp_conn_state_map,
-				  CLOCK_MONOTONIC)))
+	if ((bpf_timer_init(&value->timer, &udp_conn_state_map,
+			    CLOCK_MONOTONIC)))
 		goto retn;
 
-	if ((ret = bpf_timer_set_callback(&value->timer,
-					  refresh_udp_conn_state_timer_cb)))
+	if ((bpf_timer_set_callback(&value->timer,
+				    refresh_udp_conn_state_timer_cb)))
 		goto retn;
 
-	if ((ret = bpf_timer_start(&value->timer, TIMEOUT_UDP_CONN_STATE, 0)))
+	if ((bpf_timer_start(&value->timer, TIMEOUT_UDP_CONN_STATE, 0)))
 		goto retn;
 
 retn:
@@ -1405,11 +1405,11 @@ int tproxy_wan_egress(struct __sk_buff *skb)
 			__be32 mac[4] = {
 				0,
 				0,
-				bpf_htonl((ethh.h_source[0] << 8) +
+				bpf_htonl((ethh.h_source[0] << 8) |
 					  (ethh.h_source[1])),
-				bpf_htonl((ethh.h_source[2] << 24) +
-					  (ethh.h_source[3] << 16) +
-					  (ethh.h_source[4] << 8) +
+				bpf_htonl((ethh.h_source[2] << 24) |
+					  (ethh.h_source[3] << 16) |
+					  (ethh.h_source[4] << 8) |
 					  (ethh.h_source[5])),
 			};
 			__s64 s64_ret;
@@ -1532,10 +1532,10 @@ int tproxy_wan_egress(struct __sk_buff *skb)
 		__be32 mac[4] = {
 			0,
 			0,
-			bpf_htonl((ethh.h_source[0] << 8) + (ethh.h_source[1])),
-			bpf_htonl((ethh.h_source[2] << 24) +
-				  (ethh.h_source[3] << 16) +
-				  (ethh.h_source[4] << 8) + (ethh.h_source[5])),
+			bpf_htonl((ethh.h_source[0] << 8) | (ethh.h_source[1])),
+			bpf_htonl((ethh.h_source[2] << 24) |
+				  (ethh.h_source[3] << 16) |
+				  (ethh.h_source[4] << 8) | (ethh.h_source[5])),
 		};
 		__s64 s64_ret;
 
@@ -1639,17 +1639,23 @@ int tproxy_dae0_ingress(struct __sk_buff *skb)
 	struct redirect_tuple redirect_tuple = {};
 
 	if (skb->protocol == bpf_htons(ETH_P_IP)) {
-		bpf_skb_load_bytes(skb, ETH_HLEN + offsetof(struct iphdr, daddr),
+		bpf_skb_load_bytes(skb,
+				   ETH_HLEN + offsetof(struct iphdr, daddr),
 				   &redirect_tuple.sip.u6_addr32[3],
 				   sizeof(redirect_tuple.sip.u6_addr32[3]));
-		bpf_skb_load_bytes(skb, ETH_HLEN + offsetof(struct iphdr, saddr),
+		bpf_skb_load_bytes(skb,
+				   ETH_HLEN + offsetof(struct iphdr, saddr),
 				   &redirect_tuple.dip.u6_addr32[3],
 				   sizeof(redirect_tuple.dip.u6_addr32[3]));
 	} else {
-		bpf_skb_load_bytes(skb, ETH_HLEN + offsetof(struct ipv6hdr, daddr),
-				   &redirect_tuple.sip, sizeof(redirect_tuple.sip));
-		bpf_skb_load_bytes(skb, ETH_HLEN + offsetof(struct ipv6hdr, saddr),
-				   &redirect_tuple.dip, sizeof(redirect_tuple.dip));
+		bpf_skb_load_bytes(skb,
+				   ETH_HLEN + offsetof(struct ipv6hdr, daddr),
+				   &redirect_tuple.sip,
+				   sizeof(redirect_tuple.sip));
+		bpf_skb_load_bytes(skb,
+				   ETH_HLEN + offsetof(struct ipv6hdr, saddr),
+				   &redirect_tuple.dip,
+				   sizeof(redirect_tuple.dip));
 	}
 	struct redirect_entry *redirect_entry =
 		bpf_map_lookup_elem(&redirect_track, &redirect_tuple);
@@ -1707,7 +1713,7 @@ static __always_inline int _update_map_elem_by_cookie(const __u64 cookie)
 			// __builtin_memset(&buf, 0, MAX_ARG_SCANNER_BUFFER_SIZE);
 			unsigned long to_read = arg_end - (arg_start + j);
 
-			if (to_read >= MAX_ARG_SCANNER_BUFFER_SIZE)
+			if (to_read > MAX_ARG_SCANNER_BUFFER_SIZE)
 				to_read = MAX_ARG_SCANNER_BUFFER_SIZE;
 			else
 				buf[to_read] = 0;
@@ -1872,11 +1878,7 @@ int local_tcp_sockops(struct bpf_sock_ops *skops)
 	{
 		struct tuples_key rev_tuple = {};
 
-		rev_tuple.l4proto = IPPROTO_TCP;
-		rev_tuple.sport = tuple.dport;
-		rev_tuple.dport = tuple.sport;
-		__builtin_memcpy(&rev_tuple.sip, &tuple.dip, IPV6_BYTE_LENGTH);
-		__builtin_memcpy(&rev_tuple.dip, &tuple.sip, IPV6_BYTE_LENGTH);
+		copy_reversed_tuples(&tuple, &rev_tuple);
 
 		struct routing_result *routing_result;
 

control/tcp.go

@@ -65,6 +65,8 @@ func (c *ControlPlane) handleConn(lConn net.Conn) (err error) {
 		switch {
 		case strings.HasSuffix(err.Error(), "write: broken pipe"),
 			strings.HasSuffix(err.Error(), "i/o timeout"),
+			strings.HasPrefix(err.Error(), "EOF"),
+			strings.HasSuffix(err.Error(), "connection reset by peer"),
 			strings.HasSuffix(err.Error(), "canceled by local with error code 0"),
 			strings.HasSuffix(err.Error(), "canceled by remote with error code 0"):
 			return nil // ignore
@@ -162,10 +164,7 @@ func (c *ControlPlane) RouteDialTcp(p *RouteDialParam) (conn netproxy.Conn, err
 	}
 	ctx, cancel := context.WithTimeout(context.TODO(), consts.DefaultDialTimeout)
 	defer cancel()
-	cd := netproxy.ContextDialerConverter{
-		Dialer: d,
-	}
-	return cd.DialContext(ctx, common.MagicNetwork("tcp", routingResult.Mark, c.mptcp), dialTarget)
+	return d.DialContext(ctx, common.MagicNetwork("tcp", routingResult.Mark, c.mptcp), dialTarget)
 }
 
 type WriteCloser interface {