Add MASQUE/CONNECT-UDP support (RFC 9298)
Implement UDP tunneling over HTTP/3 using HTTP Datagrams (RFC 9297): - Add masque handler for server-side CONNECT-UDP - Add masque connector and h3-masque dialer for client-side - Add enableDatagrams option to HTTP/3 listener - Add shared utilities for datagram connections and path parsing Resource management and connection caching: - Add deferred stream cleanup in connector on error paths - Add IsClosed() and Close() methods to Client for proper session management - Clean up stale cached clients in dialer before reuse - Close underlying stream when DatagramConn is closed - Move RequestStream opening from connector to dialer to enable dead connection detection and cache invalidation (follows QUIC dialer pattern)
This commit is contained in:
David Manouchehri
committed by
ginuerzh
ginuerzh
parent
b3b5986b63
commit
7625973ca1
@@ -0,0 +1,467 @@
|
||||
package masque
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/auth"
|
||||
"github.com/go-gost/core/bypass"
|
||||
"github.com/go-gost/core/handler"
|
||||
"github.com/go-gost/core/limiter"
|
||||
"github.com/go-gost/core/limiter/traffic"
|
||||
"github.com/go-gost/core/logger"
|
||||
md "github.com/go-gost/core/metadata"
|
||||
"github.com/go-gost/core/observer"
|
||||
"github.com/go-gost/core/observer/stats"
|
||||
"github.com/go-gost/core/recorder"
|
||||
xbypass "github.com/go-gost/x/bypass"
|
||||
xctx "github.com/go-gost/x/ctx"
|
||||
ictx "github.com/go-gost/x/internal/ctx"
|
||||
"github.com/go-gost/x/internal/net/udp"
|
||||
masque_util "github.com/go-gost/x/internal/util/masque"
|
||||
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"
|
||||
traffic_wrapper "github.com/go-gost/x/limiter/traffic/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"
|
||||
"github.com/go-gost/x/registry"
|
||||
"github.com/quic-go/quic-go/http3"
|
||||
)
|
||||
|
||||
func init() {
|
||||
registry.HandlerRegistry().Register("masque", NewHandler)
|
||||
}
|
||||
|
||||
var (
|
||||
ErrBadRequest = errors.New("masque: bad request")
|
||||
ErrMethodNotAllowed = errors.New("masque: method not allowed")
|
||||
ErrCapsuleRequired = errors.New("masque: capsule-protocol header required")
|
||||
ErrDatagramNotSupport = errors.New("masque: datagrams not supported")
|
||||
)
|
||||
|
||||
type masqueHandler struct {
|
||||
md metadata
|
||||
options handler.Options
|
||||
stats *stats_util.HandlerStats
|
||||
limiter traffic.TrafficLimiter
|
||||
cancel context.CancelFunc
|
||||
recorder recorder.RecorderObject
|
||||
}
|
||||
|
||||
func NewHandler(opts ...handler.Option) handler.Handler {
|
||||
options := handler.Options{}
|
||||
for _, opt := range opts {
|
||||
opt(&options)
|
||||
}
|
||||
|
||||
return &masqueHandler{
|
||||
options: options,
|
||||
}
|
||||
}
|
||||
|
||||
func (h *masqueHandler) Init(md md.Metadata) error {
|
||||
if err := h.parseMetadata(md); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
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),
|
||||
)
|
||||
}
|
||||
|
||||
for _, ro := range h.options.Recorders {
|
||||
if ro.Record == xrecorder.RecorderServiceHandler {
|
||||
h.recorder = ro
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *masqueHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) {
|
||||
defer conn.Close()
|
||||
|
||||
start := time.Now()
|
||||
|
||||
ro := &xrecorder.HandlerRecorderObject{
|
||||
Network: "udp",
|
||||
Service: h.options.Service,
|
||||
RemoteAddr: conn.RemoteAddr().String(),
|
||||
LocalAddr: conn.LocalAddr().String(),
|
||||
Proto: "masque",
|
||||
Time: start,
|
||||
SID: xctx.SidFromContext(ctx).String(),
|
||||
}
|
||||
|
||||
if srcAddr := xctx.SrcAddrFromContext(ctx); srcAddr != nil {
|
||||
ro.ClientAddr = srcAddr.String()
|
||||
}
|
||||
|
||||
log := h.options.Logger.WithFields(map[string]any{
|
||||
"network": ro.Network,
|
||||
"remote": conn.RemoteAddr().String(),
|
||||
"local": conn.LocalAddr().String(),
|
||||
"client": ro.ClientAddr,
|
||||
"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(start)
|
||||
if h.recorder.Recorder != nil {
|
||||
if err := ro.Record(ctx, h.recorder.Recorder); err != nil {
|
||||
log.Errorf("record: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
log.WithFields(map[string]any{
|
||||
"duration": time.Since(start),
|
||||
"inputBytes": ro.InputBytes,
|
||||
"outputBytes": ro.OutputBytes,
|
||||
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
|
||||
}()
|
||||
|
||||
if !h.checkRateLimit(conn.RemoteAddr()) {
|
||||
return rate_limiter.ErrRateLimit
|
||||
}
|
||||
|
||||
// Extract request and response from context metadata
|
||||
ctxMd := ictx.MetadataFromContext(ctx)
|
||||
if ctxMd == nil {
|
||||
err := errors.New("masque: wrong connection type, requires HTTP/3")
|
||||
log.Error(err)
|
||||
return err
|
||||
}
|
||||
|
||||
w, _ := ctxMd.Get("w").(http.ResponseWriter)
|
||||
r, _ := ctxMd.Get("r").(*http.Request)
|
||||
|
||||
if w == nil || r == nil {
|
||||
return ErrBadRequest
|
||||
}
|
||||
|
||||
return h.handleConnectUDP(ctx, w, r, conn.LocalAddr(), ro, log)
|
||||
}
|
||||
|
||||
func (h *masqueHandler) Close() error {
|
||||
if h.cancel != nil {
|
||||
h.cancel()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *masqueHandler) handleConnectUDP(ctx context.Context, w http.ResponseWriter, r *http.Request, laddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
|
||||
// Validate Extended CONNECT request
|
||||
if r.Method != http.MethodConnect {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
log.Error(ErrMethodNotAllowed)
|
||||
return ErrMethodNotAllowed
|
||||
}
|
||||
|
||||
// Check for :protocol pseudo-header (Extended CONNECT)
|
||||
// In quic-go, the :protocol pseudo-header is stored in r.Proto
|
||||
if r.Proto != "connect-udp" {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
log.Error("masque: expected :protocol=connect-udp, got: ", r.Proto)
|
||||
return ErrBadRequest
|
||||
}
|
||||
|
||||
// Validate capsule-protocol header
|
||||
if r.Header.Get("Capsule-Protocol") != "?1" {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
log.Error(ErrCapsuleRequired)
|
||||
return ErrCapsuleRequired
|
||||
}
|
||||
|
||||
// Extract user for logging
|
||||
if u, _, _ := h.basicProxyAuth(r.Header.Get("Proxy-Authorization")); u != "" {
|
||||
log = log.WithFields(map[string]any{"user": u})
|
||||
ro.ClientID = u
|
||||
}
|
||||
|
||||
// Authenticate
|
||||
clientID, ok := h.authenticate(ctx, w, r, log)
|
||||
if !ok {
|
||||
return errors.New("authentication failed")
|
||||
}
|
||||
|
||||
if clientID != "" {
|
||||
log = log.WithFields(map[string]any{"clientID": clientID})
|
||||
ro.ClientID = clientID
|
||||
}
|
||||
ctx = xctx.ContextWithClientID(ctx, xctx.ClientID(clientID))
|
||||
|
||||
// Parse target from path
|
||||
host, port, err := masque_util.ParseMasquePath(r.URL.Path)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
log.Error("masque: invalid path: ", err)
|
||||
return err
|
||||
}
|
||||
|
||||
targetAddr := fmt.Sprintf("%s:%d", host, port)
|
||||
ro.Host = targetAddr
|
||||
log = log.WithFields(map[string]any{
|
||||
"target": targetAddr,
|
||||
})
|
||||
log.Debug("connect-udp request to ", targetAddr)
|
||||
|
||||
// Check bypass
|
||||
if h.options.Bypass != nil &&
|
||||
h.options.Bypass.Contains(ctx, "udp", targetAddr, bypass.WithService(h.options.Service)) {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
log.Debug("bypass: ", targetAddr)
|
||||
return xbypass.ErrBypass
|
||||
}
|
||||
|
||||
// Get HTTP/3 stream for datagrams
|
||||
streamer, ok := w.(http3.HTTPStreamer)
|
||||
if !ok {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
log.Error(ErrDatagramNotSupport)
|
||||
return ErrDatagramNotSupport
|
||||
}
|
||||
|
||||
// Send success response with capsule-protocol header
|
||||
w.Header().Set("Capsule-Protocol", "?1")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
// Flush the response headers
|
||||
if flusher, ok := w.(http.Flusher); ok {
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
// Get the underlying HTTP/3 stream
|
||||
stream := streamer.HTTPStream()
|
||||
|
||||
// Resolve target address
|
||||
raddr, err := net.ResolveUDPAddr("udp", targetAddr)
|
||||
if err != nil {
|
||||
log.Error("masque: failed to resolve target address: ", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Create datagram connection wrapping the HTTP/3 stream (client side)
|
||||
datagramConn := masque_util.NewDatagramConn(stream, laddr, raddr)
|
||||
defer datagramConn.Close()
|
||||
|
||||
// Wrap with traffic limiter
|
||||
var clientPC net.PacketConn = datagramConn
|
||||
clientPC = traffic_wrapper.WrapPacketConn(
|
||||
clientPC,
|
||||
h.limiter,
|
||||
clientID,
|
||||
limiter.ServiceOption(h.options.Service),
|
||||
limiter.ScopeOption(limiter.ScopeClient),
|
||||
limiter.NetworkOption("udp"),
|
||||
limiter.AddrOption(targetAddr),
|
||||
limiter.ClientOption(clientID),
|
||||
limiter.SrcOption(ro.RemoteAddr),
|
||||
)
|
||||
|
||||
// Track per-client connection stats
|
||||
if h.options.Observer != nil {
|
||||
pstats := h.stats.Stats(clientID)
|
||||
pstats.Add(stats.KindTotalConns, 1)
|
||||
pstats.Add(stats.KindCurrentConns, 1)
|
||||
defer pstats.Add(stats.KindCurrentConns, -1)
|
||||
clientPC = stats_wrapper.WrapPacketConn(clientPC, pstats)
|
||||
}
|
||||
|
||||
// Get target connection - either through router/chain or direct
|
||||
var targetPC net.PacketConn
|
||||
|
||||
switch h.md.hash {
|
||||
case "host":
|
||||
ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: targetAddr})
|
||||
}
|
||||
|
||||
if h.options.Router != nil {
|
||||
// Use router to dial through chain (for forwarding through upstream proxy)
|
||||
c, err := h.options.Router.Dial(ctx, "udp", targetAddr)
|
||||
if err != nil {
|
||||
log.Error("masque: failed to dial through router: ", err)
|
||||
return err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
// The connection from router should be a PacketConn (e.g., from masque connector)
|
||||
if pc, ok := c.(net.PacketConn); ok {
|
||||
targetPC = pc
|
||||
log.Debugf("relaying UDP to %s via chain", targetAddr)
|
||||
} else {
|
||||
// Wrap as PacketConn if it's a regular Conn
|
||||
targetPC = &connPacketConn{Conn: c, raddr: raddr}
|
||||
log.Debugf("relaying UDP to %s via chain (wrapped)", targetAddr)
|
||||
}
|
||||
} else {
|
||||
// Direct connection - create local UDP socket
|
||||
directConn, err := net.ListenPacket("udp", "")
|
||||
if err != nil {
|
||||
log.Error("masque: failed to create UDP socket: ", err)
|
||||
return err
|
||||
}
|
||||
defer directConn.Close()
|
||||
|
||||
// Wrap with fixed target address
|
||||
targetPC = &fixedTargetPacketConn{
|
||||
PacketConn: directConn,
|
||||
target: raddr,
|
||||
}
|
||||
log.Debugf("relaying UDP to %s directly via %s", raddr, directConn.LocalAddr())
|
||||
}
|
||||
|
||||
// Wrap target with metrics
|
||||
targetPC = metrics.WrapPacketConn(h.options.Service, targetPC)
|
||||
|
||||
// Relay UDP packets between client and target
|
||||
relay := udp.NewRelay(clientPC, targetPC).
|
||||
WithService(h.options.Service).
|
||||
WithLogger(log).
|
||||
WithBufferSize(h.md.bufferSize)
|
||||
|
||||
return relay.Run(ctx)
|
||||
}
|
||||
|
||||
// fixedTargetPacketConn wraps a PacketConn and always reads/writes to a fixed target address
|
||||
type fixedTargetPacketConn struct {
|
||||
net.PacketConn
|
||||
target net.Addr
|
||||
}
|
||||
|
||||
func (c *fixedTargetPacketConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
|
||||
n, _, err = c.PacketConn.ReadFrom(b)
|
||||
return n, c.target, err
|
||||
}
|
||||
|
||||
func (c *fixedTargetPacketConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
|
||||
return c.PacketConn.WriteTo(b, c.target)
|
||||
}
|
||||
|
||||
// connPacketConn wraps a net.Conn as a net.PacketConn for use with the UDP relay.
|
||||
// This is used when the router returns a stream-based connection (like from masque connector).
|
||||
type connPacketConn struct {
|
||||
net.Conn
|
||||
raddr net.Addr
|
||||
}
|
||||
|
||||
func (c *connPacketConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
|
||||
n, err = c.Conn.Read(b)
|
||||
return n, c.raddr, err
|
||||
}
|
||||
|
||||
func (c *connPacketConn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
|
||||
return c.Conn.Write(b)
|
||||
}
|
||||
|
||||
func (h *masqueHandler) 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 *masqueHandler) 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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *masqueHandler) basicProxyAuth(proxyAuth string) (username, password string, ok bool) {
|
||||
if proxyAuth == "" {
|
||||
return
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(proxyAuth, "Basic ") {
|
||||
return
|
||||
}
|
||||
c, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(proxyAuth, "Basic "))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
cs := string(c)
|
||||
s := strings.IndexByte(cs, ':')
|
||||
if s < 0 {
|
||||
return
|
||||
}
|
||||
|
||||
return cs[:s], cs[s+1:], true
|
||||
}
|
||||
|
||||
func (h *masqueHandler) authenticate(ctx context.Context, w http.ResponseWriter, r *http.Request, log logger.Logger) (id string, ok bool) {
|
||||
u, p, _ := h.basicProxyAuth(r.Header.Get("Proxy-Authorization"))
|
||||
if h.options.Auther == nil {
|
||||
return "", true
|
||||
}
|
||||
if id, ok = h.options.Auther.Authenticate(ctx, u, p, auth.WithService(h.options.Service)); ok {
|
||||
return
|
||||
}
|
||||
|
||||
realm := defaultRealm
|
||||
if h.md.authBasicRealm != "" {
|
||||
realm = h.md.authBasicRealm
|
||||
}
|
||||
w.Header().Set("Proxy-Authenticate", fmt.Sprintf("Basic realm=\"%s\"", realm))
|
||||
w.WriteHeader(http.StatusProxyAuthRequired)
|
||||
log.Debug("proxy authentication required")
|
||||
return
|
||||
}
|
||||
Reference in New Issue
Block a user