From b2d901e70fbcbf8993a12db4f4ceb38bcbb87f0c Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Sat, 1 Aug 2026 15:54:38 +0800 Subject: [PATCH] x: add Redis TCP sniffing with protocol-based routing Add RESP protocol detection to the sniffing framework so plaintext Redis traffic on a port-reuse listener can be identified and routed by protocol. Sniff() now peeks 1 byte first to match RESP type bytes (*+-:$) without blocking, then falls through to the existing 5-byte TLS/HTTP/SSH detection. ParseRedisMetadata consumes exactly one RESP message via io.TeeReader and records command/key in HandlerRecorderObject.Redis. HandleRedis in both the sniffing (injected-dial) and forwarder (hop-selected) packages replays the captured first message and pipes the rest, so pipelined commands beyond the first are forwarded transparently. Wired into all 5 dispatch sites: sni, forward/local, forward/remote, http CONNECT, and relay tunnel handlers. forwarder paths route via resolveRedisNode using ProtocolSelectOption(ProtoRedis), enabling `protocol: redis` hop selection. Part of two-phase plan (2026-07-29-sniffing-redis-postgres.md); PostgreSQL deferred to phase 2. --- handler/forward/local/sniffing.go | 14 +- handler/forward/remote/sniffing.go | 14 +- handler/http/connect.go | 7 + handler/relay/connect.go | 8 + handler/sni/sniffing.go | 16 +- internal/util/forwarder/sniffer_redis.go | 141 +++++++++++++++++ internal/util/sniffing/sniff.go | 21 ++- internal/util/sniffing/sniffer_redis.go | 188 +++++++++++++++++++++++ recorder/recorder.go | 7 + 9 files changed, 404 insertions(+), 12 deletions(-) create mode 100644 internal/util/forwarder/sniffer_redis.go create mode 100644 internal/util/sniffing/sniffer_redis.go diff --git a/handler/forward/local/sniffing.go b/handler/forward/local/sniffing.go index 6b39517..7e63a26 100644 --- a/handler/forward/local/sniffing.go +++ b/handler/forward/local/sniffing.go @@ -76,7 +76,7 @@ func (h *forwardHandler) sniffingDial(ctx context.Context, network, address stri // handled and (false, nil) when the caller should fall through to raw forwarding. func (h *forwardHandler) handleSniffedProtocol(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, proto string) (handled bool, err error) { switch proto { - case sniffing.ProtoHTTP, sniffing.ProtoTLS: + case sniffing.ProtoHTTP, sniffing.ProtoTLS, sniffing.ProtoRedis: dial := func(ctx context.Context, network, address string) (net.Conn, error) { return h.sniffingDial(ctx, network, address, ro) } @@ -92,7 +92,17 @@ func (h *forwardHandler) handleSniffedProtocol(ctx context.Context, conn net.Con forwarder.WithLog(log), ) } - return true, sniffer.HandleTLS(ctx, conn, + if proto == sniffing.ProtoTLS { + return true, sniffer.HandleTLS(ctx, conn, + forwarder.WithService(h.options.Service), + forwarder.WithDial(dial), + forwarder.WithHop(h.getHop()), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) + } + return true, sniffer.HandleRedis(ctx, conn, forwarder.WithService(h.options.Service), forwarder.WithDial(dial), forwarder.WithHop(h.getHop()), diff --git a/handler/forward/remote/sniffing.go b/handler/forward/remote/sniffing.go index 311d7f5..59773d1 100644 --- a/handler/forward/remote/sniffing.go +++ b/handler/forward/remote/sniffing.go @@ -76,7 +76,7 @@ func (h *forwardHandler) sniffingDial(ctx context.Context, network, address stri // handled and (false, nil) when the caller should fall through to raw forwarding. func (h *forwardHandler) handleSniffedProtocol(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, proto string) (handled bool, err error) { switch proto { - case sniffing.ProtoHTTP, sniffing.ProtoTLS: + case sniffing.ProtoHTTP, sniffing.ProtoTLS, sniffing.ProtoRedis: dial := func(ctx context.Context, network, address string) (net.Conn, error) { return h.sniffingDial(ctx, network, address, ro) } @@ -92,7 +92,17 @@ func (h *forwardHandler) handleSniffedProtocol(ctx context.Context, conn net.Con forwarder.WithLog(log), ) } - return true, sniffer.HandleTLS(ctx, conn, + if proto == sniffing.ProtoTLS { + return true, sniffer.HandleTLS(ctx, conn, + forwarder.WithService(h.options.Service), + forwarder.WithDial(dial), + forwarder.WithHop(h.getHop()), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) + } + return true, sniffer.HandleRedis(ctx, conn, forwarder.WithService(h.options.Service), forwarder.WithDial(dial), forwarder.WithHop(h.getHop()), diff --git a/handler/http/connect.go b/handler/http/connect.go index 09d6bca..e177f22 100644 --- a/handler/http/connect.go +++ b/handler/http/connect.go @@ -147,6 +147,13 @@ func (h *httpHandler) sniffAndHandle(ctx context.Context, conn net.Conn, cc net. sniffing.WithRecorderObject(ro), sniffing.WithLog(log), ) + case sniffing.ProtoRedis: + return true, sniffer.HandleRedis(ctx, "tcp", conn, + sniffing.WithService(h.options.Service), + sniffing.WithDial(dial), + sniffing.WithRecorderObject(ro), + sniffing.WithLog(log), + ) } return false, nil diff --git a/handler/relay/connect.go b/handler/relay/connect.go index f2787d9..b388f50 100644 --- a/handler/relay/connect.go +++ b/handler/relay/connect.go @@ -252,6 +252,14 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network sniffing.WithRecorderObject(ro), sniffing.WithLog(log), ) + case sniffing.ProtoRedis: + ro.Time = time.Time{} + return sniffer.HandleRedis(ctx, "tcp", conn, + sniffing.WithService(h.options.Service), + sniffing.WithDial(dial), + sniffing.WithRecorderObject(ro), + sniffing.WithLog(log), + ) } } diff --git a/handler/sni/sniffing.go b/handler/sni/sniffing.go index 4ba75e0..edc14fd 100644 --- a/handler/sni/sniffing.go +++ b/handler/sni/sniffing.go @@ -98,7 +98,7 @@ func (h *sniHandler) sniffingDialTLS(ctx context.Context, network, address strin // connection directly (which for the SNI handler means silently dropping it). func (h *sniHandler) handleSniffedProtocol(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, proto string) (handled bool, err error) { switch proto { - case sniffing.ProtoHTTP, sniffing.ProtoTLS: + case sniffing.ProtoHTTP, sniffing.ProtoTLS, sniffing.ProtoRedis: dial := func(ctx context.Context, network, address string) (net.Conn, error) { return h.sniffingDial(ctx, network, address, ro) } @@ -116,11 +116,19 @@ func (h *sniHandler) handleSniffedProtocol(ctx context.Context, conn net.Conn, r sniffing.WithLog(log), ) } - return true, sniffer.HandleTLS(ctx, "tcp", conn, + if proto == sniffing.ProtoTLS { + return true, sniffer.HandleTLS(ctx, "tcp", conn, + sniffing.WithService(h.options.Service), + sniffing.WithDial(dial), + sniffing.WithDialTLS(dialTLS), + sniffing.WithBypass(h.options.Bypass), + sniffing.WithRecorderObject(ro), + sniffing.WithLog(log), + ) + } + return true, sniffer.HandleRedis(ctx, "tcp", conn, sniffing.WithService(h.options.Service), sniffing.WithDial(dial), - sniffing.WithDialTLS(dialTLS), - sniffing.WithBypass(h.options.Bypass), sniffing.WithRecorderObject(ro), sniffing.WithLog(log), ) diff --git a/internal/util/forwarder/sniffer_redis.go b/internal/util/forwarder/sniffer_redis.go new file mode 100644 index 0000000..4a0fa07 --- /dev/null +++ b/internal/util/forwarder/sniffer_redis.go @@ -0,0 +1,141 @@ +package forwarder + +import ( + "bytes" + "context" + "fmt" + "io" + "net" + "time" + + "github.com/go-gost/core/chain" + "github.com/go-gost/core/hop" + xnet "github.com/go-gost/x/internal/net" + "github.com/go-gost/x/internal/util/sniffing" +) + +// HandleRedis proxies a sniffed Redis connection through a hop-selected node. +// It parses the first RESP message for metadata recording, selects an upstream +// node via resolveRedisNode (which passes ProtoRedis to the hop's protocol +// selector), replays the captured bytes, and pipes traffic bidirectionally. +func (h *Sniffer) HandleRedis(ctx context.Context, conn net.Conn, opts ...HandleOption) error { + var ho HandleOptions + for _, opt := range opts { + opt(&ho) + } + + buf := new(bytes.Buffer) + if err := sniffing.ParseRedisMetadata(io.TeeReader(conn, buf), ho.recorderObject); err != nil { + return err + } + + ro := ho.recorderObject + + node, cc, err := h.dialRedis(ctx, &ho) + if err != nil { + return err + } + defer cc.Close() + ho.node = node + + log := ho.log.WithFields(map[string]any{"src": cc.LocalAddr().String(), "dst": cc.RemoteAddr().String()}) + log.Debugf("connected to node %s(%s)", node.Name, node.Addr) + + ro.SrcAddr = cc.LocalAddr().String() + ro.DstAddr = cc.RemoteAddr().String() + + if _, err := buf.WriteTo(cc); err != nil { + return err + } + + log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host) + xnet.Pipe(ctx, conn, cc) + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + }).Infof("%s >-< %s", ro.RemoteAddr, ro.Host) + + return nil +} + +// resolveRedisNode selects an upstream node for a Redis connection via hop +// selection with ProtocolSelectOption. Unlike resolveTLSNode, there is no +// host to route by — routing is driven by ProtocolSelectOption(ProtoRedis) +// and ClientIPSelectOption. +func resolveRedisNode(ctx context.Context, ho *HandleOptions) (node *chain.Node, err error) { + node = &chain.Node{} + if ho.hop != nil { + var clientIP net.IP + if clientAddr, _ := net.ResolveTCPAddr("tcp", ho.recorderObject.ClientAddr); clientAddr != nil { + clientIP = clientAddr.IP + } + node = ho.hop.Select(ctx, + hop.ClientIPSelectOption(clientIP), + hop.ProtocolSelectOption(sniffing.ProtoRedis), + ) + } + if node == nil { + return nil, fmt.Errorf("node not available") + } + ho.recorderObject.Node = node.Name + return node, nil +} + +// dialRedis selects a node via resolveRedisNode and establishes a TCP +// connection to the node's address. +func (h *Sniffer) dialRedis(ctx context.Context, ho *HandleOptions) (node *chain.Node, cc net.Conn, err error) { + dial := ho.dial + if dial == nil { + dial = (&net.Dialer{}).DialContext + } + + if node = ho.node; node != nil { + cc, err = dial(ctx, "tcp", node.Addr) + return + } + + node, err = resolveRedisNode(ctx, ho) + if err != nil { + return + } + + ro := ho.recorderObject + addr := node.Addr + network := "tcp" + if opts := node.Options(); opts != nil { + switch opts.Network { + case "unix": + network = opts.Network + default: + if _, _, splitErr := net.SplitHostPort(addr); splitErr != nil { + addr += ":6379" + } + } + } else { + if _, _, splitErr := net.SplitHostPort(addr); splitErr != nil { + addr += ":6379" + } + } + ro.Host = addr + + ho.log = ho.log.WithFields(map[string]any{ + "node": node.Name, + "dst": fmt.Sprintf("%s/%s", addr, network), + }) + ho.log.Debugf("find node for redis -> %s(%s)", node.Name, addr) + + cc, err = dial(ctx, network, addr) + if err != nil { + if marker := node.Marker(); marker != nil { + marker.Mark() + } + ho.log.Warnf("connect to node %s(%s) failed: %v", node.Name, node.Addr, err) + return + } + + if marker := node.Marker(); marker != nil { + marker.Reset() + } + + cc = tlsWrapConn(cc, node.Options().TLS) + return +} diff --git a/internal/util/sniffing/sniff.go b/internal/util/sniffing/sniff.go index 19f72f3..69d97be 100644 --- a/internal/util/sniffing/sniff.go +++ b/internal/util/sniffing/sniff.go @@ -12,13 +12,26 @@ import ( ) const ( - ProtoHTTP = "http" - ProtoTLS = "tls" - ProtoSSH = "ssh" - ProtoQUIC = "quic" + ProtoHTTP = "http" + ProtoTLS = "tls" + ProtoSSH = "ssh" + ProtoQUIC = "quic" + ProtoRedis = "redis" ) func Sniff(ctx context.Context, r *bufio.Reader) (proto string, err error) { + // Tier 1: Peek 1 byte to detect protocols identifiable by first byte. + // Peek(1) does not block (unless the connection has zero bytes). + b, err := r.Peek(1) + if err != nil { + return + } + switch b[0] { + case '*', '+', '-', ':', '$': + return ProtoRedis, nil + } + + // Tier 2: Peek 5 bytes for TLS/HTTP/SSH (existing behavior). hdr, err := r.Peek(dissector.RecordHeaderLen) if err != nil { return diff --git a/internal/util/sniffing/sniffer_redis.go b/internal/util/sniffing/sniffer_redis.go new file mode 100644 index 0000000..243e9a7 --- /dev/null +++ b/internal/util/sniffing/sniffer_redis.go @@ -0,0 +1,188 @@ +package sniffing + +import ( + "bufio" + "bytes" + "context" + "fmt" + "io" + "net" + "strconv" + "strings" + "time" + + xnet "github.com/go-gost/x/internal/net" + xrecorder "github.com/go-gost/x/recorder" +) + +// ParseRedisMetadata reads a single RESP message from r and populates +// ro.Redis with the extracted command and key. +// +// r must provide exactly one RESP message — the caller is expected to +// wrap the conn with io.TeeReader(conn, buf) so that the bytes are both +// parsed and captured in buf for replay to the upstream. +// +// On success, ro.Redis is non-nil. On error (malformed RESP, truncated +// message), ro.Redis is nil and the caller should fall back to raw +// forwarding. +func ParseRedisMetadata(r io.Reader, ro *xrecorder.HandlerRecorderObject) error { + br := bufio.NewReader(r) + + b, err := br.ReadByte() + if err != nil { + return err + } + + switch b { + case '*': + return parseRESPArray(br, ro) + case '$': + s, err := readBulkStringPayload(br) + if err != nil { + return err + } + ro.Redis = &xrecorder.RedisRecorderObject{Command: strings.ToUpper(s)} + return nil + case '+', '-': + s, err := readRESPLine(br) + if err != nil { + return err + } + ro.Redis = &xrecorder.RedisRecorderObject{Command: strings.ToUpper(s)} + return nil + case ':': + s, err := readRESPLine(br) + if err != nil { + return err + } + ro.Redis = &xrecorder.RedisRecorderObject{Command: s} + return nil + default: + return fmt.Errorf("unknown RESP type byte: 0x%02x", b) + } +} + +// readRESPLine reads until '\n' and strips the trailing "\r\n". +func readRESPLine(br *bufio.Reader) (string, error) { + line, err := br.ReadString('\n') + if err != nil { + return "", err + } + return strings.TrimRight(line, "\r\n"), nil +} + +// readBulkStringPayload reads \r\n\r\n (the leading '$' must +// already be consumed by the caller). Returns the string value. +func readBulkStringPayload(br *bufio.Reader) (string, error) { + line, err := readRESPLine(br) + if err != nil { + return "", err + } + n, err := strconv.Atoi(line) + if err != nil { + return "", fmt.Errorf("invalid bulk string length: %s", line) + } + if n < 0 { + return "", nil // null bulk string + } + buf := make([]byte, n) + if _, err := io.ReadFull(br, buf); err != nil { + return "", err + } + // consume trailing \r\n + if _, err := br.Discard(2); err != nil { + return "", err + } + return string(buf), nil +} + +// parseRESPArray parses a RESP array: \r\n, then n bulk strings. +// The leading '*' must already be consumed by the caller. +func parseRESPArray(br *bufio.Reader, ro *xrecorder.HandlerRecorderObject) error { + line, err := readRESPLine(br) + if err != nil { + return err + } + n, err := strconv.Atoi(line) + if err != nil { + return fmt.Errorf("invalid RESP array length: %s", line) + } + if n < 0 { + return nil // null array + } + + for i := 0; i < n; i++ { + b, err := br.ReadByte() + if err != nil { + return err + } + if b != '$' { + return fmt.Errorf("expected bulk string in RESP array, got byte 0x%02x", b) + } + s, err := readBulkStringPayload(br) + if err != nil { + return err + } + if i == 0 { + // ponytail: first element allocated inline, second appended. + // A nil check on the first element is enough — RESP + // commands always have at least one element. + ro.Redis = &xrecorder.RedisRecorderObject{Command: strings.ToUpper(s)} + } else if i == 1 { + ro.Redis.Key = s + } + } + if ro.Redis == nil { + ro.Redis = &xrecorder.RedisRecorderObject{} + } + return nil +} + +// HandleRedis proxies a sniffed Redis connection. It parses the first RESP +// message for metadata recording, then replays the captured bytes to the +// upstream and pipes traffic bidirectionally. +// +// The network parameter is passed through to the injected dial function. +// Bypass checks are not performed here — Redis has no host concept for +// bypass to match against; protocol-based routing is handled by the +// hop selector in the forwarder path. +func (h *Sniffer) HandleRedis(ctx context.Context, network string, conn net.Conn, opts ...HandleOption) error { + var ho HandleOptions + for _, opt := range opts { + opt(&ho) + } + + buf := new(bytes.Buffer) + if err := ParseRedisMetadata(io.TeeReader(conn, buf), ho.recorderObject); err != nil { + return err + } + + log := ho.log + ro := ho.recorderObject + + dial := ho.dial + if dial == nil { + dial = (&net.Dialer{}).DialContext + } + cc, err := dial(ctx, network, ro.Host) + if err != nil { + return err + } + defer cc.Close() + + log = log.WithFields(map[string]any{"src": cc.LocalAddr().String(), "dst": cc.RemoteAddr().String()}) + ro.SrcAddr = cc.LocalAddr().String() + ro.DstAddr = cc.RemoteAddr().String() + + if _, err := buf.WriteTo(cc); err != nil { + return err + } + + log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host) + xnet.Pipe(ctx, conn, cc) + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + }).Infof("%s >-< %s", ro.RemoteAddr, ro.Host) + + return nil +} diff --git a/recorder/recorder.go b/recorder/recorder.go index 0a1d516..9f897dd 100644 --- a/recorder/recorder.go +++ b/recorder/recorder.go @@ -114,6 +114,12 @@ type DNSRecorderObject struct { Cached bool `json:"cached"` } +// RedisRecorderObject holds the recorded data of a sniffed Redis command. +type RedisRecorderObject struct { + Command string `json:"command"` + Key string `json:"key"` +} + // HandlerRecorderObject bundles traffic metadata recorded at the handler // level — addresses, protocols, transferred bytes, and optional sub-records // for HTTP, WebSocket, TLS, and DNS traffic. @@ -135,6 +141,7 @@ type HandlerRecorderObject struct { Websocket *WebsocketRecorderObject `json:"websocket,omitempty"` TLS *TLSRecorderObject `json:"tls,omitempty"` DNS *DNSRecorderObject `json:"dns,omitempty"` + Redis *RedisRecorderObject `json:"redis,omitempty"` Route string `json:"route,omitempty"` InputBytes uint64 `json:"inputBytes"` OutputBytes uint64 `json:"outputBytes"`