diff --git a/handler/relay/bind.go b/handler/relay/bind.go index 3c7cd49f..fcd5b864 100644 --- a/handler/relay/bind.go +++ b/handler/relay/bind.go @@ -24,6 +24,35 @@ import ( xservice "github.com/go-gost/x/service" ) +// handleBind processes a relay CmdBind request. +// +// BIND mode is used for reverse-proxy scenarios: the client asks the relay +// handler to listen on a local port, then forwards inbound connections back +// to the client over a mux session. +// +// Flow: +// +// handleBind() +// ├─ 1. Wrap traffic limiter + stats +// ├─ 2. Check BIND is enabled +// ├─ 3. TCP BIND → bindTCP() +// │ ├─ net.Listen on the specified address +// │ ├─ Return the listening address via relay.Response +// │ ├─ Upgrade the connection to a mux session +// │ ├─ Start tcpHandler + tcpListener as an internal service +// │ │ ├─ tcpHandler: gets a stream from mux session → writes AddrFeature +// │ │ └─ tcpListener: accepts local TCP connections +// │ ├─ Goroutine: drain unexpected mux sessions +// │ └─ service.Serve() blocks +// └─ 4. UDP BIND → bindUDP() +// ├─ net.ListenPacket on the UDP port +// ├─ Return the listening address +// └─ Create udp.Relay for datagram relay +// +// Key design: +// - TCP BIND uses mux: each inbound connection gets an independent mux stream +// carrying a relay AddrFeature identifying the peer address. +// - UDP BIND uses udp.Relay directly, bypassing mux. func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { log = log.WithFields(map[string]any{ "dst": address, @@ -32,6 +61,7 @@ func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, a log.Debugf("%s >> %s", conn.RemoteAddr(), address) + // --- Traffic limiter + stats wrapper --- { clientID := ctxvalue.ClientIDFromContext(ctx) rw := traffic_wrapper.WrapReadWriter( @@ -56,6 +86,7 @@ func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, a conn = xnet.NewReadWriteConn(rw, rw, conn) } + // Check whether BIND is enabled in config. resp := relay.Response{ Version: relay.Version1, Status: relay.StatusOK, @@ -75,6 +106,27 @@ func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, a } } +// bindTCP implements TCP BIND mode. +// +// Detailed flow: +// +// bindTCP() +// ├─ 1. net.Listen on the specified TCP address +// ├─ 2. Write the listening address into relay.Response, send to client +// ├─ 3. Upgrade the client connection to a mux session +// │ (mux multiplexes multiple independent streams over one TCP conn) +// ├─ 4. Create internal tcpListener + tcpHandler + Service +// │ ├─ tcpListener: wraps net.Listener with proxyproto/metrics/admission +// │ ├─ tcpHandler: on each inbound connection: +// │ │ ├─ Gets a stream from the mux session +// │ │ ├─ Writes the inbound peer address as AddrFeature on the stream +// │ │ └─ Pipes data bidirectionally (local conn ↔ mux stream) +// │ └─ Service: wraps listener + handler, blocks on Serve() +// ├─ 5. Goroutine: accept and discard unexpected mux connections +// └─ 6. srv.Serve() blocks until shutdown +// +// The client receives forwarded connections as streams on the mux session. +// Each stream carries a relay.AddrFeature identifying the original peer. func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { resp := relay.Response{ Version: relay.Version1, @@ -84,7 +136,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr lc := xnet.ListenConfig{ Netns: h.options.Netns, } - ln, err := lc.Listen(ctx, network, address) // strict mode: if the port already in use, it will return error + ln, err := lc.Listen(ctx, network, address) // 严格模式:端口已被占用时会返回错误 if err != nil { log.Error(err) resp.Status = relay.StatusServiceUnavailable @@ -93,6 +145,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr } defer ln.Close() + // Internal service name: "-ep-" serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr()) log = log.WithFields(map[string]any{ "service": serviceName, @@ -103,6 +156,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr }) ro.SrcAddr = ln.Addr().String() + // Return the listening address to the client. af := &relay.AddrFeature{} if err := af.ParseFrom(ln.Addr().String()); err != nil { log.Warn(err) @@ -113,7 +167,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr return err } - // Upgrade connection to multiplex session. + // Upgrade the client connection to a mux session. session, err := mux.ClientSession(conn, h.md.muxCfg) if err != nil { log.Error(err) @@ -121,6 +175,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr } defer session.Close() + // Internal endpoint listener (proxyproto → metrics → admission layers). epListener := newTCPListener(ln, listener.AddrOption(address), listener.ServiceOption(serviceName), @@ -129,6 +184,10 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr "kind": "listener", })), ) + // Internal endpoint handler — on each inbound connection: + // 1. Gets a stream from the mux session + // 2. Writes the peer address as AddrFeature on the mux stream + // 3. Pipes data bidirectionally epHandler := newTCPHandler(session, handler.ServiceOption(serviceName), handler.LoggerOption(log.WithFields(map[string]any{ @@ -145,6 +204,8 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr log = log.WithFields(map[string]any{}) log.Infof("bind on %s/%s OK", ln.Addr(), ln.Addr().Network()) + // Goroutine: accept and discard unexpected mux connections. + // Normal inbound connections are handled by tcpHandler via session.GetConn(). go func() { defer srv.Close() for { @@ -153,13 +214,23 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr log.Error(err) return } - conn.Close() // we do not handle incoming connections. + conn.Close() // 我们不处理意外入站的连接 } }() return srv.Serve() } +// bindUDP implements UDP BIND mode. +// +// Flow: +// 1. net.ListenPacket on the specified UDP address. +// 2. Return the listening address to the client. +// 3. Create a udp.Relay for bidir datagram relay between client and local port. +// +// Unlike TCP BIND, UDP BIND does not use mux. It relays datagrams directly via +// udp.Relay, wrapping the stream connection as UDPTunServerConn for datagram +// framing over the stream. func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { resp := relay.Response{ Version: relay.Version1, @@ -203,6 +274,7 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr log.Infof("bind on %s OK", pc.LocalAddr()) + // relay_util.UDPTunServerConn 将流式连接包装成数据报模式 r := udp.NewRelay(relay_util.UDPTunServerConn(conn), pc). WithService(h.options.Service). WithBypass(h.options.Bypass). @@ -216,4 +288,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/conn.go b/handler/relay/conn.go index 7c9a7b13..72da5f72 100644 --- a/handler/relay/conn.go +++ b/handler/relay/conn.go @@ -9,6 +9,16 @@ import ( "net" ) +// tcpConn wraps a TCP connection with response header buffering (wbuf). +// +// In non-noDelay mode, the relay.Response header is first written into wbuf. +// On the first Write() call, the buffered header and the data are sent together. +// This avoids sending a small relay frame before the data stream begins. +// +// Write() semantics: +// - n always returns len(b) rather than the actual bytes written. This differs +// from net.Conn's contract. The rationale: when wbuf has content, the actual +// write is "wbuf + b" but the caller only cares that "b" was "processed". type tcpConn struct { net.Conn wbuf bytes.Buffer @@ -19,9 +29,9 @@ func (c *tcpConn) Read(b []byte) (n int, err error) { } func (c *tcpConn) Write(b []byte) (n int, err error) { - n = len(b) // force byte length consistent + n = len(b) // always return len(b), not actual bytes written if c.wbuf.Len() > 0 { - c.wbuf.Write(b) // append the data to the cached header + c.wbuf.Write(b) // 将数据追加到缓存的头部之后 _, err = c.wbuf.WriteTo(c.Conn) return } @@ -29,6 +39,27 @@ func (c *tcpConn) Write(b []byte) (n int, err error) { return } +// udpConn wraps UDP datagrams over a stream connection. +// +// When the GOST relay protocol carries UDP over a TCP/stream transport, +// a 2-byte big-endian length prefix frames each datagram. TCP is a stream +// protocol without message boundaries, so this framing is necessary. +// +// Datagram wire format: +// +// [2-byte length (big-endian)][datagram payload] +// +// Read() flow: +// 1. Read 2-byte length prefix. +// 2. Read the datagram payload of the specified length. +// 3. If the caller's buffer is too small, allocate an internal buffer, +// read the full datagram, then truncate on copy. +// +// Write() flow: +// 1. Reject data exceeding MaxUint16 (65535). +// 2. If wbuf has a cached header, append the length prefix + data to the +// header and flush everything at once. +// 3. Otherwise write the length prefix then the data. type udpConn struct { net.Conn wbuf bytes.Buffer @@ -45,6 +76,7 @@ func (c *udpConn) Read(b []byte) (n int, err error) { if len(b) >= dlen { return io.ReadFull(c.Conn, b[:dlen]) } + // Caller's buffer is too small; allocate internal buffer. buf := make([]byte, dlen) _, err = io.ReadFull(c.Conn, buf) n = copy(b, buf) @@ -60,14 +92,16 @@ func (c *udpConn) Write(b []byte) (n int, err error) { n = len(b) if c.wbuf.Len() > 0 { + // Wbuf has cached header; append length prefix + data and flush together. var bb [2]byte binary.BigEndian.PutUint16(bb[:], uint16(len(b))) c.wbuf.Write(bb[:]) - c.wbuf.Write(b) // append the data to the cached header + c.wbuf.Write(b) _, err = c.wbuf.WriteTo(c.Conn) return } + // Write length prefix + data directly. var bb [2]byte binary.BigEndian.PutUint16(bb[:], uint16(len(b))) _, err = c.Conn.Write(bb[:]) @@ -75,4 +109,4 @@ func (c *udpConn) Write(b []byte) (n int, err error) { return } return c.Conn.Write(b) -} +} \ No newline at end of file diff --git a/handler/relay/connect.go b/handler/relay/connect.go index 1fd78635..cf289773 100644 --- a/handler/relay/connect.go +++ b/handler/relay/connect.go @@ -26,6 +26,41 @@ import ( xrecorder "github.com/go-gost/x/recorder" ) +// handleConnect processes a relay CmdConnect request. +// +// This is the core function of the relay handler: the client connects to a +// target address through the relay. +// +// Data flow: +// +// handleConnect() +// ├─ 1. Clean address (unix/serial special handling) +// ├─ 2. Wrap traffic limiter + stats +// ├─ 3. Check target address is non-empty +// ├─ 4. Bypass check +// ├─ 5. Set consistent-hashing source (e.g. "host") +// ├─ 6. Dial by network type: +// │ ├─ "unix" → net.Dialer.DialContext("unix") +// │ ├─ "serial" → serial.OpenPort() +// │ └─ other → h.options.Router.Dial() ← chain routing +// ├─ 7. Send response header (noDelay controls timing) +// │ ├─ noDelay=true → write relay.Response immediately +// │ └─ noDelay=false → buffer in wbuf, merged with first data packet +// ├─ 8. Wrap connection by network type: +// │ ├─ UDP → udpConn (2-byte length prefix) +// │ └─ TCP → tcpConn (passthrough) +// ├─ 9. Optional protocol sniffing: +// │ ├─ HTTP → sniffer.HandleHTTP() +// │ ├─ TLS → sniffer.HandleTLS() +// │ └─ other → continue to step 10 +// └─ 10. xnet.Pipe() bidir copy (client ↔ target) +// +// Key design decisions: +// - noDelay: low-latency mode sends each write as an independent relay frame. +// Without it, the relay response header is merged with the first data packet. +// - Sniffing reads the first few bytes after connection to detect the protocol. +// HTTP/TLS traffic can be MITM-decrypted or recorded. +// - Route info (node chain) is recorded via ictx.ContextWithBuffer into ro.Route. func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { if network == "unix" || network == "serial" { if host, _, _ := net.SplitHostPort(address); host != "" { @@ -41,6 +76,8 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network log.Debugf("%s >> %s/%s", conn.RemoteAddr(), address, network) + // --- Traffic limiter + stats wrapper --- + // clientID is used as the key for traffic limiting and stats collection. { clientID := xctx.ClientIDFromContext(ctx) rw := traffic_wrapper.WrapReadWriter( @@ -78,6 +115,7 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network return } + // Bypass check — if the target is in the bypass list, return Forbidden if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, network, address, bypass.WithService(h.options.Service)) { log.Debug("bypass: ", address) resp.Status = relay.StatusForbidden @@ -85,11 +123,13 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network return xbypass.ErrBypass } + // Consistent hashing — used for sticky sessions across upstream nodes switch h.md.hash { case "host": ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: address}) } + // --- Dial to target --- var cc net.Conn switch network { case "unix": @@ -119,6 +159,7 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network ro.SrcAddr = cc.LocalAddr().String() ro.DstAddr = cc.RemoteAddr().String() + // --- Send relay response header --- if h.md.noDelay { if _, err := resp.WriteTo(conn); err != nil { log.Error(err) @@ -126,13 +167,16 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network } } + // --- Wrap connection by network type --- + // UDP connections use a 2-byte length prefix for datagram framing. + // TCP connections pass data through directly. switch network { case "udp", "udp4", "udp6": rc := &udpConn{ Conn: conn, } if !h.md.noDelay { - // cache the header + // 缓存响应头,与第一个数据包合并发送 if _, err := resp.WriteTo(&rc.wbuf); err != nil { return err } @@ -143,7 +187,7 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network rc := &tcpConn{ Conn: conn, } - // cache the header + // 缓存响应头,与第一个数据包合并发送 if _, err := resp.WriteTo(&rc.wbuf); err != nil { return err } @@ -151,6 +195,9 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network } } + // --- Optional protocol sniffing --- + // After connection establishment, sniff the first bytes of client traffic + // to detect HTTP or TLS. When detected, MITM decryption can be applied. if h.md.sniffing { if h.md.sniffingTimeout > 0 { conn.SetReadDeadline(time.Now().Add(h.md.sniffingTimeout)) @@ -204,9 +251,9 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network } } + // --- 双向数据拷贝 --- t := time.Now() log.Infof("%s <-> %s", conn.RemoteAddr(), address) - // xnet.Transport(conn, cc) xnet.Pipe(ctx, conn, cc) log.WithFields(map[string]any{ "duration": time.Since(t), @@ -215,6 +262,10 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network return nil } +// serialConn wraps a serial port connection as net.Conn. +// Serial ports have no network addresses, so LocalAddr/RemoteAddr return +// custom serialAddr values. Deadline methods are no-ops since serial +// ports do not support deadlines. type serialConn struct { io.ReadWriteCloser port string @@ -244,6 +295,7 @@ func (c *serialConn) SetWriteDeadline(t time.Time) error { return nil } +// serialAddr is the address type for a serial port connection. type serialAddr struct { port string } @@ -254,4 +306,4 @@ func (a *serialAddr) Network() string { func (a *serialAddr) String() string { return a.port -} +} \ No newline at end of file diff --git a/handler/relay/entrypoint.go b/handler/relay/entrypoint.go index c3c5e684..bda34eea 100644 --- a/handler/relay/entrypoint.go +++ b/handler/relay/entrypoint.go @@ -16,6 +16,16 @@ import ( metrics "github.com/go-gost/x/metrics/wrapper" ) +// tcpListener is the internal TCP listener used in BIND mode. +// +// Wrapping layers (outermost first): +// - proxyproto.WrapListener — PROXY protocol support +// - metrics.WrapListener — connection metrics +// - admission.WrapListener — access control (allow/deny lists) +// - raw net.Listener +// +// This is a simplified version of the standard listener wrapping chain from +// x/config/parsing/service/parse.go. type tcpListener struct { ln net.Listener options listener.Options @@ -33,7 +43,6 @@ func newTCPListener(ln net.Listener, opts ...listener.Option) listener.Listener } func (l *tcpListener) Init(md md.Metadata) (err error) { - // l.logger.Debugf("pp: %d", l.options.ProxyProtocol) ln := l.ln ln = proxyproto.WrapListener(l.options.ProxyProtocol, ln, 10*time.Second) ln = metrics.WrapListener(l.options.Service, ln) @@ -55,6 +64,20 @@ func (l *tcpListener) Close() error { return l.ln.Close() } +// tcpHandler is the internal handler for BIND mode. +// +// When an inbound connection arrives at the listen port created by bindTCP, +// this handler forwards it back to the requesting client over a mux stream. +// +// Flow: +// 1. Gets a free stream from the mux session (session.GetConn()). +// 2. Encodes the inbound peer address as a relay.AddrFeature on the stream. +// 3. Writes a relay.Response (StatusOK). +// 4. Bidirectional Pipe (inbound conn ↔ mux stream). +// +// This is the core mechanism for reverse-proxy / tunnel traversal: +// the client that requested BIND receives forwarded connections as +// streams on the mux session. type tcpHandler struct { session *mux.Session options handler.Options @@ -92,6 +115,7 @@ func (h *tcpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() + // 从 mux 会话获取一个流 cc, err := h.session.GetConn() if err != nil { log.Error(err) @@ -99,6 +123,7 @@ func (h *tcpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. } defer cc.Close() + // 将入站连接地址编码为 AddrFeature,通过 relay 帧发送给客户端 af := &relay.AddrFeature{} af.ParseFrom(conn.RemoteAddr().String()) resp := relay.Response{ @@ -113,9 +138,8 @@ func (h *tcpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. t := time.Now() log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr()) - // xnet.Transport(conn, cc) xnet.Pipe(ctx, conn, cc) log.WithFields(map[string]any{"duration": time.Since(t)}). Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr()) return nil -} +} \ No newline at end of file diff --git a/handler/relay/forward.go b/handler/relay/forward.go index 398a00b5..8f07ab99 100644 --- a/handler/relay/forward.go +++ b/handler/relay/forward.go @@ -19,6 +19,31 @@ import ( xrecorder "github.com/go-gost/x/recorder" ) +// handleForward processes relay forward mode. +// +// When a hop is set on the relayHandler (via Forward()), this mode is used. +// Unlike handleConnect, the target address is not specified by the client; +// instead it is selected by the hop's load-balancing strategy (round-robin, +// random, hash, etc.). +// +// Data flow: +// +// handleForward() +// ├─ 1. hop.Select() picks a target node from the hop +// │ └─ No node available → return ServiceUnavailable +// ├─ 2. Wrap traffic limiter + stats +// ├─ 3. Router.Dial() dials the target node +// │ └─ Dial fails → mark the node (Mark()) → return HostUnreachable +// ├─ 4. On success, reset the node's failure marker +// ├─ 5. Send response header (noDelay controls timing) +// ├─ 6. Wrap connection by network type (tcpConn/udpConn) +// └─ 7. xnet.Pipe() bidir data copy (client ↔ target) +// +// Failure handling: +// - If Router.Dial fails and the target node has a Marker, the node is +// marked as failed. Marked nodes are skipped by FailFilter/BackupFilter +// in subsequent selections. +// - On successful dial, the marker is reset, indicating recovery. func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { resp := relay.Response{ Version: relay.Version1, @@ -40,6 +65,7 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network log.Debugf("%s >> %s", conn.RemoteAddr(), target.Addr) + // --- Traffic limiter + stats wrapper --- { clientID := ctxvalue.ClientIDFromContext(ctx) rw := wrapper.WrapReadWriter( @@ -63,12 +89,13 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network conn = xnet.NewReadWriteConn(rw, rw, conn) } + // Dial the target node, recording the route path. var buf bytes.Buffer cc, err := h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), network, target.Addr) ro.Route = buf.String() if err != nil { - // TODO: the router itself may be failed due to the failed node in the router, - // the dead marker may be a wrong operation. + // TODO: the router itself may fail because of a failed node in the route. + // Marking the node here may be incorrect in that case. if marker := target.Marker(); marker != nil { marker.Mark() } @@ -85,10 +112,12 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network ro.SrcAddr = cc.LocalAddr().String() ro.DstAddr = cc.RemoteAddr().String() + // Reset the failure marker on successful dial. if marker := target.Marker(); marker != nil { marker.Reset() } + // --- Send response header --- if h.md.noDelay { if _, err := resp.WriteTo(conn); err != nil { log.Error(err) @@ -96,13 +125,14 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network } } + // --- Wrap connection by network type --- switch network { case "udp", "udp4", "udp6": rc := &udpConn{ Conn: conn, } if !h.md.noDelay { - // cache the header + // Buffer the response header, merged with the first data packet. if _, err := resp.WriteTo(&rc.wbuf); err != nil { return err } @@ -113,7 +143,7 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network Conn: conn, } if !h.md.noDelay { - // cache the header + // Buffer the response header, merged with the first data packet. if _, err := resp.WriteTo(&rc.wbuf); err != nil { return err } @@ -123,11 +153,10 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network t := time.Now() log.Debugf("%s <-> %s", conn.RemoteAddr(), target.Addr) - // xnet.Transport(conn, cc) xnet.Pipe(ctx, conn, cc) log.WithFields(map[string]any{ "duration": time.Since(t), }).Debugf("%s >-< %s", conn.RemoteAddr(), target.Addr) return nil -} +} \ No newline at end of file diff --git a/handler/relay/handler.go b/handler/relay/handler.go index 544fe35f..84c7eb45 100644 --- a/handler/relay/handler.go +++ b/handler/relay/handler.go @@ -37,6 +37,42 @@ func init() { registry.HandlerRegistry().Register("relay", NewHandler) } +// relayHandler is the GOST relay protocol server handler. +// +// The GOST relay protocol is a custom multiplexed transport supporting three modes: +// 1. Connect — the client requests a connection to a target address. +// The handler dials via the configured Router and pipes data bidirectionally. +// - Direct: handleConnect(), target address from the relay request. +// - Forward: handleForward(), target from hop selector (load balancing). +// 2. Bind — the client asks the handler to listen on a local port and forward +// incoming connections back through a mux session. Used for reverse proxying. +// 3. Forward — when a hop is set, the target is selected by the hop's strategy +// rather than specified by the client. +// +// Data flow: +// +// ┌─────────────────────────────────────────────────────────┐ +// │ Handle() → parse relay.Request │ +// │ ├─ extract auth (UserAuthFeature) → authenticate │ +// │ ├─ extract target address (AddrFeature) │ +// │ └─ extract network type (NetworkFeature) │ +// │ │ +// │ ┌── hop set? ──→ handleForward() │ +// │ │ ├─ hop.Select() pick target node │ +// │ │ └─ Router.Dial() → Pipe bidir copy │ +// │ │ +// │ └── no hop, dispatch by command: │ +// │ ├─ CmdConnect → handleConnect() │ +// │ │ ├─ bypass check │ +// │ │ ├─ consistent-hashing │ +// │ │ ├─ Router.Dial() / net.Dial() / serial.Open │ +// │ │ ├─ send response header (noDelay) │ +// │ │ ├─ optional protocol sniffing (HTTP/TLS MITM) │ +// │ │ └─ Pipe bidir data copy │ +// │ └─ CmdBind → handleBind() │ +// │ ├─ bindTCP: net.Listen → mux session → tcpHandler │ +// │ └─ bindUDP: net.ListenPacket → udp.Relay │ +// └─────────────────────────────────────────────────────────┘ type relayHandler struct { hop hop.Hop md metadata @@ -59,6 +95,14 @@ func NewHandler(opts ...handler.Option) handler.Handler { } } +// Init 初始化 relay handler。在 handler 被注册到 service 后调用。 +// +// 初始化流程: +// 1. 解析元数据配置(超时、嗅探、mux、MITM 等) +// 2. 如果配置了 Observer,创建 stats 统计器并启动后台轮询协程 +// 3. 如果配置了 TrafficLimiter,创建带缓存的流量限制器 +// 4. 从 Recorders 列表中选取 ServiceHandler 类型的 recorder +// 5. 如果配置了 MITM 证书,创建内存证书池 func (h *relayHandler) Init(md md.Metadata) (err error) { if err := h.parseMetadata(md); err != nil { return err @@ -99,6 +143,22 @@ func (h *relayHandler) Forward(hop hop.Hop) { h.hop = hop } +// Handle is the main entry point for each inbound connection. +// +// Flow: +// 1. Create a recorder object with connection metadata. +// 2. Wrap the connection for stats collection (input/output bytes). +// 3. Rate-limit check. +// 4. Set read deadline, read the relay.Request. +// 5. Clear the read deadline. +// 6. Check the relay protocol version. +// 7. Parse request features (auth, address, network type). +// 8. Authenticate (if an Auther is configured). +// 9. Dispatch: +// - If hop is set → handleForward. +// - CmdConnect → handleConnect. +// - CmdBind → handleBind. +// 10. Deferred final stats recording. func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() @@ -220,7 +280,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle log = log.WithFields(map[string]any{"network": network}) if h.hop != nil { - // forward mode + // Forward mode: target is selected from the hop. return h.handleForward(ctx, conn, network, ro, log) } @@ -242,5 +302,4 @@ func (h *relayHandler) Close() error { h.cancel() } return nil -} - +} \ No newline at end of file diff --git a/handler/relay/metadata.go b/handler/relay/metadata.go index e61b927b..15598cdf 100644 --- a/handler/relay/metadata.go +++ b/handler/relay/metadata.go @@ -13,37 +13,85 @@ import ( "github.com/go-gost/x/registry" ) +// metadata holds the relay handler configuration parsed from the generic Metadata map. type metadata struct { - // readTimeout is the deadline for reading the initial relay protocol - // handshake (relay.Request) from the client connection. The deadline - // is cleared immediately after the handshake completes, so it does - // not affect subsequent data transfer. Also passed to SnifferBuilder - // for the upstream response header read timeout. - // 0 or negative defaults to 15s. - readTimeout time.Duration - udpBufferSize int - enableBind bool - noDelay bool - hash string - muxCfg *mux.Config + // readTimeout is the deadline for reading the initial relay.Request handshake. + // Cleared immediately after the handshake, so it does not affect data transfer. + // Also passed to SnifferBuilder for upstream response header read timeout. + // Default: 15s (0 or negative falls back to 15s). + readTimeout time.Duration - observerPeriod time.Duration + // udpBufferSize is the buffer size (in bytes) for UDP datagram relay. + // Config keys: "udp.bufferSize", "udpBufferSize". + udpBufferSize int + + // enableBind controls whether CmdBind is allowed. + // Disabled by default; set "bind: true" in config to enable. + enableBind bool + + // noDelay controls whether the relay.Response header is sent immediately. + // When enabled, each write is an independent relay frame (low-latency). + // When disabled, the response header is buffered in wbuf and merged with + // the first data write. + noDelay bool + + // hash specifies the consistent-hashing source. Currently supports "host", + // which uses the target address as the hash source for sticky sessions. + hash string + + // muxCfg is the multiplexing session configuration, used only in BIND mode + // to upgrade the client connection to a mux session. + muxCfg *mux.Config + + // observerPeriod is the stats event polling interval. + // Default: 5s, minimum: 1s. + // Config keys: "observePeriod", "observer.period", "observer.observePeriod". + observerPeriod time.Duration + + // observerResetTraffic controls whether traffic counters are reset after + // each poll. When true, reported traffic is incremental; when false, cumulative. observerResetTraffic bool - sniffing bool - sniffingTimeout time.Duration - sniffingWebsocket bool + // sniffing enables protocol sniffing. When enabled, the handler detects + // the traffic protocol (HTTP/TLS) after connection establishment and + // selects the corresponding processing path (e.g. MITM decryption). + sniffing bool + + // sniffingTimeout is the read deadline during sniffing. + sniffingTimeout time.Duration + + // sniffingWebsocket enables WebSocket upgrade detection within sniffed HTTP. + sniffingWebsocket bool + + // sniffingWebsocketSampleRate controls the sampling rate for WebSocket + // traffic recording (0.0 ~ 1.0). sniffingWebsocketSampleRate float64 + // certificate is the CA certificate used for MITM decryption. certificate *x509.Certificate - privateKey crypto.PrivateKey - alpn string - mitmBypass bypass.Bypass + // privateKey is the private key paired with the MITM certificate. + privateKey crypto.PrivateKey + + // alpn is the TLS ALPN protocol list for MITM decryption. + alpn string + + // mitmBypass is an allow/deny list of hostnames that skip MITM decryption. + mitmBypass bypass.Bypass + + // limiterRefreshInterval is the traffic limiter cache refresh interval. limiterRefreshInterval time.Duration + + // limiterCleanupInterval is the traffic limiter cache cleanup interval. limiterCleanupInterval time.Duration } +// parseMetadata extracts typed configuration from the generic Metadata map. +// +// Rules: +// - Uses mdutil Get* helpers which support multiple fallback key names. +// - Numeric durations are treated as seconds; string durations use time.ParseDuration. +// - Unset or invalid values fall back to sensible defaults. func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.readTimeout = mdutil.GetDuration(md, "readTimeout") if h.md.readTimeout <= 0 { @@ -100,4 +148,4 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") return -} +} \ No newline at end of file diff --git a/handler/relay/observe.go b/handler/relay/observe.go index 0781c10a..5547c81e 100644 --- a/handler/relay/observe.go +++ b/handler/relay/observe.go @@ -8,6 +8,11 @@ import ( "github.com/go-gost/core/observer" ) +// checkRateLimit checks whether a new connection from addr should be accepted. +// +// If a RateLimiter is configured, it looks up the rate limiter for the client +// host and calls Allow(1). Without a RateLimiter or without a limiter for the +// specific host, the connection is allowed by default. func (h *relayHandler) checkRateLimit(addr net.Addr) bool { if h.options.RateLimiter == nil { return true @@ -20,11 +25,34 @@ func (h *relayHandler) checkRateLimit(addr net.Addr) bool { return true } +// observeStats is a background goroutine that periodically collects and pushes +// stats events. It is started in Init() when an Observer is configured. +// +// Event collection flow: +// +// observeStats() +// ├─ ticker fires at observerPeriod intervals +// ├─ Each tick: +// │ ├─ If previous events failed (events != nil): +// │ │ └─ Retry sending buffered events +// │ │ └─ If it fails again → continue (retain events for next tick) +// │ ├─ Call h.stats.Events() for new events +// │ ├─ If there are new events: +// │ │ ├─ Try sending (Observe) +// │ │ └─ If it fails → save to events, retry next tick +// │ └─ On success, clear events = nil +// └─ ctx.Done() → exit +// +// Retry mechanism: +// - If Observe() returns an error, events are retained for the next retry. +// - On retry, buffered events are sent first, then new events are fetched. +// - This ensures events are not lost due to transient network issues. func (h *relayHandler) observeStats(ctx context.Context) { if h.options.Observer == nil { return } + // events holds events that failed to send on the previous tick, for retry. var events []observer.Event ticker := time.NewTicker(h.md.observerPeriod) @@ -33,20 +61,23 @@ func (h *relayHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: - // Try to flush any buffered events from a previous failed attempt. + // First, retry sending any buffered events from a previous failed attempt. if len(events) > 0 { if err := h.options.Observer.Observe(ctx, events); err != nil { + // Failed again, retain events for the next retry. continue } } - // Collect and send fresh events. + // Fetch and send new events. if evs := h.stats.Events(); len(evs) > 0 { if err := h.options.Observer.Observe(ctx, evs); err != nil { + // Failed, cache for retry on the next tick. events = evs continue } } + // Sent successfully, clear the cache. events = nil case <-ctx.Done():