init

go-gost/x/handler/tunnel/bind.go

@@ -0,0 +1,109 @@
+package tunnel
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"net"
+
+	"github.com/go-gost/core/ingress"
+	"github.com/go-gost/core/logger"
+	"github.com/go-gost/core/observer/stats"
+	"github.com/go-gost/core/sd"
+	"github.com/go-gost/relay"
+	"github.com/go-gost/x/internal/util/mux"
+	"github.com/google/uuid"
+)
+
+func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, log logger.Logger) (err error) {
+	resp := relay.Response{
+		Version: relay.Version1,
+		Status:  relay.StatusOK,
+	}
+
+	uuid, err := uuid.NewRandom()
+	if err != nil {
+		resp.Status = relay.StatusInternalServerError
+		resp.WriteTo(conn)
+		return
+	}
+	connectorID := relay.NewConnectorID(uuid[:])
+	if network == "udp" {
+		connectorID = relay.NewUDPConnectorID(uuid[:])
+	}
+	// copy weight from tunnelID
+	connectorID = connectorID.SetWeight(tunnelID.Weight())
+
+	v := md5.Sum([]byte(tunnelID.String()))
+	endpoint := hex.EncodeToString(v[:8])
+
+	host, port, _ := net.SplitHostPort(address)
+	if host == "" || h.md.ingress == nil {
+		host = endpoint
+	} else if host != endpoint {
+		if rule := h.md.ingress.GetRule(ctx, host); rule != nil && rule.Endpoint != tunnelID.String() {
+			host = endpoint
+		}
+	}
+	addr := net.JoinHostPort(host, port)
+
+	af := &relay.AddrFeature{}
+	err = af.ParseFrom(addr)
+	if err != nil {
+		log.Warn(err)
+	}
+	resp.Features = append(resp.Features, af,
+		&relay.TunnelFeature{
+			ID: connectorID,
+		},
+	)
+	resp.WriteTo(conn)
+
+	// Upgrade connection to multiplex session.
+	session, err := mux.ClientSession(conn, h.md.muxCfg)
+	if err != nil {
+		return
+	}
+
+	var stats stats.Stats
+	if h.stats != nil {
+		stats = h.stats.Stats(tunnelID.String())
+	}
+
+	c := NewConnector(connectorID, tunnelID, h.id, session, &ConnectorOptions{
+		service: h.options.Service,
+		sd:      h.md.sd,
+		stats:   stats,
+		limiter: h.limiter,
+	})
+
+	h.pool.Add(tunnelID, c, h.md.tunnelTTL)
+	if h.md.ingress != nil {
+		h.md.ingress.SetRule(ctx, &ingress.Rule{
+			Hostname: endpoint,
+			Endpoint: tunnelID.String(),
+		})
+		if host != "" {
+			h.md.ingress.SetRule(ctx, &ingress.Rule{
+				Hostname: host,
+				Endpoint: tunnelID.String(),
+			})
+		}
+	}
+	if h.md.sd != nil {
+		err := h.md.sd.Register(ctx, &sd.Service{
+			ID:      connectorID.String(),
+			Name:    tunnelID.String(),
+			Node:    h.id,
+			Network: network,
+			Address: h.md.entryPoint,
+		})
+		if err != nil {
+			h.log.Error(err)
+		}
+	}
+
+	log.Debugf("%s/%s: tunnel=%s, connector=%s, weight=%d established", addr, network, tunnelID, connectorID, connectorID.Weight())
+
+	return
+}

go-gost/x/handler/tunnel/connect.go

@@ -0,0 +1,125 @@
+package tunnel
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/go-gost/core/logger"
+	"github.com/go-gost/relay"
+	xnet "github.com/go-gost/x/internal/net"
+)
+
+func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, conn net.Conn, network, srcAddr string, dstAddr string, tunnelID relay.TunnelID, log logger.Logger) error {
+	log = log.WithFields(map[string]any{
+		"dst":    fmt.Sprintf("%s/%s", dstAddr, network),
+		"cmd":    "connect",
+		"tunnel": tunnelID.String(),
+		"host":   dstAddr,
+	})
+
+	resp := relay.Response{
+		Version: relay.Version1,
+		Status:  relay.StatusOK,
+	}
+
+	if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, network, dstAddr) {
+		log.Debug("bypass: ", dstAddr)
+		resp.Status = relay.StatusForbidden
+		_, err := resp.WriteTo(conn)
+		return err
+	}
+
+	host, _, _ := net.SplitHostPort(dstAddr)
+
+	var tid relay.TunnelID
+	if ingress := h.md.ingress; ingress != nil && host != "" {
+		if rule := ingress.GetRule(ctx, host); rule != nil {
+			tid = parseTunnelID(rule.Endpoint)
+		}
+	}
+
+	// visitor is a public entrypoint.
+	if tunnelID.Equal(h.md.entryPointID) {
+		if tid.IsZero() {
+			resp.Status = relay.StatusNetworkUnreachable
+			resp.WriteTo(conn)
+			err := fmt.Errorf("no route to host %s", host)
+			log.Error(err)
+			return err
+		}
+
+		if tid.IsPrivate() {
+			resp.Status = relay.StatusHostUnreachable
+			resp.WriteTo(conn)
+			err := fmt.Errorf("tunnel %s is private for host %s", tid, host)
+			log.Error(err)
+			return err
+		}
+	} else {
+		// direct routing
+		if h.md.directTunnel {
+			tid = tunnelID
+		}
+		if !tid.Equal(tunnelID) {
+			resp.Status = relay.StatusHostUnreachable
+			resp.WriteTo(conn)
+			err := fmt.Errorf("no route to host %s", host)
+			log.Error(err)
+			return err
+		}
+	}
+
+	d := Dialer{
+		node:    h.id,
+		pool:    h.pool,
+		sd:      h.md.sd,
+		retry:   3,
+		timeout: 15 * time.Second,
+		log:     log,
+	}
+	cc, node, cid, err := d.Dial(ctx, network, tid.String())
+	if err != nil {
+		log.Error(err)
+		resp.Status = relay.StatusServiceUnavailable
+		resp.WriteTo(conn)
+		return err
+	}
+	defer cc.Close()
+
+	log.Debugf("connect to node=%s tunnel=%s connector=%s OK", node, tid, cid)
+
+	if node == h.id {
+		if _, err := resp.WriteTo(conn); err != nil {
+			log.Error(err)
+			return err
+		}
+
+		resp = relay.Response{
+			Version: relay.Version1,
+			Status:  relay.StatusOK,
+		}
+
+		af := &relay.AddrFeature{}
+		af.ParseFrom(srcAddr)
+		resp.Features = append(resp.Features, af) // src address
+
+		af = &relay.AddrFeature{}
+		af.ParseFrom(dstAddr)
+		resp.Features = append(resp.Features, af) // dst address
+
+		resp.WriteTo(cc)
+	} else {
+		req.WriteTo(cc)
+	}
+
+	t := time.Now()
+	log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
+	xnet.Transport(conn, cc)
+	log.WithFields(map[string]any{
+		"duration": time.Since(t),
+	}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
+
+	return nil
+}

go-gost/x/handler/tunnel/dialer.go

@@ -0,0 +1,78 @@
+package tunnel
+
+import (
+	"context"
+	"net"
+	"time"
+
+	"github.com/go-gost/core/logger"
+	"github.com/go-gost/core/sd"
+)
+
+type Dialer struct {
+	node    string
+	pool    *ConnectorPool
+	sd      sd.SD
+	retry   int
+	timeout time.Duration
+	log     logger.Logger
+}
+
+func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net.Conn, node string, cid string, err error) {
+	retry := d.retry
+	if retry <= 0 {
+		retry = 1
+	}
+
+	for i := 0; i < retry; i++ {
+		c := d.pool.Get(network, tid)
+		if c == nil {
+			break
+		}
+
+		conn, err = c.GetConn()
+		if err != nil {
+			d.log.Error(err)
+			continue
+		}
+		node = d.node
+		cid = c.id.String()
+
+		break
+	}
+	if conn != nil || err != nil {
+		return
+	}
+
+	if d.sd == nil {
+		err = ErrTunnelNotAvailable
+		return
+	}
+
+	ss, err := d.sd.Get(ctx, tid)
+	if err != nil {
+		return
+	}
+
+	var service *sd.Service
+	for _, s := range ss {
+		d.log.Debugf("%+v", s)
+		if s.Node != d.node && s.Network == network {
+			service = s
+			break
+		}
+	}
+	if service == nil || service.Address == "" {
+		err = ErrTunnelNotAvailable
+		return
+	}
+
+	node = service.Node
+	cid = service.ID
+
+	dialer := net.Dialer{
+		Timeout: d.timeout,
+	}
+	conn, err = dialer.DialContext(ctx, "tcp", service.Address)
+	return
+}

go-gost/x/handler/tunnel/entrypoint.go

@@ -0,0 +1,759 @@
+package tunnel
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-gost/core/handler"
+	"github.com/go-gost/core/ingress"
+	"github.com/go-gost/core/listener"
+	"github.com/go-gost/core/logger"
+	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer/stats"
+	"github.com/go-gost/core/recorder"
+	"github.com/go-gost/core/sd"
+	"github.com/go-gost/relay"
+	dissector "github.com/go-gost/tls-dissector"
+	admission "github.com/go-gost/x/admission/wrapper"
+	ctxvalue "github.com/go-gost/x/ctx"
+	ctx_internal "github.com/go-gost/x/internal/ctx"
+	xio "github.com/go-gost/x/internal/io"
+	xnet "github.com/go-gost/x/internal/net"
+	xhttp "github.com/go-gost/x/internal/net/http"
+	"github.com/go-gost/x/internal/net/proxyproto"
+	"github.com/go-gost/x/internal/util/sniffing"
+	tls_util "github.com/go-gost/x/internal/util/tls"
+	ws_util "github.com/go-gost/x/internal/util/ws"
+	climiter "github.com/go-gost/x/limiter/conn/wrapper"
+	metrics "github.com/go-gost/x/metrics/wrapper"
+	xstats "github.com/go-gost/x/observer/stats"
+	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
+	xrecorder "github.com/go-gost/x/recorder"
+	"golang.org/x/net/http/httpguts"
+	"golang.org/x/time/rate"
+)
+
+const (
+	httpHeaderSID           = "Gost-Sid"
+	httpHeaderForwardedNode = "Gost-Forwarded-Node"
+)
+
+type entrypoint struct {
+	node      string
+	service   string
+	pool      *ConnectorPool
+	ingress   ingress.Ingress
+	sd        sd.SD
+	log       logger.Logger
+	recorder  recorder.RecorderObject
+	transport http.RoundTripper
+
+	sniffingWebsocket   bool
+	websocketSampleRate float64
+
+	readTimeout time.Duration
+}
+
+func (ep *entrypoint) Handle(ctx context.Context, conn net.Conn) (err error) {
+	defer conn.Close()
+
+	ro := &xrecorder.HandlerRecorderObject{
+		Node:       ep.node,
+		Service:    ep.service,
+		RemoteAddr: conn.RemoteAddr().String(),
+		LocalAddr:  conn.LocalAddr().String(),
+		Network:    "tcp",
+		Time:       time.Now(),
+		SID:        string(ctxvalue.SidFromContext(ctx)),
+	}
+	ro.ClientIP, _, _ = net.SplitHostPort(conn.RemoteAddr().String())
+
+	log := ep.log.WithFields(map[string]any{
+		"remote": conn.RemoteAddr().String(),
+		"local":  conn.LocalAddr().String(),
+		"sid":    ro.SID,
+	})
+	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
+
+	pStats := xstats.Stats{}
+	conn = stats_wrapper.WrapConn(conn, &pStats)
+
+	defer func() {
+		if err != nil {
+			ro.Err = err.Error()
+		}
+		ro.InputBytes = pStats.Get(stats.KindInputBytes)
+		ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
+		ro.Duration = time.Since(ro.Time)
+		if err := ro.Record(ctx, ep.recorder.Recorder); err != nil {
+			log.Errorf("record: %v", err)
+		}
+
+		log.WithFields(map[string]any{
+			"duration":    time.Since(ro.Time),
+			"inputBytes":  ro.InputBytes,
+			"outputBytes": ro.OutputBytes,
+		}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
+	}()
+
+	br := bufio.NewReader(conn)
+	v, err := br.Peek(1)
+	if err != nil {
+		return err
+	}
+
+	conn = xnet.NewReadWriteConn(br, conn, conn)
+	if v[0] == relay.Version1 {
+		return ep.handleConnect(ctx, conn, ro, log)
+	}
+	if v[0] == dissector.Handshake {
+		return ep.HandleTLS(ctx, conn, ro, log)
+	}
+	return ep.handleHTTP(ctx, conn, ro, log)
+}
+
+func (ep *entrypoint) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) {
+	var tunnelID relay.TunnelID
+	if ep.ingress != nil {
+		if rule := ep.ingress.GetRule(ctx, addr); rule != nil {
+			tunnelID = parseTunnelID(rule.Endpoint)
+		}
+	}
+
+	log := ctxvalue.LoggerFromContext(ctx)
+	if log == nil {
+		log = ep.log
+	}
+	log.Debugf("dial: new connection to host %s", addr)
+
+	if tunnelID.IsZero() {
+		return nil, fmt.Errorf("%w %s", ErrTunnelRoute, addr)
+	}
+
+	ro := ctx_internal.RecorderObjectFromContext(ctx)
+	ro.ClientID = tunnelID.String()
+
+	if tunnelID.IsPrivate() {
+		return nil, fmt.Errorf("%w: tunnel %s is private for host %s", ErrPrivateTunnel, tunnelID, addr)
+	}
+
+	log = log.WithFields(map[string]any{
+		"tunnel": tunnelID.String(),
+	})
+
+	d := &Dialer{
+		node:    ep.node,
+		pool:    ep.pool,
+		sd:      ep.sd,
+		retry:   3,
+		timeout: 15 * time.Second,
+		log:     log,
+	}
+	conn, node, cid, err := d.Dial(ctx, "tcp", tunnelID.String())
+	if err != nil {
+		return
+	}
+	log.Debugf("dial: connected to host %s, tunnel: %s, connector: %s", addr, tunnelID, cid)
+
+	if node == ep.node {
+		ro.Redirect = ""
+
+		clientAddr := ctxvalue.ClientAddrFromContext(ctx)
+		var features []relay.Feature
+		af := &relay.AddrFeature{}
+		af.ParseFrom(string(clientAddr))
+		features = append(features, af) // src address
+
+		af = &relay.AddrFeature{}
+		af.ParseFrom(addr)
+		features = append(features, af) // dst address
+
+		if _, err = (&relay.Response{
+			Version:  relay.Version1,
+			Status:   relay.StatusOK,
+			Features: features,
+		}).WriteTo(conn); err != nil {
+			conn.Close()
+			return nil, err
+		}
+	} else {
+		ro.Redirect = node
+	}
+
+	return conn, nil
+}
+
+func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
+	pStats := xstats.Stats{}
+	conn = stats_wrapper.WrapConn(conn, &pStats)
+
+	br := bufio.NewReader(conn)
+	req, err := http.ReadRequest(br)
+	if err != nil {
+		return err
+	}
+
+	if log.IsLevelEnabled(logger.TraceLevel) {
+		dump, _ := httputil.DumpRequest(req, false)
+		log.Trace(string(dump))
+	}
+
+	ro.HTTP = &xrecorder.HTTPRecorderObject{
+		Host:   req.Host,
+		Proto:  req.Proto,
+		Scheme: req.URL.Scheme,
+		Method: req.Method,
+		URI:    req.RequestURI,
+		Request: xrecorder.HTTPRequestRecorderObject{
+			ContentLength: req.ContentLength,
+			Header:        req.Header.Clone(),
+		},
+	}
+
+	ro.Time = time.Time{}
+
+	if err := ep.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), req, ro, &pStats, log); err != nil {
+		log.Error(err)
+		return err
+	}
+
+	for {
+		pStats.Reset()
+
+		req, err := http.ReadRequest(br)
+		if err != nil {
+			if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
+				return nil
+			}
+			return err
+		}
+
+		if log.IsLevelEnabled(logger.TraceLevel) {
+			dump, _ := httputil.DumpRequest(req, false)
+			log.Trace(string(dump))
+		}
+
+		if err := ep.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), req, ro, &pStats, log); err != nil {
+			return err
+		}
+	}
+}
+
+func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) {
+	ro2 := &xrecorder.HandlerRecorderObject{}
+	*ro2 = *ro
+	ro = ro2
+
+	if sid := req.Header.Get(httpHeaderSID); sid != "" {
+		ro.SID = sid
+	} else {
+		req.Header.Set(httpHeaderSID, ro.SID)
+	}
+	req.Header.Set(httpHeaderForwardedNode, ro.Node)
+
+	host := req.Host
+	if _, port, _ := net.SplitHostPort(host); port == "" {
+		host = net.JoinHostPort(strings.Trim(host, "[]"), "80")
+	}
+	ro.Host = host
+	ro.Time = time.Now()
+
+	log = log.WithFields(map[string]any{
+		"host": host,
+	})
+	log.Infof("%s <-> %s", ro.RemoteAddr, req.Host)
+	defer func() {
+		if err != nil {
+			ro.Err = err.Error()
+		}
+		ro.InputBytes = pStats.Get(stats.KindInputBytes)
+		ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
+		ro.Duration = time.Since(ro.Time)
+		if err := ro.Record(ctx, ep.recorder.Recorder); err != nil {
+			log.Errorf("record: %v", err)
+		}
+
+		log.WithFields(map[string]any{
+			"duration":    time.Since(ro.Time),
+			"inputBytes":  ro.InputBytes,
+			"outputBytes": ro.OutputBytes,
+		}).Infof("%s >-< %s", ro.RemoteAddr, req.Host)
+	}()
+
+	if req.URL.Scheme == "" {
+		req.URL.Scheme = "http"
+	}
+	if req.URL.Host == "" {
+		req.URL.Host = req.Host
+	}
+
+	ro.HTTP = &xrecorder.HTTPRecorderObject{
+		Host:   req.Host,
+		Proto:  req.Proto,
+		Scheme: req.URL.Scheme,
+		Method: req.Method,
+		URI:    req.RequestURI,
+		Request: xrecorder.HTTPRequestRecorderObject{
+			ContentLength: req.ContentLength,
+			Header:        req.Header.Clone(),
+		},
+	}
+
+	if clientIP := xhttp.GetClientIP(req); clientIP != nil {
+		ro.ClientIP = clientIP.String()
+	}
+
+	clientAddr := ro.RemoteAddr
+	if ro.ClientIP != "" {
+		if _, port, _ := net.SplitHostPort(ro.RemoteAddr); port != "" {
+			clientAddr = net.JoinHostPort(ro.ClientIP, port)
+		}
+	}
+
+	res := &http.Response{
+		ProtoMajor: req.ProtoMajor,
+		ProtoMinor: req.ProtoMinor,
+		Header:     http.Header{},
+		StatusCode: http.StatusServiceUnavailable,
+	}
+	ro.HTTP.StatusCode = res.StatusCode
+
+	var reqBody *xhttp.Body
+	if opts := ep.recorder.Options; opts != nil && opts.HTTPBody {
+		if req.Body != nil {
+			bodySize := opts.MaxBodySize
+			if bodySize <= 0 {
+				bodySize = sniffing.DefaultBodySize
+			}
+			if bodySize > sniffing.MaxBodySize {
+				bodySize = sniffing.MaxBodySize
+			}
+			reqBody = xhttp.NewBody(req.Body, bodySize)
+			req.Body = reqBody
+		}
+	}
+
+	ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr))
+	ctx = ctx_internal.ContextWithRecorderObject(ctx, ro)
+	ctx = ctxvalue.ContextWithLogger(ctx, log)
+
+	resp, err := ep.transport.RoundTrip(req.WithContext(ctx))
+
+	if reqBody != nil {
+		ro.HTTP.Request.Body = reqBody.Content()
+		ro.HTTP.Request.ContentLength = reqBody.Length()
+	}
+
+	if err != nil {
+		if errors.Is(err, ErrTunnelRoute) || errors.Is(err, ErrPrivateTunnel) {
+			res.StatusCode = http.StatusBadGateway
+			ro.HTTP.StatusCode = http.StatusBadGateway
+		}
+		res.Write(rw)
+		return
+	}
+	defer resp.Body.Close()
+
+	ro.HTTP.StatusCode = resp.StatusCode
+	ro.HTTP.Response.Header = resp.Header
+	ro.HTTP.Response.ContentLength = resp.ContentLength
+
+	if log.IsLevelEnabled(logger.TraceLevel) {
+		dump, _ := httputil.DumpResponse(resp, false)
+		log.Trace(string(dump))
+	}
+
+	if resp.StatusCode == http.StatusSwitchingProtocols {
+		return ep.handleUpgradeResponse(ctx, rw, req, resp, ro, log)
+	}
+
+	var respBody *xhttp.Body
+	if opts := ep.recorder.Options; opts != nil && opts.HTTPBody {
+		bodySize := opts.MaxBodySize
+		if bodySize <= 0 {
+			bodySize = sniffing.DefaultBodySize
+		}
+		if bodySize > sniffing.MaxBodySize {
+			bodySize = sniffing.MaxBodySize
+		}
+		respBody = xhttp.NewBody(resp.Body, bodySize)
+		resp.Body = respBody
+	}
+
+	err = resp.Write(rw)
+
+	if respBody != nil {
+		ro.HTTP.Response.Body = respBody.Content()
+		ro.HTTP.Response.ContentLength = respBody.Length()
+	}
+
+	if err != nil {
+		return fmt.Errorf("write response: %v", err)
+	}
+
+	return
+}
+
+func upgradeType(h http.Header) string {
+	if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") {
+		return ""
+	}
+	return h.Get("Upgrade")
+}
+
+func (ep *entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
+	reqUpType := upgradeType(req.Header)
+	resUpType := upgradeType(res.Header)
+	if !strings.EqualFold(reqUpType, resUpType) {
+		return fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType)
+	}
+
+	backConn, ok := res.Body.(io.ReadWriteCloser)
+	if !ok {
+		return fmt.Errorf("internal error: 101 switching protocols response with non-writable body")
+	}
+	defer backConn.Close()
+
+	res.Body = nil
+	if err := res.Write(rw); err != nil {
+		return fmt.Errorf("response write: %v", err)
+	}
+
+	if reqUpType == "websocket" && ep.sniffingWebsocket {
+		return ep.sniffingWebsocketFrame(ctx, rw, backConn, ro, log)
+	}
+
+	return xnet.Transport(rw, backConn)
+}
+
+func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
+	errc := make(chan error, 1)
+
+	sampleRate := ep.websocketSampleRate
+	if sampleRate == 0 {
+		sampleRate = sniffing.DefaultSampleRate
+	}
+	if sampleRate < 0 {
+		sampleRate = math.MaxFloat64
+	}
+
+	go func() {
+		ro2 := &xrecorder.HandlerRecorderObject{}
+		*ro2 = *ro
+		ro := ro2
+
+		limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate))
+
+		buf := &bytes.Buffer{}
+		for {
+			start := time.Now()
+
+			if err := ep.copyWebsocketFrame(cc, rw, buf, "client", ro); err != nil {
+				errc <- err
+				return
+			}
+
+			if limiter.Allow() {
+				ro.Duration = time.Since(start)
+				ro.Time = time.Now()
+				if err := ro.Record(ctx, ep.recorder.Recorder); err != nil {
+					log.Errorf("record: %v", err)
+				}
+			}
+		}
+	}()
+
+	go func() {
+		ro2 := &xrecorder.HandlerRecorderObject{}
+		*ro2 = *ro
+		ro := ro2
+
+		limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate))
+
+		buf := &bytes.Buffer{}
+		for {
+			start := time.Now()
+
+			if err := ep.copyWebsocketFrame(rw, cc, buf, "server", ro); err != nil {
+				errc <- err
+				return
+			}
+
+			if limiter.Allow() {
+				ro.Duration = time.Since(start)
+				ro.Time = time.Now()
+				if err := ro.Record(ctx, ep.recorder.Recorder); err != nil {
+					log.Errorf("record: %v", err)
+				}
+			}
+		}
+	}()
+
+	<-errc
+	return nil
+}
+
+func (ep *entrypoint) copyWebsocketFrame(w io.Writer, r io.Reader, buf *bytes.Buffer, from string, ro *xrecorder.HandlerRecorderObject) (err error) {
+	fr := ws_util.Frame{}
+	if _, err = fr.ReadFrom(r); err != nil {
+		return err
+	}
+
+	ws := &xrecorder.WebsocketRecorderObject{
+		From:    from,
+		Fin:     fr.Header.Fin,
+		Rsv1:    fr.Header.Rsv1,
+		Rsv2:    fr.Header.Rsv2,
+		Rsv3:    fr.Header.Rsv3,
+		OpCode:  int(fr.Header.OpCode),
+		Masked:  fr.Header.Masked,
+		MaskKey: fr.Header.MaskKey,
+		Length:  fr.Header.PayloadLength,
+	}
+	if opts := ep.recorder.Options; opts != nil && opts.HTTPBody {
+		bodySize := opts.MaxBodySize
+		if bodySize <= 0 {
+			bodySize = sniffing.DefaultBodySize
+		}
+		if bodySize > sniffing.MaxBodySize {
+			bodySize = sniffing.MaxBodySize
+		}
+
+		buf.Reset()
+		if _, err := io.Copy(buf, io.LimitReader(fr.Data, int64(bodySize))); err != nil {
+			return err
+		}
+		ws.Payload = buf.Bytes()
+	}
+
+	ro.Websocket = ws
+	length := uint64(fr.Header.Length()) + uint64(fr.Header.PayloadLength)
+	if from == "client" {
+		ro.InputBytes = length
+		ro.OutputBytes = 0
+	} else {
+		ro.InputBytes = 0
+		ro.OutputBytes = length
+	}
+
+	fr.Data = io.MultiReader(bytes.NewReader(buf.Bytes()), fr.Data)
+	if _, err := fr.WriteTo(w); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (ep *entrypoint) HandleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
+	buf := new(bytes.Buffer)
+	clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf))
+	if err != nil {
+		return err
+	}
+
+	ro.TLS = &xrecorder.TLSRecorderObject{
+		ServerName:  clientHello.ServerName,
+		ClientHello: hex.EncodeToString(buf.Bytes()),
+	}
+	if len(clientHello.SupportedProtos) > 0 {
+		ro.TLS.Proto = clientHello.SupportedProtos[0]
+	}
+
+	host := clientHello.ServerName
+	if host != "" {
+		if _, _, err := net.SplitHostPort(host); err != nil {
+			host = net.JoinHostPort(strings.Trim(host, "[]"), "443")
+		}
+		ro.Host = host
+	}
+
+	ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(ro.RemoteAddr))
+	ctx = ctx_internal.ContextWithRecorderObject(ctx, ro)
+	ctx = ctxvalue.ContextWithLogger(ctx, log)
+
+	cc, err := ep.dial(ctx, "tcp", host)
+	if err != nil {
+		return err
+	}
+	defer cc.Close()
+
+	if _, err := buf.WriteTo(cc); err != nil {
+		return err
+	}
+
+	xio.SetReadDeadline(cc, time.Now().Add(ep.readTimeout))
+	serverHello, err := dissector.ParseServerHello(io.TeeReader(cc, buf))
+	xio.SetReadDeadline(cc, time.Time{})
+
+	if serverHello != nil {
+		ro.TLS.CipherSuite = tls_util.CipherSuite(serverHello.CipherSuite).String()
+		ro.TLS.CompressionMethod = serverHello.CompressionMethod
+		if serverHello.Proto != "" {
+			ro.TLS.Proto = serverHello.Proto
+		}
+		if serverHello.Version > 0 {
+			ro.TLS.Version = tls_util.Version(serverHello.Version).String()
+		}
+	}
+
+	if buf.Len() > 0 {
+		ro.TLS.ServerHello = hex.EncodeToString(buf.Bytes())
+	}
+
+	if _, err := buf.WriteTo(conn); err != nil {
+		return err
+	}
+
+	xnet.Transport(conn, cc)
+	return err
+}
+
+func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
+	req := relay.Request{}
+	if _, err := req.ReadFrom(conn); err != nil {
+		return err
+	}
+
+	resp := relay.Response{
+		Version: relay.Version1,
+		Status:  relay.StatusOK,
+	}
+
+	var srcAddr, dstAddr string
+	network := "tcp"
+	var tunnelID relay.TunnelID
+	for _, f := range req.Features {
+		switch f.Type() {
+		case relay.FeatureAddr:
+			if feature, _ := f.(*relay.AddrFeature); feature != nil {
+				v := net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port)))
+				if srcAddr != "" {
+					dstAddr = v
+				} else {
+					srcAddr = v
+				}
+			}
+		case relay.FeatureTunnel:
+			if feature, _ := f.(*relay.TunnelFeature); feature != nil {
+				tunnelID = relay.NewTunnelID(feature.ID[:])
+			}
+		case relay.FeatureNetwork:
+			if feature, _ := f.(*relay.NetworkFeature); feature != nil {
+				network = feature.Network.String()
+			}
+		}
+	}
+
+	if tunnelID.IsZero() {
+		resp.Status = relay.StatusBadRequest
+		resp.WriteTo(conn)
+		return ErrTunnelID
+	}
+
+	ro.ClientID = tunnelID.String()
+
+	d := Dialer{
+		pool:    ep.pool,
+		retry:   3,
+		timeout: 15 * time.Second,
+		log:     log,
+	}
+	cc, _, cid, err := d.Dial(ctx, network, tunnelID.String())
+	if err != nil {
+		log.Error(err)
+		resp.Status = relay.StatusServiceUnavailable
+		resp.WriteTo(conn)
+		return err
+	}
+	defer cc.Close()
+
+	log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid)
+
+	if _, err := resp.WriteTo(conn); err != nil {
+		log.Error(err)
+		return err
+	}
+
+	af := &relay.AddrFeature{}
+	af.ParseFrom(srcAddr)
+	resp.Features = append(resp.Features, af) // src address
+
+	af = &relay.AddrFeature{}
+	af.ParseFrom(dstAddr)
+	resp.Features = append(resp.Features, af) // dst address
+
+	resp.WriteTo(cc)
+
+	t := time.Now()
+	log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
+	xnet.Transport(conn, cc)
+	log.WithFields(map[string]any{
+		"duration": time.Since(t),
+	}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
+
+	return nil
+}
+
+type tcpListener struct {
+	ln      net.Listener
+	options listener.Options
+}
+
+func newTCPListener(ln net.Listener, opts ...listener.Option) listener.Listener {
+	options := listener.Options{}
+	for _, opt := range opts {
+		opt(&options)
+	}
+	return &tcpListener{
+		ln:      ln,
+		options: options,
+	}
+}
+
+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)
+	ln = admission.WrapListener(l.options.Admission, ln)
+	ln = climiter.WrapListener(l.options.ConnLimiter, ln)
+	l.ln = ln
+
+	return
+}
+
+func (l *tcpListener) Accept() (conn net.Conn, err error) {
+	return l.ln.Accept()
+}
+
+func (l *tcpListener) Addr() net.Addr {
+	return l.ln.Addr()
+}
+
+func (l *tcpListener) Close() error {
+	return l.ln.Close()
+}
+
+type entrypointHandler struct {
+	ep *entrypoint
+}
+
+func (h *entrypointHandler) Init(md md.Metadata) (err error) {
+	return
+}
+
+func (h *entrypointHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
+	return h.ep.Handle(ctx, conn)
+}

go-gost/x/handler/tunnel/handler.go

@@ -0,0 +1,354 @@
+package tunnel
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/go-gost/core/handler"
+	"github.com/go-gost/core/limiter"
+	"github.com/go-gost/core/limiter/traffic"
+	"github.com/go-gost/core/listener"
+	"github.com/go-gost/core/logger"
+	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
+	"github.com/go-gost/core/service"
+	"github.com/go-gost/relay"
+	ctxvalue "github.com/go-gost/x/ctx"
+	xnet "github.com/go-gost/x/internal/net"
+	stats_util "github.com/go-gost/x/internal/util/stats"
+	rate_limiter "github.com/go-gost/x/limiter/rate"
+	cache_limiter "github.com/go-gost/x/limiter/traffic/cache"
+	xrecorder "github.com/go-gost/x/recorder"
+	"github.com/go-gost/x/registry"
+	xservice "github.com/go-gost/x/service"
+	"github.com/google/uuid"
+)
+
+var (
+	ErrBadVersion         = errors.New("bad version")
+	ErrUnknownCmd         = errors.New("unknown command")
+	ErrTunnelID           = errors.New("invalid tunnel ID")
+	ErrTunnelNotAvailable = errors.New("tunnel not available")
+	ErrUnauthorized       = errors.New("unauthorized")
+	ErrTunnelRoute        = errors.New("no route to host")
+	ErrPrivateTunnel      = errors.New("private tunnel")
+)
+
+func init() {
+	registry.HandlerRegistry().Register("tunnel", NewHandler)
+}
+
+type tunnelHandler struct {
+	id      string
+	options handler.Options
+	pool    *ConnectorPool
+	epSvc   service.Service
+	ep      *entrypoint
+	md      metadata
+	log     logger.Logger
+	stats   *stats_util.HandlerStats
+	limiter traffic.TrafficLimiter
+	cancel  context.CancelFunc
+}
+
+func NewHandler(opts ...handler.Option) handler.Handler {
+	options := handler.Options{}
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	return &tunnelHandler{
+		options: options,
+	}
+}
+
+func (h *tunnelHandler) Init(md md.Metadata) (err error) {
+	if err := h.parseMetadata(md); err != nil {
+		return err
+	}
+
+	uuid, err := uuid.NewRandom()
+	if err != nil {
+		return err
+	}
+	h.id = uuid.String()
+
+	h.log = h.options.Logger.WithFields(map[string]any{
+		"node": h.id,
+	})
+	h.pool = NewConnectorPool(h.id)
+
+	h.ep = &entrypoint{
+		node:    h.id,
+		service: h.options.Service,
+		pool:    h.pool,
+		ingress: h.md.ingress,
+		sd:      h.md.sd,
+		log: h.log.WithFields(map[string]any{
+			"kind": "entrypoint",
+		}),
+		sniffingWebsocket:   h.md.sniffingWebsocket,
+		websocketSampleRate: h.md.sniffingWebsocketSampleRate,
+		readTimeout:         h.md.entryPointReadTimeout,
+	}
+	h.ep.transport = &http.Transport{
+		DialContext:           h.ep.dial,
+		IdleConnTimeout:       30 * time.Second,
+		ResponseHeaderTimeout: h.md.entryPointReadTimeout,
+		DisableKeepAlives:     !h.md.entryPointKeepalive,
+		DisableCompression:    !h.md.entryPointCompression,
+	}
+
+	for _, ro := range h.options.Recorders {
+		if ro.Record == xrecorder.RecorderServiceHandler {
+			h.ep.recorder = ro
+			break
+		}
+	}
+	if err = h.initEntrypoint(); err != nil {
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	h.cancel = cancel
+
+	if h.options.Observer != nil {
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
+		go h.observeStats(ctx)
+	}
+
+	if h.options.Limiter != nil {
+		h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter,
+			cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval),
+			cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval),
+			cache_limiter.ScopeOption(limiter.ScopeClient),
+		)
+	}
+
+	return nil
+}
+
+func (h *tunnelHandler) initEntrypoint() (err error) {
+	if h.md.entryPoint == "" {
+		return
+	}
+
+	network := "tcp"
+	if xnet.IsIPv4(h.md.entryPoint) {
+		network = "tcp4"
+	}
+
+	ln, err := net.Listen(network, h.md.entryPoint)
+	if err != nil {
+		h.log.Error(err)
+		return
+	}
+
+	serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr())
+	log := h.log.WithFields(map[string]any{
+		"service":  serviceName,
+		"listener": "tcp",
+		"handler":  "tunnel-ep",
+		"kind":     "service",
+	})
+	epListener := newTCPListener(ln,
+		listener.AddrOption(h.md.entryPoint),
+		listener.ServiceOption(serviceName),
+		listener.ProxyProtocolOption(h.md.entryPointProxyProtocol),
+		listener.LoggerOption(log.WithFields(map[string]any{
+			"kind": "listener",
+		})),
+	)
+	if err = epListener.Init(nil); err != nil {
+		return
+	}
+	epHandler := &entrypointHandler{
+		ep: h.ep,
+	}
+	if err = epHandler.Init(nil); err != nil {
+		return
+	}
+
+	h.epSvc = xservice.NewService(
+		serviceName, epListener, epHandler,
+		xservice.LoggerOption(log),
+	)
+	go h.epSvc.Serve()
+	log.Infof("entrypoint: %s", h.epSvc.Addr())
+
+	return
+}
+
+func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) {
+	start := time.Now()
+	log := h.log.WithFields(map[string]any{
+		"remote": conn.RemoteAddr().String(),
+		"local":  conn.LocalAddr().String(),
+		"sid":    ctxvalue.SidFromContext(ctx),
+	})
+
+	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
+
+	defer func() {
+		if err != nil {
+			conn.Close()
+		}
+		log.WithFields(map[string]any{
+			"duration": time.Since(start),
+		}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
+	}()
+
+	if !h.checkRateLimit(conn.RemoteAddr()) {
+		return rate_limiter.ErrRateLimit
+	}
+
+	if h.md.readTimeout > 0 {
+		conn.SetReadDeadline(time.Now().Add(h.md.readTimeout))
+	}
+
+	req := relay.Request{}
+	if _, err := req.ReadFrom(conn); err != nil {
+		return err
+	}
+
+	conn.SetReadDeadline(time.Time{})
+
+	resp := relay.Response{
+		Version: relay.Version1,
+		Status:  relay.StatusOK,
+	}
+
+	if req.Version != relay.Version1 {
+		resp.Status = relay.StatusBadRequest
+		resp.WriteTo(conn)
+		return ErrBadVersion
+	}
+
+	var user, pass string
+	var srcAddr, dstAddr string
+	network := "tcp"
+	var tunnelID relay.TunnelID
+	for _, f := range req.Features {
+		switch f.Type() {
+		case relay.FeatureUserAuth:
+			if feature, _ := f.(*relay.UserAuthFeature); feature != nil {
+				user, pass = feature.Username, feature.Password
+			}
+		case relay.FeatureAddr:
+			if feature, _ := f.(*relay.AddrFeature); feature != nil {
+				v := net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port)))
+				if srcAddr != "" {
+					dstAddr = v
+				} else {
+					srcAddr = v
+				}
+			}
+		case relay.FeatureTunnel:
+			if feature, _ := f.(*relay.TunnelFeature); feature != nil {
+				tunnelID = feature.ID
+			}
+		case relay.FeatureNetwork:
+			if feature, _ := f.(*relay.NetworkFeature); feature != nil {
+				network = feature.Network.String()
+			}
+		}
+	}
+
+	if tunnelID.IsZero() {
+		resp.Status = relay.StatusBadRequest
+		resp.WriteTo(conn)
+		return ErrTunnelID
+	}
+
+	if user != "" {
+		log = log.WithFields(map[string]any{"user": user})
+	}
+
+	if h.options.Auther != nil {
+		clientID, ok := h.options.Auther.Authenticate(ctx, user, pass)
+		if !ok {
+			resp.Status = relay.StatusUnauthorized
+			resp.WriteTo(conn)
+			return ErrUnauthorized
+		}
+		ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(clientID))
+	}
+
+	switch req.Cmd & relay.CmdMask {
+	case relay.CmdConnect:
+		defer conn.Close()
+
+		log.Debugf("connect: %s >> %s/%s", srcAddr, dstAddr, network)
+		return h.handleConnect(ctx, &req, conn, network, srcAddr, dstAddr, tunnelID, log)
+
+	case relay.CmdBind:
+		log.Debugf("bind: %s >> %s/%s", srcAddr, dstAddr, network)
+		return h.handleBind(ctx, conn, network, dstAddr, tunnelID, log)
+	default:
+		resp.Status = relay.StatusBadRequest
+		resp.WriteTo(conn)
+		return ErrUnknownCmd
+	}
+}
+
+// Close implements io.Closer interface.
+func (h *tunnelHandler) Close() error {
+	if h.epSvc != nil {
+		h.epSvc.Close()
+	}
+	h.pool.Close()
+
+	if h.cancel != nil {
+		h.cancel()
+	}
+
+	return nil
+}
+
+func (h *tunnelHandler) checkRateLimit(addr net.Addr) bool {
+	if h.options.RateLimiter == nil {
+		return true
+	}
+	host, _, _ := net.SplitHostPort(addr.String())
+	if limiter := h.options.RateLimiter.Limiter(host); limiter != nil {
+		return limiter.Allow(1)
+	}
+
+	return true
+}
+
+func (h *tunnelHandler) observeStats(ctx context.Context) {
+	if h.options.Observer == nil {
+		return
+	}
+
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}

go-gost/x/handler/tunnel/metadata.go

@@ -0,0 +1,125 @@
+package tunnel
+
+import (
+	"strings"
+	"time"
+
+	"github.com/go-gost/core/ingress"
+	"github.com/go-gost/core/logger"
+	mdata "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/sd"
+	"github.com/go-gost/relay"
+	xingress "github.com/go-gost/x/ingress"
+	"github.com/go-gost/x/internal/util/mux"
+	mdutil "github.com/go-gost/x/metadata/util"
+	"github.com/go-gost/x/registry"
+)
+
+const (
+	defaultTTL = 15 * time.Second
+)
+
+type metadata struct {
+	readTimeout time.Duration
+
+	entryPoint                  string
+	entryPointID                relay.TunnelID
+	entryPointProxyProtocol     int
+	entryPointKeepalive         bool
+	entryPointCompression       bool
+	entryPointReadTimeout       time.Duration
+	sniffingWebsocket           bool
+	sniffingWebsocketSampleRate float64
+
+	directTunnel           bool
+	tunnelTTL              time.Duration
+	ingress                ingress.Ingress
+	sd                     sd.SD
+	muxCfg                 *mux.Config
+
+	observerPeriod       time.Duration
+	observerResetTraffic bool
+
+	limiterRefreshInterval time.Duration
+	limiterCleanupInterval time.Duration
+}
+
+func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
+	h.md.readTimeout = mdutil.GetDuration(md, "readTimeout")
+
+	h.md.entryPoint = mdutil.GetString(md, "entrypoint")
+	h.md.entryPointID = parseTunnelID(mdutil.GetString(md, "entrypoint.id"))
+	h.md.entryPointProxyProtocol = mdutil.GetInt(md, "entrypoint.ProxyProtocol")
+
+	h.md.entryPointKeepalive = mdutil.GetBool(md, "entrypoint.keepalive")
+	h.md.entryPointCompression = mdutil.GetBool(md, "entrypoint.compression")
+
+	h.md.entryPointReadTimeout = mdutil.GetDuration(md, "entrypoint.readTimeout")
+	if h.md.entryPointReadTimeout <= 0 {
+		h.md.entryPointReadTimeout = 15 * time.Second
+	}
+
+	h.md.sniffingWebsocket = mdutil.GetBool(md, "sniffing.websocket")
+	h.md.sniffingWebsocketSampleRate = mdutil.GetFloat(md, "sniffing.websocket.sampleRate")
+
+	h.md.tunnelTTL = mdutil.GetDuration(md, "tunnel.ttl")
+	if h.md.tunnelTTL <= 0 {
+		h.md.tunnelTTL = defaultTTL
+	}
+	h.md.directTunnel = mdutil.GetBool(md, "tunnel.direct")
+
+	h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))
+	if h.md.ingress == nil {
+		var rules []*ingress.Rule
+		for _, s := range strings.Split(mdutil.GetString(md, "tunnel"), ",") {
+			ss := strings.SplitN(s, ":", 2)
+			if len(ss) != 2 {
+				continue
+			}
+			rules = append(rules, &ingress.Rule{
+				Hostname: ss[0],
+				Endpoint: ss[1],
+			})
+		}
+		if len(rules) > 0 {
+			h.md.ingress = xingress.NewIngress(
+				xingress.RulesOption(rules),
+				xingress.LoggerOption(logger.Default().WithFields(map[string]any{
+					"kind":    "ingress",
+					"ingress": "@internal",
+				})),
+			)
+		}
+	}
+	h.md.sd = registry.SDRegistry().Get(mdutil.GetString(md, "sd"))
+
+	h.md.muxCfg = &mux.Config{
+		Version:           mdutil.GetInt(md, "mux.version"),
+		KeepAliveInterval: mdutil.GetDuration(md, "mux.keepaliveInterval"),
+		KeepAliveDisabled: mdutil.GetBool(md, "mux.keepaliveDisabled"),
+		KeepAliveTimeout:  mdutil.GetDuration(md, "mux.keepaliveTimeout"),
+		MaxFrameSize:      mdutil.GetInt(md, "mux.maxFrameSize"),
+		MaxReceiveBuffer:  mdutil.GetInt(md, "mux.maxReceiveBuffer"),
+		MaxStreamBuffer:   mdutil.GetInt(md, "mux.maxStreamBuffer"),
+	}
+	if h.md.muxCfg.Version == 0 {
+		h.md.muxCfg.Version = 2
+	}
+	if h.md.muxCfg.MaxStreamBuffer == 0 {
+		h.md.muxCfg.MaxStreamBuffer = 1048576
+	}
+
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
+	if h.md.observerPeriod == 0 {
+		h.md.observerPeriod = 5 * time.Second
+	}
+	if h.md.observerPeriod < time.Second {
+		h.md.observerPeriod = time.Second
+	}
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
+
+	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
+	h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval")
+
+	return
+}

go-gost/x/handler/tunnel/tunnel.go

@@ -0,0 +1,394 @@
+package tunnel
+
+import (
+	"context"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/go-gost/core/limiter"
+	"github.com/go-gost/core/limiter/traffic"
+	"github.com/go-gost/core/logger"
+	"github.com/go-gost/core/sd"
+	"github.com/go-gost/relay"
+	"github.com/go-gost/x/internal/util/mux"
+
+	"github.com/go-gost/core/observer/stats"
+	traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper"
+	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
+	"github.com/go-gost/x/selector"
+	"github.com/google/uuid"
+)
+
+const (
+	MaxWeight uint8 = 0xff
+)
+
+type ConnectorOptions struct {
+	service string
+	sd      sd.SD
+	stats   stats.Stats
+	limiter traffic.TrafficLimiter
+}
+
+type Connector struct {
+	id   relay.ConnectorID
+	tid  relay.TunnelID
+	node string
+	s    *mux.Session
+	t    time.Time
+	opts *ConnectorOptions
+	log  logger.Logger
+}
+
+func NewConnector(id relay.ConnectorID, tid relay.TunnelID, node string, s *mux.Session, opts *ConnectorOptions) *Connector {
+	if opts == nil {
+		opts = &ConnectorOptions{}
+	}
+
+	c := &Connector{
+		id:   id,
+		tid:  tid,
+		node: node,
+		s:    s,
+		t:    time.Now(),
+		opts: opts,
+		log: logger.Default().WithFields(map[string]any{
+			"node":      node,
+			"tunnel":    tid.String(),
+			"connector": id.String(),
+		}),
+	}
+
+	go c.waitClose()
+	return c
+}
+
+func (c *Connector) waitClose() {
+	for {
+		conn, err := c.s.Accept()
+		if err != nil {
+			c.log.Errorf("connector %s: %v", c.id, err)
+			c.Close()
+			if c.opts.sd != nil {
+				c.opts.sd.Deregister(context.Background(), &sd.Service{
+					ID:   c.id.String(),
+					Name: c.tid.String(),
+					Node: c.node,
+				})
+				c.log.Debugf("deregister connector %s from sd", c.id.String())
+			}
+			return
+		}
+		conn.Close()
+	}
+}
+
+func (c *Connector) ID() relay.ConnectorID {
+	return c.id
+}
+
+func (c *Connector) GetConn() (net.Conn, error) {
+	if c == nil || c.s == nil {
+		return nil, nil
+	}
+
+	conn, err := c.s.GetConn()
+	if err != nil {
+		return nil, err
+	}
+
+	conn = stats_wrapper.WrapConn(conn, c.opts.stats)
+
+	network := "tcp"
+	if c.id.IsUDP() {
+		network = "udp"
+	}
+	conn = traffic_wrapper.WrapConn(
+		conn,
+		c.opts.limiter,
+		c.tid.String(),
+		limiter.ScopeOption(limiter.ScopeClient),
+		limiter.ServiceOption(c.opts.service),
+		limiter.ClientOption(c.tid.String()),
+		limiter.NetworkOption(network),
+		limiter.SrcOption(conn.RemoteAddr().String()),
+	)
+	return conn, nil
+}
+
+func (c *Connector) Close() error {
+	if c == nil || c.s == nil {
+		return nil
+	}
+
+	return c.s.Close()
+}
+
+func (c *Connector) IsClosed() bool {
+	if c == nil || c.s == nil {
+		return true
+	}
+
+	return c.s.IsClosed()
+}
+
+type Tunnel struct {
+	node       string
+	id         relay.TunnelID
+	connectors []*Connector
+	t          time.Time
+	close      chan struct{}
+	mu         sync.RWMutex
+	ttl        time.Duration
+}
+
+func NewTunnel(node string, tid relay.TunnelID, ttl time.Duration) *Tunnel {
+	t := &Tunnel{
+		node:  node,
+		id:    tid,
+		t:     time.Now(),
+		close: make(chan struct{}),
+		ttl:   ttl,
+	}
+	if t.ttl <= 0 {
+		t.ttl = defaultTTL
+	}
+	go t.clean()
+	return t
+}
+
+func (t *Tunnel) ID() relay.TunnelID {
+	return t.id
+}
+
+func (t *Tunnel) AddConnector(c *Connector) {
+	if c == nil {
+		return
+	}
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	t.connectors = append(t.connectors, c)
+}
+
+func (t *Tunnel) GetConnector(network string) *Connector {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	if len(t.connectors) == 1 {
+		if t.connectors[0].IsClosed() {
+			return nil
+		}
+		return t.connectors[0]
+	}
+
+	rw := selector.NewRandomWeighted[*Connector]()
+
+	found := false
+	for _, c := range t.connectors {
+		if c.IsClosed() {
+			continue
+		}
+
+		weight := c.ID().Weight()
+		if weight == 0 {
+			weight = 1
+		}
+
+		if network == "udp" && c.id.IsUDP() ||
+			network != "udp" && !c.id.IsUDP() {
+			if weight == MaxWeight && !found {
+				rw.Reset()
+				found = true
+			}
+
+			if weight == MaxWeight || !found {
+				rw.Add(c, int(weight))
+			}
+		}
+	}
+
+	return rw.Next()
+}
+
+func (t *Tunnel) Close() error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	select {
+	case <-t.close:
+	default:
+		for _, c := range t.connectors {
+			c.Close()
+		}
+		close(t.close)
+	}
+
+	return nil
+}
+
+func (t *Tunnel) CloseOnIdle() bool {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	select {
+	case <-t.close:
+	default:
+		if len(t.connectors) == 0 {
+			close(t.close)
+			return true
+		}
+	}
+	return false
+}
+
+func (t *Tunnel) clean() {
+	ticker := time.NewTicker(t.ttl)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			t.mu.Lock()
+			if len(t.connectors) == 0 {
+				t.mu.Unlock()
+				break
+			}
+
+			var connectors []*Connector
+			for _, c := range t.connectors {
+				if c.IsClosed() {
+					c.log.Debugf("remove connector: %s %s", t.id, c.id)
+					continue
+				}
+
+				connectors = append(connectors, c)
+				if c.opts.sd != nil {
+					c.opts.sd.Renew(context.Background(), &sd.Service{
+						ID:   c.id.String(),
+						Name: t.id.String(),
+						Node: t.node,
+					})
+				}
+			}
+			if len(connectors) != len(t.connectors) {
+				t.connectors = connectors
+			}
+			t.mu.Unlock()
+
+		case <-t.close:
+			return
+		}
+	}
+}
+
+type ConnectorPool struct {
+	node    string
+	tunnels map[string]*Tunnel
+	mu      sync.RWMutex
+	cancel  context.CancelFunc
+}
+
+func NewConnectorPool(node string) *ConnectorPool {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	p := &ConnectorPool{
+		node:    node,
+		tunnels: make(map[string]*Tunnel),
+		cancel:  cancel,
+	}
+
+	go p.closeIdles(ctx)
+	return p
+}
+
+func (p *ConnectorPool) Add(tid relay.TunnelID, c *Connector, ttl time.Duration) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	s := tid.String()
+
+	t := p.tunnels[s]
+	if t == nil {
+		t = NewTunnel(p.node, tid, ttl)
+		p.tunnels[s] = t
+	}
+	t.AddConnector(c)
+}
+
+func (p *ConnectorPool) Get(network string, tid string) *Connector {
+	if p == nil {
+		return nil
+	}
+
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	t := p.tunnels[tid]
+	if t == nil {
+		return nil
+	}
+
+	return t.GetConnector(network)
+}
+
+func (p *ConnectorPool) Close() error {
+	if p == nil {
+		return nil
+	}
+
+	if p.cancel != nil {
+		p.cancel()
+	}
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	for k, v := range p.tunnels {
+		v.Close()
+		delete(p.tunnels, k)
+	}
+
+	return nil
+}
+
+func (p *ConnectorPool) closeIdles(ctx context.Context) {
+	ticker := time.NewTicker(15 * time.Minute)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			p.mu.Lock()
+			for k, v := range p.tunnels {
+				if v.CloseOnIdle() {
+					delete(p.tunnels, k)
+					logger.Default().Debugf("remove idle tunnel: %s", k)
+				}
+			}
+			p.mu.Unlock()
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func parseTunnelID(s string) (tid relay.TunnelID) {
+	if s == "" {
+		return
+	}
+	private := false
+	if s[0] == '$' {
+		private = true
+		s = s[1:]
+	}
+	uuid, _ := uuid.Parse(s)
+
+	if private {
+		return relay.NewPrivateTunnelID(uuid[:])
+	}
+	return relay.NewTunnelID(uuid[:])
+}