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() {