From 8b0806bad4973dbc1a25b24070832bfae15bc5c1 Mon Sep 17 00:00:00 2001 From: Yuan Tong <13075180+tongyuantongyu@users.noreply.github.com> Date: Sat, 13 Jun 2026 13:39:14 +0800 Subject: [PATCH] feat(listener/runix): support remote UDS forwarding (#102) --- chain/route.go | 6 ++ config/cmd/cmd.go | 10 +- connector/relay/bind.go | 19 +++- connector/relay/listener.go | 8 +- connector/sshd/connector.go | 6 +- handler/forward/local/forward.go | 12 +-- handler/forward/remote/forward.go | 12 +-- handler/forward/remote/handler.go | 7 +- handler/forward/remote/util.go | 2 +- handler/forward/remote/util_test.go | 13 +++ handler/relay/bind.go | 13 ++- handler/relay/handler.go | 13 ++- listener/runix/listener.go | 149 ++++++++++++++++++++++++++++ listener/runix/metadata.go | 11 ++ 14 files changed, 247 insertions(+), 34 deletions(-) create mode 100644 listener/runix/listener.go create mode 100644 listener/runix/metadata.go diff --git a/chain/route.go b/chain/route.go index e582a9f4..010ed090 100644 --- a/chain/route.go +++ b/chain/route.go @@ -86,6 +86,12 @@ func (*defaultRoute) Bind(ctx context.Context, network, address string, opts ... Logger: logger, }) return ln, err + case "unix": + addr, err := net.ResolveUnixAddr(network, address) + if err != nil { + return nil, err + } + return net.ListenUnix(network, addr) default: err := fmt.Errorf("network %s unsupported", network) return nil, err diff --git a/config/cmd/cmd.go b/config/cmd/cmd.go index d198514e..87c24b4c 100644 --- a/config/cmd/cmd.go +++ b/config/cmd/cmd.go @@ -209,7 +209,7 @@ func BuildConfigFromCmd(serviceList, nodeList []string) (*config.Config, error) for i, service := range services { service.Name = fmt.Sprintf("%sservice-%d", namePrefix, i) if chain != nil { - if service.Listener.Type == "rtcp" || service.Listener.Type == "rudp" { + if service.Listener.Type == "rtcp" || service.Listener.Type == "rudp" || service.Listener.Type == "runix" { service.Listener.Chain = chain.Name } else { service.Handler.Chain = chain.Name @@ -397,8 +397,8 @@ func buildServiceConfig(url *url.URL) ([]*config.ServiceConfig, error) { listener = schemes[1] } - // For path-based protocols (unix socket, serial), use path as address - isPathBasedProtocol := listener == "unix" || listener == "serial" + // For path-based protocols (Unix socket listeners and serial), use path as address + isPathBasedProtocol := listener == "unix" || listener == "runix" || listener == "serial" var addrs []string if isPathBasedProtocol { @@ -457,7 +457,7 @@ func buildServiceConfig(url *url.URL) ([]*config.ServiceConfig, error) { if listener == "tcp" || listener == "udp" || listener == "rtcp" || listener == "rudp" || listener == "tun" || listener == "tap" || - listener == "dns" || listener == "unix" || + listener == "dns" || listener == "unix" || listener == "runix" || listener == "serial" { handler = listener } else { @@ -714,7 +714,7 @@ func buildNodeConfig(url *url.URL, m map[string]any) (*config.NodeConfig, error) nodeMd[k] = v } - // For path-based protocols (unix socket, serial), use path as address + // For path-based protocols (Unix socket dialers and serial), use path as address var nodeAddr string isPathBasedProtocol := dialer == "unix" || dialer == "serial" if isPathBasedProtocol { diff --git a/connector/relay/bind.go b/connector/relay/bind.go index 1ab8d3a6..33c04c13 100644 --- a/connector/relay/bind.go +++ b/connector/relay/bind.go @@ -28,7 +28,7 @@ func (c *relayConnector) Bind(ctx context.Context, conn net.Conn, network, addre } switch network { - case "tcp", "tcp4", "tcp6": + case "tcp", "tcp4", "tcp6", "unix": return c.bindTCP(ctx, conn, network, address, log) case "udp", "udp4", "udp6": return c.bindUDP(ctx, conn, network, address, &options, log) @@ -52,6 +52,7 @@ func (c *relayConnector) bindTCP(ctx context.Context, conn net.Conn, network, ad } return &bindListener{ + network: network, addr: laddr, session: session, logger: log, @@ -97,13 +98,17 @@ func (c *relayConnector) bind(conn net.Conn, cmd relay.CmdType, network, address nid := relay.NetworkTCP if network == "udp" || network == "udp4" || network == "udp6" { nid = relay.NetworkUDP + } else if network == "unix" { + nid = relay.NetworkUnix } req.Features = append(req.Features, &relay.NetworkFeature{ Network: nid, }) fa := &relay.AddrFeature{} - if h, _, e := net.SplitHostPort(address); e == nil && h == "" { - address = net.JoinHostPort("0.0.0.0", address) + if network != "unix" { + if h, _, e := net.SplitHostPort(address); e == nil && h == "" { + address = net.JoinHostPort("0.0.0.0", address) + } } fa.ParseFrom(address) req.Features = append(req.Features, fa) @@ -125,7 +130,11 @@ func (c *relayConnector) bind(conn net.Conn, cmd relay.CmdType, network, address for _, f := range resp.Features { if f.Type() == relay.FeatureAddr { if fa, ok := f.(*relay.AddrFeature); ok { - addr = net.JoinHostPort(fa.Host, strconv.Itoa(int(fa.Port))) + if network == "unix" { + addr = fa.Host + } else { + addr = net.JoinHostPort(fa.Host, strconv.Itoa(int(fa.Port))) + } } } } @@ -137,6 +146,8 @@ func (c *relayConnector) bind(conn net.Conn, cmd relay.CmdType, network, address baddr, err = net.ResolveTCPAddr(network, addr) case "udp", "udp4", "udp6": baddr, err = net.ResolveUDPAddr(network, addr) + case "unix": + baddr, err = net.ResolveUnixAddr(network, addr) default: err = fmt.Errorf("unknown network %s", network) } diff --git a/connector/relay/listener.go b/connector/relay/listener.go index f29e1fa5..71039b6f 100644 --- a/connector/relay/listener.go +++ b/connector/relay/listener.go @@ -62,7 +62,13 @@ func (p *bindListener) getPeerConn(conn net.Conn) (net.Conn, error) { } } - raddr, err := net.ResolveTCPAddr("tcp", address) + var raddr net.Addr + var err error + if p.network == "unix" { + raddr, err = net.ResolveUnixAddr("unix", address) + } else { + raddr, err = net.ResolveTCPAddr("tcp", address) + } if err != nil { return nil, err } diff --git a/connector/sshd/connector.go b/connector/sshd/connector.go index 928da07a..79613541 100644 --- a/connector/sshd/connector.go +++ b/connector/sshd/connector.go @@ -74,8 +74,10 @@ func (c *sshdConnector) Bind(ctx context.Context, conn net.Conn, network, addres return nil, errors.New("ssh: invalid connection") } - if host, port, _ := net.SplitHostPort(address); host == "" { - address = net.JoinHostPort("0.0.0.0", port) + if network != "unix" { + if host, port, err := net.SplitHostPort(address); err == nil && host == "" { + address = net.JoinHostPort("0.0.0.0", port) + } } return cc.Client().Listen(network, address) diff --git a/handler/forward/local/forward.go b/handler/forward/local/forward.go index 35fb1d6a..5355108a 100644 --- a/handler/forward/local/forward.go +++ b/handler/forward/local/forward.go @@ -42,13 +42,13 @@ func (h *forwardHandler) dialTarget(ctx context.Context, conn net.Conn, ro *xrec } addr := target.Addr if opts := target.Options(); opts != nil { - switch opts.Network { - case "unix": + if opts.Network != "" { network = opts.Network - default: - if _, _, err := net.SplitHostPort(addr); err != nil { - addr += ":0" - } + } + } + if network != "unix" { + if _, _, err := net.SplitHostPort(addr); err != nil { + addr += ":0" } } diff --git a/handler/forward/remote/forward.go b/handler/forward/remote/forward.go index 3c3a0dba..07c5dad3 100644 --- a/handler/forward/remote/forward.go +++ b/handler/forward/remote/forward.go @@ -28,13 +28,13 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, } addr := target.Addr if opts := target.Options(); opts != nil { - switch opts.Network { - case "unix": + if opts.Network != "" { network = opts.Network - default: - if _, _, err := net.SplitHostPort(addr); err != nil { - addr += ":0" - } + } + } + if network != "unix" { + if _, _, err := net.SplitHostPort(addr); err != nil { + addr += ":0" } } diff --git a/handler/forward/remote/handler.go b/handler/forward/remote/handler.go index 7a195f4b..120acd6c 100644 --- a/handler/forward/remote/handler.go +++ b/handler/forward/remote/handler.go @@ -1,8 +1,8 @@ // Package remote implements a reverse forwarding handler for connections -// received from remote GOST nodes. It listens for "rtcp" and "rudp" protocols +// received from remote GOST nodes. It listens for "rtcp", "rudp", and "runix" protocols // and forwards the accepted connections to the configured hop. // -// The handler is registered under the names "rtcp" and "rudp" via +// The handler is registered under the names "rtcp", "rudp", and "runix" via // NewHandler in init(). // // # Connection processing flow @@ -85,6 +85,7 @@ import ( func init() { registry.HandlerRegistry().Register("rtcp", NewHandler) registry.HandlerRegistry().Register("rudp", NewHandler) + registry.HandlerRegistry().Register("runix", NewHandler) } type forwardHandler struct { @@ -98,7 +99,7 @@ type forwardHandler struct { } // NewHandler creates a remote forwarding handler with the given options. -// The handler registers for "rtcp" and "rudp" protocols. +// The handler registers for "rtcp", "rudp", and "runix" protocols. func NewHandler(opts ...handler.Option) handler.Handler { options := handler.Options{} for _, opt := range opts { diff --git a/handler/forward/remote/util.go b/handler/forward/remote/util.go index a72c1c3c..afc6a126 100644 --- a/handler/forward/remote/util.go +++ b/handler/forward/remote/util.go @@ -22,7 +22,7 @@ func (h *forwardHandler) newRecorderObject(ctx context.Context, conn net.Conn, s Service: h.options.Service, RemoteAddr: conn.RemoteAddr().String(), LocalAddr: conn.LocalAddr().String(), - Network: "tcp", + Network: conn.LocalAddr().Network(), Time: start, SID: xctx.SidFromContext(ctx).String(), } diff --git a/handler/forward/remote/util_test.go b/handler/forward/remote/util_test.go index 1bb76149..b62f4743 100644 --- a/handler/forward/remote/util_test.go +++ b/handler/forward/remote/util_test.go @@ -82,6 +82,19 @@ func TestNewRecorderObject_UDP(t *testing.T) { } } +func TestNewRecorderObject_Unix(t *testing.T) { + h := newInitdHandler() + conn := newStringConn(nil) + conn.local = &net.UnixAddr{Name: "/tmp/local.sock", Net: "unix"} + conn.remote = &net.UnixAddr{Name: "/tmp/remote.sock", Net: "unix"} + + ro := h.newRecorderObject(context.Background(), conn, time.Now()) + + if ro.Network != "unix" { + t.Errorf("expected network unix, got %s", ro.Network) + } +} + // --------------------------------------------------------------------------- // checkRateLimit // --------------------------------------------------------------------------- diff --git a/handler/relay/bind.go b/handler/relay/bind.go index 9b6eb80d..9ea435e3 100644 --- a/handler/relay/bind.go +++ b/handler/relay/bind.go @@ -99,10 +99,15 @@ func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, a return err } - if network == "tcp" { + switch network { + case "tcp", "tcp4", "tcp6", "unix": return h.bindTCP(ctx, conn, network, address, ro, log) - } else { + case "udp", "udp4", "udp6": return h.bindUDP(ctx, conn, network, address, ro, log) + default: + resp.Status = relay.StatusBadRequest + resp.WriteTo(conn) + return fmt.Errorf("network %s is unsupported", network) } } @@ -149,7 +154,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr()) log = log.WithFields(map[string]any{ "service": serviceName, - "listener": "tcp", + "listener": network, "handler": "ep-tcp", "bind": fmt.Sprintf("%s/%s", ln.Addr(), ln.Addr().Network()), "src": ln.Addr().String(), @@ -288,4 +293,4 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr "duration": time.Since(t), }).Debugf("%s >-< %s", conn.RemoteAddr(), pc.LocalAddr()) return nil -} \ No newline at end of file +} diff --git a/handler/relay/handler.go b/handler/relay/handler.go index dd6e29f2..f3a507c0 100644 --- a/handler/relay/handler.go +++ b/handler/relay/handler.go @@ -236,6 +236,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle var user, pass string var address string + var addrFeature *relay.AddrFeature var networkID relay.NetworkID for _, f := range req.Features { switch f.Type() { @@ -245,7 +246,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle } case relay.FeatureAddr: if feature, _ := f.(*relay.AddrFeature); feature != nil { - address = net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port))) + addrFeature = feature } case relay.FeatureNetwork: if feature, _ := f.(*relay.NetworkFeature); feature != nil { @@ -275,6 +276,14 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle if (req.Cmd & relay.FUDP) == relay.FUDP { network = "udp" } + if addrFeature != nil { + switch network { + case "unix", "serial": + address = addrFeature.Host + default: + address = net.JoinHostPort(addrFeature.Host, strconv.Itoa(int(addrFeature.Port))) + } + } ro.Network = network ro.Host = address log = log.WithFields(map[string]any{"network": network}) @@ -302,4 +311,4 @@ func (h *relayHandler) Close() error { h.cancel() } return nil -} \ No newline at end of file +} diff --git a/listener/runix/listener.go b/listener/runix/listener.go new file mode 100644 index 00000000..a24f9a56 --- /dev/null +++ b/listener/runix/listener.go @@ -0,0 +1,149 @@ +package runix + +import ( + "context" + "net" + "sync" + + "github.com/go-gost/core/limiter" + "github.com/go-gost/core/listener" + "github.com/go-gost/core/logger" + md "github.com/go-gost/core/metadata" + admission "github.com/go-gost/x/admission/wrapper" + climiter "github.com/go-gost/x/limiter/conn/wrapper" + limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" + metrics "github.com/go-gost/x/metrics/wrapper" + stats "github.com/go-gost/x/observer/stats/wrapper" + "github.com/go-gost/x/registry" +) + +func init() { + registry.ListenerRegistry().Register("runix", NewListener) +} + +type runixListener struct { + laddr net.Addr + ln net.Listener + logger logger.Logger + closed chan struct{} + md metadata + options listener.Options + mu sync.Mutex +} + +func NewListener(opts ...listener.Option) listener.Listener { + options := listener.Options{} + for _, opt := range opts { + opt(&options) + } + return &runixListener{ + closed: make(chan struct{}), + logger: options.Logger, + options: options, + } +} + +func (l *runixListener) Init(md md.Metadata) (err error) { + if err = l.parseMetadata(md); err != nil { + return + } + + if laddr, _ := net.ResolveUnixAddr("unix", l.options.Addr); laddr != nil { + l.laddr = laddr + } + if l.laddr == nil { + l.laddr = &bindAddr{addr: l.options.Addr} + } + + return +} + +func (l *runixListener) Accept() (conn net.Conn, err error) { + select { + case <-l.closed: + return nil, net.ErrClosed + default: + } + + ln := l.getListener() + if ln == nil { + ln, err = l.options.Router.Bind(context.Background(), "unix", l.laddr.String()) + if err != nil { + return nil, listener.NewBindError(err) + } + + ln = metrics.WrapListener(l.options.Service, ln) + ln = stats.WrapListener(ln, l.options.Stats) + ln = admission.WrapListener(l.options.Service, l.options.Admission, ln) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) + ln = climiter.WrapListener(l.options.ConnLimiter, ln) + l.setListener(ln) + } + + select { + case <-l.closed: + ln.Close() + return nil, net.ErrClosed + default: + } + + conn, err = ln.Accept() + if err != nil { + ln.Close() + l.setListener(nil) + return nil, listener.NewAcceptError(err) + } + + conn = limiter_wrapper.WrapConn( + conn, + l.options.TrafficLimiter, + conn.RemoteAddr().String(), + limiter.ScopeOption(limiter.ScopeConn), + limiter.ServiceOption(l.options.Service), + limiter.NetworkOption(conn.LocalAddr().Network()), + limiter.SrcOption(conn.RemoteAddr().String()), + ) + + return +} + +func (l *runixListener) Addr() net.Addr { + return l.laddr +} + +func (l *runixListener) Close() error { + select { + case <-l.closed: + default: + close(l.closed) + if ln := l.getListener(); ln != nil { + ln.Close() + } + } + + return nil +} + +func (l *runixListener) setListener(ln net.Listener) { + l.mu.Lock() + defer l.mu.Unlock() + l.ln = ln +} + +func (l *runixListener) getListener() net.Listener { + l.mu.Lock() + defer l.mu.Unlock() + return l.ln +} + +type bindAddr struct { + addr string +} + +func (p *bindAddr) Network() string { + return "unix" +} + +func (p *bindAddr) String() string { + return p.addr +} diff --git a/listener/runix/metadata.go b/listener/runix/metadata.go new file mode 100644 index 00000000..483691e9 --- /dev/null +++ b/listener/runix/metadata.go @@ -0,0 +1,11 @@ +package runix + +import ( + mdata "github.com/go-gost/core/metadata" +) + +type metadata struct{} + +func (l *runixListener) parseMetadata(md mdata.Metadata) (err error) { + return +}