From 55963aa393ba4b9438010ca568cf01ddff801be5 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 07:26:51 +0800 Subject: [PATCH 1/2] fix(web): preserve origin encoding and H3 overload semantics --- docs/WEB_COVER.md | 20 + internal/cover/handler.go | 5 + internal/cover/public_origin.go | 3 +- internal/cover/representation_test.go | 349 ++++++++++++++++++ .../tunnel/web_cover_representation_test.go | 267 ++++++++++++++ .../tunnel/web_h3_admission_error_test.go | 303 +++++++++++++++ internal/tunnel/web_limits.go | 5 +- 7 files changed, 950 insertions(+), 2 deletions(-) create mode 100644 internal/cover/representation_test.go create mode 100644 internal/tunnel/web_cover_representation_test.go create mode 100644 internal/tunnel/web_h3_admission_error_test.go diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index e3f8f6d..7ed446f 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -40,6 +40,15 @@ HTTP/2 CONNECT presented to the public origin is always treated as cover, including when it carries an otherwise valid ticket: the handler removes `Proxy-Authorization` before delegation and never dials its authority. +Public physical-connection limits are separate from authenticated tunnel-stream +limits, and the combined TCP/UDP server shares its global and per-source +connection allowance. Source builds after v1.0.1 reject an excess H3 connection +with `H3_EXCESSIVE_LOAD` (`0x107`), rather than reusing the native relay's code, +which means `H3_INTERNAL_ERROR` in [HTTP/3](https://www.rfc-editor.org/rfc/rfc9114.html#section-8.1). +Rejection happens before HTTP dispatch; +the existing admitted connection remains usable and closing it releases capacity. +This is overload handling, not a tunnel authentication response. + ### Website WebSocket support in source builds Source builds after v1.0.1 also forward valid HTTP/1.1 WebSocket upgrades to @@ -202,6 +211,17 @@ the upstream cannot be reached. Consequently, an upstream that requires an `Authorization` request header is not suitable without a separate authorized front end. +Source builds after v1.0.1 disable automatic compression negotiation and response +decompression on the proxy's private default upstream transport. The visitor's +`Accept-Encoding` remains unchanged: explicitly requested gzip still works, and +the origin's encoded bytes, `Content-Encoding`, length, digest and ETag remain +together. This avoids rewriting `no-transform` content while retaining metadata +for the old bytes; see [HTTP message transformations](https://www.rfc-editor.org/rfc/rfc9110.html#section-7.7). +It does not verify the website's digest or establish browser-like fingerprints. +Applications supplying a custom RoundTripper retain its own negotiation/decoding +policy and must configure transparent forwarding themselves. Static cover, +authenticated tunnel payloads and the published v1.0.1 binary are unchanged. + #### Optional fixed public origin in source builds Source builds after v1.0.1 can separate the website's public HTTP identity diff --git a/internal/cover/handler.go b/internal/cover/handler.go index 034656b..ac31d6b 100644 --- a/internal/cover/handler.go +++ b/internal/cover/handler.go @@ -31,6 +31,8 @@ var hopByHopHeaders = [...]string{ // NewReverseProxyHandler returns a reverse proxy that can dial only origin. // The requester controls the path, query, and ordinary end-to-end headers, but // never the upstream scheme, authority, or Host header. +// A nil transport does not add compression negotiation or decode responses. +// Supplied transports retain their own compression policy. func NewReverseProxyHandler(origin *url.URL, transport http.RoundTripper) (http.Handler, error) { target, err := normalizeOrigin(origin) if err != nil { @@ -43,6 +45,9 @@ func newReverseProxyHandler(target *url.URL, transport http.RoundTripper, public if transport == nil { defaultTransport := http.DefaultTransport.(*http.Transport).Clone() defaultTransport.Proxy = nil + // Forward the visitor's encoding preferences and the origin's encoded + // bytes together with their metadata, including no-transform/digests. + defaultTransport.DisableCompression = true transport = defaultTransport } diff --git a/internal/cover/public_origin.go b/internal/cover/public_origin.go index 1a58d3f..77f42f9 100644 --- a/internal/cover/public_origin.go +++ b/internal/cover/public_origin.go @@ -14,7 +14,8 @@ import ( // and TLS still use upstream; public supplies only the outbound Host and the // trusted X-Forwarded-Host/Proto values. Both URLs must be root origins. // -// Origin, Referer, cookies, redirects, and content are never rewritten. The +// Origin, Referer, cookies, redirects, and content are not rewritten by the +// handler. Supplied transports retain their own compression policy. The // upstream must generate its own public URLs and retain its CSRF/Origin checks; // an absent Origin is not evidence that a request is safe. Its trusted-header // whitelist must use only the proxy-owned forwarding fields, not arbitrary diff --git a/internal/cover/representation_test.go b/internal/cover/representation_test.go new file mode 100644 index 0000000..17d82a9 --- /dev/null +++ b/internal/cover/representation_test.go @@ -0,0 +1,349 @@ +package cover + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "encoding/base64" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "reflect" + "strconv" + "sync/atomic" + "testing" + "time" +) + +// These are owned HTTP/1.1 wire tests, not a browser or fingerprint experiment. +// Client transports never request or decode gzip on the test's behalf. A +// synthetic website generates Content-Digest from actual message content; +// the proxy must forward the representation, not authenticate that website. +func TestReverseProxyRepresentationOnWire(t *testing.T) { + cases := []struct { + name, method, path, acceptEncoding, byteRange string + }{ + {"absent_accept_encoding", http.MethodGet, "/negotiated", "", ""}, + {"explicit_gzip", http.MethodGet, "/negotiated", "gzip", ""}, + {"explicit_identity", http.MethodGet, "/negotiated", "identity", ""}, + {"head", http.MethodHead, "/negotiated", "", ""}, + {"range", http.MethodGet, "/negotiated", "", "bytes=2-8"}, + // With no Accept-Encoding, a server may legitimately choose gzip. + // This separately exposes decoding, not just changed negotiation. + {"always_gzip_without_accept_encoding", http.MethodGet, "/always-gzip", "", ""}, + } + for _, mode := range []string{"default", "public_origin"} { + t.Run(mode, func(t *testing.T) { + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newRepresentationFixture(t, mode, nil) + direct, directSeen := f.exchange(f.origin.URL, tc.method, tc.path, tc.acceptEncoding, tc.byteRange) + proxied, proxySeen := f.exchange(f.front.URL, tc.method, tc.path, tc.acceptEncoding, tc.byteRange) + if !reflect.DeepEqual(proxySeen.acceptEncoding, directSeen.acceptEncoding) { + t.Errorf("upstream Accept-Encoding changed: direct=%q proxy=%q", directSeen.acceptEncoding, proxySeen.acceptEncoding) + } + if proxySeen.method != tc.method || proxySeen.byteRange != tc.byteRange { + t.Errorf("upstream method/Range changed: %q/%q", proxySeen.method, proxySeen.byteRange) + } + assertRepresentationTransparent(t, proxied, direct, proxySeen) + if tc.method == http.MethodHead && (len(proxied.body) != 0 || proxied.header.Get("Content-Digest") != representationDigest(nil)) { + t.Error("HEAD must carry no message content and its digest must describe that empty content") + } + if tc.byteRange != "" && (proxied.status != http.StatusPartialContent || proxied.header.Get("Content-Range") != fmt.Sprintf("bytes 2-8/%d", len(f.identity))) { + t.Errorf("ordinary range status/metadata = %d/%q", proxied.status, proxied.header.Get("Content-Range")) + } + t.Logf("owned HTTP/1.1 method=%s path=%s directAE=%q proxyAE=%q directEncoding=%q proxyEncoding=%q directBytes=%d proxyBytes=%d", + tc.method, tc.path, directSeen.acceptEncoding, proxySeen.acceptEncoding, direct.header.Get("Content-Encoding"), proxied.header.Get("Content-Encoding"), len(direct.body), len(proxied.body)) + }) + } + }) + } +} + +func TestReverseProxyRepresentationCustomTransportPolicy(t *testing.T) { + for _, mode := range []string{"default", "public_origin"} { + t.Run(mode, func(t *testing.T) { + // An explicitly supplied transport owns its policy, including Go's + // automatic gzip decoding. The constructor must not mutate it to + // enforce the default policy; callers needing transparency must + // configure that policy themselves. + custom := &http.Transport{Proxy: nil, DisableCompression: false, ResponseHeaderTimeout: time.Second, IdleConnTimeout: time.Second} + t.Cleanup(custom.CloseIdleConnections) + f := newRepresentationFixture(t, mode, custom) + custom.DialContext = representationOwnedDial(f.origin.Listener.Addr().String()) + dialPointer := reflect.ValueOf(custom.DialContext).Pointer() + got, seen := f.exchange(f.front.URL, http.MethodGet, "/negotiated", "", "") + if custom.DisableCompression || custom.Proxy != nil || custom.ResponseHeaderTimeout != time.Second || custom.IdleConnTimeout != time.Second || reflect.ValueOf(custom.DialContext).Pointer() != dialPointer { + t.Error("constructor or request processing mutated caller-owned transport configuration") + } + if !reflect.DeepEqual(seen.acceptEncoding, []string{"gzip"}) || !bytes.Equal(got.body, f.identity) || got.header.Get("Content-Encoding") != "" { + t.Errorf("explicit custom gzip policy changed: AE=%q encoding=%q body=%q", seen.acceptEncoding, got.header.Get("Content-Encoding"), got.body) + } + if seen.encoding != "gzip" || bytes.Equal(got.body, seen.body) || got.header.Get("Content-Digest") != seen.digest || got.header.Get("ETag") != seen.etag { + t.Error("custom transport's decoded body / untouched upstream metadata behavior was not preserved") + } + t.Log("explicit caller transport retains its automatic gzip policy; this is not a transparent-policy assertion for that caller") + }) + } +} + +type representationSeen struct { + acceptEncoding []string + method string + byteRange string + encoding string + digest string + etag string + length int64 + body []byte + writeErr error + joined <-chan struct{} +} + +type representationResult struct { + status int + contentLength int64 + header http.Header + body []byte +} + +type representationFixture struct { + t *testing.T + origin *httptest.Server + front *httptest.Server + client *http.Client + identity []byte + gzipped []byte + seen chan representationSeen + requests atomic.Int32 + unexpected atomic.Int32 + frontDone chan struct{} +} + +func newRepresentationFixture(t *testing.T, mode string, supplied http.RoundTripper) *representationFixture { + t.Helper() + global := http.DefaultTransport.(*http.Transport) + globalCompression := global.Clone().DisableCompression + t.Cleanup(func() { + if global.Clone().DisableCompression != globalCompression { + t.Error("fixture/constructor changed global DefaultTransport compression configuration") + } + }) + f := &representationFixture{t: t, identity: []byte("owned website payload: preserve encoded content and ordinary metadata, not client-side transparent decompression"), seen: make(chan representationSeen, 2), frontDone: make(chan struct{}, 1)} + wantRequests := int32(2) + if supplied != nil { + wantRequests = 1 + } + t.Cleanup(func() { + if got := f.requests.Load(); got != wantRequests || f.unexpected.Load() != 0 { + t.Errorf("owned request/observer counts: requests=%d want=%d unexpected=%d", got, wantRequests, f.unexpected.Load()) + } + }) + var compressed bytes.Buffer + gz := gzip.NewWriter(&compressed) + if _, err := gz.Write(f.identity); err != nil { + t.Fatal(err) + } + if err := gz.Close(); err != nil { + t.Fatal(err) + } + f.gzipped = bytes.Clone(compressed.Bytes()) + f.origin = httptest.NewUnstartedServer(http.HandlerFunc(f.serveOrigin)) + representationServerBudgets(f.origin) + f.origin.Start() + t.Cleanup(func() { f.origin.CloseClientConnections(); f.origin.Close() }) + originURL, err := url.Parse(f.origin.URL) + if err != nil { + t.Fatal(err) + } + f.front = httptest.NewUnstartedServer(nil) + representationServerBudgets(f.front) + publicURL := &url.URL{Scheme: "https", Host: f.front.Listener.Addr().String()} + var handler http.Handler + if mode == "default" { + handler, err = NewReverseProxyHandler(originURL, supplied) + } else { + handler, err = NewReverseProxyHandlerWithPublicOrigin(originURL, publicURL, supplied) + } + if err != nil { + _ = f.front.Listener.Close() + t.Fatal(err) + } + // Inspect the private wrapper solely to close the fixture-owned default + // transport. Wire assertions above depend only on the public constructors. + if supplied == nil { + proxy, ok := handler.(*httputil.ReverseProxy) + if !ok { + proxy = handler.(*publicOriginHandler).proxy.(*httputil.ReverseProxy) + } + owned := proxy.Transport.(*informationalHeaderTransport).base.(*http.Transport) + t.Cleanup(owned.CloseIdleConnections) + } + f.front.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer func() { + select { + case f.frontDone <- struct{}{}: + default: + f.unexpected.Add(1) + } + }() + handler.ServeHTTP(w, r) + }) + f.front.Start() + t.Cleanup(func() { f.front.CloseClientConnections(); f.front.Close() }) + clientTransport := &http.Transport{Proxy: nil, DisableCompression: true, ResponseHeaderTimeout: time.Second, IdleConnTimeout: time.Second} + clientTransport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.origin.Listener.Addr().String() && address != f.front.Listener.Addr().String() { + return nil, fmt.Errorf("representation fixture refuses non-owned destination %s/%s", network, address) + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, network, address) + } + t.Cleanup(clientTransport.CloseIdleConnections) + f.client = &http.Client{Transport: clientTransport, Timeout: 2 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + return f +} + +func representationServerBudgets(server *httptest.Server) { + server.Config.ReadHeaderTimeout = time.Second + server.Config.ReadTimeout = 2 * time.Second + server.Config.WriteTimeout = 2 * time.Second + server.Config.IdleTimeout = time.Second +} + +func representationOwnedDial(target string) func(context.Context, string, string) (net.Conn, error) { + return func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != target { + return nil, fmt.Errorf("custom representation transport refuses non-owned destination %s/%s", network, address) + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, network, address) + } +} + +func (f *representationFixture) serveOrigin(w http.ResponseWriter, r *http.Request) { + f.requests.Add(1) + joined := make(chan struct{}) + defer close(joined) + s := representationSeen{acceptEncoding: append([]string(nil), r.Header.Values("Accept-Encoding")...), method: r.Method, byteRange: r.Header.Get("Range"), joined: joined} + body := f.identity + status := http.StatusOK + if r.URL.Path == "/always-gzip" || r.Header.Get("Accept-Encoding") == "gzip" { + body = f.gzipped + s.encoding = "gzip" + w.Header().Set("Content-Encoding", "gzip") + } + s.etag = fmt.Sprintf("\"%x\"", sha256.Sum256(body)) + if s.byteRange != "" { + if s.byteRange != "bytes=2-8" || s.encoding != "" { + http.Error(w, "invalid owned range", http.StatusBadRequest) + return + } + status = http.StatusPartialContent + w.Header().Set("Content-Range", fmt.Sprintf("bytes 2-8/%d", len(body))) + body = body[2:9] + } + s.length = int64(len(body)) + // HEAD has no message content. Content-Length and ETag still describe + // the selected representation; Content-Digest describes empty content. + if r.Method == http.MethodHead { + s.body = nil + } else { + s.body = bytes.Clone(body) + } + s.digest = representationDigest(s.body) + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Cache-Control", "no-transform") + w.Header().Set("Content-Digest", s.digest) + w.Header().Set("Content-Length", strconv.FormatInt(s.length, 10)) + w.Header().Set("ETag", s.etag) + w.Header().Set("Vary", "Accept-Encoding") + w.WriteHeader(status) + if r.Method != http.MethodHead { + _, s.writeErr = w.Write(s.body) + } + select { + case f.seen <- s: + default: + f.unexpected.Add(1) + } +} + +func (f *representationFixture) exchange(base, method, path, acceptEncoding, byteRange string) (representationResult, representationSeen) { + f.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + request, err := http.NewRequestWithContext(ctx, method, base+path, nil) + if err != nil { + f.t.Fatal(err) + } + if acceptEncoding != "" { + request.Header.Set("Accept-Encoding", acceptEncoding) + } + if byteRange != "" { + request.Header.Set("Range", byteRange) + } + response, err := f.client.Do(request) + if err != nil { + f.t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + closeErr := response.Body.Close() + if readErr != nil || closeErr != nil { + f.t.Fatalf("owned HTTP response read/close: %v/%v", readErr, closeErr) + } + var seen representationSeen + select { + case seen = <-f.seen: + case <-ctx.Done(): + f.t.Fatal("owned origin result did not arrive") + } + select { + case <-seen.joined: + case <-ctx.Done(): + f.t.Fatal("owned origin handler did not join") + } + if seen.writeErr != nil { + f.t.Fatalf("owned origin write: %v", seen.writeErr) + } + if base == f.front.URL { + select { + case <-f.frontDone: + case <-ctx.Done(): + f.t.Fatal("owned front handler did not join") + } + } + if response.ProtoMajor != 1 || response.ProtoMinor != 1 { + f.t.Errorf("actual public protocol = %s, want HTTP/1.1", response.Proto) + } + if f.unexpected.Load() != 0 { + f.t.Errorf("unexpected observer notifications=%d", f.unexpected.Load()) + } + return representationResult{status: response.StatusCode, contentLength: response.ContentLength, header: response.Header.Clone(), body: body}, seen +} + +func assertRepresentationTransparent(t *testing.T, got, direct representationResult, seen representationSeen) { + t.Helper() + if got.status != direct.status || !bytes.Equal(got.body, direct.body) { + t.Errorf("direct/proxied status/body changed: status=%d/%d bytes=%d/%d", direct.status, got.status, len(direct.body), len(got.body)) + } + if !bytes.Equal(got.body, seen.body) || got.header.Get("Content-Encoding") != seen.encoding || got.contentLength != seen.length { + t.Errorf("no-transform upstream representation changed: bytes=%d/%d encoding=%q/%q length=%d/%d", len(seen.body), len(got.body), seen.encoding, got.header.Get("Content-Encoding"), seen.length, got.contentLength) + } + for _, field := range []string{"Content-Encoding", "Content-Type", "Content-Length", "Cache-Control", "Content-Digest", "ETag", "Vary", "Content-Range"} { + if !reflect.DeepEqual(got.header.Values(field), direct.header.Values(field)) { + t.Errorf("direct/proxied %s changed: %q/%q", field, direct.header.Values(field), got.header.Values(field)) + } + } + if got.header.Get("Content-Digest") != representationDigest(got.body) { + t.Errorf("forwarded Content-Digest does not describe delivered message content: declared=%q actual=%q", got.header.Get("Content-Digest"), representationDigest(got.body)) + } +} + +func representationDigest(content []byte) string { + sum := sha256.Sum256(content) + return "sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":" +} diff --git a/internal/tunnel/web_cover_representation_test.go b/internal/tunnel/web_cover_representation_test.go new file mode 100644 index 0000000..b684705 --- /dev/null +++ b/internal/tunnel/web_cover_representation_test.go @@ -0,0 +1,267 @@ +package tunnel + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "sync/atomic" + "testing" + "time" + + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" + "github.com/cppla/autocar/internal/cover" +) + +// The frontend transports disable automatic compression and decompression. +// Each constructor uses its real nil-transport path, so this exercises the +// default origin transport rather than a test-configured replacement. +func TestWebCoverRepresentationTransparentAcrossProtocols(t *testing.T) { + identity := []byte("ordinary site

representation bytes stay intact

\n") + var encoded bytes.Buffer + zipped := gzip.NewWriter(&encoded) + if _, err := zipped.Write(identity); err != nil { + t.Fatal(err) + } + if err := zipped.Close(); err != nil { + t.Fatal(err) + } + gzipBody := append([]byte(nil), encoded.Bytes()...) + for _, constructor := range []string{"fixed_origin", "public_origin"} { + t.Run(constructor, func(t *testing.T) { + for _, proto := range []int{1, 2, 3} { + t.Run(fmt.Sprintf("h%d", proto), func(t *testing.T) { + for _, invalidCredential := range []bool{false, true} { + t.Run(fmt.Sprintf("invalid_ticket_%t", invalidCredential), func(t *testing.T) { + webCoverRepresentationExchange(t, constructor, proto, invalidCredential, identity, gzipBody) + }) + } + }) + } + }) + } +} + +type webCoverRepresentationObservation struct { + phase, encoding, host, method, uri, credential string + physical any + joined <-chan struct{} +} + +func webCoverRepresentationExchange(t *testing.T, constructor string, proto int, invalidCredential bool, identity, gzipBody []byte) { + t.Helper() + originRequests := make(chan webCoverRepresentationObservation, 2) + coverRequests := make(chan webCoverRepresentationObservation, 2) + var unexpected atomic.Int64 + origin := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + defer close(joined) + phase, encoding := r.Header.Get("X-Representation-Phase"), r.Header.Get("Accept-Encoding") + observation := webCoverRepresentationObservation{ + phase: phase, encoding: encoding, host: r.Host, method: r.Method, uri: r.RequestURI, + credential: r.Header.Get("Proxy-Authorization"), joined: joined, + } + select { + case originRequests <- observation: + default: + unexpected.Add(1) + } + body := identity + if encoding == "gzip" { + body = gzipBody + w.Header().Set("Content-Encoding", "gzip") + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.Header().Set("Content-Digest", webCoverRepresentationDigest(body)) + w.Header().Set("ETag", webCoverRepresentationETag(body)) + w.Header().Set("Vary", "Accept-Encoding") + w.Header().Set("Cache-Control", "public, max-age=120, no-transform") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body) + })) + origin.Config.ReadHeaderTimeout = 2 * time.Second + origin.Config.ReadTimeout = 2 * time.Second + origin.Config.WriteTimeout = 2 * time.Second + origin.Config.IdleTimeout = 2 * time.Second + origin.Start() + // Closing the exact owned origin also releases the nil-transport proxy's + // idle origin sockets. Its server waits for handlers; no unbounded worker + // or response-body gate is used in this fixture. + t.Cleanup(func() { origin.CloseClientConnections(); origin.Close() }) + target, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + // Both routing and the configured public authority stay numeric owned + // loopback addresses, even if an authority regression occurs. + public := &url.URL{Scheme: "https", Host: target.Host} + var website http.Handler + if constructor == "public_origin" { + website, err = cover.NewReverseProxyHandlerWithPublicOrigin(target, public, nil) + } else { + website, err = cover.NewReverseProxyHandler(target, nil) + } + if err != nil { + t.Fatal(err) + } + observingCover := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + defer close(joined) + var physical any + if proto == 3 { + physical, _ = r.Context().Value(webH3ConnectionContextKey{}).(*quic.Conn) + } else { + physical, _ = r.Context().Value(webTLSConnectionContextKey{}).(*tls.Conn) + } + observation := webCoverRepresentationObservation{ + phase: r.Header.Get("X-Representation-Phase"), credential: r.Header.Get("Proxy-Authorization"), + physical: physical, joined: joined, + } + select { + case coverRequests <- observation: + default: + unexpected.Add(1) + } + website.ServeHTTP(w, r) + }) + var dials, resolves atomic.Int64 + // This existing helper owns a real shared-port TCP/UDP server and joins its + // Serve worker within three seconds after exact server Close/cancellation. + server, clientTLS := webAltSvcCombinedServer(t, observingCover, webAltSvcForbiddenDialer(&dials), &resolves) + rt := webAltSvcPublicTransport(t, clientTLS, proto) + switch client := rt.(type) { + case *http.Transport: + client.DisableCompression = true + case *http3.Transport: + client.DisableCompression = true + default: + t.Fatalf("unexpected frontend transport type %T", rt) + } + var firstPhysical any + for _, phase := range []string{"identity", "gzip"} { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + request, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+webAltSvcAddress(server, proto)+"/page?phase="+phase, nil) + if err != nil { + cancel() + t.Fatal(err) + } + request.Host = public.Host + request.Header.Set("X-Representation-Phase", phase) + if phase == "gzip" { + request.Header.Set("Accept-Encoding", "gzip") + } + if invalidCredential { + request.Header.Set("Proxy-Authorization", "Bearer fictional-invalid-ticket") + } + response, err := rt.RoundTrip(request) + if err != nil { + cancel() + t.Fatal(err) + } + body, readErr := io.ReadAll(io.LimitReader(response.Body, 4096)) + closeErr := response.Body.Close() + cancel() + if readErr != nil || closeErr != nil { + t.Fatalf("bounded public response Read/Close=%v/%v", readErr, closeErr) + } + if response.StatusCode != 200 || response.ProtoMajor != proto || response.TLS == nil || response.Uncompressed { + t.Errorf("actual public status/protocol/TLS/auto-decoded=%d/%s/%t/%t", response.StatusCode, response.Proto, response.TLS != nil, response.Uncompressed) + } + wantBody, wantEncoding := identity, "" + if phase == "gzip" { + wantBody, wantEncoding = gzipBody, "gzip" + } + if !bytes.Equal(body, wantBody) { + t.Errorf("%s wire representation differs: actual %d bytes, expected %d", phase, len(body), len(wantBody)) + } + for field, want := range map[string]string{ + "Content-Encoding": wantEncoding, + "Content-Digest": webCoverRepresentationDigest(wantBody), + "ETag": webCoverRepresentationETag(wantBody), + "Vary": "Accept-Encoding", + "Cache-Control": "public, max-age=120, no-transform", + "Content-Type": "text/html; charset=utf-8", + "Content-Length": strconv.Itoa(len(wantBody)), + "Alt-Svc": webH3AltSvcValue(server.UDPAddr()), + } { + if got := response.Header.Get(field); got != want { + t.Errorf("%s %s=%q, want %q", phase, field, got, want) + } + } + if response.ContentLength != int64(len(wantBody)) || response.Header.Get(webAuthResponseHeader) != "" || response.Header.Get("Proxy-Authenticate") != "" { + t.Errorf("%s representation length/private proof/challenge changed", phase) + } + gotOrigin := webCoverRepresentationReceive(t, originRequests) + if gotOrigin.phase != phase || gotOrigin.encoding != wantEncoding || gotOrigin.host != target.Host || gotOrigin.method != http.MethodGet || gotOrigin.uri != "/page?phase="+phase || gotOrigin.credential != "" { + t.Errorf("actual origin phase/AE/Host/method/URI/credential=%q/%q/%q/%q/%q/%q", gotOrigin.phase, gotOrigin.encoding, gotOrigin.host, gotOrigin.method, gotOrigin.uri, gotOrigin.credential) + } + webCoverRepresentationJoin(t, gotOrigin.joined, "origin handler") + gotCover := webCoverRepresentationReceive(t, coverRequests) + if gotCover.phase != phase || gotCover.credential != "" { + t.Error("cover request phase changed or tunnel credential survived") + } + // Typed nils inside an interface are not a usable physical witness. + switch physical := gotCover.physical.(type) { + case *tls.Conn: + if physical == nil { + t.Fatal("cover did not observe its actual TLS connection") + } + case *quic.Conn: + if physical == nil { + t.Fatal("cover did not observe its actual QUIC connection") + } + default: + t.Fatalf("cover physical identity has unexpected type %T", gotCover.physical) + } + if firstPhysical == nil { + firstPhysical = gotCover.physical + } else if firstPhysical != gotCover.physical { + t.Error("second ordinary GET did not reuse the same physical frontend connection") + } + webCoverRepresentationJoin(t, gotCover.joined, "cover handler") + } + if dials.Load() != 0 || resolves.Load() != 0 || unexpected.Load() != 0 || len(originRequests) != 0 || len(coverRequests) != 0 { + t.Errorf("tunnel dial/resolve/unexpected/remaining origin/cover=%d/%d/%d/%d/%d", dials.Load(), resolves.Load(), unexpected.Load(), len(originRequests), len(coverRequests)) + } + t.Logf("actual two-GET identity/gzip exchange observed; tunnel dial/resolve counters=%d/%d; physical reuse oracle checked", dials.Load(), resolves.Load()) +} + +func webCoverRepresentationDigest(body []byte) string { + sum := sha256.Sum256(body) + return "sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":" +} + +func webCoverRepresentationETag(body []byte) string { + sum := sha256.Sum256(body) + return fmt.Sprintf("\"sha256-%x\"", sum) +} + +func webCoverRepresentationReceive(t *testing.T, observations <-chan webCoverRepresentationObservation) webCoverRepresentationObservation { + t.Helper() + select { + case observation := <-observations: + return observation + case <-time.After(time.Second): + t.Fatal("completed public response has no actual handler observation") + return webCoverRepresentationObservation{} + } +} + +func webCoverRepresentationJoin(t *testing.T, joined <-chan struct{}, name string) { + t.Helper() + select { + case <-joined: + case <-time.After(time.Second): + t.Errorf("%s did not return after its actual response", name) + } +} diff --git a/internal/tunnel/web_h3_admission_error_test.go b/internal/tunnel/web_h3_admission_error_test.go new file mode 100644 index 0000000..ef6f330 --- /dev/null +++ b/internal/tunnel/web_h3_admission_error_test.go @@ -0,0 +1,303 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "io" + "net" + "net/http" + "net/netip" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" + "github.com/cppla/autocar/internal/transport" +) + +type webH3AdmissionErrorPeer struct { + packet net.PacketConn + transport *quic.Transport + conn *quic.Conn + http *http3.Transport + client *http3.ClientConn + closeOnce sync.Once + closeErr error +} + +func (p *webH3AdmissionErrorPeer) close() error { + p.closeOnce.Do(func() { + if p.conn != nil { + p.closeErr = errors.Join(p.closeErr, p.conn.CloseWithError(quic.ApplicationErrorCode(http3.ErrCodeNoError), "")) + } + if p.http != nil { + p.closeErr = errors.Join(p.closeErr, p.http.Close()) + } + p.closeErr = errors.Join(p.closeErr, p.transport.Close(), p.packet.Close()) + }) + return p.closeErr +} + +type webH3AdmissionErrorAccepted struct { + conn *quic.Conn + auth *webServerConnectionAuth +} + +type webH3AdmissionErrorFixture struct { + t *testing.T + ctx context.Context + clientTLS *tls.Config + server *WebH3Server + peers []*webH3AdmissionErrorPeer + mu sync.Mutex + accepted []webH3AdmissionErrorAccepted + covers atomic.Int64 + dials atomic.Int64 + resolves atomic.Int64 + encoding atomic.Bool +} + +func newWebH3AdmissionErrorFixture(t *testing.T, global, perSource int) *webH3AdmissionErrorFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second) + serverTLS, clientTLS := testTLSConfigs(t) + clientTLS = clientTLS.Clone() + clientTLS.NextProtos = []string{http3.NextProtoH3} + f := &webH3AdmissionErrorFixture{t: t, ctx: ctx, clientTLS: clientTLS} + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + MaxConnections: global, MaxClientConnections: perSource, + HandshakeTimeout: time.Second, DialTimeout: time.Second, + Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + f.dials.Add(1) + return nil, errors.New("admission fixture forbids destination dialing") + }), + UDPResolver: UDPResolverFunc(func(context.Context, string) ([]netip.AddrPort, error) { + f.resolves.Add(1) + return nil, errors.New("admission fixture forbids destination resolution") + }), + Cover: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.covers.Add(1) + if r.Header.Get("Accept-Encoding") != "" { + f.encoding.Store(true) + } + w.Header().Set("Content-Type", "text/plain") + w.Header().Set("X-Owned-Cover", "admission") + _, _ = io.WriteString(w, "ordinary:"+r.URL.Path) + }), + }) + if err != nil { + cancel() + t.Fatal(err) + } + f.server = server + originalContext := server.server.ConnContext + // Installed before Serve. Only a genuinely admitted physical connection + // reaches this hook; the original transport-owned auth state is preserved. + server.server.ConnContext = func(ctx context.Context, conn *quic.Conn) context.Context { + ctx = originalContext(ctx, conn) + auth, _ := ctx.Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + f.mu.Lock() + f.accepted = append(f.accepted, webH3AdmissionErrorAccepted{conn, auth}) + f.mu.Unlock() + return ctx + } + serveResult, serveJoined := make(chan error, 1), make(chan struct{}) + go func() { + defer close(serveJoined) + serveResult <- server.Serve(ctx) + }() + t.Cleanup(func() { + for _, peer := range f.peers { + if err := peer.close(); err != nil { + t.Errorf("owned peer cleanup: %v", err) + } + } + if err := server.Close(); err != nil { + t.Errorf("owned server cleanup: %v", err) + } + // Finish exact-owned socket/connection Close before canceling Serve; + // otherwise its Shutdown can race our Close on the packet socket. + cancel() + select { + case <-serveJoined: + if err := <-serveResult; err != nil { + t.Errorf("owned Serve: %v", err) + } + case <-time.After(2 * time.Second): + t.Error("owned Serve worker did not join") + } + f.waitReleased() + }) + return f +} + +func (f *webH3AdmissionErrorFixture) dial() (*webH3AdmissionErrorPeer, error) { + f.t.Helper() + packet, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + f.t.Fatal(err) + } + peer := &webH3AdmissionErrorPeer{packet: packet, transport: &quic.Transport{Conn: packet}} + f.peers = append(f.peers, peer) + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + peer.conn, err = peer.transport.Dial(ctx, f.server.Addr(), f.clientTLS, &quic.Config{ + Versions: []quic.Version{quic.Version1}, HandshakeIdleTimeout: time.Second, MaxIdleTimeout: 3 * time.Second, + }) + return peer, err +} + +func (f *webH3AdmissionErrorFixture) get(peer *webH3AdmissionErrorPeer, path string) { + f.t.Helper() + // NewClientConn does not register this connection in a Transport pool. The + // fixture explicitly owns and closes its QUIC connection/Transport/socket. + peer.http = &http3.Transport{DisableCompression: true, MaxResponseHeaderBytes: defaultWebClientMaxResponseHeaderBytes} + peer.client = peer.http.NewClientConn(peer.conn) + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + request, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+f.server.Addr().String()+path, nil) + if err != nil { + f.t.Fatal(err) + } + response, err := peer.client.RoundTrip(request) + if err != nil { + f.t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + closeErr := response.Body.Close() + if readErr != nil || closeErr != nil || response.StatusCode != http.StatusOK || response.ProtoMajor != 3 || + response.Header.Get("X-Owned-Cover") != "admission" || string(body) != "ordinary:"+path { + f.t.Fatalf("real H3 GET status=%d protocol=%d body=%q read=%v close=%v", response.StatusCode, response.ProtoMajor, body, readErr, closeErr) + } + f.t.Logf("healthy real H3 GET %s: status=200 complete body=%q", path, body) +} + +func (f *webH3AdmissionErrorFixture) sourceKey() string { + return tlsSourceKey(&net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) +} + +func (f *webH3AdmissionErrorFixture) waitReleased() { + f.t.Helper() + // Low-frequency observation of actual release state, not a mock Close or + // an entry witness. An independent cleanup budget also works after f.ctx + // has been canceled. No packet retry or sleep-based scheduling assumption. + deadline := time.NewTimer(2 * time.Second) + defer deadline.Stop() + tick := time.NewTicker(10 * time.Millisecond) + defer tick.Stop() + for { + if len(f.server.listener.admission.slots) == 0 && f.server.listener.admission.clients.count(f.sourceKey()) == 0 { + return + } + select { + case <-tick.C: + case <-deadline.C: + f.t.Errorf("actual admission did not release: global=%d source=%d", len(f.server.listener.admission.slots), f.server.listener.admission.clients.count(f.sourceKey())) + return + } + } +} + +func (f *webH3AdmissionErrorFixture) assertState(covers, accepted, occupied int) { + f.t.Helper() + f.mu.Lock() + connections := append([]webH3AdmissionErrorAccepted(nil), f.accepted...) + f.mu.Unlock() + for _, connection := range connections { + if connection.auth == nil || connection.auth.phaseSnapshot() != webServerConnectionAuthFresh { + f.t.Error("public GET changed actual physical connection authentication state") + } + } + handler := f.server.server.Handler.(*webTunnelHandler) + handler.auth.mu.Lock() + nonces := len(handler.auth.nonces) + handler.auth.mu.Unlock() + if f.covers.Load() != int64(covers) || len(connections) != accepted || nonces != 0 || + len(handler.core.sem) != 0 || f.dials.Load() != 0 || f.resolves.Load() != 0 || f.encoding.Load() || + len(f.server.listener.admission.slots) != occupied || f.server.listener.admission.clients.count(f.sourceKey()) != occupied { + f.t.Errorf("state cover=%d/%d admitted=%d/%d accepted-nonces=%d tunnel-slots=%d dials=%d resolves=%d compressed=%t physical-slots=%d/%d source-slots=%d/%d", + f.covers.Load(), covers, len(connections), accepted, nonces, len(handler.core.sem), f.dials.Load(), f.resolves.Load(), f.encoding.Load(), + len(f.server.listener.admission.slots), occupied, f.server.listener.admission.clients.count(f.sourceKey()), occupied) + } +} + +func TestWebH3AdmissionRejectsExcessiveLoad(t *testing.T) { + for _, test := range []struct { + name string + global int + }{ + {name: "global_capacity", global: 1}, + {name: "per_source_capacity", global: 2}, + } { + t.Run(test.name, func(t *testing.T) { + f := newWebH3AdmissionErrorFixture(t, test.global, 1) + first, err := f.dial() + if err != nil { + t.Fatal(err) + } + f.get(first, "/first") + f.assertState(1, 1, 1) + f.mu.Lock() + if len(f.accepted) != 1 { + f.mu.Unlock() + t.Fatal("first GET did not have exactly one actual admitted physical connection") + } + firstServer := f.accepted[0].conn + f.mu.Unlock() + + second, err := f.dial() + if err == nil { + if second.conn == nil || second.conn == first.conn { + t.Fatal("second dial did not create an independent physical QUIC connection") + } + select { + case <-second.conn.Context().Done(): + err = context.Cause(second.conn.Context()) + case <-f.ctx.Done(): + t.Fatal("second physical connection was not rejected within independent fixture budget") + } + } + var rejection *quic.ApplicationError + if !errors.As(err, &rejection) || !rejection.Remote { + t.Errorf("second physical connection did not receive a remote application rejection: %v", err) + } else { + t.Logf("actual remote rejection error-code=%#x", uint64(rejection.ErrorCode)) + if rejection.ErrorCode != quic.ApplicationErrorCode(http3.ErrCodeExcessiveLoad) { + // Keep the real release/recovery controls running on the old + // wrong-code baseline; do not inspect a private reason string. + t.Errorf("actual H3 admission code=%#x want=%#x (H3_EXCESSIVE_LOAD)", uint64(rejection.ErrorCode), uint64(http3.ErrCodeExcessiveLoad)) + } + } + f.assertState(1, 1, 1) + if err := second.close(); err != nil { + t.Fatal(err) + } + if err := first.close(); err != nil { + t.Fatal(err) + } + select { + case <-firstServer.Context().Done(): + case <-f.ctx.Done(): + t.Fatal("actual first server connection did not close") + } + f.waitReleased() + f.assertState(1, 1, 0) + third, err := f.dial() + if err != nil { + t.Fatal(err) + } + f.get(third, "/third") + f.assertState(2, 2, 1) + if err := third.close(); err != nil { + t.Fatal(err) + } + f.waitReleased() + f.assertState(2, 2, 0) + t.Logf("healthy recovery: cover calls=2, admitted connections=2, source/global slots=0, destination calls=0/0; limit=%d/1", test.global) + }) + } +} diff --git a/internal/tunnel/web_limits.go b/internal/tunnel/web_limits.go index 8d01e32..bdd2f4a 100644 --- a/internal/tunnel/web_limits.go +++ b/internal/tunnel/web_limits.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" ) // webConnectionAdmission bounds public web-cover connections before they can @@ -125,7 +126,9 @@ func (l *webAdmissionQUICListener) Accept(ctx context.Context) (*quic.Conn, erro } release, ok := l.admission.acquire(conn.RemoteAddr()) if !ok { - _ = conn.CloseWithError(quic.ApplicationErrorCode(connectionRejected), "connection unavailable") + // This listener speaks HTTP/3, not the native relay protocol. Its + // connection limit is an overload condition, not an internal error. + _ = conn.CloseWithError(quic.ApplicationErrorCode(http3.ErrCodeExcessiveLoad), "connection unavailable") continue } go func() { From b16c0338eda66dd60638a987e07718c11d017416 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 07:32:09 +0800 Subject: [PATCH 2/2] docs(web): clarify custom transport policy wording --- docs/WEB_COVER.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index 7ed446f..fa8ab58 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -218,7 +218,7 @@ the origin's encoded bytes, `Content-Encoding`, length, digest and ETag remain together. This avoids rewriting `no-transform` content while retaining metadata for the old bytes; see [HTTP message transformations](https://www.rfc-editor.org/rfc/rfc9110.html#section-7.7). It does not verify the website's digest or establish browser-like fingerprints. -Applications supplying a custom RoundTripper retain its own negotiation/decoding +Applications supplying a custom RoundTripper retain their own negotiation/decoding policy and must configure transparent forwarding themselves. Static cover, authenticated tunnel payloads and the published v1.0.1 binary are unchanged.