From e45d9a8cc81a3245c2f2540ebef502b5a777f2e5 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Fri, 5 Jun 2026 23:26:03 +0800 Subject: [PATCH] feat: add stateless UDP forwarding mode (#853) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add stateless=true metadata option to UDP listener/handler for raw datagram forwarding without per-client session tracking, similar to NGINX Stream Module behavior. - udp.NewListener: add Stateless field to ListenConfig; branch in NewListener/Accept to skip connPool/listenLoop when stateless - datagramConn: lightweight net.Conn+PacketConn wrapping a single UDP datagram — no channels, mutexes, or pool logic - forward handler: add handleRawDatagram for single request-response cycle; extract dialTarget to share hop selection/dial/proxyproto preamble with handleRawForwarding Usage: gost -L "udp://:10000/127.0.0.1:2000?stateless=true" --- handler/forward/local/forward.go | 102 ++++++++++++++++++++++++++--- handler/forward/local/handler.go | 4 ++ handler/forward/local/metadata.go | 9 +++ internal/net/udp/listener.go | 105 +++++++++++++++++++++++++++++- listener/udp/listener.go | 1 + listener/udp/metadata.go | 3 + 6 files changed, 213 insertions(+), 11 deletions(-) diff --git a/handler/forward/local/forward.go b/handler/forward/local/forward.go index a20ab4e3..35fb1d6a 100644 --- a/handler/forward/local/forward.go +++ b/handler/forward/local/forward.go @@ -16,10 +16,20 @@ import ( xrecorder "github.com/go-gost/x/recorder" ) -// handleRawForwarding performs node selection, dials the target through the -// router, and pipes the raw connection. It is used when sniffing is disabled -// or the protocol was not HTTP/TLS. -func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, network, proto string) error { +// dialResult is returned by dialTarget, carrying the dialed connection +// and associated state that callers need for subsequent I/O and logging. +type dialResult struct { + cc net.Conn + log logger.Logger + target *chain.Node +} + +// dialTarget performs node selection, Router.Dial, proxy protocol wrapping, +// and recorder population — the shared preamble for both stream forwarding +// (handleRawForwarding) and datagram forwarding (handleRawDatagram). +// +// The caller must close cc when done. +func (h *forwardHandler) dialTarget(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, network, proto string) (*dialResult, error) { target := &chain.Node{} if curHop := h.getHop(); curHop != nil { target = curHop.Select(ctx, @@ -28,7 +38,7 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, } if target == nil { log.Error(errNodeNotAvailable) - return errNodeNotAvailable + return nil, errNodeNotAvailable } addr := target.Addr if opts := target.Options(); opts != nil { @@ -63,12 +73,11 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, if marker := target.Marker(); marker != nil { marker.Mark() } - return err + return nil, err } if marker := target.Marker(); marker != nil { marker.Reset() } - defer cc.Close() cc = proxyproto.WrapClientConn( h.md.proxyProtocol, @@ -80,14 +89,87 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, ro.SrcAddr = cc.LocalAddr().String() ro.DstAddr = cc.RemoteAddr().String() + return &dialResult{ + cc: cc, + log: log, + target: target, + }, nil +} + +// handleRawForwarding performs node selection, dials the target through the +// router, and pipes the raw connection. It is used when sniffing is disabled +// or the protocol was not HTTP/TLS. +func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, network, proto string) error { + dr, err := h.dialTarget(ctx, conn, ro, log, network, proto) + if err != nil { + return err + } + defer dr.cc.Close() + t := time.Now() - log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr) - if err := xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)); err != nil { + log.Infof("%s <-> %s", conn.RemoteAddr(), dr.target.Addr) + if err := xnet.Pipe(ctx, conn, dr.cc, xnet.WithReadTimeout(h.md.idleTimeout)); err != nil { log.Debugf("pipe: %v", err) } log.WithFields(map[string]any{ "duration": time.Since(t), - }).Infof("%s >-< %s", conn.RemoteAddr(), target.Addr) + }).Infof("%s >-< %s", conn.RemoteAddr(), dr.target.Addr) + + return nil +} + +// handleRawDatagram forwards a single UDP datagram to the selected node and +// writes the response back. Unlike handleRawForwarding (which uses xnet.Pipe +// for bidirectional stream copy), this performs a single request-response +// cycle — appropriate for stateless UDP forwarding where each packet is +// independent and no session is maintained. +// +// The method reuses dialTarget for the shared hop-selection, Router.Dial, +// proxyproto, and recorder preamble. +func (h *forwardHandler) handleRawDatagram(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, network, proto string) error { + dr, err := h.dialTarget(ctx, conn, ro, log, network, proto) + if err != nil { + return err + } + defer dr.cc.Close() + + t := time.Now() + log.Infof("%s <-> %s", conn.RemoteAddr(), dr.target.Addr) + + bufp := make([]byte, h.md.bufferSize) + n, err := conn.Read(bufp) + if err != nil { + log.Debugf("conn read: %v", err) + return err + } + + if _, err := dr.cc.Write(bufp[:n]); err != nil { + log.Debugf("outbound write: %v", err) + return err + } + + // Read the response into the same buffer to avoid a second allocation. + if err := dr.cc.SetReadDeadline(time.Now().Add(h.md.readTimeout)); err != nil { + log.Debugf("set read deadline: %v", err) + return err + } + n, err = dr.cc.Read(bufp) + if err != nil { + log.Debugf("outbound read: %v", err) + return err + } + // Clear the deadline so it doesn't affect subsequent use of the + // underlying connection (e.g., keep-alive pools). + dr.cc.SetReadDeadline(time.Time{}) + + if _, err := conn.Write(bufp[:n]); err != nil { + log.Debugf("conn write: %v", err) + return err + } + + log.WithFields(map[string]any{ + "duration": time.Since(t), + }).Infof("%s >-< %s", conn.RemoteAddr(), dr.target.Addr) return nil } diff --git a/handler/forward/local/handler.go b/handler/forward/local/handler.go index 6ba45a6f..c8e158c5 100644 --- a/handler/forward/local/handler.go +++ b/handler/forward/local/handler.go @@ -206,6 +206,10 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand return errRouterNotAvailable } + if h.md.stateless { + return h.handleRawDatagram(ctx, conn, ro, log, network, "udp") + } + var proto string if network == "tcp" && h.md.sniffing { if h.md.sniffingTimeout > 0 { diff --git a/handler/forward/local/metadata.go b/handler/forward/local/metadata.go index 7359bda7..9c41dc8a 100644 --- a/handler/forward/local/metadata.go +++ b/handler/forward/local/metadata.go @@ -36,6 +36,9 @@ type metadata struct { privateKey crypto.PrivateKey alpn string mitmBypass bypass.Bypass + + stateless bool + bufferSize int } func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) { @@ -70,5 +73,11 @@ func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.alpn = mdutil.GetString(md, "mitm.alpn") h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) + h.md.stateless = mdutil.GetBool(md, "stateless") + h.md.bufferSize = mdutil.GetInt(md, "bufferSize", "readBufferSize", "udp.bufferSize") + if h.md.bufferSize <= 0 { + h.md.bufferSize = 4096 + } + return } diff --git a/internal/net/udp/listener.go b/internal/net/udp/listener.go index e2a2acea..d8b7b4a7 100644 --- a/internal/net/udp/listener.go +++ b/internal/net/udp/listener.go @@ -19,6 +19,7 @@ type ListenConfig struct { ReadBufferSize int TTL time.Duration Keepalive bool + Stateless bool Logger logger.Logger } type listener struct { @@ -33,6 +34,11 @@ type listener struct { // NewListener creates a net.Listener from a net.PacketConn by demultiplexing // UDP datagrams into per-client net.Conn streams. Idle connections are cleaned // up according to cfg.TTL. +// +// When cfg.Stateless is true, the listener operates in raw datagram mode: +// no connPool, no listenLoop goroutine, no per-client session tracking. +// Each Accept call blocks on a ReadFrom and returns a lightweight +// datagramConn wrapping a single packet. func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener { if cfg == nil { cfg = &ListenConfig{} @@ -40,11 +46,16 @@ func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener { ln := &listener{ conn: conn, - cqueue: make(chan net.Conn, cfg.Backlog), closed: make(chan struct{}), errChan: make(chan error, 1), config: cfg, } + + if cfg.Stateless { + return ln + } + + ln.cqueue = make(chan net.Conn, cfg.Backlog) ln.connPool = newConnPool(cfg.TTL).WithLogger(cfg.Logger) go ln.listenLoop() @@ -52,6 +63,10 @@ func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener { } func (ln *listener) Accept() (conn net.Conn, err error) { + if ln.config.Stateless { + return ln.acceptStateless() + } + select { case conn = <-ln.cqueue: return @@ -65,6 +80,30 @@ func (ln *listener) Accept() (conn net.Conn, err error) { } } +func (ln *listener) acceptStateless() (net.Conn, error) { + b := bufpool.Get(ln.config.ReadBufferSize) + + n, raddr, err := ln.conn.ReadFrom(b) + if err != nil { + bufpool.Put(b) + // Surface the error so the service loop can decide whether to + // continue accepting. + select { + case <-ln.closed: + return nil, net.ErrClosed + default: + } + return nil, err + } + + return &datagramConn{ + pc: ln.conn, + data: b[:n], + localAddr: ln.Addr(), + remoteAddr: raddr, + }, nil +} + func (ln *listener) listenLoop() { for { select { @@ -243,3 +282,67 @@ func (c *conn) WriteQueue(b []byte) error { return errors.New("recv queue is full") } } + +// datagramConn is a lightweight net.Conn that wraps a single UDP datagram. +// It is used in stateless mode where each Accept returns a new datagramConn +// for a single packet — no channels, no mutexes, no pool tracking. +type datagramConn struct { + pc net.PacketConn + data []byte + offset int + localAddr net.Addr + remoteAddr net.Addr +} + +func (c *datagramConn) Read(b []byte) (n int, err error) { + if c.data == nil || c.offset >= len(c.data) { + return 0, net.ErrClosed + } + n = copy(b, c.data[c.offset:]) + c.offset += n + return +} + +func (c *datagramConn) Write(b []byte) (n int, err error) { + return c.pc.WriteTo(b, c.remoteAddr) +} + +// ReadFrom implements net.PacketConn. It reads the buffered datagram and +// returns the sender's address. +func (c *datagramConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) { + n, err = c.Read(b) + addr = c.remoteAddr + return +} + +// WriteTo implements net.PacketConn. It sends b to addr via the underlying +// PacketConn, ignoring the stored remoteAddr. +func (c *datagramConn) WriteTo(b []byte, addr net.Addr) (n int, err error) { + return c.pc.WriteTo(b, addr) +} + +func (c *datagramConn) Close() error { + bufpool.Put(c.data) + c.data = nil + return nil +} + +func (c *datagramConn) LocalAddr() net.Addr { + return c.localAddr +} + +func (c *datagramConn) RemoteAddr() net.Addr { + return c.remoteAddr +} + +func (c *datagramConn) SetDeadline(t time.Time) error { + return c.pc.SetReadDeadline(t) +} + +func (c *datagramConn) SetReadDeadline(t time.Time) error { + return c.pc.SetReadDeadline(t) +} + +func (c *datagramConn) SetWriteDeadline(t time.Time) error { + return c.pc.SetWriteDeadline(t) +} diff --git a/listener/udp/listener.go b/listener/udp/listener.go index 6f21f7c5..76a00c79 100644 --- a/listener/udp/listener.go +++ b/listener/udp/listener.go @@ -76,6 +76,7 @@ func (l *udpListener) Init(md md.Metadata) (err error) { ReadBufferSize: l.md.readBufferSize, Keepalive: l.md.keepalive, TTL: l.md.ttl, + Stateless: l.md.stateless, Logger: l.logger, }) return diff --git a/listener/udp/metadata.go b/listener/udp/metadata.go index 1e5ec75e..71fdc3e4 100644 --- a/listener/udp/metadata.go +++ b/listener/udp/metadata.go @@ -20,6 +20,7 @@ type metadata struct { backlog int keepalive bool ttl time.Duration + stateless bool } func (l *udpListener) parseMetadata(md mdata.Metadata) (err error) { @@ -43,5 +44,7 @@ func (l *udpListener) parseMetadata(md mdata.Metadata) (err error) { l.md.readQueueSize = defaultReadQueueSize } + l.md.stateless = mdutil.GetBool(md, "stateless") + return }