diff --git a/cmd/autocar/preflight_test.go b/cmd/autocar/preflight_test.go index 2bbb0b0..92f5330 100644 --- a/cmd/autocar/preflight_test.go +++ b/cmd/autocar/preflight_test.go @@ -245,6 +245,8 @@ func TestPreflightServerRejectsInvalidConfiguration(t *testing.T) { {"handshake timeout", "--handshake-timeout", []string{"--handshake-timeout", "-1s"}}, {"destination write timeout", "--destination-write-timeout", []string{"--destination-write-timeout", "-1s"}}, {"connections", "--max-connections", []string{"--max-connections", "0"}}, + {"fallback above connection limit", "--max-client-fallback-connections", []string{"--max-connections", "1", "--max-client-connections", "1", "--max-streams", "3", "--max-client-fallback-connections", "2"}}, + {"fallback above stream limit", "--max-client-fallback-connections", []string{"--max-streams", "1", "--max-client-fallback-connections", "2"}}, {"UDP sessions", "--max-client-udp-sessions", []string{"--max-udp-sessions", "1"}}, {"rate bound", "protocol maximum", []string{"--pacing", "fixed-rate", "--max-upload-mbps", "8000001", "--max-download-mbps", "1"}}, {"web missing root", "static cover", []string{"--protocol", "web", "--cover-root", missing}}, diff --git a/cmd/autocar/server.go b/cmd/autocar/server.go index 53aa648..a91329c 100644 --- a/cmd/autocar/server.go +++ b/cmd/autocar/server.go @@ -41,8 +41,8 @@ func runServer(parent context.Context, args []string) error { deniedCIDRsText := fs.String("deny-cidrs", "", "additional comma-separated denied destination CIDRs/IPs") maxStreams := fs.Int("max-streams", 1024, "maximum active tunnel streams") maxConnections := fs.Int("max-connections", 256, "global maximum accepted relay connections") - maxClientConnections := fs.Int("max-client-connections", 32, "maximum relay connections per source IPv4 or IPv6 /64") - maxClientFallbackConnections := fs.Int("max-client-fallback-connections", 0, "native protocol: maximum TLS fallback connections per source IPv4 or IPv6 /64; zero uses min(32, --max-streams)") + maxClientConnections := fs.Int("max-client-connections", 32, "maximum connections per source IPv4 or IPv6 /64: native QUIC, or shared web H2/H3") + maxClientFallbackConnections := fs.Int("max-client-fallback-connections", 0, "native protocol: maximum TLS fallback connections per source IPv4 or IPv6 /64; zero uses min(32, --max-streams, --max-connections)") maxUDPSessions := fs.Int("max-udp-sessions", 256, "maximum live UDP associations across all QUIC clients") maxClientUDPSessions := fs.Int("max-client-udp-sessions", 32, "maximum live UDP associations per source IPv4 or IPv6 /64") maxUDPDestinations := fs.Int("max-udp-destinations", 64, "maximum numeric destinations authorized per UDP association") @@ -90,8 +90,8 @@ func runServer(parent context.Context, args []string) error { if *maxClientConnections <= 0 || *maxClientConnections > *maxConnections { return errors.New("--max-client-connections must be positive and no greater than --max-connections") } - if !*disableFallback && (*maxClientFallbackConnections < 0 || *maxClientFallbackConnections > *maxStreams) { - return errors.New("--max-client-fallback-connections must be zero or positive and no greater than --max-streams") + if !*disableFallback && (*maxClientFallbackConnections < 0 || *maxClientFallbackConnections > min(*maxStreams, *maxConnections)) { + return errors.New("--max-client-fallback-connections must be zero or positive and no greater than --max-streams or --max-connections") } if serverProtocol == "web" && *maxClientFallbackConnections != 0 { return errors.New("--max-client-fallback-connections is valid only with --protocol=native; web H2/H3 share --max-client-connections") @@ -233,6 +233,10 @@ func runServer(parent context.Context, args []string) error { return nil } + connectionAdmission, err := tunnel.NewConnectionAdmission(*maxConnections) + if err != nil { + return fmt.Errorf("configure connection admission: %w", err) + } quicServer, err := tunnel.ListenQUIC(tunnel.QUICServerConfig{ Address: *listen, Token: token, @@ -248,6 +252,7 @@ func runServer(parent context.Context, args []string) error { MaxConcurrentStreams: *maxStreams, StreamAdmission: streamAdmission, MaxConnections: *maxConnections, + ConnectionAdmission: connectionAdmission, MaxClientConnections: *maxClientConnections, MaxUDPSessions: *maxUDPSessions, MaxClientUDPSessions: *maxClientUDPSessions, @@ -274,6 +279,8 @@ func runServer(parent context.Context, args []string) error { DestinationWriteTimeout: *destinationWriteTimeout, MaxConcurrentStreams: *maxStreams, StreamAdmission: streamAdmission, + MaxConnections: *maxConnections, + ConnectionAdmission: connectionAdmission, MaxClientConnections: *maxClientFallbackConnections, }) if err != nil { diff --git a/cmd/autocar/server_connection_admission_test.go b/cmd/autocar/server_connection_admission_test.go new file mode 100644 index 0000000..43b946f --- /dev/null +++ b/cmd/autocar/server_connection_admission_test.go @@ -0,0 +1,147 @@ +package main + +import ( + "context" + "crypto/tls" + "errors" + "log/slog" + "net" + "testing" + "time" + + "github.com/cppla/autocar/internal/protocol" + "github.com/cppla/autocar/internal/security" + quic "github.com/quic-go/quic-go" +) + +func TestNativeServerMaxConnectionsIncludesTLSFallback(t *testing.T) { + clearPreflightEnvironment(t) + files := newPreflightFiles(t) + addresses := make(chan nativeListenerAddress, 2) + previousLogger := slog.Default() + slog.SetDefault(slog.New(&nativeAddressHandler{addresses: addresses})) + t.Cleanup(func() { slog.SetDefault(previousLogger) }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + var serverErr error + go func() { + defer close(done) + serverErr = runServer(ctx, []string{ + "--protocol=native", "--listen=127.0.0.1:0", "--tcp-listen=127.0.0.1:0", + "--cert", files.cert, "--key", files.key, "--token-file", files.token, + "--max-connections=1", "--max-client-connections=1", "--max-streams=3", + }) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + if serverErr != nil { + t.Errorf("server shutdown: %v", serverErr) + } + case <-time.After(3 * time.Second): + t.Error("server did not stop") + } + }) + listeners := make(map[string]string) + for len(listeners) < 2 { + select { + case listener := <-addresses: + listeners[listener.transport] = listener.address + case <-done: + t.Fatalf("server failed before listening: %v", serverErr) + case <-time.After(3 * time.Second): + t.Fatal("native listeners did not start") + } + } + roots, err := security.LoadCertPool(files.cert) + if err != nil { + t.Fatal(err) + } + clientTLS := &tls.Config{ + MinVersion: tls.VersionTLS13, RootCAs: roots, ServerName: "relay.invalid", + NextProtos: []string{protocol.ALPN}, + } + dialTLS := func() (*tls.Conn, error) { + return tls.DialWithDialer(&net.Dialer{Timeout: time.Second}, "tcp", listeners["tls"], clientTLS.Clone()) + } + first, err := dialTLS() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = first.Close() }) + // A verified handshake proves the accepted TLS connection is holding the + // connection budget, even though it has not consumed an active stream. + second, err := dialTLS() + if second != nil { + _ = second.Close() + } + if err == nil { + t.Fatal("--max-connections=1 admitted a second TLS fallback connection with --max-streams=3") + } + if timeout, ok := err.(net.Error); ok && timeout.Timeout() { + t.Fatalf("second TLS connection timed out instead of being rejected: %v", err) + } + quicCtx, stopQUIC := context.WithTimeout(context.Background(), 2*time.Second) + defer stopQUIC() + q, err := quic.DialAddr(quicCtx, listeners["quic"], clientTLS.Clone(), nil) + if q != nil { + defer q.CloseWithError(0, "test cleanup") + select { + case <-q.Context().Done(): + err = context.Cause(q.Context()) + case <-quicCtx.Done(): + t.Fatal("QUIC was admitted while TLS held the global connection budget") + } + } + var appErr *quic.ApplicationError + if !errors.As(err, &appErr) || appErr.ErrorMessage != "connection limit reached" { + t.Fatalf("QUIC rejection = %v, want global connection limit", err) + } + _ = first.Close() + deadline := time.Now().Add(3 * time.Second) + for { + replacement, err := dialTLS() + if err == nil { + _ = replacement.Close() + break + } + if time.Now().After(deadline) { + t.Fatalf("TLS connection slot was not returned after close: %v", err) + } + time.Sleep(5 * time.Millisecond) + } +} + +type nativeListenerAddress struct { + transport string + address string +} + +type nativeAddressHandler struct { + addresses chan<- nativeListenerAddress +} + +func (h *nativeAddressHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *nativeAddressHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *nativeAddressHandler) WithGroup(string) slog.Handler { return h } +func (h *nativeAddressHandler) Handle(_ context.Context, record slog.Record) error { + if record.Message != "relay listener started" { + return nil + } + var listener nativeListenerAddress + record.Attrs(func(attribute slog.Attr) bool { + switch attribute.Key { + case "transport": + listener.transport = attribute.Value.String() + case "address": + listener.address = attribute.Value.String() + } + return true + }) + select { + case h.addresses <- listener: + default: + } + return nil +} diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 809ab97..2cf7d4d 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -125,9 +125,14 @@ services. Add deployment-specific blocks with `--deny-cidrs` and `--deny-ports`. Resource controls include `--max-connections`, `--max-client-connections`, -`--max-streams` and `--max-client-fallback-connections`. `--max-streams` is one -global admission budget shared by the QUIC and TLS listeners; enabling fallback -does not double the configured stream capacity. +`--max-streams` and `--max-client-fallback-connections`. Native QUIC and TLS +listeners share one `--max-connections` budget for all accepted connections, +including unauthenticated connections and authenticated idle QUIC sessions. +`--max-streams` independently limits active streams across both listeners. +The per-source limits remain separate: `--max-client-connections` caps QUIC +sessions, while `--max-client-fallback-connections` caps TLS connections and +defaults to `min(32, --max-streams, --max-connections)`. An explicit fallback +limit cannot exceed either `--max-streams` or `--max-connections`. QUIC UDP associations have separate bounded controls: diff --git a/internal/proxy/http.go b/internal/proxy/http.go index 9b19595..f2297f5 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -283,10 +283,16 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) { return } defer response.Body.Close() + // Preserve connection-specific trailer names before stripping Connection. + // The body can add both trailer values and previously unknown names at EOF. + blockedTrailers := httpForwardTrailerNominations(response.Header, response.Trailer) removeHopByHopHeaders(response.Header) copyHeaders(w.Header(), response.Header) + declaredTrailers := make(map[string]bool, len(response.Trailer)) for key := range response.Trailer { - if !isHopByHopHeader(key) { + if httpForwardTrailerAllowed(key, blockedTrailers) { + key = textproto.CanonicalMIMEHeaderKey(key) + declaredTrailers[key] = true w.Header().Add("Trailer", key) } } @@ -305,11 +311,22 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) { // response; net/http handles the sentinel without logging a stack trace. panic(http.ErrAbortHandler) } + for key := range httpForwardTrailerNominations(response.Trailer) { + blockedTrailers[key] = true + } + // A newly blocked declared trailer must not inherit an earlier header value. + for key := range declaredTrailers { + w.Header().Del(key) + } for key, values := range response.Trailer { - if isHopByHopHeader(key) { + if !httpForwardTrailerAllowed(key, blockedTrailers) { continue } - w.Header()[textproto.CanonicalMIMEHeaderKey(key)] = append([]string(nil), values...) + key = textproto.CanonicalMIMEHeaderKey(key) + if !declaredTrailers[key] { + key = http.TrailerPrefix + key + } + w.Header()[key] = append([]string(nil), values...) } } diff --git a/internal/proxy/http_response_trailer_test.go b/internal/proxy/http_response_trailer_test.go new file mode 100644 index 0000000..083aed0 --- /dev/null +++ b/internal/proxy/http_response_trailer_test.go @@ -0,0 +1,215 @@ +package proxy + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" +) + +func TestHTTPForwardResponseTrailers(t *testing.T) { + for _, mode := range []string{"declared", "undeclared", "mixed"} { + for _, body := range []string{"body", ""} { + name := mode + "/body" + if body == "" { + name = mode + "/empty" + } + t.Run(name, func(t *testing.T) { + want := make(http.Header) + if mode != "undeclared" { + want["X-Declared"] = []string{"first", "second"} + } + if mode != "declared" { + want["X-Undeclared"] = []string{"late-first", "late-second"} + } + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if mode != "undeclared" { + w.Header().Set("Trailer", "X-Declared") + } + _, _ = io.WriteString(w, body) + // Flush even an empty body so late metadata cannot become + // ordinary headers or an inferred Content-Length response. + w.(http.Flusher).Flush() + for field, values := range want { + if field == "X-Undeclared" { + field = http.TrailerPrefix + field + } + w.Header()[field] = append([]string(nil), values...) + } + })) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + defer client.CloseIdleConnections() + for _, route := range []struct { + name string + client *http.Client + }{{"direct", origin.Client()}, {"proxy", client}} { + t.Run(route.name, func(t *testing.T) { + response, err := route.client.Get(origin.URL) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + for field := range want { + if len(response.Header.Values(field)) != 0 || len(response.Trailer.Values(field)) != 0 { + t.Fatalf("trailer %s published before body EOF: headers=%v trailers=%v", field, response.Header, response.Trailer) + } + } + got, err := io.ReadAll(response.Body) + if err != nil || string(got) != body || response.StatusCode != http.StatusOK { + t.Fatalf("response: status=%d body=%q error=%v", response.StatusCode, got, err) + } + if !reflect.DeepEqual(response.Trailer, want) { + t.Fatalf("trailers=%v, want %v", response.Trailer, want) + } + }) + } + }) + } + } +} + +func TestHTTPForwardResponseTrailerPolicy(t *testing.T) { + // The raw origin deliberately sends fields net/http's ResponseWriter + // refuses to emit, but which its response parser accepts as trailers. + unsafe := []string{ + "Authorization", "Host", "If-Match", "Content-Type", "Content-Length", + "Transfer-Encoding", "Trailer", "Keep-Alive", "Te", "Upgrade", + "Proxy-Authorization", "Proxy-Authenticate", "Proxy-Authentication-Info", "Proxy-Connection", + } + declared := []string{"X-End", "X-Initial-Hop", "X-Late-Hop"} + for _, field := range unsafe { + // These three declarations are rejected by the upstream parser; + // send them only in the terminal trailer block instead. + if field != "Content-Length" && field != "Transfer-Encoding" && field != "Trailer" { + declared = append(declared, field) + } + } + var wire strings.Builder + wire.WriteString("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n") + wire.WriteString("Connection: x-INITIAL-hop, X-Initial-Undeclared\r\n") + // If this declared field is blocked only at EOF, its ordinary header + // value must not be mistaken for a trailer by the downstream writer. + wire.WriteString("X-Late-Hop: initial-header-must-not-become-trailer\r\n") + wire.WriteString("Trailer: " + strings.Join(declared, ", ") + "\r\n\r\n") + wire.WriteString("4\r\nbody\r\n0\r\nX-End: declared-control\r\nX-Late-End: undeclared-control\r\n") + wire.WriteString("Connection: x-LATE-hop, X-Late-Undeclared\r\n") + blocked := append([]string{"X-Initial-Hop", "X-Initial-Undeclared", "X-Late-Hop", "X-Late-Undeclared"}, unsafe...) + for _, field := range blocked { + wire.WriteString(field + ": must-not-forward\r\n") + } + wire.WriteString("\r\n") + origin := httpResponseTrailerOrigin(t, wire.String()) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + defer client.CloseIdleConnections() + for _, route := range []struct { + name string + client *http.Client + }{{"direct", origin.Client()}, {"proxy", client}} { + t.Run(route.name, func(t *testing.T) { + response, err := route.client.Get(origin.URL) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + initialTrailers := response.Trailer.Clone() + body, err := io.ReadAll(response.Body) + if err != nil || string(body) != "body" || response.StatusCode != http.StatusOK { + t.Fatalf("response: status=%d body=%q error=%v", response.StatusCode, body, err) + } + for field, want := range map[string]string{"X-End": "declared-control", "X-Late-End": "undeclared-control"} { + if got := response.Trailer.Get(field); got != want { + t.Errorf("control trailer %s=%q, want %q", field, got, want) + } + } + for _, field := range blocked { + got := response.Trailer.Get(field) + if route.name == "direct" { + if got != "must-not-forward" { + t.Errorf("origin control trailer %s=%q", field, got) + } + } else { + wantHeader := "" + if field == "X-Late-Hop" { + // This ordinary header was sent before the terminal + // Connection nomination became available. + wantHeader = "initial-header-must-not-become-trailer" + } + if got != "" || response.Header.Get(field) != wantHeader { + t.Errorf("blocked %s: header=%q, want %q; trailer=%q, want empty", field, response.Header.Get(field), wantHeader, got) + } + } + } + if route.name == "proxy" { + for _, field := range append([]string{"X-Initial-Hop"}, unsafe...) { + if _, ok := initialTrailers[http.CanonicalHeaderKey(field)]; ok { + t.Errorf("blocked trailer %s announced in response headers", field) + } + } + if response.Header.Get("Connection") != "" || response.Trailer.Get("Connection") != "" { + t.Errorf("Connection leaked: headers=%v trailers=%v", response.Header, response.Trailer) + } + } + }) + } +} + +func TestHTTPForwardResponseTrailersRequireCompleteBody(t *testing.T) { + for _, ending := range []struct { + name string + wire string + }{ + {"missing_terminal_chunk", ""}, + {"incomplete_trailer_block", "0\r\nX-End: must-not-publish\r\nX-Late-End: must-not-publish\r\n"}, + } { + t.Run(ending.name, func(t *testing.T) { + origin := httpResponseTrailerOrigin(t, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nTrailer: X-End\r\n\r\n4\r\nbody\r\n"+ending.wire) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + defer client.CloseIdleConnections() + response, err := client.Get(origin.URL) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if !errors.Is(err, io.ErrUnexpectedEOF) || string(body) != "body" || response.StatusCode != http.StatusOK { + t.Fatalf("incomplete response: status=%d body=%q error=%v", response.StatusCode, body, err) + } + for _, field := range []string{"X-End", "X-Late-End"} { + if response.Header.Get(field) != "" || response.Trailer.Get(field) != "" { + t.Errorf("trailer %s published after incomplete body: headers=%v trailers=%v", field, response.Header, response.Trailer) + } + } + }) + } +} + +func httpResponseTrailerOrigin(t *testing.T, wire string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + conn, rw, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + defer conn.Close() + if _, err := rw.WriteString(wire); err != nil { + t.Error(err) + return + } + if err := rw.Flush(); err != nil { + t.Error(err) + } + })) +} diff --git a/internal/tunnel/connection_admission.go b/internal/tunnel/connection_admission.go new file mode 100644 index 0000000..6d68ae9 --- /dev/null +++ b/internal/tunnel/connection_admission.go @@ -0,0 +1,43 @@ +package tunnel + +import ( + "errors" + "fmt" +) + +// ConnectionAdmission bounds accepted native relay connections, including +// unauthenticated connections and idle QUIC sessions. Pass the same instance +// to QUIC and TLS servers to share one global limit. Closing a server returns +// its slots without closing the budget or affecting other servers using it. +// +// The zero value is invalid; construct values with NewConnectionAdmission. +type ConnectionAdmission struct { + sem chan struct{} +} + +// NewConnectionAdmission creates a connection budget with a positive limit. +func NewConnectionAdmission(limit int) (*ConnectionAdmission, error) { + if limit <= 0 { + return nil, errors.New("tunnel: connection admission limit must be positive") + } + return &ConnectionAdmission{sem: make(chan struct{}, limit)}, nil +} + +func resolveConnectionAdmission(limit, fallback int, admission *ConnectionAdmission) (*ConnectionAdmission, error) { + if limit < 0 { + return nil, errors.New("tunnel: maximum connections cannot be negative") + } + if admission == nil { + if limit == 0 { + limit = fallback + } + return NewConnectionAdmission(limit) + } + if cap(admission.sem) == 0 { + return nil, errors.New("tunnel: invalid zero-value connection admission") + } + if limit != 0 && limit != cap(admission.sem) { + return nil, fmt.Errorf("tunnel: maximum connections (%d) does not match shared connection admission limit (%d)", limit, cap(admission.sem)) + } + return admission, nil +} diff --git a/internal/tunnel/connection_admission_test.go b/internal/tunnel/connection_admission_test.go new file mode 100644 index 0000000..02b9ae6 --- /dev/null +++ b/internal/tunnel/connection_admission_test.go @@ -0,0 +1,288 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "net" + "testing" + "time" + + quic "github.com/quic-go/quic-go" +) + +func TestConnectionAdmissionValidationAndCompatibility(t *testing.T) { + for _, limit := range []int{-1, 0} { + if admission, err := NewConnectionAdmission(limit); err == nil || admission != nil { + t.Fatalf("NewConnectionAdmission(%d) = (%v, %v), want error", limit, admission, err) + } + } + serverTLS, _ := testTLSConfigs(t) + shared, err := NewConnectionAdmission(2) + if err != nil { + t.Fatal(err) + } + for _, test := range []struct { + name string + limit int + admission *ConnectionAdmission + wantErr bool + }{ + {name: "negative", limit: -1, wantErr: true}, + {name: "negative with shared", limit: -1, admission: shared, wantErr: true}, + {name: "zero-value", admission: &ConnectionAdmission{}, wantErr: true}, + {name: "mismatch", limit: 1, admission: shared, wantErr: true}, + {name: "shared inferred", admission: shared}, + {name: "shared explicit", limit: 2, admission: shared}, + {name: "independent default"}, + {name: "independent explicit", limit: 3}, + } { + t.Run(test.name, func(t *testing.T) { + q, qErr := ListenQUIC(QUICServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + MaxConcurrentStreams: 4, MaxConnections: test.limit, ConnectionAdmission: test.admission, + }) + if q != nil { + defer q.Close() + } + s, sErr := ListenTLS(TLSServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + MaxConcurrentStreams: 4, MaxConnections: test.limit, ConnectionAdmission: test.admission, + }) + if s != nil { + defer s.Close() + } + if (qErr != nil) != test.wantErr || (sErr != nil) != test.wantErr { + t.Fatalf("QUIC error = %v, TLS error = %v, wantErr %t", qErr, sErr, test.wantErr) + } + if test.wantErr { + return + } + if test.admission != nil { + if q.connSem != shared.sem || s.connSem != shared.sem { + t.Fatal("native transports do not share the configured connection budget") + } + } else { + if q.connSem == s.connSem { + t.Fatal("independent servers unexpectedly share a connection budget") + } + qLimit, sLimit := test.limit, test.limit + if test.limit == 0 { + qLimit, sLimit = defaultMaxConnections, 4 + } + if cap(q.connSem) != qLimit || cap(s.connSem) != sLimit { + t.Fatalf("connection limits = (%d, %d), want (%d, %d)", cap(q.connSem), cap(s.connSem), qLimit, sLimit) + } + } + }) + } +} + +func TestQUICAndTLSShareGlobalConnectionAdmission(t *testing.T) { + f := newConnectionAdmissionFixture(t, 2) + firstTLS := f.dialTLS() + f.awaitUsage(1) + firstQUIC := f.dialQUIC() + f.awaitUsage(2) + if len(f.tlsServer.core.sem) != 0 || len(f.quicServer.core.sem) != 0 { + t.Fatal("idle unauthenticated connections consumed active-stream slots") + } + f.rejectTLS() + f.rejectQUIC() + f.awaitUsage(2) + + _ = firstTLS.Close() + f.awaitUsage(1) + secondQUIC := f.dialQUIC() + f.awaitUsage(2) + _ = firstQUIC.CloseWithError(0, "test connection release") + f.awaitUsage(1) + _ = f.dialTLS() + f.awaitUsage(2) + _ = secondQUIC.CloseWithError(0, "test connection release") + f.awaitUsage(1) + _ = f.dialQUIC() + f.awaitUsage(2) + + // Closing one transport must release only its own connections. The shared + // budget remains usable by the other transport until it too shuts down. + if err := f.quicServer.Close(); err != nil { + t.Fatal(err) + } + f.awaitUsage(1) + _ = f.dialTLS() + f.awaitUsage(2) + if err := f.tlsServer.Close(); err != nil { + t.Fatal(err) + } + f.awaitUsage(0) +} + +func TestConnectionAdmissionRollsBackPerSourceRejection(t *testing.T) { + for _, mode := range []string{"quic", "tls"} { + t.Run(mode, func(t *testing.T) { + f := newConnectionAdmissionFixture(t, 1) + if mode == "quic" { + _ = f.dialQUIC() + } else { + _ = f.dialTLS() + } + f.awaitUsage(1) + if mode == "quic" { + f.rejectQUIC() + f.awaitUsage(1) + _ = f.dialTLS() + } else { + f.rejectTLS() + f.awaitUsage(1) + _ = f.dialQUIC() + } + f.awaitUsage(2) + }) + } +} + +func TestConnectionAdmissionReturnsFailedTLSHandshake(t *testing.T) { + f := newConnectionAdmissionFixture(t, 2) + conn, err := net.DialTimeout("tcp", f.tlsServer.Addr().String(), time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + f.awaitUsage(1) + if _, err := conn.Write([]byte("invalid TLS handshake\r\n")); err != nil { + t.Fatal(err) + } + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := conn.Read(make([]byte, 1)); err == nil { + t.Fatal("invalid TLS handshake remained open") + } else if timeout, ok := err.(net.Error); ok && timeout.Timeout() { + t.Fatal("invalid TLS handshake was not rejected promptly") + } + f.awaitUsage(0) + _ = f.dialQUIC() + f.awaitUsage(1) +} + +type connectionAdmissionFixture struct { + t *testing.T + admission *ConnectionAdmission + quicServer *QUICServer + tlsServer *TLSServer + clientTLS *tls.Config +} + +func newConnectionAdmissionFixture(t *testing.T, sourceLimit int) *connectionAdmissionFixture { + t.Helper() + serverTLS, clientTLS := testTLSConfigs(t) + shared, err := NewConnectionAdmission(2) + if err != nil { + t.Fatal(err) + } + q, err := ListenQUIC(QUICServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + HandshakeTimeout: 30 * time.Second, MaxConcurrentStreams: 4, + ConnectionAdmission: shared, MaxClientConnections: sourceLimit, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = q.Close() }) + s, err := ListenTLS(TLSServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, + HandshakeTimeout: 30 * time.Second, MaxConcurrentStreams: 4, + ConnectionAdmission: shared, MaxClientConnections: sourceLimit, + }) + if err != nil { + t.Fatal(err) + } + clientTLS, err = clientTLSConfig(clientTLS, s.Addr().String()) + if err != nil { + _ = s.Close() + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + qDone, sDone := make(chan error, 1), make(chan error, 1) + go func() { qDone <- q.Serve(ctx) }() + go func() { sDone <- s.Serve(ctx) }() + t.Cleanup(func() { + cancel() + _ = q.Close() + _ = s.Close() + for name, done := range map[string]<-chan error{"QUIC": qDone, "TLS": sDone} { + select { + case err := <-done: + if err != nil { + t.Errorf("%s Serve: %v", name, err) + } + case <-time.After(3 * time.Second): + t.Errorf("%s Serve did not stop", name) + } + } + if got := len(shared.sem); got != 0 { + t.Errorf("connection admission after shutdown = %d, want 0", got) + } + }) + return &connectionAdmissionFixture{t: t, admission: shared, quicServer: q, tlsServer: s, clientTLS: clientTLS} +} + +func (f *connectionAdmissionFixture) awaitUsage(want int) { + f.t.Helper() + waitForCondition(f.t, 3*time.Second, func() bool { return len(f.admission.sem) == want }, "shared connection budget") +} + +func (f *connectionAdmissionFixture) dialTLS() net.Conn { + f.t.Helper() + conn, err := tls.DialWithDialer(&net.Dialer{Timeout: 2 * time.Second}, "tcp", f.tlsServer.Addr().String(), f.clientTLS.Clone()) + if err != nil { + f.t.Fatal(err) + } + f.t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func (f *connectionAdmissionFixture) rejectTLS() { + f.t.Helper() + conn, err := tls.DialWithDialer(&net.Dialer{Timeout: 2 * time.Second}, "tcp", f.tlsServer.Addr().String(), f.clientTLS.Clone()) + if conn != nil { + _ = conn.Close() + } + if err == nil { + f.t.Fatal("TLS connection was admitted while the connection budget was full") + } + if timeout, ok := err.(net.Error); ok && timeout.Timeout() { + f.t.Fatalf("TLS rejection timed out: %v", err) + } +} + +func (f *connectionAdmissionFixture) dialQUIC() *quic.Conn { + f.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + conn, err := quic.DialAddr(ctx, f.quicServer.Addr().String(), f.clientTLS.Clone(), hardenedQUICClientConfig(nil)) + if err != nil { + f.t.Fatal(err) + } + f.t.Cleanup(func() { _ = conn.CloseWithError(0, "test cleanup") }) + return conn +} + +func (f *connectionAdmissionFixture) rejectQUIC() { + f.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + conn, err := quic.DialAddr(ctx, f.quicServer.Addr().String(), f.clientTLS.Clone(), hardenedQUICClientConfig(nil)) + if conn != nil { + defer conn.CloseWithError(0, "test cleanup") + select { + case <-conn.Context().Done(): + err = context.Cause(conn.Context()) + case <-ctx.Done(): + f.t.Fatal("QUIC connection was not rejected promptly") + } + } + var appErr *quic.ApplicationError + if !errors.As(err, &appErr) || appErr.ErrorCode != connectionRejected { + f.t.Fatalf("QUIC connection rejection = %v, want connection limit", err) + } +} diff --git a/internal/tunnel/quic.go b/internal/tunnel/quic.go index ee01214..cbd9810 100644 --- a/internal/tunnel/quic.go +++ b/internal/tunnel/quic.go @@ -43,8 +43,13 @@ type QUICServerConfig struct { // to the admission limit. Nil preserves the independent-server behavior. StreamAdmission *StreamAdmission // MaxConnections bounds accepted QUIC connections, including authenticated - // idle sessions. Zero uses a conservative default. + // idle sessions. Zero with nil ConnectionAdmission uses a conservative + // default; otherwise zero uses the shared admission limit. MaxConnections int + // ConnectionAdmission optionally shares the accepted-connection budget + // with other native transports. MaxConnections must be zero or match its + // limit. Nil preserves an independent budget for this server. + ConnectionAdmission *ConnectionAdmission // MaxClientConnections bounds concurrent QUIC sessions per source IPv4 or // IPv6 /64. Zero uses a conservative default no larger than MaxConnections. MaxClientConnections int @@ -134,13 +139,11 @@ func ListenQUIC(config QUICServerConfig) (*QUICServer, error) { if _, err := newServerPacingNegotiator(config.Pacing, config.MaxTx, config.MaxRx, config.AllowClientRates); err != nil { return nil, err } - if config.MaxConnections < 0 { - return nil, errors.New("tunnel: maximum QUIC connections cannot be negative") - } - maxConnections := config.MaxConnections - if maxConnections == 0 { - maxConnections = defaultMaxConnections + connectionAdmission, err := resolveConnectionAdmission(config.MaxConnections, defaultMaxConnections, config.ConnectionAdmission) + if err != nil { + return nil, err } + maxConnections := cap(connectionAdmission.sem) maxClientConnections := config.MaxClientConnections if maxClientConnections < 0 { return nil, errors.New("tunnel: maximum QUIC client connections cannot be negative") @@ -167,7 +170,7 @@ func ListenQUIC(config QUICServerConfig) (*QUICServer, error) { ctx: ctx, cancel: cancel, conns: make(map[*quic.Conn]struct{}), - connSem: make(chan struct{}, maxConnections), + connSem: connectionAdmission.sem, streamSem: make(chan struct{}, cap(core.sem)), clients: newSourceConnectionLimiter(maxClientConnections), udp: udp, diff --git a/internal/tunnel/tls.go b/internal/tunnel/tls.go index a11c192..a8e1850 100644 --- a/internal/tunnel/tls.go +++ b/internal/tunnel/tls.go @@ -30,7 +30,17 @@ type TLSServerConfig struct { // StreamAdmission optionally shares the active-stream budget with other // server transports. When set, MaxConcurrentStreams must be zero or equal // to the admission limit. Nil preserves the independent-server behavior. - StreamAdmission *StreamAdmission + StreamAdmission *StreamAdmission + // MaxConnections bounds accepted TLS connections, including those awaiting + // authentication. Zero with nil ConnectionAdmission preserves the previous + // independent budget equal to the active-stream limit. + MaxConnections int + // ConnectionAdmission optionally shares the accepted-connection budget + // with other native transports. MaxConnections must be zero or match its + // limit. Nil preserves an independent budget for this server. + ConnectionAdmission *ConnectionAdmission + // MaxClientConnections separately bounds TLS connections per source IPv4 + // or IPv6 /64. Zero uses min(32, stream limit, connection limit). MaxClientConnections int } @@ -78,15 +88,20 @@ func ListenTLS(config TLSServerConfig) (*TLSServer, error) { if err != nil { return nil, err } + connectionAdmission, err := resolveConnectionAdmission(config.MaxConnections, cap(core.sem), config.ConnectionAdmission) + if err != nil { + return nil, err + } + maxClientLimit := min(cap(core.sem), cap(connectionAdmission.sem)) maxClientConnections := config.MaxClientConnections if maxClientConnections < 0 { return nil, errors.New("tunnel: maximum TLS client connections cannot be negative") } if maxClientConnections == 0 { - maxClientConnections = min(defaultMaxClientConnections, cap(core.sem)) + maxClientConnections = min(defaultMaxClientConnections, maxClientLimit) } - if maxClientConnections > cap(core.sem) { - return nil, fmt.Errorf("tunnel: maximum TLS client connections (%d) exceeds maximum concurrent streams (%d)", maxClientConnections, cap(core.sem)) + if maxClientConnections > maxClientLimit { + return nil, fmt.Errorf("tunnel: maximum TLS client connections (%d) exceeds connection or concurrent stream limit (%d)", maxClientConnections, maxClientLimit) } listener, err := net.Listen("tcp", config.Address) if err != nil { @@ -98,7 +113,7 @@ func ListenTLS(config TLSServerConfig) (*TLSServer, error) { tlsConfig: tlsConfig, core: core, clients: newSourceConnectionLimiter(maxClientConnections), - connSem: make(chan struct{}, cap(core.sem)), + connSem: connectionAdmission.sem, ctx: ctx, cancel: cancel, conns: make(map[net.Conn]struct{}), diff --git a/internal/tunnel/tls_limits_test.go b/internal/tunnel/tls_limits_test.go index 9d9a88f..513b906 100644 --- a/internal/tunnel/tls_limits_test.go +++ b/internal/tunnel/tls_limits_test.go @@ -11,14 +11,17 @@ func TestTLSServerClientConnectionLimitValidationAndDefaults(t *testing.T) { serverTLS, _ := testTLSConfigs(t) for _, test := range []struct { - name string - streams int - clients int - wantClient int + name string + streams int + connections int + clients int + wantClient int }{ {name: "default below 32", streams: 2, wantClient: 2}, {name: "default capped at 32", streams: 64, wantClient: 32}, {name: "explicit", streams: 4, clients: 3, wantClient: 3}, + {name: "default capped at connections", streams: 4, connections: 1, wantClient: 1}, + {name: "explicit connection budget", streams: 4, connections: 3, clients: 3, wantClient: 3}, } { t.Run(test.name, func(t *testing.T) { server, err := ListenTLS(TLSServerConfig{ @@ -26,6 +29,7 @@ func TestTLSServerClientConnectionLimitValidationAndDefaults(t *testing.T) { Token: testToken, TLSConfig: serverTLS, MaxConcurrentStreams: test.streams, + MaxConnections: test.connections, MaxClientConnections: test.clients, }) if err != nil { @@ -39,12 +43,15 @@ func TestTLSServerClientConnectionLimitValidationAndDefaults(t *testing.T) { } for _, test := range []struct { - name string - streams int - clients int + name string + streams int + connections int + clients int }{ {name: "negative", streams: 4, clients: -1}, {name: "exceeds global limit", streams: 2, clients: 3}, + {name: "exceeds connection limit", streams: 4, connections: 2, clients: 3}, + {name: "exceeds stream limit with explicit connections", streams: 2, connections: 4, clients: 3}, } { t.Run(test.name, func(t *testing.T) { server, err := ListenTLS(TLSServerConfig{ @@ -52,6 +59,7 @@ func TestTLSServerClientConnectionLimitValidationAndDefaults(t *testing.T) { Token: testToken, TLSConfig: serverTLS, MaxConcurrentStreams: test.streams, + MaxConnections: test.connections, MaxClientConnections: test.clients, }) if server != nil {