x/handler/http/connect.go

package http

import (
	"bufio"
	"bytes"
	"context"
	"crypto"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"net"
	"net/http"
	"net/http/httputil"
	"time"

	"github.com/go-gost/core/bypass"
	"github.com/go-gost/core/limiter"
	"github.com/go-gost/core/recorder"
	stats "github.com/go-gost/core/observer/stats"
	"github.com/go-gost/core/logger"
	xctx "github.com/go-gost/x/ctx"
	ictx "github.com/go-gost/x/internal/ctx"
	xnet "github.com/go-gost/x/internal/net"
	"github.com/go-gost/x/internal/util/sniffing"
	tls_util "github.com/go-gost/x/internal/util/tls"
	traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper"
	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
	xrecorder "github.com/go-gost/x/recorder"
)

// handleConnect handles HTTP CONNECT tunnel requests. It dials the target
// address through the proxy chain router, sends a "200 Connection established"
// response to the client, and then relays raw bytes bidirectionally between
// client and upstream.
//
// When sniffing is enabled, the initial bytes from the client are inspected.
// If they match HTTP or TLS, the connection is handed off to the sniffer for
// protocol-aware forwarding (HTTP request routing, TLS MITM decryption).
// If the sniffed protocol is unknown, or sniffing is disabled, raw pipe
// forwarding is used.
func (h *httpHandler) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger, addr string, resp *http.Response) error {
	ctx = ictx.ContextWithRecorderObject(ctx, ro)
	ctx = ictx.ContextWithLogger(ctx, log)
	cc, err := h.dial(ctx, "tcp", addr)
	if err != nil {
		resp.StatusCode = http.StatusServiceUnavailable

		if log.IsLevelEnabled(logger.TraceLevel) {
			dump, _ := httputil.DumpResponse(resp, false)
			log.Trace(string(dump))
		}
		if err := resp.Write(conn); err != nil {
			log.Error("write error response: ", err)
		}
		return err
	}
	// snifferHandled tracks whether the upstream connection has been taken
	// over by the sniffer. If the sniffer doesn't claim it, we close cc.
	snifferHandled := false
	defer func() {
		if !snifferHandled {
			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()

	b := buildConnectResponse(h.md.proxyAgent)
	if log.IsLevelEnabled(logger.TraceLevel) {
		log.Trace(string(b))
	}
	if _, err = conn.Write(b); err != nil {
		log.Error(err)
		return err
	}

	if h.md.sniffing {
		snifferHandled, err = h.sniffAndHandle(ctx, conn, cc, ro, log)
		if snifferHandled {
			return err
		}
	}

	start := time.Now()
	log.Infof("%s <-> %s", conn.RemoteAddr(), addr)
	xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout))
	log.WithFields(map[string]any{
		"duration": time.Since(start),
	}).Infof("%s >-< %s", conn.RemoteAddr(), addr)

	return nil
}

// sniffAndHandle peeks at the initial bytes of the client connection to
// determine the protocol. If HTTP or TLS is detected, the connection is
// handed off to the appropriate sniffer handler for protocol-aware forwarding.
// The dial and dialTLS closures return the already-established upstream
// connection so that the sniffer uses the same tunnel.
//
// Returns (true, err) when the sniffer took over the connection. Returns
// (false, nil) when the protocol is unrecognised and the caller should
// fall back to raw pipe forwarding.
func (h *httpHandler) sniffAndHandle(ctx context.Context, conn net.Conn, cc net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (handled bool, err error) {
	if h.md.sniffingTimeout > 0 {
		conn.SetReadDeadline(time.Now().Add(h.md.sniffingTimeout))
	}

	br := bufio.NewReader(conn)
	proto, _ := sniffing.Sniff(ctx, br)
	ro.Proto = proto

	if h.md.sniffingTimeout > 0 {
		conn.SetReadDeadline(time.Time{})
	}

	// Both dial closures return the existing upstream connection so the
	// sniffer uses the same tunnel rather than dialling a new one.
	dial := func(ctx context.Context, network, address string) (net.Conn, error) {
		return cc, nil
	}
	dialTLS := func(ctx context.Context, network, address string, cfg *tls.Config) (net.Conn, error) {
		return cc, nil
	}
	sniffer := h.sniffer.Build()

	conn = xnet.NewReadWriteConn(br, conn, conn)
	switch proto {
	case sniffing.ProtoHTTP:
		return true, sniffer.HandleHTTP(ctx, "tcp", conn,
			sniffing.WithService(h.options.Service),
			sniffing.WithDial(dial),
			sniffing.WithDialTLS(dialTLS),
			sniffing.WithRecorderObject(ro),
			sniffing.WithLog(log),
		)
	case sniffing.ProtoTLS:
		return true, sniffer.HandleTLS(ctx, "tcp", conn,
			sniffing.WithService(h.options.Service),
			sniffing.WithDial(dial),
			sniffing.WithDialTLS(dialTLS),
			sniffing.WithRecorderObject(ro),
			sniffing.WithLog(log),
		)
	}

	return false, nil
}

// dial establishes an upstream connection through the proxy chain router.
// It attaches the recorder object and logger from the context so that the
// router can annotate the route path and source/destination addresses.
//
// When h.md.hash is "host", the target address is used as the hash source
// for consistent hop selection across the chain.
func (h *httpHandler) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) {
	switch h.md.hash {
	case "host":
		ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: addr})
	}

	if log := ictx.LoggerFromContext(ctx); log != nil {
		log.Debugf("dial: new connection to host %s", addr)
	}

	if h.options.Router == nil {
		return nil, &net.OpError{Op: "dial", Net: network, Addr: nil, Err: errors.New("nil router")}
	}

	var buf bytes.Buffer
	conn, err = h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), network, addr)
	if ro := ictx.RecorderObjectFromContext(ctx); ro != nil {
		ro.Route = buf.String()

		if conn != nil {
			ro.SrcAddr = conn.LocalAddr().String()
			ro.DstAddr = conn.RemoteAddr().String()
		}
	}

	return
}

// setupTrafficLimiter wraps the connection with per-client traffic shaping
// and optional stats tracking. When an Observer is configured, per-client
// stats counters (total connections, current connections) are updated.
//
// The returned net.Conn reads and writes through the limiter and stats
// wrappers while still satisfying the net.Conn interface.
//
// The cleanup function (non-nil when an Observer is configured) must be
// deferred by the caller to decrement KindCurrentConns when the connection
// handling completes. It captures the per-client stats object so the
// deferred call fires in the caller's scope, not when setupTrafficLimiter
// returns.
func (h *httpHandler) setupTrafficLimiter(conn net.Conn, clientID, network, addr string) (net.Conn, func()) {
	rw := traffic_wrapper.WrapReadWriter(
		h.limiter,
		conn,
		clientID,
		limiter.ScopeOption(limiter.ScopeClient),
		limiter.ServiceOption(h.options.Service),
		limiter.NetworkOption(network),
		limiter.AddrOption(addr),
		limiter.ClientOption(clientID),
		limiter.SrcOption(conn.RemoteAddr().String()),
	)
	var cleanup func()
	if h.options.Observer != nil {
		pstats := h.stats.Stats(clientID)
		pstats.Add(stats.KindTotalConns, 1)
		pstats.Add(stats.KindCurrentConns, 1)
		cleanup = func() { pstats.Add(stats.KindCurrentConns, -1) }
		rw = stats_wrapper.WrapReadWriter(rw, pstats)
	}

	return xnet.NewReadWriteConn(rw, rw, conn), cleanup
}

// SnifferBuilder holds all configuration needed to construct a sniffing.Sniffer.
// It is populated once during Init and reused for each sniffed connection.
type SnifferBuilder struct {
	Websocket           bool
	WebsocketSampleRate float64
	Recorder            recorder.Recorder
	RecorderOptions     *recorder.Options
	Certificate         *x509.Certificate
	PrivateKey          crypto.PrivateKey
	ALPN                string
	CertPool            tls_util.CertPool
	MitmBypass          bypass.Bypass
	ReadTimeout         time.Duration
}

// Build creates a new sniffing.Sniffer from the builder's configuration.
func (b *SnifferBuilder) Build() *sniffing.Sniffer {
	return &sniffing.Sniffer{
		Websocket:           b.Websocket,
		WebsocketSampleRate: b.WebsocketSampleRate,
		Recorder:            b.Recorder,
		RecorderOptions:     b.RecorderOptions,
		Certificate:         b.Certificate,
		PrivateKey:          b.PrivateKey,
		NegotiatedProtocol:  b.ALPN,
		CertPool:            b.CertPool,
		MitmBypass:          b.MitmBypass,
		ReadTimeout:         b.ReadTimeout,
	}
}