From 1752b29df9ce43923669d4690c2d100517c8c391 Mon Sep 17 00:00:00 2001 From: David Manouchehri Date: Sun, 28 Dec 2025 21:13:30 +0000 Subject: [PATCH] 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) --- connector/masque/connector.go | 180 ++++++++++++ connector/masque/metadata.go | 17 ++ dialer/http3/masque/client.go | 150 ++++++++++ dialer/http3/masque/dialer.go | 136 ++++++++++ dialer/http3/masque/metadata.go | 43 +++ handler/masque/handler.go | 467 ++++++++++++++++++++++++++++++++ handler/masque/metadata.go | 49 ++++ internal/util/masque/conn.go | 176 ++++++++++++ internal/util/masque/path.go | 60 ++++ listener/http3/listener.go | 4 +- listener/http3/metadata.go | 2 + 11 files changed, 1283 insertions(+), 1 deletion(-) create mode 100644 connector/masque/connector.go create mode 100644 connector/masque/metadata.go create mode 100644 dialer/http3/masque/client.go create mode 100644 dialer/http3/masque/dialer.go create mode 100644 dialer/http3/masque/metadata.go create mode 100644 handler/masque/handler.go create mode 100644 handler/masque/metadata.go create mode 100644 internal/util/masque/conn.go create mode 100644 internal/util/masque/path.go diff --git a/connector/masque/connector.go b/connector/masque/connector.go new file mode 100644 index 00000000..9a2b8971 --- /dev/null +++ b/connector/masque/connector.go @@ -0,0 +1,180 @@ +package masque + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/go-gost/core/connector" + "github.com/go-gost/core/logger" + md "github.com/go-gost/core/metadata" + ctxvalue "github.com/go-gost/x/ctx" + masque_dialer "github.com/go-gost/x/dialer/http3/masque" + masque_util "github.com/go-gost/x/internal/util/masque" + "github.com/go-gost/x/registry" +) + +func init() { + registry.ConnectorRegistry().Register("masque", NewConnector) +} + +var ( + ErrInvalidConnection = errors.New("masque: invalid connection type, expected MasqueConn") + ErrConnectFailed = errors.New("masque: CONNECT-UDP failed") + ErrCapsuleRequired = errors.New("masque: server did not confirm capsule-protocol") +) + +type masqueConnector struct { + md metadata + options connector.Options +} + +func NewConnector(opts ...connector.Option) connector.Connector { + options := connector.Options{} + for _, opt := range opts { + opt(&options) + } + + return &masqueConnector{ + options: options, + } +} + +func (c *masqueConnector) Init(md md.Metadata) (err error) { + return c.parseMetadata(md) +} + +func (c *masqueConnector) Connect(ctx context.Context, conn net.Conn, network, address string, opts ...connector.ConnectOption) (net.Conn, error) { + log := c.options.Logger.WithFields(map[string]any{ + "remote": conn.RemoteAddr().String(), + "local": conn.LocalAddr().String(), + "network": network, + "address": address, + "sid": string(ctxvalue.SidFromContext(ctx)), + }) + + // Only support UDP + if !strings.HasPrefix(network, "udp") { + return nil, fmt.Errorf("masque connector only supports UDP, got %s", network) + } + + log.Debugf("connect-udp %s/%s", address, network) + + // Get the MasqueConn from the dialer + masqueConn, ok := conn.(*masque_dialer.MasqueConn) + if !ok { + log.Error(ErrInvalidConnection) + return nil, ErrInvalidConnection + } + + // Get pre-opened stream from dialer (stream opening happens there for dead connection detection) + reqStream := masqueConn.GetRequestStream() + proxyHost := masqueConn.GetHost() + + // Apply connect timeout to the actual stream + if c.md.connectTimeout > 0 { + reqStream.SetDeadline(time.Now().Add(c.md.connectTimeout)) + defer reqStream.SetDeadline(time.Time{}) + } + + // Ensure stream is closed on any error + success := false + defer func() { + if !success { + reqStream.Close() + } + }() + + // Build CONNECT-UDP path + host, port, err := net.SplitHostPort(address) + if err != nil { + return nil, fmt.Errorf("masque: invalid address %s: %w", address, err) + } + + path := masque_util.BuildMasquePath(host, mustAtoi(port)) + + // Create CONNECT-UDP request + // For Extended CONNECT (RFC 9220), set Proto to the :protocol value + // quic-go translates req.Proto to the :protocol pseudo-header + req := &http.Request{ + Method: http.MethodConnect, + URL: &url.URL{Path: path}, + Host: proxyHost, + Header: http.Header{ + "Capsule-Protocol": []string{"?1"}, + }, + Proto: "connect-udp", // This becomes the :protocol pseudo-header + ProtoMajor: 3, + } + + // Add proxy authentication if configured + // Use Proxy-Authorization header (not Authorization) for proxy auth + if c.options.Auth != nil { + u := c.options.Auth.Username() + p, _ := c.options.Auth.Password() + auth := u + ":" + p + encoded := base64.StdEncoding.EncodeToString([]byte(auth)) + req.Header.Set("Proxy-Authorization", "Basic "+encoded) + } + + if log.IsLevelEnabled(logger.TraceLevel) { + log.Tracef("CONNECT-UDP request: %s %s Host=%s", req.Method, path, proxyHost) + } + log.Debugf("sending CONNECT-UDP request to %s", path) + + // Send request headers + if err := reqStream.SendRequestHeader(req); err != nil { + log.Error("masque: failed to send request:", err) + return nil, err + } + + // Read response + resp, err := reqStream.ReadResponse() + if err != nil { + log.Error("masque: failed to read response:", err) + return nil, err + } + + if log.IsLevelEnabled(logger.TraceLevel) { + log.Tracef("CONNECT-UDP response: %d %s", resp.StatusCode, resp.Status) + } + + if resp.StatusCode != http.StatusOK { + log.Errorf("masque: proxy returned status %d", resp.StatusCode) + return nil, fmt.Errorf("%w: status %d", ErrConnectFailed, resp.StatusCode) + } + + if resp.Header.Get("Capsule-Protocol") != "?1" { + log.Error(ErrCapsuleRequired) + return nil, ErrCapsuleRequired + } + + log.Debugf("CONNECT-UDP established to %s", address) + + // Get the underlying HTTP/3 stream for datagrams + stream := reqStream + + // Resolve target address + raddr, err := net.ResolveUDPAddr("udp", address) + if err != nil { + return nil, err + } + + // Create datagram connection wrapping the request stream + datagramConn := masque_util.NewDatagramConnFromRequestStream(stream, conn.LocalAddr(), raddr) + + success = true // Prevent defer from closing stream - datagramConn now owns it + return datagramConn, nil +} + +func mustAtoi(s string) int { + n, _ := strconv.Atoi(s) + return n +} diff --git a/connector/masque/metadata.go b/connector/masque/metadata.go new file mode 100644 index 00000000..b4ed17df --- /dev/null +++ b/connector/masque/metadata.go @@ -0,0 +1,17 @@ +package masque + +import ( + "time" + + mdata "github.com/go-gost/core/metadata" + mdutil "github.com/go-gost/x/metadata/util" +) + +type metadata struct { + connectTimeout time.Duration +} + +func (c *masqueConnector) parseMetadata(md mdata.Metadata) (err error) { + c.md.connectTimeout = mdutil.GetDuration(md, "timeout", "connectTimeout") + return nil +} diff --git a/dialer/http3/masque/client.go b/dialer/http3/masque/client.go new file mode 100644 index 00000000..5ae28a54 --- /dev/null +++ b/dialer/http3/masque/client.go @@ -0,0 +1,150 @@ +package masque + +import ( + "context" + "crypto/tls" + "net" + "time" + + "github.com/go-gost/core/logger" + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/http3" +) + +// Client manages HTTP/3 connections for MASQUE proxying. +// It wraps a QUIC connection and HTTP/3 client, following the session pattern +// used by other gost dialers (ssh, quic, mws, etc.). +type Client struct { + host string + addr string + transport *http3.Transport + quicConn *quic.Conn + clientConn *http3.ClientConn + dialer func(ctx context.Context, addr string, tlsCfg *tls.Config, cfg *quic.Config) (*quic.Conn, error) + log logger.Logger +} + +// IsClosed returns true if the underlying connection is closed. +// This follows the pattern used by other gost session wrappers. +func (c *Client) IsClosed() bool { + if c.clientConn == nil { + return true + } + select { + case <-c.clientConn.Context().Done(): + return true + default: + } + return false +} + +// Close closes the client connection. +func (c *Client) Close() error { + if c.quicConn != nil { + return (*c.quicConn).CloseWithError(0, "closed") + } + return nil +} + +// Dial establishes an HTTP/3 connection to the proxy server. +// It returns a MasqueConn that can be used by the MASQUE connector. +// Following the QUIC dialer pattern, this opens a request stream immediately +// to detect dead connections and allow cache invalidation. +func (c *Client) Dial(ctx context.Context, addr string) (net.Conn, error) { + // Create connection if we don't have one (caller checks IsClosed first) + if c.clientConn == nil { + // Use the transport's Dial function to establish QUIC connection + quicConn, err := c.dialer(ctx, addr, c.transport.TLSClientConfig, c.transport.QUICConfig) + if err != nil { + return nil, err + } + c.quicConn = quicConn + + // Create HTTP/3 client connection on top of QUIC + c.clientConn = c.transport.NewClientConn(quicConn) + } + + // Open request stream NOW (following QUIC dialer pattern). + // This allows dead connection detection - if this fails, caller can invalidate cache. + reqStream, err := c.clientConn.OpenRequestStream(ctx) + if err != nil { + // Connection is dead - clear our state so next dial creates fresh connection + c.clientConn = nil + c.quicConn = nil + return nil, err + } + + // Return a connection wrapper with pre-opened stream + return &MasqueConn{ + clientConn: c.clientConn, + reqStream: reqStream, + host: c.host, + log: c.log, + }, nil +} + +// MasqueConn wraps an HTTP/3 client connection for use with the MASQUE connector. +// It implements net.Conn for compatibility with the gost connector interface, +// but the actual data transfer happens via datagrams through the pre-opened RequestStream. +type MasqueConn struct { + clientConn *http3.ClientConn + reqStream *http3.RequestStream // Pre-opened stream for CONNECT-UDP + host string + log logger.Logger +} + +// GetRequestStream returns the pre-opened HTTP/3 request stream. +// The connector uses this for the CONNECT-UDP handshake. +func (c *MasqueConn) GetRequestStream() *http3.RequestStream { + return c.reqStream +} + +// GetHost returns the proxy host. +func (c *MasqueConn) GetHost() string { + return c.host +} + +// Read implements net.Conn but is not used for MASQUE. +// The actual data transfer happens via datagrams. +func (c *MasqueConn) Read(b []byte) (n int, err error) { + // This should not be called - datagrams are used for data transfer + return 0, nil +} + +// Write implements net.Conn but is not used for MASQUE. +// The actual data transfer happens via datagrams. +func (c *MasqueConn) Write(b []byte) (n int, err error) { + // This should not be called - datagrams are used for data transfer + return len(b), nil +} + +// Close closes the connection. +func (c *MasqueConn) Close() error { + // Don't close the underlying clientConn as it's shared/pooled + return nil +} + +// LocalAddr returns the local network address. +func (c *MasqueConn) LocalAddr() net.Addr { + return &net.UDPAddr{} +} + +// RemoteAddr returns the remote network address. +func (c *MasqueConn) RemoteAddr() net.Addr { + return &net.UDPAddr{} +} + +// SetDeadline sets the read and write deadlines. +func (c *MasqueConn) SetDeadline(t time.Time) error { + return nil +} + +// SetReadDeadline sets the read deadline. +func (c *MasqueConn) SetReadDeadline(t time.Time) error { + return nil +} + +// SetWriteDeadline sets the write deadline. +func (c *MasqueConn) SetWriteDeadline(t time.Time) error { + return nil +} diff --git a/dialer/http3/masque/dialer.go b/dialer/http3/masque/dialer.go new file mode 100644 index 00000000..5c26879c --- /dev/null +++ b/dialer/http3/masque/dialer.go @@ -0,0 +1,136 @@ +package masque + +import ( + "context" + "crypto/tls" + "net" + "sync" + + "github.com/go-gost/core/dialer" + md "github.com/go-gost/core/metadata" + "github.com/go-gost/x/registry" + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/http3" +) + +func init() { + registry.DialerRegistry().Register("h3-masque", NewDialer) +} + +type masqueDialer struct { + clients map[string]*Client + clientMutex sync.Mutex + md metadata + options dialer.Options +} + +func NewDialer(opts ...dialer.Option) dialer.Dialer { + options := dialer.Options{} + for _, opt := range opts { + opt(&options) + } + + return &masqueDialer{ + clients: make(map[string]*Client), + options: options, + } +} + +func (d *masqueDialer) Init(md md.Metadata) (err error) { + return d.parseMetadata(md) +} + +// Multiplex implements dialer.Multiplexer interface. +func (d *masqueDialer) Multiplex() bool { + return true +} + +func (d *masqueDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialOption) (net.Conn, error) { + d.clientMutex.Lock() + defer d.clientMutex.Unlock() + + // Check if cached client is still alive + client := d.clients[addr] + if client != nil && client.IsClosed() { + client.Close() + delete(d.clients, addr) + client = nil + } + + if client == nil { + var options dialer.DialOptions + for _, opt := range opts { + opt(&options) + } + + host := d.md.host + if host == "" { + host = options.Host + } + if h, _, _ := net.SplitHostPort(host); h != "" { + host = h + } + + dialFn := func(ctx context.Context, adr string, tlsCfg *tls.Config, cfg *quic.Config) (*quic.Conn, error) { + udpAddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + + udpConn, err := options.Dialer.Dial(ctx, "udp", "") + if err != nil { + return nil, err + } + + return quic.DialEarly(ctx, udpConn.(net.PacketConn), udpAddr, tlsCfg, cfg) + } + + // Ensure TLS config has HTTP/3 ALPN + tlsCfg := d.options.TLSConfig + if tlsCfg == nil { + tlsCfg = &tls.Config{} + } + tlsCfg = tlsCfg.Clone() + if len(tlsCfg.NextProtos) == 0 { + tlsCfg.NextProtos = []string{http3.NextProtoH3} + } + if tlsCfg.ServerName == "" && host != "" { + tlsCfg.ServerName = host + } + + client = &Client{ + log: d.options.Logger, + host: host, + addr: addr, + dialer: dialFn, + transport: &http3.Transport{ + TLSClientConfig: tlsCfg, + EnableDatagrams: true, + QUICConfig: &quic.Config{ + KeepAlivePeriod: d.md.keepAlivePeriod, + HandshakeIdleTimeout: d.md.handshakeTimeout, + MaxIdleTimeout: d.md.maxIdleTimeout, + Versions: []quic.Version{ + quic.Version1, + quic.Version2, + }, + MaxIncomingStreams: int64(d.md.maxStreams), + EnableDatagrams: true, + }, + }, + } + d.clients[addr] = client + } + + // Dial opens a request stream - if this fails, connection is dead + conn, err := client.Dial(ctx, addr) + if err != nil { + // Stream opening failed - connection is dead, remove from cache + d.options.Logger.Error(err) + client.Close() + delete(d.clients, addr) + return nil, err + } + + return conn, nil +} diff --git a/dialer/http3/masque/metadata.go b/dialer/http3/masque/metadata.go new file mode 100644 index 00000000..34a9f3b7 --- /dev/null +++ b/dialer/http3/masque/metadata.go @@ -0,0 +1,43 @@ +package masque + +import ( + "time" + + mdata "github.com/go-gost/core/metadata" + mdutil "github.com/go-gost/x/metadata/util" +) + +const ( + mdKeyHost = "host" + mdKeyKeepAlive = "keepAlive" + mdKeyKeepAlivePeriod = "ttl" + mdKeyHandshakeTimeout = "handshakeTimeout" + mdKeyMaxIdleTimeout = "maxIdleTimeout" + mdKeyMaxStreams = "maxStreams" +) + +type metadata struct { + host string + + // QUIC config options + keepAlivePeriod time.Duration + maxIdleTimeout time.Duration + handshakeTimeout time.Duration + maxStreams int +} + +func (d *masqueDialer) parseMetadata(md mdata.Metadata) (err error) { + d.md.host = mdutil.GetString(md, mdKeyHost) + + if mdutil.GetBool(md, mdKeyKeepAlive) { + d.md.keepAlivePeriod = mdutil.GetDuration(md, mdKeyKeepAlivePeriod) + if d.md.keepAlivePeriod <= 0 { + d.md.keepAlivePeriod = 10 * time.Second + } + } + d.md.handshakeTimeout = mdutil.GetDuration(md, mdKeyHandshakeTimeout) + d.md.maxIdleTimeout = mdutil.GetDuration(md, mdKeyMaxIdleTimeout) + d.md.maxStreams = mdutil.GetInt(md, mdKeyMaxStreams) + + return nil +} diff --git a/handler/masque/handler.go b/handler/masque/handler.go new file mode 100644 index 00000000..4330e415 --- /dev/null +++ b/handler/masque/handler.go @@ -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 +} diff --git a/handler/masque/metadata.go b/handler/masque/metadata.go new file mode 100644 index 00000000..b12eaa0d --- /dev/null +++ b/handler/masque/metadata.go @@ -0,0 +1,49 @@ +package masque + +import ( + "time" + + mdata "github.com/go-gost/core/metadata" + mdutil "github.com/go-gost/x/metadata/util" +) + +const ( + defaultBufferSize = 4096 + defaultRealm = "gost" +) + +type metadata struct { + hash string + bufferSize int + authBasicRealm string + + observerPeriod time.Duration + observerResetTraffic bool + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration +} + +func (h *masqueHandler) parseMetadata(md mdata.Metadata) (err error) { + h.md.hash = mdutil.GetString(md, "hash") + + h.md.bufferSize = mdutil.GetInt(md, "bufferSize", "udp.bufferSize") + if h.md.bufferSize <= 0 { + h.md.bufferSize = defaultBufferSize + } + + h.md.authBasicRealm = mdutil.GetString(md, "authBasicRealm") + + 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 nil +} diff --git a/internal/util/masque/conn.go b/internal/util/masque/conn.go new file mode 100644 index 00000000..3f31c6f6 --- /dev/null +++ b/internal/util/masque/conn.go @@ -0,0 +1,176 @@ +package masque + +import ( + "context" + "io" + "net" + "sync" + "time" + + "github.com/quic-go/quic-go/http3" +) + +// DatagramStreamer is an interface for sending and receiving HTTP/3 datagrams. +// Both http3.Stream and http3.RequestStream implement this interface. +type DatagramStreamer interface { + SendDatagram(b []byte) error + ReceiveDatagram(ctx context.Context) ([]byte, error) +} + +// DatagramConn wraps an HTTP/3 stream's datagram methods as net.PacketConn. +// This allows UDP packets to be tunneled over HTTP/3 datagrams per RFC 9297/9298. +type DatagramConn struct { + stream DatagramStreamer + closer io.Closer // Optional closer for the underlying stream + localAddr net.Addr + remoteAddr net.Addr + closed chan struct{} + closeOnce sync.Once + + mu sync.RWMutex + readDeadline time.Time +} + +// NewDatagramConn creates a new DatagramConn wrapping an HTTP/3 stream. +func NewDatagramConn(stream *http3.Stream, laddr, raddr net.Addr) *DatagramConn { + return &DatagramConn{ + stream: stream, + localAddr: laddr, + remoteAddr: raddr, + closed: make(chan struct{}), + } +} + +// NewDatagramConnFromRequestStream creates a new DatagramConn wrapping an HTTP/3 request stream. +// This is used by the client-side connector. The stream will be closed when Close() is called. +func NewDatagramConnFromRequestStream(stream *http3.RequestStream, laddr, raddr net.Addr) *DatagramConn { + return &DatagramConn{ + stream: stream, + closer: stream, // RequestStream implements io.Closer + localAddr: laddr, + remoteAddr: raddr, + closed: make(chan struct{}), + } +} + +// ReadFrom reads a UDP datagram from the HTTP/3 stream. +// Per RFC 9297, HTTP datagrams have a context ID prefix. +// For CONNECT-UDP (RFC 9298), the context ID is 0. +func (c *DatagramConn) ReadFrom(b []byte) (n int, addr net.Addr, err error) { + select { + case <-c.closed: + return 0, nil, net.ErrClosed + default: + } + + ctx := context.Background() + c.mu.RLock() + deadline := c.readDeadline + c.mu.RUnlock() + + if !deadline.IsZero() { + var cancel context.CancelFunc + ctx, cancel = context.WithDeadline(ctx, deadline) + defer cancel() + } + + data, err := c.stream.ReceiveDatagram(ctx) + if err != nil { + return 0, nil, err + } + + if len(data) == 0 { + return 0, c.remoteAddr, nil + } + + // Per RFC 9297: datagram format is context-id (varint) + payload + // For CONNECT-UDP with context ID 0, the first byte is 0x00 + // We strip the context ID prefix + if data[0] == 0x00 { + data = data[1:] + } + + n = copy(b, data) + return n, c.remoteAddr, nil +} + +// WriteTo sends a UDP datagram via the HTTP/3 stream. +// Per RFC 9297, we prepend the context ID (0x00 for CONNECT-UDP). +func (c *DatagramConn) WriteTo(b []byte, addr net.Addr) (n int, err error) { + select { + case <-c.closed: + return 0, net.ErrClosed + default: + } + + // Prepend context ID (0x00 for context ID 0) + datagram := make([]byte, 1+len(b)) + datagram[0] = 0x00 // Context ID = 0 + copy(datagram[1:], b) + + if err := c.stream.SendDatagram(datagram); err != nil { + return 0, err + } + + return len(b), nil +} + +// Read reads data from the connection (net.Conn interface). +func (c *DatagramConn) Read(b []byte) (n int, err error) { + n, _, err = c.ReadFrom(b) + return +} + +// Write writes data to the connection (net.Conn interface). +func (c *DatagramConn) Write(b []byte) (n int, err error) { + return c.WriteTo(b, c.remoteAddr) +} + +// Close closes the datagram connection and the underlying stream. +func (c *DatagramConn) Close() error { + var err error + c.closeOnce.Do(func() { + close(c.closed) + if c.closer != nil { + err = c.closer.Close() + } + }) + return err +} + +// LocalAddr returns the local network address. +func (c *DatagramConn) LocalAddr() net.Addr { + return c.localAddr +} + +// RemoteAddr returns the remote network address. +func (c *DatagramConn) RemoteAddr() net.Addr { + return c.remoteAddr +} + +// SetDeadline sets the read and write deadlines. +func (c *DatagramConn) SetDeadline(t time.Time) error { + c.SetReadDeadline(t) + c.SetWriteDeadline(t) + return nil +} + +// SetReadDeadline sets the read deadline. +func (c *DatagramConn) SetReadDeadline(t time.Time) error { + c.mu.Lock() + c.readDeadline = t + c.mu.Unlock() + return nil +} + +// SetWriteDeadline sets the write deadline. +// Note: Write deadlines are not used for HTTP/3 datagrams as SendDatagram is non-blocking. +func (c *DatagramConn) SetWriteDeadline(t time.Time) error { + return nil +} + +// Ensure DatagramConn implements both net.Conn and net.PacketConn +var ( + _ net.Conn = (*DatagramConn)(nil) + _ net.PacketConn = (*DatagramConn)(nil) +) diff --git a/internal/util/masque/path.go b/internal/util/masque/path.go new file mode 100644 index 00000000..caedb79d --- /dev/null +++ b/internal/util/masque/path.go @@ -0,0 +1,60 @@ +package masque + +import ( + "errors" + "net/url" + "strconv" + "strings" +) + +const ( + // WellKnownPath is the well-known path prefix for MASQUE UDP proxying + WellKnownPath = "/.well-known/masque/udp/" +) + +var ( + ErrInvalidPath = errors.New("masque: invalid path template") + ErrInvalidPort = errors.New("masque: invalid port number") +) + +// ParseMasquePath parses a MASQUE UDP proxy path template. +// The expected format is: /.well-known/masque/udp/{host}/{port}/ +// Returns the host and port extracted from the path. +func ParseMasquePath(path string) (host string, port int, err error) { + if !strings.HasPrefix(path, WellKnownPath) { + return "", 0, ErrInvalidPath + } + + remainder := strings.TrimPrefix(path, WellKnownPath) + remainder = strings.TrimSuffix(remainder, "/") + + parts := strings.Split(remainder, "/") + if len(parts) != 2 { + return "", 0, ErrInvalidPath + } + + host = parts[0] + if host == "" { + return "", 0, ErrInvalidPath + } + + // URL decode the host in case it contains percent-encoded characters + host, err = url.PathUnescape(host) + if err != nil { + return "", 0, ErrInvalidPath + } + + port, err = strconv.Atoi(parts[1]) + if err != nil || port <= 0 || port > 65535 { + return "", 0, ErrInvalidPort + } + + return host, port, nil +} + +// BuildMasquePath constructs a MASQUE UDP proxy path from host and port. +func BuildMasquePath(host string, port int) string { + // URL encode the host in case it contains special characters (like IPv6 addresses) + encodedHost := url.PathEscape(host) + return WellKnownPath + encodedHost + "/" + strconv.Itoa(port) + "/" +} diff --git a/listener/http3/listener.go b/listener/http3/listener.go index 963d1da7..3511845b 100644 --- a/listener/http3/listener.go +++ b/listener/http3/listener.go @@ -75,8 +75,10 @@ func (l *http3Listener) Init(md md.Metadata) (err error) { }, MaxIncomingStreams: int64(l.md.maxStreams), Allow0RTT: true, + EnableDatagrams: l.md.enableDatagrams, }, - Handler: http.HandlerFunc(l.handleFunc), + EnableDatagrams: l.md.enableDatagrams, + Handler: http.HandlerFunc(l.handleFunc), } ln, err := quic.ListenAddrEarly(addr, http3.ConfigureTLSConfig(l.server.TLSConfig), l.server.QUICConfig.Clone()) diff --git a/listener/http3/metadata.go b/listener/http3/metadata.go index 3b5877e2..c8de9825 100644 --- a/listener/http3/metadata.go +++ b/listener/http3/metadata.go @@ -19,6 +19,7 @@ type metadata struct { maxIdleTimeout time.Duration handshakeTimeout time.Duration maxStreams int + enableDatagrams bool } func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) { @@ -46,6 +47,7 @@ func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) { l.md.handshakeTimeout = mdutil.GetDuration(md, handshakeTimeout) l.md.maxIdleTimeout = mdutil.GetDuration(md, maxIdleTimeout) l.md.maxStreams = mdutil.GetInt(md, maxStreams) + l.md.enableDatagrams = mdutil.GetBool(md, "enableDatagrams") return }