feat: add stateless UDP forwarding mode (#853)
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"
This commit is contained in:
@@ -16,10 +16,20 @@ import (
|
|||||||
xrecorder "github.com/go-gost/x/recorder"
|
xrecorder "github.com/go-gost/x/recorder"
|
||||||
)
|
)
|
||||||
|
|
||||||
// handleRawForwarding performs node selection, dials the target through the
|
// dialResult is returned by dialTarget, carrying the dialed connection
|
||||||
// router, and pipes the raw connection. It is used when sniffing is disabled
|
// and associated state that callers need for subsequent I/O and logging.
|
||||||
// or the protocol was not HTTP/TLS.
|
type dialResult struct {
|
||||||
func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, network, proto string) error {
|
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{}
|
target := &chain.Node{}
|
||||||
if curHop := h.getHop(); curHop != nil {
|
if curHop := h.getHop(); curHop != nil {
|
||||||
target = curHop.Select(ctx,
|
target = curHop.Select(ctx,
|
||||||
@@ -28,7 +38,7 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn,
|
|||||||
}
|
}
|
||||||
if target == nil {
|
if target == nil {
|
||||||
log.Error(errNodeNotAvailable)
|
log.Error(errNodeNotAvailable)
|
||||||
return errNodeNotAvailable
|
return nil, errNodeNotAvailable
|
||||||
}
|
}
|
||||||
addr := target.Addr
|
addr := target.Addr
|
||||||
if opts := target.Options(); opts != nil {
|
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 {
|
if marker := target.Marker(); marker != nil {
|
||||||
marker.Mark()
|
marker.Mark()
|
||||||
}
|
}
|
||||||
return err
|
return nil, err
|
||||||
}
|
}
|
||||||
if marker := target.Marker(); marker != nil {
|
if marker := target.Marker(); marker != nil {
|
||||||
marker.Reset()
|
marker.Reset()
|
||||||
}
|
}
|
||||||
defer cc.Close()
|
|
||||||
|
|
||||||
cc = proxyproto.WrapClientConn(
|
cc = proxyproto.WrapClientConn(
|
||||||
h.md.proxyProtocol,
|
h.md.proxyProtocol,
|
||||||
@@ -80,14 +89,87 @@ func (h *forwardHandler) handleRawForwarding(ctx context.Context, conn net.Conn,
|
|||||||
ro.SrcAddr = cc.LocalAddr().String()
|
ro.SrcAddr = cc.LocalAddr().String()
|
||||||
ro.DstAddr = cc.RemoteAddr().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()
|
t := time.Now()
|
||||||
log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr)
|
log.Infof("%s <-> %s", conn.RemoteAddr(), dr.target.Addr)
|
||||||
if err := xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout)); err != nil {
|
if err := xnet.Pipe(ctx, conn, dr.cc, xnet.WithReadTimeout(h.md.idleTimeout)); err != nil {
|
||||||
log.Debugf("pipe: %v", err)
|
log.Debugf("pipe: %v", err)
|
||||||
}
|
}
|
||||||
log.WithFields(map[string]any{
|
log.WithFields(map[string]any{
|
||||||
"duration": time.Since(t),
|
"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
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -206,6 +206,10 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
|
|||||||
return errRouterNotAvailable
|
return errRouterNotAvailable
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if h.md.stateless {
|
||||||
|
return h.handleRawDatagram(ctx, conn, ro, log, network, "udp")
|
||||||
|
}
|
||||||
|
|
||||||
var proto string
|
var proto string
|
||||||
if network == "tcp" && h.md.sniffing {
|
if network == "tcp" && h.md.sniffing {
|
||||||
if h.md.sniffingTimeout > 0 {
|
if h.md.sniffingTimeout > 0 {
|
||||||
|
|||||||
@@ -36,6 +36,9 @@ type metadata struct {
|
|||||||
privateKey crypto.PrivateKey
|
privateKey crypto.PrivateKey
|
||||||
alpn string
|
alpn string
|
||||||
mitmBypass bypass.Bypass
|
mitmBypass bypass.Bypass
|
||||||
|
|
||||||
|
stateless bool
|
||||||
|
bufferSize int
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) {
|
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.alpn = mdutil.GetString(md, "mitm.alpn")
|
||||||
h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass"))
|
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
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ type ListenConfig struct {
|
|||||||
ReadBufferSize int
|
ReadBufferSize int
|
||||||
TTL time.Duration
|
TTL time.Duration
|
||||||
Keepalive bool
|
Keepalive bool
|
||||||
|
Stateless bool
|
||||||
Logger logger.Logger
|
Logger logger.Logger
|
||||||
}
|
}
|
||||||
type listener struct {
|
type listener struct {
|
||||||
@@ -33,6 +34,11 @@ type listener struct {
|
|||||||
// NewListener creates a net.Listener from a net.PacketConn by demultiplexing
|
// NewListener creates a net.Listener from a net.PacketConn by demultiplexing
|
||||||
// UDP datagrams into per-client net.Conn streams. Idle connections are cleaned
|
// UDP datagrams into per-client net.Conn streams. Idle connections are cleaned
|
||||||
// up according to cfg.TTL.
|
// 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 {
|
func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener {
|
||||||
if cfg == nil {
|
if cfg == nil {
|
||||||
cfg = &ListenConfig{}
|
cfg = &ListenConfig{}
|
||||||
@@ -40,11 +46,16 @@ func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener {
|
|||||||
|
|
||||||
ln := &listener{
|
ln := &listener{
|
||||||
conn: conn,
|
conn: conn,
|
||||||
cqueue: make(chan net.Conn, cfg.Backlog),
|
|
||||||
closed: make(chan struct{}),
|
closed: make(chan struct{}),
|
||||||
errChan: make(chan error, 1),
|
errChan: make(chan error, 1),
|
||||||
config: cfg,
|
config: cfg,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if cfg.Stateless {
|
||||||
|
return ln
|
||||||
|
}
|
||||||
|
|
||||||
|
ln.cqueue = make(chan net.Conn, cfg.Backlog)
|
||||||
ln.connPool = newConnPool(cfg.TTL).WithLogger(cfg.Logger)
|
ln.connPool = newConnPool(cfg.TTL).WithLogger(cfg.Logger)
|
||||||
go ln.listenLoop()
|
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) {
|
func (ln *listener) Accept() (conn net.Conn, err error) {
|
||||||
|
if ln.config.Stateless {
|
||||||
|
return ln.acceptStateless()
|
||||||
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case conn = <-ln.cqueue:
|
case conn = <-ln.cqueue:
|
||||||
return
|
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() {
|
func (ln *listener) listenLoop() {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
@@ -243,3 +282,67 @@ func (c *conn) WriteQueue(b []byte) error {
|
|||||||
return errors.New("recv queue is full")
|
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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -76,6 +76,7 @@ func (l *udpListener) Init(md md.Metadata) (err error) {
|
|||||||
ReadBufferSize: l.md.readBufferSize,
|
ReadBufferSize: l.md.readBufferSize,
|
||||||
Keepalive: l.md.keepalive,
|
Keepalive: l.md.keepalive,
|
||||||
TTL: l.md.ttl,
|
TTL: l.md.ttl,
|
||||||
|
Stateless: l.md.stateless,
|
||||||
Logger: l.logger,
|
Logger: l.logger,
|
||||||
})
|
})
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ type metadata struct {
|
|||||||
backlog int
|
backlog int
|
||||||
keepalive bool
|
keepalive bool
|
||||||
ttl time.Duration
|
ttl time.Duration
|
||||||
|
stateless bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *udpListener) parseMetadata(md mdata.Metadata) (err error) {
|
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.readQueueSize = defaultReadQueueSize
|
||||||
}
|
}
|
||||||
|
|
||||||
|
l.md.stateless = mdutil.GetBool(md, "stateless")
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user