From 281295d02f7216ad94fd2ccca984e3113c02abfb Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Wed, 23 Jul 2025 22:44:02 +0800 Subject: [PATCH] fix tun device --- handler/socks/v4/handler.go | 2 +- handler/tun/handler.go | 8 ++- handler/vtun/endpoint.go | 40 +++++-------- handler/vtun/handler.go | 25 ++++---- handler/vtun/stack.go | 113 +++++++++++++++++++++++++++-------- listener/tun/tun_darwin.go | 4 +- listener/tun/tun_unix.go | 4 +- listener/tun/tun_windows.go | 8 +-- listener/vtun/conn.go | 19 ++++++ listener/vtun/listener.go | 4 ++ listener/vtun/tun_darwin.go | 4 +- listener/vtun/tun_unix.go | 4 +- listener/vtun/tun_windows.go | 8 +-- 13 files changed, 161 insertions(+), 82 deletions(-) diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index 744ecdbd..7d7132f5 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -123,9 +123,9 @@ func (h *socks4Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), - "sid": ctxvalue.SidFromContext(ctx), "client": ro.ClientIP, "network": ro.Network, + "sid": ro.SID, }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) diff --git a/handler/tun/handler.go b/handler/tun/handler.go index dd77e934..1c6cbe5c 100644 --- a/handler/tun/handler.go +++ b/handler/tun/handler.go @@ -62,13 +62,15 @@ func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. log := h.options.Logger - v, _ := conn.(md.Metadatable) - if v == nil { + var config *tun_util.Config + if v, _ := conn.(md.Metadatable); v != nil { + config = v.Metadata().Get("config").(*tun_util.Config) + } + if config == nil { err := errors.New("tun: wrong connection type") log.Error(err) return err } - config := v.Metadata().Get("config").(*tun_util.Config) start := time.Now() log = log.WithFields(map[string]any{ diff --git a/handler/vtun/endpoint.go b/handler/vtun/endpoint.go index f39bfb2d..c8170dba 100644 --- a/handler/vtun/endpoint.go +++ b/handler/vtun/endpoint.go @@ -4,7 +4,6 @@ package tun import ( "context" - "errors" "io" "sync" @@ -23,7 +22,7 @@ const ( ) // Endpoint implements the interface of stack.LinkEndpoint from io.ReadWriter. -type Endpoint struct { +type endpoint struct { *channel.Endpoint // rw is the io.ReadWriter for reading and writing packets. @@ -45,31 +44,18 @@ type Endpoint struct { } // New returns stack.LinkEndpoint(.*Endpoint) and error. -func newEndpoint(rw io.ReadWriter, mtu uint32, offset int, log logger.Logger) (*Endpoint, error) { - if mtu == 0 { - return nil, errors.New("MTU size is zero") - } - - if rw == nil { - return nil, errors.New("RW interface is nil") - } - - if offset < 0 { - return nil, errors.New("offset must be non-negative") - } - - return &Endpoint{ - Endpoint: channel.New(defaultOutQueueLen, mtu, ""), +func newEndpoint(rw io.ReadWriter, mtu int, log logger.Logger) *endpoint { + return &endpoint{ + Endpoint: channel.New(defaultOutQueueLen, uint32(mtu), ""), rw: rw, - mtu: mtu, - offset: offset, + mtu: uint32(mtu), log: log, - }, nil + } } // Attach launches the goroutine that reads packets from io.Reader and // dispatches them via the provided dispatcher. -func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) { +func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { e.Endpoint.Attach(dispatcher) e.once.Do(func() { ctx, cancel := context.WithCancel(context.Background()) @@ -85,12 +71,12 @@ func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) { }) } -func (e *Endpoint) Wait() { +func (e *endpoint) Wait() { e.wg.Wait() } // dispatchLoop dispatches packets to upper layer. -func (e *Endpoint) dispatchLoop(cancel context.CancelFunc) { +func (e *endpoint) dispatchLoop(cancel context.CancelFunc) { // Call cancel() to ensure (*Endpoint).outboundLoop(context.Context) exits // gracefully after (*Endpoint).dispatchLoop(context.CancelFunc) returns. defer cancel() @@ -106,7 +92,9 @@ func (e *Endpoint) dispatchLoop(cancel context.CancelFunc) { break } - e.log.Debugf("read tun: (%d) % x", n, data[:n]) + if e.log.IsLevelEnabled(logger.TraceLevel) { + e.log.Tracef("read: (%d) % x", n, data[:n]) + } if n == 0 || n > mtu { continue @@ -132,7 +120,7 @@ func (e *Endpoint) dispatchLoop(cancel context.CancelFunc) { // outboundLoop reads outbound packets from channel, and then it calls // writePacket to send those packets back to lower layer. -func (e *Endpoint) outboundLoop(ctx context.Context) { +func (e *endpoint) outboundLoop(ctx context.Context) { for { pkt := e.ReadContext(ctx) if pkt == nil { @@ -143,7 +131,7 @@ func (e *Endpoint) outboundLoop(ctx context.Context) { } // writePacket writes outbound packets to the io.Writer. -func (e *Endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error { +func (e *endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error { defer pkt.DecRef() buf := pkt.ToBuffer() diff --git a/handler/vtun/handler.go b/handler/vtun/handler.go index e3fb51d9..fc37114b 100644 --- a/handler/vtun/handler.go +++ b/handler/vtun/handler.go @@ -2,12 +2,14 @@ package tun import ( "context" + "errors" "net" "time" "github.com/go-gost/core/handler" md "github.com/go-gost/core/metadata" ctxvalue "github.com/go-gost/x/ctx" + tun_util "github.com/go-gost/x/internal/util/tun" "github.com/go-gost/x/registry" "github.com/xjasonlyu/tun2socks/v2/core" ) @@ -44,11 +46,21 @@ func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. log := h.options.Logger + var config *tun_util.Config + if v, _ := conn.(md.Metadatable); v != nil { + config = v.Metadata().Get("config").(*tun_util.Config) + } + if config == nil { + err := errors.New("tun: wrong connection type") + log.Error(err) + return err + } + start := time.Now() log = log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), - "sid": ctxvalue.SidFromContext(ctx), + "sid": string(ctxvalue.SidFromContext(ctx)), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) @@ -58,17 +70,12 @@ func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() - ep, err := newEndpoint(conn, 1420, 0, log) - if err != nil { - return err - } - - th := newTransportHandler(log) + th := newTransportHandler(&h.options) th.ProcessAsync() defer th.Close() stack, err := core.CreateStack(&core.Config{ - LinkEndpoint: ep, + LinkEndpoint: newEndpoint(conn, config.MTU, log), TransportHandler: th, }) if err != nil { @@ -76,8 +83,6 @@ func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. } defer stack.Close() - log.Debugf("is attached: %v", ep.IsAttached()) - <-ctx.Done() return nil diff --git a/handler/vtun/stack.go b/handler/vtun/stack.go index 66a7fe39..874c9187 100644 --- a/handler/vtun/stack.go +++ b/handler/vtun/stack.go @@ -2,11 +2,19 @@ package tun import ( "context" + "net" "net/netip" "sync" "time" - "github.com/go-gost/core/logger" + "github.com/go-gost/core/handler" + "github.com/go-gost/core/observer/stats" + ctxvalue "github.com/go-gost/x/ctx" + xnet "github.com/go-gost/x/internal/net" + xstats "github.com/go-gost/x/observer/stats" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" + "github.com/rs/xid" "github.com/xjasonlyu/tun2socks/v2/core/adapter" "go.uber.org/atomic" ) @@ -33,34 +41,35 @@ type transportHandler struct { procOnce sync.Once procCancel context.CancelFunc - log logger.Logger + opts *handler.Options } -func newTransportHandler(log logger.Logger) *transportHandler { +func newTransportHandler(opts *handler.Options) *transportHandler { return &transportHandler{ tcpQueue: make(chan adapter.TCPConn), udpQueue: make(chan adapter.UDPConn), udpTimeout: atomic.NewDuration(udpSessionTimeout), procCancel: func() { /* nop */ }, - log: log, + + opts: opts, } } -func (t *transportHandler) HandleTCP(conn adapter.TCPConn) { - t.tcpQueue <- conn +func (h *transportHandler) HandleTCP(conn adapter.TCPConn) { + h.tcpQueue <- conn } -func (t *transportHandler) HandleUDP(conn adapter.UDPConn) { - t.udpQueue <- conn +func (h *transportHandler) HandleUDP(conn adapter.UDPConn) { + h.udpQueue <- conn } -func (t *transportHandler) process(ctx context.Context) { +func (h *transportHandler) process(ctx context.Context) { for { select { - case conn := <-t.tcpQueue: - go t.handleTCPConn(conn) - case conn := <-t.udpQueue: - go t.handleUDPConn(conn) + case conn := <-h.tcpQueue: + go h.handleTCPConn(conn) + case conn := <-h.udpQueue: + go h.handleUDPConn(conn) case <-ctx.Done(): return } @@ -68,35 +77,87 @@ func (t *transportHandler) process(ctx context.Context) { } // ProcessAsync can be safely called multiple times, but will only be effective once. -func (t *transportHandler) ProcessAsync() { - t.procOnce.Do(func() { +func (h *transportHandler) ProcessAsync() { + h.procOnce.Do(func() { ctx, cancel := context.WithCancel(context.Background()) - t.procCancel = cancel - go t.process(ctx) + h.procCancel = cancel + go h.process(ctx) }) } // Close closes the Tunnel and releases its resources. -func (t *transportHandler) Close() { - t.procCancel() +func (h *transportHandler) Close() { + h.procCancel() } -func (t *transportHandler) SetUDPTimeout(timeout time.Duration) { - t.udpTimeout.Store(timeout) +func (h *transportHandler) SetUDPTimeout(timeout time.Duration) { + h.udpTimeout.Store(timeout) } -func (t *transportHandler) handleTCPConn(originConn adapter.TCPConn) { +func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) { defer originConn.Close() id := originConn.ID() - srcIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice()) + remoteIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice()) dstIP, _ := netip.AddrFromSlice(id.LocalAddress.AsSlice()) - raddr := netip.AddrPortFrom(srcIP, id.RemotePort) - laddr := netip.AddrPortFrom(dstIP, id.LocalPort) + remoteAddr := netip.AddrPortFrom(remoteIP, id.RemotePort) + dstAddr := netip.AddrPortFrom(dstIP, id.LocalPort) - t.log.Debugf("[TCP] %s <-> %s", raddr.String(), laddr.String()) + start := time.Now() + + sid := xid.New().String() + ctx := ctxvalue.ContextWithSid(context.Background(), ctxvalue.Sid(sid)) + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.opts.Service, + Network: "tcp", + RemoteAddr: remoteAddr.String(), + Dst: dstAddr.String(), + Time: start, + SID: sid, + } + log := h.opts.Logger.WithFields(map[string]any{"network": "tcp", "sid": ro.SID}) + + log.Debugf("%s <> %s", remoteAddr.String(), dstAddr.String()) + + var err error + var conn net.Conn = originConn + + pStats := xstats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) + ro.Duration = time.Since(start) + + log.WithFields(map[string]any{ + "duration": time.Since(start), + "inputBytes": ro.InputBytes, + "outputBytes": ro.OutputBytes, + }).Infof("%s >< %s", remoteAddr.String(), dstAddr.String()) + }() + + cc, err := h.opts.Router.Dial(ctx, "tcp", dstAddr.String()) + if err != nil { + log.Errorf("dial %s: %v", dstAddr.String(), err) + return + } + defer cc.Close() + + ro.Src = cc.LocalAddr().String() + + t := time.Now() + log.Infof("%s <-> %s", remoteAddr, dstAddr) + xnet.Transport(conn, cc) + log.WithFields(map[string]any{ + "duration": time.Since(t), + }).Infof("%s >-< %s", remoteAddr, dstAddr) } // TODO: Port Restricted NAT support. diff --git a/listener/tun/tun_darwin.go b/listener/tun/tun_darwin.go index 96db894b..9cc16bc3 100644 --- a/listener/tun/tun_darwin.go +++ b/listener/tun/tun_darwin.go @@ -30,7 +30,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. if len(l.md.config.Net) > 0 { cmd := fmt.Sprintf("ifconfig %s inet %s %s mtu %d up", name, l.md.config.Net[0].String(), l.md.config.Peer, l.md.config.MTU) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if err = exec.Command(args[0], args[1:]...).Run(); err != nil { return @@ -48,7 +48,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. func (l *tunListener) addRoutes(ifName string) error { for _, route := range l.routes { cmd := fmt.Sprintf("route add -net %s -interface %s", route.Net.String(), ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if err := exec.Command(args[0], args[1:]...).Run(); err != nil { return err diff --git a/listener/tun/tun_unix.go b/listener/tun/tun_unix.go index 087bb725..fd9b801d 100644 --- a/listener/tun/tun_unix.go +++ b/listener/tun/tun_unix.go @@ -28,7 +28,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. if len(l.md.config.Net) > 0 { cmd := fmt.Sprintf("ifconfig %s inet %s mtu %d up", name, l.md.config.Net[0].String(), l.md.config.MTU) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -48,7 +48,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. func (l *tunListener) addRoutes(ifName string) error { for _, route := range l.routes { cmd := fmt.Sprintf("route add -net %s -interface %s", route.Net.String(), ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { return fmt.Errorf("%s: %v", cmd, er) diff --git a/listener/tun/tun_windows.go b/listener/tun/tun_windows.go index 0562f83f..19cc1b20 100644 --- a/listener/tun/tun_windows.go +++ b/listener/tun/tun_windows.go @@ -28,7 +28,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. cmd := fmt.Sprintf("netsh interface ip set address name=%s "+ "source=static addr=%s mask=%s gateway=none", name, ipNet.IP.String(), ipMask(ipNet.Mask)) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -44,7 +44,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. for _, dns := range l.md.config.DNS { cmd := fmt.Sprintf("netsh interface ip add dnsservers name=%s address=%s validate=no", name, dns.String()) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -65,7 +65,7 @@ func (l *tunListener) addRoutes(ifName string, gw net.IP) error { if gw != nil { cmd += " nexthop=" + gw.String() } - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { return fmt.Errorf("%s: %v", cmd, er) @@ -77,7 +77,7 @@ func (l *tunListener) addRoutes(ifName string, gw net.IP) error { func (l *tunListener) deleteRoute(ifName string, route string) error { cmd := fmt.Sprintf("netsh interface ip delete route prefix=%s interface=%s store=active", route, ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") return exec.Command(args[0], args[1:]...).Run() } diff --git a/listener/vtun/conn.go b/listener/vtun/conn.go index 43a322c1..20c01857 100644 --- a/listener/vtun/conn.go +++ b/listener/vtun/conn.go @@ -6,6 +6,8 @@ import ( "io" "net" "time" + + mdata "github.com/go-gost/core/metadata" ) type conn struct { @@ -49,3 +51,20 @@ func (c *conn) Close() (err error) { } return c.ifce.Close() } + +type metadataConn struct { + net.Conn + md mdata.Metadata +} + +// Metadata implements metadata.Metadatable interface. +func (c *metadataConn) Metadata() mdata.Metadata { + return c.md +} + +func withMetadata(md mdata.Metadata, c net.Conn) net.Conn { + return &metadataConn{ + Conn: c, + md: md, + } +} diff --git a/listener/vtun/listener.go b/listener/vtun/listener.go index 5f0e8478..b6de36d9 100644 --- a/listener/vtun/listener.go +++ b/listener/vtun/listener.go @@ -12,6 +12,7 @@ import ( "github.com/go-gost/core/router" traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" + mdx "github.com/go-gost/x/metadata" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" "github.com/go-gost/x/registry" @@ -95,6 +96,9 @@ func (l *tunListener) listenLoop() { limiter.ServiceOption(l.options.Service), limiter.NetworkOption(c.LocalAddr().Network()), ) + c = withMetadata(mdx.NewMetadata(map[string]any{ + "config": l.md.config, + }), c) l.cqueue <- c return nil diff --git a/listener/vtun/tun_darwin.go b/listener/vtun/tun_darwin.go index 96db894b..9cc16bc3 100644 --- a/listener/vtun/tun_darwin.go +++ b/listener/vtun/tun_darwin.go @@ -30,7 +30,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. if len(l.md.config.Net) > 0 { cmd := fmt.Sprintf("ifconfig %s inet %s %s mtu %d up", name, l.md.config.Net[0].String(), l.md.config.Peer, l.md.config.MTU) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if err = exec.Command(args[0], args[1:]...).Run(); err != nil { return @@ -48,7 +48,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. func (l *tunListener) addRoutes(ifName string) error { for _, route := range l.routes { cmd := fmt.Sprintf("route add -net %s -interface %s", route.Net.String(), ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if err := exec.Command(args[0], args[1:]...).Run(); err != nil { return err diff --git a/listener/vtun/tun_unix.go b/listener/vtun/tun_unix.go index 087bb725..fd9b801d 100644 --- a/listener/vtun/tun_unix.go +++ b/listener/vtun/tun_unix.go @@ -28,7 +28,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. if len(l.md.config.Net) > 0 { cmd := fmt.Sprintf("ifconfig %s inet %s mtu %d up", name, l.md.config.Net[0].String(), l.md.config.MTU) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -48,7 +48,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. func (l *tunListener) addRoutes(ifName string) error { for _, route := range l.routes { cmd := fmt.Sprintf("route add -net %s -interface %s", route.Net.String(), ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { return fmt.Errorf("%s: %v", cmd, er) diff --git a/listener/vtun/tun_windows.go b/listener/vtun/tun_windows.go index 0562f83f..19cc1b20 100644 --- a/listener/vtun/tun_windows.go +++ b/listener/vtun/tun_windows.go @@ -28,7 +28,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. cmd := fmt.Sprintf("netsh interface ip set address name=%s "+ "source=static addr=%s mask=%s gateway=none", name, ipNet.IP.String(), ipMask(ipNet.Mask)) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -44,7 +44,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net. for _, dns := range l.md.config.DNS { cmd := fmt.Sprintf("netsh interface ip add dnsservers name=%s address=%s validate=no", name, dns.String()) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { @@ -65,7 +65,7 @@ func (l *tunListener) addRoutes(ifName string, gw net.IP) error { if gw != nil { cmd += " nexthop=" + gw.String() } - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") if er := exec.Command(args[0], args[1:]...).Run(); er != nil { return fmt.Errorf("%s: %v", cmd, er) @@ -77,7 +77,7 @@ func (l *tunListener) addRoutes(ifName string, gw net.IP) error { func (l *tunListener) deleteRoute(ifName string, route string) error { cmd := fmt.Sprintf("netsh interface ip delete route prefix=%s interface=%s store=active", route, ifName) - l.logger.Debug(cmd) + l.log.Debug(cmd) args := strings.Split(cmd, " ") return exec.Command(args[0], args[1:]...).Run() }