diff --git a/caddytest/integration/webtransport_test.go b/caddytest/integration/webtransport_test.go index 17294e697..6a8eebbe4 100644 --- a/caddytest/integration/webtransport_test.go +++ b/caddytest/integration/webtransport_test.go @@ -16,8 +16,16 @@ package integration import ( "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "fmt" "io" + "math/big" + "net" "net/http" "strings" "testing" @@ -286,3 +294,209 @@ func TestWebTransport_ReverseProxyEndToEnd(t *testing.T) { t.Fatalf("echo mismatch:\n got: %q\n want: %q", strings.TrimSpace(string(got)), payload) } } + +// TestWebTransport_ReverseProxyForwardsHeaders proves that the WebTransport +// proxy path applies the same request-preparation pipeline as the normal +// reverse_proxy path: `headers.request.set` lands on the upstream CONNECT, +// X-Forwarded-For is added, and a Via header is appended. The upstream here +// is a standalone webtransport.Server (not another Caddy) so we can observe +// the raw headers of the Extended CONNECT that Caddy forwarded. +func TestWebTransport_ReverseProxyForwardsHeaders(t *testing.T) { + if testing.Short() { + t.Skip() + } + + // Capture the first Extended CONNECT's headers. + gotHeaders := make(chan http.Header, 1) + upstreamAddr, stopUpstream := startStandaloneWebTransport(t, func(sess *webtransport.Session, r *http.Request) { + select { + case gotHeaders <- r.Header.Clone(): + default: + } + _ = sess.CloseWithError(0, "") + }) + t.Cleanup(stopUpstream) + + config := fmt.Sprintf(`{ + "admin": {"listen": "localhost:2999"}, + "apps": { + "http": { + "http_port": 9080, + "https_port": 9443, + "grace_period": 1, + "servers": { + "proxy": { + "listen": [":9443"], + "protocols": ["h3"], + "routes": [ + { + "handle": [ + { + "handler": "reverse_proxy", + "transport": { + "protocol": "http", + "versions": ["3"], + "webtransport": true, + "tls": {"insecure_skip_verify": true} + }, + "headers": { + "request": { + "set": {"X-Caddy-Test": ["caddy-wt-hdr"]} + } + }, + "upstreams": [{"dial": "127.0.0.1:%d"}] + } + ] + } + ], + "tls_connection_policies": [ + { + "certificate_selection": {"any_tag": ["cert0"]}, + "default_sni": "a.caddy.localhost" + } + ] + } + } + }, + "tls": { + "certificates": { + "load_files": [ + { + "certificate": "/a.caddy.localhost.crt", + "key": "/a.caddy.localhost.key", + "tags": ["cert0"] + } + ] + } + }, + "pki": {"certificate_authorities": {"local": {"install_trust": false}}} + } +}`, upstreamAddr.Port) + + tester := caddytest.NewTester(t) + tester.InitServer(config, "json") + + dialer := &webtransport.Dialer{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // local CA + ServerName: "a.caddy.localhost", + NextProtos: []string{http3.NextProtoH3}, + }, + QUICConfig: &quic.Config{ + EnableDatagrams: true, + EnableStreamResetPartialDelivery: true, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var sess *webtransport.Session + deadline := time.Now().Add(3 * time.Second) + for { + _, s, err := dialer.Dial(ctx, "https://127.0.0.1:9443/", nil) + if err == nil { + sess = s + break + } + if time.Now().After(deadline) { + t.Fatalf("webtransport dial through proxy failed: %v", err) + } + time.Sleep(100 * time.Millisecond) + } + defer sess.CloseWithError(0, "") + + select { + case hdr := <-gotHeaders: + if got := hdr.Get("X-Caddy-Test"); got != "caddy-wt-hdr" { + t.Errorf("upstream did not receive `headers.request.set` value; got X-Caddy-Test=%q", got) + } + if got := hdr.Get("X-Forwarded-For"); !strings.Contains(got, "127.0.0.1") { + t.Errorf("upstream did not receive X-Forwarded-For=127.0.0.1; got %q", got) + } + if got := hdr.Get("Via"); got == "" { + t.Errorf("upstream did not receive Via header") + } + case <-time.After(3 * time.Second): + t.Fatal("upstream did not observe forwarded CONNECT headers in time") + } +} + +// startStandaloneWebTransport starts a webtransport.Server on a random UDP +// port with a self-signed cert. handler runs after a successful Upgrade. +// Returns the listener addr and a shutdown func. +func startStandaloneWebTransport(t *testing.T, handler func(s *webtransport.Session, r *http.Request)) (*net.UDPAddr, func()) { + t.Helper() + tlsCfg := newSelfSignedTLSConfig(t, "localhost") + + mux := http.NewServeMux() + h3 := &http3.Server{ + TLSConfig: tlsCfg, + Handler: mux, + QUICConfig: &quic.Config{ + EnableDatagrams: true, + EnableStreamResetPartialDelivery: true, + }, + } + webtransport.ConfigureHTTP3Server(h3) + wtServer := &webtransport.Server{H3: h3} + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + sess, err := wtServer.Upgrade(w, r) + if err != nil { + t.Logf("standalone WebTransport upgrade failed: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + handler(sess, r) + }) + + udpAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + conn, err := net.ListenUDP("udp", udpAddr) + if err != nil { + t.Fatal(err) + } + servErr := make(chan error, 1) + go func() { servErr <- wtServer.Serve(conn) }() + shutdown := func() { + _ = wtServer.Close() + <-servErr + _ = conn.Close() + } + return conn.LocalAddr().(*net.UDPAddr), shutdown +} + +// newSelfSignedTLSConfig produces a self-signed TLS config suitable for +// 127.0.0.1 and the given common name, with the H3 ALPN advertised. +func newSelfSignedTLSConfig(t *testing.T, cn string) *tls.Config { + t.Helper() + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: cn}, + NotBefore: time.Now().Add(-time.Minute), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + DNSNames: []string{cn}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, + IsCA: true, + } + der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, priv.Public(), priv) + if err != nil { + t.Fatal(err) + } + cert, err := x509.ParseCertificate(der) + if err != nil { + t.Fatal(err) + } + return &tls.Config{ + Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: priv, Leaf: cert}}, + NextProtos: []string{http3.NextProtoH3}, + } +} diff --git a/modules/caddyhttp/reverseproxy/webtransport_transport.go b/modules/caddyhttp/reverseproxy/webtransport_transport.go index 8f564ac00..47d0dddc7 100644 --- a/modules/caddyhttp/reverseproxy/webtransport_transport.go +++ b/modules/caddyhttp/reverseproxy/webtransport_transport.go @@ -20,11 +20,14 @@ import ( "errors" "fmt" "net/http" + "time" "github.com/quic-go/quic-go" "github.com/quic-go/webtransport-go" "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "github.com/caddyserver/caddy/v2" "github.com/caddyserver/caddy/v2/modules/caddyhttp" caddywt "github.com/caddyserver/caddy/v2/modules/caddyhttp/webtransport" ) @@ -42,9 +45,20 @@ func isWebTransportExtendedConnect(r *http.Request) bool { // // Unlike the regular HTTP proxy path, there are no retries: a failed // dial closes the client's session and returns (so the handler chain -// can finish). Requests that reach this function are already known to -// be WebTransport; callers should gate with isWebTransportProxyRequest. +// can finish). The outgoing CONNECT is prepared with the same Rewrite, +// hop-by-hop stripping, X-Forwarded-*/Via, transport- and user-configured +// header ops as the normal proxy path so operators see consistent +// behavior. Requests that reach this function are already known to be +// WebTransport; callers should gate with isWebTransportExtendedConnect. func (h *Handler) serveWebTransport(w http.ResponseWriter, r *http.Request) error { + repl := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer) + start := time.Now() + defer func() { + d := time.Since(start) + repl.Set("http.reverse_proxy.duration", d) + repl.Set("http.reverse_proxy.duration_ms", d.Seconds()*1e3) + }() + srv, ok := r.Context().Value(caddyhttp.ServerCtxKey).(*caddyhttp.Server) if !ok || srv == nil { return caddyhttp.Error(http.StatusInternalServerError, @@ -56,16 +70,64 @@ func (h *Handler) serveWebTransport(w http.ResponseWriter, r *http.Request) erro errors.New("webtransport: HTTP/3 is not enabled on this server; WebTransport requires H3")) } - // Select an upstream via the configured LB policy. No retries. - upstreams := h.Upstreams if h.LoadBalancing == nil || h.LoadBalancing.SelectionPolicy == nil { return caddyhttp.Error(http.StatusInternalServerError, errors.New("webtransport: load balancer is not configured")) } + + // Resolve the candidate upstream set (static or dynamic) and select + // one. WT sessions are long-lived and not idempotent, so there are no + // retries; picking once matches how operators expect WT to behave. + upstreams := h.Upstreams + if h.DynamicUpstreams != nil { + dynUpstreams, err := h.DynamicUpstreams.GetUpstreams(r) + if err != nil { + if c := h.logger.Check(zapcore.WarnLevel, "webtransport: dynamic upstreams failed; falling back to static"); c != nil { + c.Write(zap.Error(err)) + } + } else { + upstreams = dynUpstreams + for _, dUp := range dynUpstreams { + h.provisionUpstream(dUp, true) + } + } + } upstream := h.LoadBalancing.SelectionPolicy.Select(upstreams, r, w) if upstream == nil { - return caddyhttp.Error(http.StatusBadGateway, - errors.New("webtransport: no upstream available")) + return caddyhttp.Error(http.StatusBadGateway, errNoUpstream) + } + + // Resolve per-upstream placeholders (addresses may include them) and + // publish the {http.reverse_proxy.upstream.*} replacer values before + // we commit to upgrading — so any client-side failure logs downstream + // see the selected upstream too. + dialInfo, err := upstream.fillDialInfo(repl) + if err != nil { + return caddyhttp.Error(http.StatusInternalServerError, + fmt.Errorf("webtransport: making dial info: %w", err)) + } + repl.Set("http.reverse_proxy.upstream.address", dialInfo.String()) + repl.Set("http.reverse_proxy.upstream.hostport", dialInfo.Address) + repl.Set("http.reverse_proxy.upstream.host", dialInfo.Host) + repl.Set("http.reverse_proxy.upstream.port", dialInfo.Port) + repl.Set("http.reverse_proxy.upstream.requests", upstream.Host.NumRequests()) + repl.Set("http.reverse_proxy.upstream.max_requests", upstream.MaxRequests) + repl.Set("http.reverse_proxy.upstream.fails", upstream.Host.Fails()) + + // Prepare the outgoing request the same way normal proxying does — + // Rewrite, hop-by-hop stripping, X-Forwarded-*, Via, etc. — then apply + // transport and user header ops. prepareRequest's body-buffering and + // Early-Data paths are no-ops for a CONNECT request (empty body). + clonedReq, err := h.prepareRequest(r, repl) + if err != nil { + return caddyhttp.Error(http.StatusInternalServerError, + fmt.Errorf("webtransport: preparing request: %w", err)) + } + if h.transportHeaderOps != nil { + h.transportHeaderOps.ApplyToRequest(clonedReq) + } + if h.Headers != nil && h.Headers.Request != nil { + h.Headers.Request.ApplyToRequest(clonedReq) } // Reach the naked http3 response writer so Upgrade's type assertions @@ -78,18 +140,24 @@ func (h *Handler) serveWebTransport(w http.ResponseWriter, r *http.Request) erro clientSess, err := wtServer.Upgrade(naked, r) if err != nil { - h.logger.Debug("webtransport client upgrade failed", zap.Error(err)) + if c := h.logger.Check(zapcore.DebugLevel, "webtransport client upgrade failed"); c != nil { + c.Write(zap.Error(err)) + } return caddyhttp.Error(http.StatusBadRequest, fmt.Errorf("webtransport upgrade: %w", err)) } ht := h.Transport.(*HTTPTransport) - upstreamURL := buildWebTransportUpstreamURL(upstream.Dial, r) - _, upstreamSess, err := dialUpstreamWebTransport(r.Context(), ht.h3Transport.TLSClientConfig, upstreamURL, r.Header.Clone()) + upstreamURL := buildWebTransportUpstreamURL(dialInfo.Address, clonedReq) + _, upstreamSess, err := dialUpstreamWebTransport(r.Context(), ht.h3Transport.TLSClientConfig, upstreamURL, clonedReq.Header) if err != nil { - h.logger.Error("webtransport upstream dial failed", - zap.String("upstream", upstreamURL), - zap.Error(err)) + h.countFailure(upstream) + if c := h.logger.Check(zapcore.ErrorLevel, "webtransport upstream dial failed"); c != nil { + c.Write( + zap.String("upstream", upstreamURL), + zap.Error(err), + ) + } _ = clientSess.CloseWithError(0, "upstream dial failed") return nil }