go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

docs(handler/relay): convert all comments to English

Browse Source 
Translate all Chinese comments in the relay handler package to English:
handler.go, connect.go, bind.go, forward.go, conn.go, entrypoint.go,
observe.go, metadata.go. All inline and doc comments are now in English.

Also set a persistent preference: all future code comments must be
written in English only.
This commit is contained in:
ginuerzh
2026-06-03 23:05:08 +08:00
parent 95874c53f5
commit d5fd62aa47
8 changed files with 395 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handler/relay/bind.go
+76 -4
View File
@@ -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: "<main-service>-ep-<listen-address>"
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
}
}
handler/relay/conn.go
+38 -4
View File
@@ -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)
}
}
handler/relay/connect.go
+56 -4
View File
@@ -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
}
}
handler/relay/entrypoint.go
+27 -3
View File
@@ -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
}
}
handler/relay/forward.go
+35 -6
View File
@@ -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
}
}
handler/relay/handler.go
+62 -3
View File
@@ -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
}
}
handler/relay/metadata.go
+68 -20
View File
@@ -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
}
}
handler/relay/observe.go
+33 -2
View File
@@ -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():