From a54d96cd72b453f15807b665ceed83ddbd1c2b0e Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Tue, 15 Sep 2026 00:28:01 +0900 Subject: [PATCH 1/9] [proxy] Make the upstream HTTP version configurable (#7410) --- proxy/internal/roundtrip/multi.go | 9 +- proxy/internal/roundtrip/multi_test.go | 64 ++- proxy/internal/roundtrip/netbird.go | 10 +- proxy/internal/roundtrip/transport.go | 108 ++++ proxy/internal/roundtrip/upstream.go | 455 +++++++++++++++++ proxy/internal/roundtrip/upstream_test.go | 587 ++++++++++++++++++++++ 6 files changed, 1218 insertions(+), 15 deletions(-) create mode 100644 proxy/internal/roundtrip/upstream.go create mode 100644 proxy/internal/roundtrip/upstream_test.go diff --git a/proxy/internal/roundtrip/multi.go b/proxy/internal/roundtrip/multi.go index 567249437..1abf54a8d 100644 --- a/proxy/internal/roundtrip/multi.go +++ b/proxy/internal/roundtrip/multi.go @@ -26,8 +26,8 @@ import ( // branch at all), construct the MultiTransport via NewDirectOnly. type MultiTransport struct { embedded http.RoundTripper - direct *http.Transport - insecure *http.Transport + direct *upstreamTransport + insecure *upstreamTransport } // errNoEmbeddedTransport is returned when a request reaches the @@ -53,7 +53,6 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra } direct := &http.Transport{ DialContext: dialWithTimeout(dialer.DialContext), - ForceAttemptHTTP2: true, MaxIdleConns: cfg.maxIdleConns, MaxIdleConnsPerHost: cfg.maxIdleConnsPerHost, MaxConnsPerHost: cfg.maxConnsPerHost, @@ -70,8 +69,8 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra return &MultiTransport{ embedded: embedded, - direct: direct, - insecure: insecure, + direct: newUpstreamTransport(direct, cfg.upstreamHTTPVersion, logger), + insecure: newUpstreamTransport(insecure, cfg.upstreamHTTPVersion, logger), } } diff --git a/proxy/internal/roundtrip/multi_test.go b/proxy/internal/roundtrip/multi_test.go index 5c6cf1c97..7b8b50e81 100644 --- a/proxy/internal/roundtrip/multi_test.go +++ b/proxy/internal/roundtrip/multi_test.go @@ -5,6 +5,7 @@ import ( "io" "net/http" "net/http/httptest" + "os" "strings" "testing" "time" @@ -75,16 +76,71 @@ func TestMultiTransport_AppliesEnvOverridesToDirect(t *testing.T) { mt := NewMultiTransport(&stubRoundTripper{body: "embedded"}, nil) - assert.Equal(t, 42, mt.direct.MaxIdleConns, + assert.Equal(t, 42, mt.direct.primary.MaxIdleConns, "NB_PROXY_MAX_IDLE_CONNS must propagate to the direct transport") - assert.Equal(t, 11*time.Second, mt.direct.IdleConnTimeout, + assert.Equal(t, 11*time.Second, mt.direct.primary.IdleConnTimeout, "NB_PROXY_IDLE_CONN_TIMEOUT must propagate to the direct transport") - assert.Equal(t, 7*time.Second, mt.direct.TLSHandshakeTimeout, + assert.Equal(t, 7*time.Second, mt.direct.primary.TLSHandshakeTimeout, "NB_PROXY_TLS_HANDSHAKE_TIMEOUT must propagate to the direct transport") - assert.Equal(t, 42, mt.insecure.MaxIdleConns, + assert.Equal(t, 42, mt.insecure.primary.MaxIdleConns, "env tuning must also apply to the insecure-skip-verify direct transport") } +// TestMultiTransport_UpstreamHTTPVersion pins the protocol actually +// negotiated with an HTTPS upstream that offers both h2 and http/1.1. +// The request rides the insecure clone, so this also covers the version +// surviving http.Transport.Clone. +func TestMultiTransport_UpstreamHTTPVersion(t *testing.T) { + tests := []struct { + name string + env string + wantProto string + }{ + {name: "unset negotiates h2", env: "", wantProto: "HTTP/2.0"}, + {name: "auto negotiates h2", env: "auto", wantProto: "HTTP/2.0"}, + {name: "1.1 pins http/1.1", env: "1.1", wantProto: "HTTP/1.1"}, + {name: "2 negotiates h2", env: "2", wantProto: "HTTP/2.0"}, + {name: "unsupported value keeps the default", env: "http3", wantProto: "HTTP/2.0"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // t.Setenv registers the restore for whatever the process + // inherited; unsetting afterwards lets the default row + // exercise a genuinely absent variable. + t.Setenv(EnvUpstreamHTTPVersion, tc.env) + if tc.env == "" { + require.NoError(t, os.Unsetenv(EnvUpstreamHTTPVersion)) + } + + // The test server's certificate isn't in any root pool, so the + // request rides the insecure branch via WithSkipTLSVerify. + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, r.Proto) + })) + srv.EnableHTTP2 = true + srv.StartTLS() + defer srv.Close() + + mt := NewDirectOnly(nil) + ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background())) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil) + require.NoError(t, err) + + resp, err := mt.RoundTrip(req) + require.NoError(t, err) + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + require.NoError(t, err) + + assert.Equal(t, tc.wantProto, resp.Proto, + "client-side protocol must follow %s=%q", EnvUpstreamHTTPVersion, tc.env) + assert.Equal(t, tc.wantProto, string(body), + "the upstream must see the same protocol the client negotiated") + }) + } +} + // TestMultiTransport_NilEmbeddedErrorsWhenWGPathRequested guards // against the previous silent fallback: a MultiTransport constructed // without an embedded transport must reject requests that don't diff --git a/proxy/internal/roundtrip/netbird.go b/proxy/internal/roundtrip/netbird.go index ae3308a3e..d7b464182 100644 --- a/proxy/internal/roundtrip/netbird.go +++ b/proxy/internal/roundtrip/netbird.go @@ -82,10 +82,10 @@ type serviceNotification struct { // clientEntry holds an embedded NetBird client and tracks which services use it. type clientEntry struct { client *embed.Client - transport *http.Transport + transport *upstreamTransport // insecureTransport is a clone of transport with TLS verification disabled, // used when per-target skip_tls_verify is set. - insecureTransport *http.Transport + insecureTransport *upstreamTransport services map[ServiceKey]serviceInfo createdAt time.Time started bool @@ -414,7 +414,6 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account // not work with reverse proxied requests. transport := &http.Transport{ DialContext: dialWithTimeout(client.DialContext), - ForceAttemptHTTP2: true, MaxIdleConns: n.transportCfg.maxIdleConns, MaxIdleConnsPerHost: n.transportCfg.maxIdleConnsPerHost, MaxConnsPerHost: n.transportCfg.maxConnsPerHost, @@ -426,15 +425,14 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account ReadBufferSize: n.transportCfg.readBufferSize, DisableCompression: n.transportCfg.disableCompression, } - insecureTransport := transport.Clone() insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec return &clientEntry{ client: client, services: map[ServiceKey]serviceInfo{key: si}, - transport: transport, - insecureTransport: insecureTransport, + transport: newUpstreamTransport(transport, n.transportCfg.upstreamHTTPVersion, n.logger), + insecureTransport: newUpstreamTransport(insecureTransport, n.transportCfg.upstreamHTTPVersion, n.logger), createdAt: time.Now(), started: false, inflightMap: make(map[backendKey]chan struct{}), diff --git a/proxy/internal/roundtrip/transport.go b/proxy/internal/roundtrip/transport.go index 7c450bbb7..9e872e447 100644 --- a/proxy/internal/roundtrip/transport.go +++ b/proxy/internal/roundtrip/transport.go @@ -1,8 +1,11 @@ package roundtrip import ( + "crypto/tls" + "net/http" "os" "strconv" + "strings" "time" log "github.com/sirupsen/logrus" @@ -21,6 +24,30 @@ const ( EnvReadBufferSize = "NB_PROXY_READ_BUFFER_SIZE" EnvDisableCompression = "NB_PROXY_DISABLE_COMPRESSION" EnvMaxInflight = "NB_PROXY_MAX_INFLIGHT" + EnvUpstreamHTTPVersion = "NB_PROXY_UPSTREAM_HTTP_VERSION" +) + +// upstreamHTTPVersion selects the HTTP version the proxy uses towards an +// upstream. The explicit values are absolute: they mean the same thing +// however the transports are dialled and whatever the default becomes, +// so operator configuration survives a change of default. +type upstreamHTTPVersion string + +const ( + // upstreamHTTPAuto leaves the choice to the upstream: h2 is offered + // alongside http/1.1 in the TLS handshake and the upstream picks. + // An upstream that picks h2 and then fails to serve it is moved to + // HTTP/1.1 on its own (see upstreamTransport), which is the part + // ALPN cannot express. This is the only value whose meaning tracks + // the proxy's default. + upstreamHTTPAuto upstreamHTTPVersion = "auto" + // upstreamHTTP11 never offers h2, so the upstream sees HTTP/1.1. + upstreamHTTP11 upstreamHTTPVersion = "1.1" + // upstreamHTTP2 offers h2 in the TLS handshake and keeps it there: + // an upstream that negotiates h2 and then breaks is never moved to + // HTTP/1.1. Cleartext upstreams stay on HTTP/1.1 regardless: the + // proxy speaks no h2c. + upstreamHTTP2 upstreamHTTPVersion = "2" ) // transportConfig holds tunable parameters for the per-account HTTP transport. @@ -37,6 +64,11 @@ type transportConfig struct { disableCompression bool // maxInflight limits per-backend concurrent requests. 0 means unlimited. maxInflight int + // upstreamHTTPVersion selects the HTTP version used towards HTTPS + // upstreams. The default negotiates it with each upstream; the + // explicit values are for backends whose advertised h2 support is + // unusable and whose failure mode the negotiation cannot see. + upstreamHTTPVersion upstreamHTTPVersion } func defaultTransportConfig() transportConfig { @@ -47,6 +79,7 @@ func defaultTransportConfig() transportConfig { idleConnTimeout: 90 * time.Second, tlsHandshakeTimeout: 10 * time.Second, expectContinueTimeout: 1 * time.Second, + upstreamHTTPVersion: upstreamHTTPAuto, } } @@ -86,6 +119,9 @@ func loadTransportConfig(logger *log.Logger) transportConfig { if v, ok := envInt(EnvMaxInflight, logger); ok { cfg.maxInflight = v } + if v, ok := envUpstreamHTTPVersion(EnvUpstreamHTTPVersion, logger); ok { + cfg.upstreamHTTPVersion = v + } logger.WithFields(log.Fields{ "max_idle_conns": cfg.maxIdleConns, @@ -99,11 +135,83 @@ func loadTransportConfig(logger *log.Logger) transportConfig { "read_buffer_size": cfg.readBufferSize, "disable_compression": cfg.disableCompression, "max_inflight": cfg.maxInflight, + "upstream_http_version": cfg.upstreamHTTPVersion, }).Debug("backend transport configuration") return cfg } +// applyUpstreamHTTPVersion configures t's ALPN offer for the requested +// HTTP version. It is the single place that decides which protocols a +// transport offers, so changing the proxy's default only touches this +// function and leaves every explicit operator setting intact. What +// happens when a negotiated h2 upstream then fails belongs to +// upstreamTransport, which owns the runtime half of "auto". +// +// HTTP/1.1 is pinned by clearing ForceAttemptHTTP2 and installing an +// empty TLSNextProto, which disables h2 regardless of how the transport +// is dialled. Relying on net/http's conservative default (h2 off +// whenever a custom dialer is set) would silently start negotiating h2 +// again the day a transport switches to DialTLSContext. +func applyUpstreamHTTPVersion(t *http.Transport, version upstreamHTTPVersion) { + if version == upstreamHTTP11 { + t.ForceAttemptHTTP2 = false + t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{} + t.TLSClientConfig = withoutHTTP2ALPN(t.TLSClientConfig) + return + } + t.ForceAttemptHTTP2 = true +} + +// withoutHTTP2ALPN drops h2 from the ALPN offer. Configuring h2 makes +// net/http append h2 to the transport's TLSClientConfig, so a transport +// cloned from one that already served a request carries that offer with +// it. Left in place, the upstream would select a protocol this +// transport then refuses to speak, and the response would come back as +// h2 frames parsed as an HTTP/1.1 message. +func withoutHTTP2ALPN(cfg *tls.Config) *tls.Config { + // A nil config offers no ALPN at all, which is already HTTP/1.1. + if cfg == nil || len(cfg.NextProtos) == 0 { + return cfg + } + + protos := make([]string, 0, len(cfg.NextProtos)) + for _, proto := range cfg.NextProtos { + if proto == "h2" { + continue + } + protos = append(protos, proto) + } + if len(protos) == len(cfg.NextProtos) { + return cfg + } + + // Clone rather than edit in place: the caller may share this config + // with the transport it was cloned from. + stripped := cfg.Clone() + stripped.NextProtos = protos + + return stripped +} + +// envUpstreamHTTPVersion reads an upstream HTTP version from the +// environment. An unrecognised value warns and leaves the default in +// place rather than guessing at the operator's intent. +func envUpstreamHTTPVersion(key string, logger *log.Logger) (upstreamHTTPVersion, bool) { + s := strings.TrimSpace(os.Getenv(key)) + if s == "" { + return "", false + } + switch v := upstreamHTTPVersion(strings.ToLower(s)); v { + case upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2: + return v, true + default: + logger.Warnf("ignoring unsupported %s=%q, expected one of %q, %q, %q", + key, s, upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2) + return "", false + } +} + func envInt(key string, logger *log.Logger) (int, bool) { s := os.Getenv(key) if s == "" { diff --git a/proxy/internal/roundtrip/upstream.go b/proxy/internal/roundtrip/upstream.go new file mode 100644 index 000000000..6602c836e --- /dev/null +++ b/proxy/internal/roundtrip/upstream.go @@ -0,0 +1,455 @@ +package roundtrip + +import ( + "errors" + "net" + "net/http" + "net/netip" + "net/url" + "strings" + "sync" + "time" + + log "github.com/sirupsen/logrus" +) + +// upstreamDowngradeTTL is how long an upstream stays pinned to HTTP/1.1 +// after an h2 failure that only implied it cannot serve h2. Bounded so a +// fixed or replaced backend returns to h2 without restarting the proxy. +// A pin the upstream asked for itself does not expire — see downgrade. +const upstreamDowngradeTTL = 10 * time.Minute + +// downgrade is an upstream's HTTP/1.1 pin. +type downgrade struct { + // expiry is when the pin lapses and the upstream is offered h2 + // again. The zero time means it never does: the upstream answered + // HTTP_1_1_REQUIRED, which is a statement about how it is + // configured, not a fault that may clear on its own. Re-probing + // that every upstreamDowngradeTTL would buy nothing but a failed + // request per interval, so the pin holds until the transport goes + // away with the proxy or the account's client. + expiry time.Time +} + +// permanent reports whether the upstream asked for this pin itself. +func (d downgrade) permanent() bool { + return d.expiry.IsZero() +} + +// active reports whether the pin still stands at now. +func (d downgrade) active(now time.Time) bool { + return d.permanent() || now.Before(d.expiry) +} + +// upstreamTransport carries requests to a single upstream family (one +// TLS configuration) and implements what upstreamHTTPAuto means. +// +// ALPN already lets the upstream pick the protocol: primary offers both +// h2 and http/1.1 and the server chooses. What ALPN cannot express is +// an upstream that selects h2 and then fails to speak it — the case +// this type handles. The first h2-level failure for a host pins that +// host to fallback, an HTTP/1.1-only clone of primary, and the request +// is retried there when it can be replayed. +// +// The downgrade is per upstream host, not per transport: one broken +// backend must not drop every other backend to HTTP/1.1. +type upstreamTransport struct { + // primary is the configured transport: h2 offered in ALPN for + // upstreamHTTPAuto and upstreamHTTP2, HTTP/1.1-only for + // upstreamHTTP11. + primary *http.Transport + // version decides whether a downgrade may happen at all. Only + // upstreamHTTPAuto downgrades; the explicit values are absolute. + version upstreamHTTPVersion + logger *log.Logger + + // fallbackMu guards the lazy fallback clone: most deployments never + // hit a broken h2 upstream and should not pay for a second + // connection pool. + fallbackMu sync.Mutex + fallback *http.Transport + + mu sync.RWMutex + // downgraded maps an upstream host to its HTTP/1.1 pin. + downgraded map[string]downgrade +} + +// newUpstreamTransport wraps base for the requested HTTP version. base +// must not be used directly afterwards: the wrapper owns it, including +// its connection pool. +func newUpstreamTransport(base *http.Transport, version upstreamHTTPVersion, logger *log.Logger) *upstreamTransport { + if logger == nil { + logger = log.StandardLogger() + } + applyUpstreamHTTPVersion(base, version) + + return &upstreamTransport{ + primary: base, + version: version, + logger: logger, + downgraded: make(map[string]downgrade), + } +} + +// RoundTrip implements http.RoundTripper. +func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if !t.mayDowngrade(req) { + return t.primary.RoundTrip(req) + } + + host := upstreamKey(req.URL) + if t.isDowngraded(host) { + return t.http1().RoundTrip(req) + } + + resp, err := t.primary.RoundTrip(req) + if err == nil || !isHTTP2ProtocolError(err) { + return resp, err + } + + // HTTP_1_1_REQUIRED is the upstream saying it will not serve this + // request over h2 however often it is asked — IIS answers it for + // Windows Authentication and for client-certificate sites, where + // the cause is site configuration rather than a passing fault. + t.markDowngraded(host, isHTTP11Required(err)) + + if !safeToRetry(req, err) { + // The upstream may have carried out the request before failing + // to answer over h2, and repeating it could duplicate whatever + // it did. The host is pinned either way, so the next request + // goes out over HTTP/1.1. + t.logger.WithFields(log.Fields{ + "upstream": host, + "method": req.Method, + }).Debug("not retrying over HTTP/1.1: the upstream may already have applied this request") + return nil, err + } + + retry, ok := replayable(req) + if !ok { + // The body is already consumed and cannot be regenerated, so + // this request fails, and the pin carries the next one. + return nil, err + } + return t.http1().RoundTrip(retry) +} + +// CloseIdleConnections closes idle connections on both pools. +func (t *upstreamTransport) CloseIdleConnections() { + t.primary.CloseIdleConnections() + if fallback := t.existingHTTP1(); fallback != nil { + fallback.CloseIdleConnections() + } +} + +// mayDowngrade reports whether a failed request is a downgrade +// candidate. Only upstreamHTTPAuto downgrades, and only for TLS +// upstreams: the proxy speaks no h2c, so a cleartext upstream is +// already on HTTP/1.1 and an error there says nothing about h2. +func (t *upstreamTransport) mayDowngrade(req *http.Request) bool { + return t.version == upstreamHTTPAuto && req.URL != nil && req.URL.Scheme == "https" +} + +func (t *upstreamTransport) isDowngraded(host string) bool { + t.mu.RLock() + pin, ok := t.downgraded[host] + t.mu.RUnlock() + + if !ok { + return false + } + if pin.active(time.Now()) { + return true + } + + t.mu.Lock() + defer t.mu.Unlock() + + // Re-read under the write lock rather than trusting the expired pin + // from above: a concurrent request may have re-pinned the host since, + // and that pin decides this request too. Reporting the stale read + // would send one request back to h2 against a live pin. + pin, ok = t.downgraded[host] + if !ok { + return false + } + if pin.active(time.Now()) { + return true + } + delete(t.downgraded, host) + + return false +} + +// markDowngraded pins host to HTTP/1.1. permanent marks a pin the +// upstream asked for; anything else lapses after upstreamDowngradeTTL so +// a repaired backend is offered h2 again. +func (t *upstreamTransport) markDowngraded(host string, permanent bool) { + now := time.Now() + pin := downgrade{expiry: now.Add(upstreamDowngradeTTL)} + if permanent { + pin = downgrade{} + } + + t.mu.Lock() + previous, pinned := t.downgraded[host] + // A permanent pin is never weakened back into an expiring one: the + // upstream has already said h2 is not on offer. + promoted := pinned && !previous.permanent() && permanent + if !pinned || !previous.permanent() { + t.downgraded[host] = pin + } + for h, existing := range t.downgraded { + if !existing.active(now) { + delete(t.downgraded, h) + } + } + t.mu.Unlock() + + // Log a new pin, and a pin the upstream has since asked to make + // permanent — otherwise an operator would only ever see the "for the + // next 10m" line and never learn the upstream settled the question. + if pinned && !promoted { + return + } + + entry := t.logger.WithField("upstream", host) + if permanent { + entry.Warnf("upstream answered HTTP_1_1_REQUIRED, using HTTP/1.1 for it from now on") + return + } + entry.Warnf("upstream negotiated HTTP/2 but failed to serve it, using HTTP/1.1 for the next %s (set %s=1.1 to pin it)", + upstreamDowngradeTTL, EnvUpstreamHTTPVersion) +} + +// http1 returns the HTTP/1.1-only clone, creating it on first use. +func (t *upstreamTransport) http1() *http.Transport { + t.fallbackMu.Lock() + defer t.fallbackMu.Unlock() + + if t.fallback == nil { + fallback := t.primary.Clone() + applyUpstreamHTTPVersion(fallback, upstreamHTTP11) + t.fallback = fallback + } + + return t.fallback +} + +// existingHTTP1 returns the fallback transport only if it was already +// created, so housekeeping never allocates a second connection pool for +// an upstream that never needed one. +func (t *upstreamTransport) existingHTTP1() *http.Transport { + t.fallbackMu.Lock() + defer t.fallbackMu.Unlock() + + return t.fallback +} + +// upstreamKey normalizes an authority for use as a pin key, so one +// upstream cannot end up with two independent pins. DNS labels compare +// case-insensitively, and the default HTTPS port is implied — every +// downgrade path is TLS-only, so a bare host and the same host on :443 +// are the same upstream. +func upstreamKey(u *url.URL) string { + host := normalizeUpstreamHost(u.Hostname()) + + port := u.Port() + if port == "" || port == "443" { + return host + } + + // JoinHostPort rather than concatenation: an IPv6 literal needs its + // brackets back after Hostname stripped them. + return net.JoinHostPort(host, port) +} + +// normalizeUpstreamHost folds the spellings of one host onto a single +// key. An IP literal goes through netip so that the several textual +// forms of one address (case, leading zeroes, a compressed run) collapse +// and a v4-mapped address keys as the v4 address it is. A zone +// identifier is left exactly as written: it names an interface, and +// interface names are case-sensitive on the systems that have them, so +// %eth0 and %ETH0 may be different links and must not share a pin. +// Anything that is not an IP literal is a DNS name, which compares +// case-insensitively. +func normalizeUpstreamHost(host string) string { + if addr, err := netip.ParseAddr(host); err == nil { + return addr.Unmap().String() + } + + return strings.ToLower(host) +} + +// safeToRetry reports whether req may be sent a second time over +// HTTP/1.1 after err ended its h2 attempt. +// +// A failure at the h2 layer does not say whether the upstream already +// carried out the request, so replaying one that changes state could +// duplicate it. Two cases are safe: a request whose repetition is +// harmless by definition, and an upstream that told us it processed +// nothing on the connection. The second is what makes the IIS case work +// for every method — a site requiring HTTP/1.1 refuses at stream 0, +// before the request is looked at. +func safeToRetry(req *http.Request, err error) bool { + return idempotent(req) || upstreamProcessedNothing(err) +} + +// idempotent reports whether repeating req is defined to be harmless. +// It mirrors net/http's own retry rule (Request.isReplayable): a method +// with no side effects, or a caller that promised the upstream +// deduplicates by key. +func idempotent(req *http.Request) bool { + if req.Header.Get("Idempotency-Key") != "" || req.Header.Get("X-Idempotency-Key") != "" { + return true + } + + switch req.Method { + // An empty method means GET, as in net/http. + case "", http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace: + return true + } + + return false +} + +// upstreamProcessedNothing reports whether err describes a GOAWAY that +// named this request's stream as one the upstream had not received, so +// it cannot have acted on it. A stream error says the opposite: the +// stream was open, so the request had been delivered. +func upstreamProcessedNothing(err error) bool { + if err == nil { + return false + } + + msg := transportError(err).Error() + + return strings.Contains(msg, goAwayStreamNotReceivedMarker) || + strings.Contains(msg, goAwayNothingProcessedMarker) +} + +// replayable returns a request that can be sent a second time, or +// ok=false when the body is gone. A RoundTripper consumes and closes +// the body it was given, so a retry needs either no body at all or +// GetBody to produce a fresh one. +func replayable(req *http.Request) (*http.Request, bool) { + if req.Body == nil || req.Body == http.NoBody { + return req, true + } + if req.GetBody == nil { + return nil, false + } + + body, err := req.GetBody() + if err != nil { + return nil, false + } + + retry := req.Clone(req.Context()) + retry.Body = body + + return retry, true +} + +// http2ErrorMarkers are the substrings that identify an HTTP/2 protocol +// failure. net/http bundles its own private copy of the http2 package, +// so its errors cannot be matched by type from here: http2.StreamError +// and friends in x/net are different types from the ones a +// bundled-h2 transport returns. The strings below are the formats those +// bundled errors print, and they are specific to h2 framing — a +// downgrade must never be triggered by an ordinary network or TLS +// error, which retrying on HTTP/1.1 would not fix. +var http2ErrorMarkers = []string{ + // Transport-level h2 failures, e.g. + // "http2: server sent GOAWAY and closed the connection". + "http2:", + // http2.StreamError, e.g. "stream error: stream ID 1; PROTOCOL_ERROR". + "stream error: stream ID", + // http2.ConnectionError, e.g. "connection error: PROTOCOL_ERROR". + "connection error: ", + // The code an upstream sends to say the request must be retried + // over HTTP/1.1, as a GOAWAY or on the stream. + http11RequiredMarker, +} + +const ( + // http11RequiredMarker is the error code an upstream sends to say the + // request belongs on HTTP/1.1. Unlike the other markers it is not a + // fault: the upstream is describing its own configuration. + http11RequiredMarker = "HTTP_1_1_REQUIRED" + + // A GOAWAY carrying NO_ERROR closes a connection without complaint: + // a server draining before shutdown, recycling an application pool, + // capping requests per connection. The upstream speaks h2 perfectly + // well, so this must never pin it. The two spellings are the two + // formats the bundled transport prints the code in. + goAwayNoErrorEqualsMarker = "ErrCode=NO_ERROR" + goAwayNoErrorColonMarker = "ErrCode:NO_ERROR" + // gracefulGoAwayMarker is errClientConnGotGoAway, which the bundled + // transport raises for a stream the server never received on a + // connection it is shutting down gracefully. It normally retries + // those itself on a new connection and this never surfaces. + gracefulGoAwayMarker = "Transport received Server's graceful shutdown GOAWAY" + + // goAwayStreamNotReceivedMarker is the bundled transport's abort for + // the first stream on a connection whose GOAWAY carried a real error + // code — the IIS case. It sits in the same "streamID > LastStreamID" + // branch as the graceful abort, so the server had not received the + // stream (see net/http's h2_bundle.go). + goAwayStreamNotReceivedMarker = "Transport received GOAWAY from server ErrCode:" + // goAwayNothingProcessedMarker is a GoAwayError naming stream 0 as + // the last one received, which says the same thing. The trailing + // comma keeps it from matching LastStreamID=10 and the rest. + goAwayNothingProcessedMarker = "LastStreamID=0," +) + +// isHTTP11Required reports whether the upstream itself asked for +// HTTP/1.1, rather than merely failing at h2. +func isHTTP11Required(err error) bool { + return err != nil && strings.Contains(transportError(err).Error(), http11RequiredMarker) +} + +// isHTTP2ProtocolError reports whether err says the upstream cannot +// serve the h2 it negotiated. +func isHTTP2ProtocolError(err error) bool { + if err == nil { + return false + } + + msg := transportError(err).Error() + + // A graceful GOAWAY is routine connection management, not an + // upstream that cannot serve h2. + if isGracefulGoAway(msg) { + return false + } + + for _, marker := range http2ErrorMarkers { + if strings.Contains(msg, marker) { + return true + } + } + + return false +} + +// isGracefulGoAway reports whether msg describes a GOAWAY sent to close +// a healthy connection rather than to report an inability to serve h2. +func isGracefulGoAway(msg string) bool { + return strings.Contains(msg, goAwayNoErrorEqualsMarker) || + strings.Contains(msg, goAwayNoErrorColonMarker) || + strings.Contains(msg, gracefulGoAwayMarker) +} + +// transportError strips a *url.Error wrapper, which prefixes the request +// URL to the message. Markers are matched as substrings, so a URL left +// in place could classify an ordinary dial or TLS failure as an h2 one +// on the strength of the path alone. +func transportError(err error) error { + var urlErr *url.Error + if errors.As(err, &urlErr) && urlErr.Err != nil { + return urlErr.Err + } + + return err +} diff --git a/proxy/internal/roundtrip/upstream_test.go b/proxy/internal/roundtrip/upstream_test.go new file mode 100644 index 000000000..a83570ef3 --- /dev/null +++ b/proxy/internal/roundtrip/upstream_test.go @@ -0,0 +1,587 @@ +package roundtrip + +import ( + "bufio" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "errors" + "fmt" + "io" + "math/big" + "net" + "net/http" + "net/url" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/net/http2" +) + +// TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2 covers the case ALPN +// cannot express: the upstream advertises h2, picks it, and then cannot +// serve it. The request must still succeed, over HTTP/1.1, and the +// upstream must stay on HTTP/1.1 for the requests that follow. +func TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2(t *testing.T) { + t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto)) + srv := startBrokenHTTP2Server(t) + + mt := NewDirectOnly(nil) + ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background())) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil) + require.NoError(t, err) + + resp, err := mt.RoundTrip(req) + require.NoError(t, err, "a replayable request must be retried on HTTP/1.1 instead of failing") + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + require.NoError(t, err) + assert.Equal(t, "HTTP/1.1", resp.Proto, "the retry must ride the HTTP/1.1 transport") + assert.Equal(t, "http/1.1", string(body), "the upstream must see an http/1.1 ALPN offer on the retry") + + assert.True(t, mt.insecure.isDowngraded(srv.addr), + "the upstream must stay pinned to HTTP/1.1 after proving it cannot serve h2") + mt.insecure.mu.RLock() + pin := mt.insecure.downgraded[srv.addr] + mt.insecure.mu.RUnlock() + assert.True(t, pin.permanent(), + "an upstream answering HTTP_1_1_REQUIRED must not be re-probed for h2") + + // The second request must not repeat the h2 attempt: the server + // counts h2 handshakes, so a repeat would show up here. + h2Attempts := srv.http2Handshakes() + req, err = http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil) + require.NoError(t, err) + resp, err = mt.RoundTrip(req) + require.NoError(t, err) + _ = resp.Body.Close() + + assert.Equal(t, "HTTP/1.1", resp.Proto, "a pinned upstream must go straight to HTTP/1.1") + assert.Equal(t, h2Attempts, srv.http2Handshakes(), + "a pinned upstream must not be probed for h2 again until the pin expires") +} + +// TestUpstreamTransport_ExplicitHTTP2NeverDowngrades pins the promise +// that the explicit values are absolute: an operator who asked for h2 +// keeps h2, broken upstream or not. +func TestUpstreamTransport_ExplicitHTTP2NeverDowngrades(t *testing.T) { + t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTP2)) + srv := startBrokenHTTP2Server(t) + + mt := NewDirectOnly(nil) + ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background())) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil) + require.NoError(t, err) + + resp, err := mt.RoundTrip(req) + if err == nil { + _ = resp.Body.Close() + } + require.Error(t, err, "NB_PROXY_UPSTREAM_HTTP_VERSION=2 must not fall back to HTTP/1.1") + assert.False(t, mt.insecure.isDowngraded(srv.addr), "an explicit version must never pin an upstream") +} + +// TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests covers the +// other half of the fallback: an h2 failure says nothing about whether +// the upstream already applied the request, so a state-changing one is +// not replayed. The host is still pinned, so the next request rides +// HTTP/1.1 without a second h2 attempt. +func TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests(t *testing.T) { + t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto)) + srv := startBrokenHTTP2Server(t) + // A stream error means the stream was open, so the upstream had the + // request in hand — unlike the GOAWAY at stream 0 the fake server + // sends, which states it processed nothing. + streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol} + + mt := NewDirectOnly(nil) + transport := mt.insecure + ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background())) + + assert.False(t, safeToRetry(newTestRequest(t, ctx, http.MethodPost, srv.addr), streamErr), + "a POST must not be replayed after a failure that may have been applied") + assert.True(t, safeToRetry(newTestRequest(t, ctx, http.MethodGet, srv.addr), streamErr), + "a GET is safe to replay whatever the failure was") + + // The fake server's GOAWAY names stream 0, so even a POST is safe + // there and the request must succeed over HTTP/1.1. + resp, err := transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr)) + require.NoError(t, err, "a GOAWAY at stream 0 means the upstream applied nothing, so the POST may be replayed") + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + require.NoError(t, err) + assert.Equal(t, "http/1.1", string(body), "the retry must reach the upstream over http/1.1") + + h2Attempts := srv.http2Handshakes() + resp, err = transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr)) + require.NoError(t, err) + _ = resp.Body.Close() + assert.Equal(t, h2Attempts, srv.http2Handshakes(), + "the pin must carry later requests without another h2 attempt") +} + +func newTestRequest(t *testing.T, ctx context.Context, method, addr string) *http.Request { + t.Helper() + + req, err := http.NewRequestWithContext(ctx, method, "https://"+addr, strings.NewReader("payload")) + require.NoError(t, err) + + return req +} + +func TestUpstreamKey(t *testing.T) { + tests := []struct { + name string + url string + want string + }{ + {name: "host", url: "https://backend.invalid/path", want: "backend.invalid"}, + // DNS is case-insensitive, so these are one upstream and must + // share one pin. + {name: "mixed case", url: "https://Backend.INVALID/path", want: "backend.invalid"}, + // The default port is implied on every path that can downgrade. + {name: "explicit default port", url: "https://backend.invalid:443/", want: "backend.invalid"}, + {name: "non-default port", url: "https://backend.invalid:8443/", want: "backend.invalid:8443"}, + // An IPv6 literal needs its brackets back after Hostname strips + // them, or the key is not a dialable authority. + {name: "ipv6 default port", url: "https://[2001:db8::1]/", want: "2001:db8::1"}, + {name: "ipv6 with port", url: "https://[2001:db8::1]:8443/", want: "[2001:db8::1]:8443"}, + // One address in three spellings: hex case, a leading zero and an + // uncompressed zero run are all the same upstream. + {name: "ipv6 upper case", url: "https://[2001:DB8::1]/", want: "2001:db8::1"}, + {name: "ipv6 leading zero", url: "https://[2001:0db8::1]/", want: "2001:db8::1"}, + {name: "ipv6 uncompressed", url: "https://[2001:db8:0:0:0:0:0:1]/", want: "2001:db8::1"}, + // A v4-mapped address is the v4 address, not a second upstream. + {name: "v4-mapped", url: "https://[::ffff:192.0.2.1]/", want: "192.0.2.1"}, + // A zone names an interface, and interface names are + // case-sensitive, so these two are different links. + {name: "ipv6 zone", url: "https://[fe80::1%25eth0]/", want: "fe80::1%eth0"}, + {name: "ipv6 zone upper case", url: "https://[fe80::1%25ETH0]/", want: "fe80::1%ETH0"}, + // The address before the zone still normalizes. + {name: "ipv6 zone with upper-case address", url: "https://[FE80::1%25eth0]/", want: "fe80::1%eth0"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + parsed, err := url.Parse(tc.url) + require.NoError(t, err) + + assert.Equal(t, tc.want, upstreamKey(parsed)) + }) + } +} + +func TestSafeToRetry(t *testing.T) { + streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol} + goAwayAtZero := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`) + goAwayLater := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=11, ErrCode=PROTOCOL_ERROR, debug=""`) + + tests := []struct { + name string + method string + headers map[string]string + err error + want bool + }{ + {name: "get", method: http.MethodGet, err: streamErr, want: true}, + {name: "head", method: http.MethodHead, err: streamErr, want: true}, + {name: "options", method: http.MethodOptions, err: streamErr, want: true}, + {name: "trace", method: http.MethodTrace, err: streamErr, want: true}, + {name: "post", method: http.MethodPost, err: streamErr, want: false}, + {name: "put", method: http.MethodPut, err: streamErr, want: false}, + {name: "patch", method: http.MethodPatch, err: streamErr, want: false}, + {name: "delete", method: http.MethodDelete, err: streamErr, want: false}, + // The upstream reported it handled nothing, so repeating the + // request cannot duplicate anything. + {name: "post with goaway at stream 0", method: http.MethodPost, err: goAwayAtZero, want: true}, + // What the bundled transport actually raises for the first + // stream on a connection the upstream GOAWAYs with a real error + // code, which is the shape a real IIS site produces. + { + name: "post with first-stream goaway abort", + method: http.MethodPost, + err: errors.New("http2: Transport received GOAWAY from server ErrCode:HTTP_1_1_REQUIRED"), + want: true, + }, + // It handled earlier streams, so this one may have been applied. + {name: "post with goaway after other streams", method: http.MethodPost, err: goAwayLater, want: false}, + { + name: "post with idempotency key", + method: http.MethodPost, + headers: map[string]string{"Idempotency-Key": "abc"}, + err: streamErr, + want: true, + }, + { + name: "post with prefixed idempotency key", + method: http.MethodPost, + headers: map[string]string{"X-Idempotency-Key": "abc"}, + err: streamErr, + want: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + req, err := http.NewRequest(tc.method, "https://backend.invalid", nil) + require.NoError(t, err) + for k, v := range tc.headers { + req.Header.Set(k, v) + } + + assert.Equal(t, tc.want, safeToRetry(req, tc.err)) + }) + } +} + +func TestUpstreamTransport_MayDowngrade(t *testing.T) { + tests := []struct { + name string + version upstreamHTTPVersion + url string + want bool + }{ + {name: "auto over TLS", version: upstreamHTTPAuto, url: "https://backend.invalid", want: true}, + // The proxy speaks no h2c, so a cleartext upstream is already on + // HTTP/1.1 and its failures say nothing about h2. + {name: "auto cleartext", version: upstreamHTTPAuto, url: "http://backend.invalid", want: false}, + {name: "explicit 1.1", version: upstreamHTTP11, url: "https://backend.invalid", want: false}, + {name: "explicit 2", version: upstreamHTTP2, url: "https://backend.invalid", want: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + transport := newUpstreamTransport(&http.Transport{}, tc.version, nil) + req, err := http.NewRequest(http.MethodGet, tc.url, nil) + require.NoError(t, err) + + assert.Equal(t, tc.want, transport.mayDowngrade(req)) + }) + } +} + +func TestUpstreamTransport_DowngradeExpires(t *testing.T) { + transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil) + transport.markDowngraded("backend.invalid:443", false) + require.True(t, transport.isDowngraded("backend.invalid:443")) + + transport.mu.Lock() + transport.downgraded["backend.invalid:443"] = downgrade{expiry: time.Now().Add(-time.Second)} + transport.mu.Unlock() + + assert.False(t, transport.isDowngraded("backend.invalid:443"), + "an expired pin must let the upstream be offered h2 again") + transport.mu.RLock() + _, stillTracked := transport.downgraded["backend.invalid:443"] + transport.mu.RUnlock() + assert.False(t, stillTracked, "an expired pin must not be kept around") +} + +func TestUpstreamTransport_DowngradeIsPerUpstream(t *testing.T) { + transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil) + transport.markDowngraded("broken.invalid:443", false) + + assert.True(t, transport.isDowngraded("broken.invalid:443")) + assert.False(t, transport.isDowngraded("healthy.invalid:443"), + "one broken upstream must not drop the others to HTTP/1.1") +} + +// TestUpstreamTransport_HTTP11RequiredPinIsPermanent covers the IIS +// case: HTTP_1_1_REQUIRED describes how the upstream is configured +// (Windows Authentication, client certificates), so re-probing it every +// upstreamDowngradeTTL would only buy a failed request per interval. +func TestUpstreamTransport_HTTP11RequiredPinIsPermanent(t *testing.T) { + transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil) + transport.markDowngraded("iis.invalid:443", true) + + transport.mu.RLock() + pin := transport.downgraded["iis.invalid:443"] + transport.mu.RUnlock() + + assert.True(t, pin.permanent(), "an upstream that asked for HTTP/1.1 must not be re-probed") + assert.True(t, pin.active(time.Now().Add(100*upstreamDowngradeTTL)), + "a permanent pin must outlive any TTL") +} + +func TestUpstreamTransport_PermanentPinSurvivesLaterFailures(t *testing.T) { + transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil) + transport.markDowngraded("iis.invalid:443", true) + // A later ambiguous failure for the same upstream must not turn the + // permanent pin into an expiring one. + transport.markDowngraded("iis.invalid:443", false) + + transport.mu.RLock() + pin := transport.downgraded["iis.invalid:443"] + transport.mu.RUnlock() + + assert.True(t, pin.permanent(), "a permanent pin must never be weakened") +} + +func TestIsHTTP11Required(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "goaway", + err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`), + want: true, + }, + { + name: "stream error", + err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeHTTP11Required}, + want: true, + }, + { + name: "other h2 failure", + err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}, + want: false, + }, + {name: "nil", err: nil, want: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, isHTTP11Required(tc.err)) + }) + } +} + +func TestIsHTTP2ProtocolError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + { + name: "goaway demanding http/1.1", + err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`), + want: true, + }, + { + name: "stream error", + err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}, + want: true, + }, + { + name: "connection error", + err: http2.ConnectionError(http2.ErrCodeProtocol), + want: true, + }, + { + name: "wrapped h2 error", + err: errors.New("Get \"https://backend.invalid\": http2: client connection lost"), + want: true, + }, + // A GOAWAY with NO_ERROR is a server draining a connection — + // recycling an application pool, capping requests per + // connection, shutting down gracefully. It speaks h2 fine. + { + name: "graceful goaway", + err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=9, ErrCode=NO_ERROR, debug=""`), + want: false, + }, + { + name: "graceful shutdown abort", + err: errors.New("http2: Transport received Server's graceful shutdown GOAWAY"), + want: false, + }, + // Markers are substrings, so a URL carried by a *url.Error must + // not be able to classify a plain failure as an h2 one. + { + name: "url error whose path looks like a marker", + err: &url.Error{ + Op: "Get", + URL: "https://backend.invalid/http2:/connection error: x", + Err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), + }, + want: false, + }, + // Retrying these on HTTP/1.1 fixes nothing, so they must never + // pin an upstream. + {name: "dial failure", err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), want: false}, + {name: "tls failure", err: errors.New("tls: failed to verify certificate: x509: certificate signed by unknown authority"), want: false}, + {name: "context cancelled", err: context.Canceled, want: false}, + {name: "eof", err: io.EOF, want: false}, + {name: "nil", err: nil, want: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, isHTTP2ProtocolError(tc.err)) + }) + } +} + +func TestReplayable(t *testing.T) { + t.Run("bodyless request", func(t *testing.T) { + req, err := http.NewRequest(http.MethodGet, "https://backend.invalid", nil) + require.NoError(t, err) + + retry, ok := replayable(req) + require.True(t, ok) + assert.Same(t, req, retry, "a bodyless request needs no clone") + }) + + t.Run("request with GetBody", func(t *testing.T) { + req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", strings.NewReader("payload")) + require.NoError(t, err) + // Consume the body the way a failed RoundTrip would. + _, err = io.ReadAll(req.Body) + require.NoError(t, err) + + retry, ok := replayable(req) + require.True(t, ok) + body, err := io.ReadAll(retry.Body) + require.NoError(t, err) + assert.Equal(t, "payload", string(body), "the retry must carry a fresh copy of the body") + }) + + t.Run("streamed request", func(t *testing.T) { + req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", io.NopCloser(strings.NewReader("payload"))) + require.NoError(t, err) + require.Nil(t, req.GetBody, "an opaque reader must not get a GetBody") + + _, ok := replayable(req) + assert.False(t, ok, "a body that cannot be regenerated must not be replayed") + }) +} + +// brokenHTTP2Server advertises h2 in ALPN, accepts it, and then refuses +// to serve it — the upstream behaviour that motivated the fallback. Over +// http/1.1 it answers normally, so a downgraded request succeeds. +type brokenHTTP2Server struct { + addr string + + handshakes chan struct{} +} + +func (s *brokenHTTP2Server) http2Handshakes() int { + return len(s.handshakes) +} + +func startBrokenHTTP2Server(t *testing.T) *brokenHTTP2Server { + t.Helper() + + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{ + Certificates: []tls.Certificate{selfSignedCert(t)}, + NextProtos: []string{"h2", "http/1.1"}, + MinVersion: tls.VersionTLS12, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + + srv := &brokenHTTP2Server{ + addr: ln.Addr().String(), + // Buffered well past what the test drives so a stuck server + // never blocks the accept loop. + handshakes: make(chan struct{}, 64), + } + + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go srv.handle(conn) + } + }() + + return srv +} + +func (s *brokenHTTP2Server) handle(conn net.Conn) { + defer func() { _ = conn.Close() }() + + tlsConn, ok := conn.(*tls.Conn) + if !ok { + return + } + if err := tlsConn.Handshake(); err != nil { + return + } + + proto := tlsConn.ConnectionState().NegotiatedProtocol + if proto == "h2" { + select { + case s.handshakes <- struct{}{}: + default: + } + s.refuseHTTP2(tlsConn) + return + } + + s.serveHTTP1(tlsConn, proto) +} + +// refuseHTTP2 completes just enough of the h2 handshake for the client +// to accept the connection, then sends the GOAWAY an upstream uses to +// say the request belongs on HTTP/1.1. +// +// The client is still writing its preface and request while the GOAWAY +// goes out, so the connection is drained before the caller closes it. +// Closing a socket with unread bytes still in its receive buffer makes +// the kernel answer with RST, which reaches the client as a write error +// rather than the GOAWAY — no h2 error, so no downgrade, and the test +// fails on the error the client saw first. +func (s *brokenHTTP2Server) refuseHTTP2(conn net.Conn) { + framer := http2.NewFramer(conn, conn) + if err := framer.WriteSettings(); err != nil { + return + } + if err := framer.WriteGoAway(0, http2.ErrCodeHTTP11Required, nil); err != nil { + return + } + + // The client closes its side once it has read the GOAWAY, which ends + // the drain; the deadline is only a backstop against a client that + // never does. + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + _, _ = io.Copy(io.Discard, conn) +} + +// serveHTTP1 answers a single request with the ALPN protocol the +// upstream actually settled on, so a test asserting on the body is +// checking what the upstream saw rather than a constant. +func (s *brokenHTTP2Server) serveHTTP1(conn net.Conn, alpn string) { + reader := bufio.NewReader(conn) + if _, err := http.ReadRequest(reader); err != nil { + return + } + + _, _ = fmt.Fprintf(conn, + "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s", + len(alpn), alpn) +} + +func selfSignedCert(t *testing.T) tls.Certificate { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + IsCA: true, + } + der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key) + require.NoError(t, err) + + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} From ea216f8e732311d237e5d884a11d4459d9f5841f Mon Sep 17 00:00:00 2001 From: Maycon Santos Date: Mon, 14 Sep 2026 19:42:02 +0200 Subject: [PATCH 2/9] [management] Speed up test store setup and summarize the unit test run (#7518) Management / Unit (amd64, mysql) hit the 20 minute go test budget on #7516. The package was not hung: each of the 133 test store creations in management/server paid about 1.6s on MySQL for CREATE DATABASE, the pre-migrations, a 40-table AutoMigrate and the post-migrations, which puts the package at 10 minutes on a healthy runner and over the budget on a slow one. The migration now runs once per test binary into a template database and each test database is cloned from it, with CREATE DATABASE ... TEMPLATE on Postgres and a replay of SHOW CREATE TABLE on MySQL. The MySQL test container also drops the binary log, doublewrite buffer and per-commit redo fsync. Two goroutine leaks in the test helpers are fixed. tools/gotestsummary turns the go test -json stream into a readable log, and the Management unit and integration jobs now pipe through it, so a timeout names the tests still running. On MySQL, management/server went from 10m16s to 6m36s. --- .github/workflows/golang-test-linux.yml | 39 ++- management/server/account_test.go | 10 +- management/server/store/sql_store.go | 69 ++-- management/server/store/store.go | 253 ++++++++++++++- management/server/testutil/store.go | 12 + tools/gotestsummary/main.go | 402 ++++++++++++++++++++++++ tools/gotestsummary/main_test.go | 180 +++++++++++ 7 files changed, 923 insertions(+), 42 deletions(-) create mode 100644 tools/gotestsummary/main.go create mode 100644 tools/gotestsummary/main_test.go diff --git a/.github/workflows/golang-test-linux.yml b/.github/workflows/golang-test-linux.yml index 9e3caa17a..449eb14fa 100644 --- a/.github/workflows/golang-test-linux.yml +++ b/.github/workflows/golang-test-linux.yml @@ -514,14 +514,32 @@ jobs: if: matrix.store == 'mysql' run: docker pull mlsmaycon/warmed-mysql:8 + # The -json stream goes through tools/gotestsummary so the log shows one + # line per test, the output of failed tests, the head of a timeout panic + # with the still-running tests, and the slowest tests per package. - name: Test + shell: bash run: | + set -o pipefail CGO_ENABLED=1 GOARCH=${{ matrix.arch }} \ NETBIRD_STORE_ENGINE=${{ matrix.store }} \ CI=true \ - go test -tags=devcert -coverprofile=coverage.txt \ + go test -json -tags=devcert -coverprofile=coverage.txt \ -exec "sudo --preserve-env=CI,NETBIRD_STORE_ENGINE" \ - -timeout 20m ./management/... ./shared/management/... + -timeout 20m ./management/... ./shared/management/... \ + | tee management-test-events.jsonl \ + | go run ./tools/gotestsummary + + # The summary trims long outputs; the raw stream keeps every line for + # the failures that need it. A green run has no use for it. + - name: Upload raw test events + if: failure() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a #v7.0.1 + with: + name: management-unit-test-events-${{ matrix.store }} + path: management-test-events.jsonl + if-no-files-found: ignore + retention-days: 14 - name: Upload coverage reports to Codecov if: matrix.arch == 'amd64' @@ -771,12 +789,27 @@ jobs: - name: check git status run: git --no-pager diff --exit-code + # Same summary as the unit job: a timeout here names the tests still + # running instead of ending in a goroutine dump. - name: Test + shell: bash run: | + set -o pipefail CGO_ENABLED=1 GOARCH=${{ matrix.arch }} \ NETBIRD_STORE_ENGINE=${{ matrix.store }} \ CI=true \ - mage integrationtest:all -gotestflags="-coverprofile=coverage.txt" + mage integrationtest:all -gotestflags="-json -coverprofile=coverage.txt" \ + | tee management-integration-test-events.jsonl \ + | go run ./tools/gotestsummary + + - name: Upload raw test events + if: failure() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a #v7.0.1 + with: + name: management-integration-test-events-${{ matrix.store }} + path: management-integration-test-events.jsonl + if-no-files-found: ignore + retention-days: 14 - name: Upload coverage reports to Codecov if: matrix.arch == 'amd64' diff --git a/management/server/account_test.go b/management/server/account_test.go index bd7bf2d97..c63782ca8 100644 --- a/management/server/account_test.go +++ b/management/server/account_test.go @@ -3516,7 +3516,13 @@ func buildTestManager(t testing.TB, store store.Store, nmdataStore *networkmapdb eventStore := &activity.InMemoryEventStore{} - metrics, err := telemetry.NewDefaultAppMetrics(context.Background()) + // Everything built here watches this context; cancelling it on cleanup stops + // the metrics flushers, caches and controllers instead of leaking them for + // the rest of the package run. + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + metrics, err := telemetry.NewDefaultAppMetrics(ctx) if err != nil { return nil, nil, err } @@ -3535,8 +3541,6 @@ func buildTestManager(t testing.TB, store store.Store, nmdataStore *networkmapdb Return(nil). AnyTimes() - ctx := context.Background() - cacheStore, err := cache.NewStore(ctx, 100*time.Millisecond, 300*time.Millisecond, 100) if err != nil { return nil, nil, err diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go index 08ec45395..d2d197a03 100644 --- a/management/server/store/sql_store.go +++ b/management/server/store/sql_store.go @@ -3154,7 +3154,12 @@ func NewMysqlStore(ctx context.Context, dsn string, metrics telemetry.AppMetrics return nil, err } - return NewSqlStore(ctx, db, types.MysqlStoreEngine, metrics, skipMigration) + store, err := NewSqlStore(ctx, db, types.MysqlStoreEngine, metrics, skipMigration) + if err != nil { + closeGormDB(db) + return nil, err + } + return store, nil } func getGormConfig() *gorm.Config { @@ -3213,23 +3218,20 @@ func NewSqliteStoreFromFileStore(ctx context.Context, fileStore *FileStore, data // NewPostgresqlStoreFromSqlStore restores a store from SqlStore and stores Postgres DB. func NewPostgresqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics) (*SqlStore, error) { - store, err := NewPostgresqlStoreForTests(ctx, dsn, metrics, false) + return newPostgresqlStoreFromSqlStore(ctx, sqliteStore, dsn, metrics, false) +} + +func newPostgresqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) { + store, err := NewPostgresqlStoreForTests(ctx, dsn, metrics, skipMigration) if err != nil { return nil, err } - err = store.SaveInstallationID(ctx, sqliteStore.GetInstallationID()) - if err != nil { + if err := seedFromSqliteStore(ctx, store, sqliteStore); err != nil { + closeStore(ctx, store) return nil, err } - for _, account := range sqliteStore.GetAllAccounts(ctx) { - err := store.SaveAccount(ctx, account) - if err != nil { - return nil, err - } - } - return store, nil } @@ -3241,11 +3243,14 @@ func NewPostgresqlStoreForTests(ctx context.Context, dsn string, metrics telemet } pool, err := connectToPgDbForTests(context.Background(), dsn) if err != nil { + closeGormDB(db) return nil, err } store, err := NewSqlStore(ctx, db, types.PostgresStoreEngine, metrics, skipMigration) if err != nil { + // Release the sessions, or the caller cannot drop the database. pool.Close() + closeGormDB(db) return nil, err } store.pool = pool @@ -3279,22 +3284,42 @@ func connectToPgDbForTests(ctx context.Context, dsn string) (*pgxpool.Pool, erro // NewMysqlStoreFromSqlStore restores a store from SqlStore and stores MySQL DB. func NewMysqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics) (*SqlStore, error) { - store, err := NewMysqlStore(ctx, dsn, metrics, false) - if err != nil { - return nil, err - } + return newMysqlStoreFromSqlStore(ctx, sqliteStore, dsn, metrics, false) +} - err = store.SaveInstallationID(ctx, sqliteStore.GetInstallationID()) - if err != nil { - return nil, err +// seedFromSqliteStore copies the installation ID and the accounts of the +// sqlite seed store into a freshly created engine store. +func seedFromSqliteStore(ctx context.Context, store, sqliteStore *SqlStore) error { + if err := store.SaveInstallationID(ctx, sqliteStore.GetInstallationID()); err != nil { + return err } - for _, account := range sqliteStore.GetAllAccounts(ctx) { - err := store.SaveAccount(ctx, account) - if err != nil { - return nil, err + if err := store.SaveAccount(ctx, account); err != nil { + return err } } + return nil +} + +// closeStore releases a store that is not handed to the caller, so a failed +// seed does not leak its connection and pool. +func closeStore(ctx context.Context, store *SqlStore) { + store.Close(ctx) + if store.pool != nil { + store.pool.Close() + } +} + +func newMysqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) { + store, err := NewMysqlStore(ctx, dsn, metrics, skipMigration) + if err != nil { + return nil, err + } + + if err := seedFromSqliteStore(ctx, store, sqliteStore); err != nil { + closeStore(ctx, store) + return nil, err + } return store, nil } diff --git a/management/server/store/store.go b/management/server/store/store.go index 6886536b9..7a92de0a6 100644 --- a/management/server/store/store.go +++ b/management/server/store/store.go @@ -4,6 +4,7 @@ package store import ( "context" + "database/sql" "errors" "fmt" "net" @@ -15,6 +16,7 @@ import ( "runtime" "slices" "strings" + "sync" "time" "github.com/google/uuid" @@ -732,6 +734,7 @@ func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) ( time.Sleep(100 * time.Millisecond) } } + store.Close(ctx) return nil, nil, fmt.Errorf("failed to create test store after %d attempts: %v", maxRetries, err) } @@ -758,14 +761,15 @@ func addAllGroupToAccount(ctx context.Context, store Store) error { return nil } -func getSqlStoreEngine(ctx context.Context, store *SqlStore, kind types.Engine) (Store, func(), error) { +func getSqlStoreEngine(ctx context.Context, sqliteStore *SqlStore, kind types.Engine) (Store, func(), error) { + store := sqliteStore var cleanup func() var err error switch kind { case types.PostgresStoreEngine: - store, cleanup, err = newReusedPostgresStore(ctx, store, kind) + store, cleanup, err = newReusedPostgresStore(ctx, sqliteStore, kind) case types.MysqlStoreEngine: - store, cleanup, err = newReusedMysqlStore(ctx, store, kind) + store, cleanup, err = newReusedMysqlStore(ctx, sqliteStore, kind) default: cleanup = func() { // sqlite doesn't need to be cleaned up @@ -781,6 +785,11 @@ func getSqlStoreEngine(ctx context.Context, store *SqlStore, kind types.Engine) if store.pool != nil { store.pool.Close() } + if store != sqliteStore { + // The sqlite store only seeded the engine under test; without this + // every test leaks its connection and the opener goroutines. + sqliteStore.Close(ctx) + } } return store, closeConnection, nil @@ -805,19 +814,23 @@ func newReusedPostgresStore(ctx context.Context, store *SqlStore, kind types.Eng return nil, nil, fmt.Errorf("failed to open postgres connection: %v", err) } - dsn, cleanup, err := createRandomDB(dsn, db, kind) - - sqlDB, _ := db.DB() - if sqlDB != nil { - sqlDB.Close() + template, err := postgresSchemaTemplate(ctx, dsn, db) + if err != nil { + closeGormDB(db) + return nil, nil, err } + dsn, cleanup, err := createRandomDB(dsn, db, kind, template) + + closeGormDB(db) + if err != nil { return nil, nil, err } - store, err = NewPostgresqlStoreFromSqlStore(ctx, store, dsn, nil) + store, err = newPostgresqlStoreFromSqlStore(ctx, store, dsn, nil, true) if err != nil { + cleanup() return nil, nil, err } @@ -850,7 +863,13 @@ func newReusedMysqlStore(ctx context.Context, store *SqlStore, kind types.Engine sqlDB.SetMaxOpenConns(1) sqlDB.SetMaxIdleConns(1) - dsn, cleanup, err := createRandomDB(dsn, db, kind) + tableDDL, err := mysqlSchemaTemplate(ctx, dsn, db) + if err != nil { + sqlDB.Close() + return nil, nil, err + } + + dsn, cleanup, err := createRandomDB(dsn, db, kind, "") sqlDB.Close() @@ -858,14 +877,200 @@ func newReusedMysqlStore(ctx context.Context, store *SqlStore, kind types.Engine return nil, nil, err } - store, err = NewMysqlStoreFromSqlStore(ctx, store, dsn, nil) + if err := cloneMysqlSchema(ctx, dsn, tableDDL); err != nil { + cleanup() + return nil, nil, err + } + + store, err = newMysqlStoreFromSqlStore(ctx, store, dsn, nil, true) if err != nil { + cleanup() return nil, nil, err } return store, cleanup, nil } +// schemaTemplates remembers, per engine and server, a database that went +// through the full migration once in this process. Every later test database +// is cloned from it, so a test pays for CREATE DATABASE and a schema copy +// instead of the 40-table AutoMigrate plus every pre and post migration, which +// is what made each MySQL test store cost well over a second in CI. +var ( + schemaTemplatesMu sync.Mutex + schemaTemplates = map[string]*schemaTemplate{} +) + +type schemaTemplate struct { + dbName string + // tableDDL holds the CREATE TABLE statements of the template. MySQL has no + // server-side database template, so the schema is replayed statement by + // statement into each test database. + tableDDL []string +} + +func schemaTemplateKey(engine types.Engine, dsn string) string { + return string(engine) + "|" + dsn +} + +func newTestDBName(prefix string) string { + return fmt.Sprintf("%s_%s", prefix, strings.ReplaceAll(uuid.New().String(), "-", "_")) +} + +// postgresSchemaTemplate returns the name of a fully migrated database that +// CREATE DATABASE ... TEMPLATE can copy, creating it on first use. +func postgresSchemaTemplate(ctx context.Context, baseDSN string, admin *gorm.DB) (string, error) { + schemaTemplatesMu.Lock() + defer schemaTemplatesMu.Unlock() + + key := schemaTemplateKey(types.PostgresStoreEngine, baseDSN) + if tpl, ok := schemaTemplates[key]; ok { + return tpl.dbName, nil + } + + name := newTestDBName("test_template") + if err := admin.Exec(fmt.Sprintf("CREATE DATABASE %s", name)).Error; err != nil { + return "", fmt.Errorf("create postgres template database: %w", err) + } + + tplStore, err := NewPostgresqlStoreForTests(ctx, replaceDBName(baseDSN, name), nil, false) + if err != nil { + dropDatabase(admin, name) + return "", fmt.Errorf("migrate postgres template database: %w", err) + } + // TEMPLATE refuses a source that still has sessions, so release both handles + // before the first clone. + tplStore.Close(ctx) + if tplStore.pool != nil { + tplStore.pool.Close() + } + + schemaTemplates[key] = &schemaTemplate{dbName: name} + return name, nil +} + +// mysqlSchemaTemplate returns the CREATE TABLE statements of a fully migrated +// database, migrating one on first use. +func mysqlSchemaTemplate(ctx context.Context, baseDSN string, admin *gorm.DB) ([]string, error) { + schemaTemplatesMu.Lock() + defer schemaTemplatesMu.Unlock() + + key := schemaTemplateKey(types.MysqlStoreEngine, baseDSN) + if tpl, ok := schemaTemplates[key]; ok { + return tpl.tableDDL, nil + } + + name := newTestDBName("test_template") + if err := admin.Exec(fmt.Sprintf("CREATE DATABASE %s", name)).Error; err != nil { + return nil, fmt.Errorf("create mysql template database: %w", err) + } + + tplStore, err := NewMysqlStore(ctx, replaceDBName(baseDSN, name), nil, false) + if err != nil { + dropDatabase(admin, name) + return nil, fmt.Errorf("migrate mysql template database: %w", err) + } + tableDDL, err := mysqlTableDDL(ctx, tplStore.db, name) + tplStore.Close(ctx) + if err != nil { + dropDatabase(admin, name) + return nil, err + } + + schemaTemplates[key] = &schemaTemplate{dbName: name, tableDDL: tableDDL} + return tableDDL, nil +} + +func mysqlTableDDL(ctx context.Context, db *gorm.DB, dbName string) ([]string, error) { + sqlDB, err := db.DB() + if err != nil { + return nil, err + } + + tables, err := mysqlTableNames(ctx, sqlDB, dbName) + if err != nil { + return nil, err + } + + tableDDL := make([]string, 0, len(tables)) + for _, table := range tables { + var name, createStmt string + row := sqlDB.QueryRowContext(ctx, fmt.Sprintf("SHOW CREATE TABLE %s.%s", dbName, table)) + if err := row.Scan(&name, &createStmt); err != nil { + return nil, fmt.Errorf("read create statement of %s: %w", table, err) + } + tableDDL = append(tableDDL, createStmt) + } + return tableDDL, nil +} + +func mysqlTableNames(ctx context.Context, sqlDB *sql.DB, dbName string) ([]string, error) { + rows, err := sqlDB.QueryContext(ctx, fmt.Sprintf("SHOW TABLES FROM %s", dbName)) + if err != nil { + return nil, fmt.Errorf("list template tables: %w", err) + } + defer rows.Close() + + var tables []string + for rows.Next() { + var table string + if err := rows.Scan(&table); err != nil { + return nil, fmt.Errorf("scan template table name: %w", err) + } + tables = append(tables, table) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list template tables: %w", err) + } + return tables, nil +} + +// cloneMysqlSchema replays the template's CREATE TABLE statements into the +// database the DSN points at. +func cloneMysqlSchema(ctx context.Context, dsn string, tableDDL []string) error { + db, err := gorm.Open(mysql.Open(dsn+"?charset=utf8&parseTime=True&loc=Local"), getGormConfig()) + if err != nil { + return fmt.Errorf("connect to test database: %w", err) + } + sqlDB, err := db.DB() + if err != nil { + return err + } + defer sqlDB.Close() + + // The statements come out of SHOW TABLES in name order, not dependency + // order, and their foreign keys reference tables of the session's default + // database. Pin a single connection so the session setting below covers + // every statement, and connect straight to the new database so unqualified + // references land there. + sqlDB.SetMaxOpenConns(1) + if _, err := sqlDB.ExecContext(ctx, "SET FOREIGN_KEY_CHECKS = 0"); err != nil { + return fmt.Errorf("disable foreign key checks: %w", err) + } + for _, stmt := range tableDDL { + if _, err := sqlDB.ExecContext(ctx, stmt); err != nil { + return fmt.Errorf("replay table definition: %w", err) + } + } + return nil +} + +// dropDatabase removes a template that never became usable, so a failed setup +// does not leave it behind on a shared server. The server may still be tearing +// down the sessions the failed migration held, so the drop retries while +// Postgres reports the database as in use. +func dropDatabase(admin *gorm.DB, name string) { + if err := execWithTemplateRetry(admin, fmt.Sprintf("DROP DATABASE IF EXISTS %s", name)); err != nil { + log.Warnf("failed to drop template database %s: %v", name, err) + } +} + +func closeGormDB(db *gorm.DB) { + if sqlDB, _ := db.DB(); sqlDB != nil { + sqlDB.Close() + } +} + func openDBWithRetry(dsn string, engine types.Engine, maxRetries int) (*gorm.DB, error) { var db *gorm.DB var err error @@ -891,10 +1096,16 @@ func openDBWithRetry(dsn string, engine types.Engine, maxRetries int) (*gorm.DB, return nil, err } -func createRandomDB(dsn string, db *gorm.DB, engine types.Engine) (string, func(), error) { - dbName := fmt.Sprintf("test_db_%s", strings.ReplaceAll(uuid.New().String(), "-", "_")) +// createRandomDB creates a uniquely named database for one test. On postgres a +// non-empty template is copied server-side with CREATE DATABASE ... TEMPLATE. +func createRandomDB(dsn string, db *gorm.DB, engine types.Engine, template string) (string, func(), error) { + dbName := newTestDBName("test_db") - if err := db.Exec(fmt.Sprintf("CREATE DATABASE %s", dbName)).Error; err != nil { + createStmt := fmt.Sprintf("CREATE DATABASE %s", dbName) + if template != "" && engine == types.PostgresStoreEngine { + createStmt = fmt.Sprintf("CREATE DATABASE %s TEMPLATE %s", dbName, template) + } + if err := execWithTemplateRetry(db, createStmt); err != nil { return "", nil, fmt.Errorf("failed to create database: %v", err) } @@ -960,6 +1171,20 @@ func createRandomDB(dsn string, db *gorm.DB, engine types.Engine) (string, func( return replaceDBName(dsn, dbName), cleanup, nil } +// execWithTemplateRetry runs a statement, retrying briefly when postgres still +// sees the template's just-closed sessions and refuses to copy it. +func execWithTemplateRetry(db *gorm.DB, stmt string) error { + var err error + for attempt := 0; attempt < 20; attempt++ { + err = db.Exec(stmt).Error + if err == nil || !strings.Contains(err.Error(), "is being accessed by other users") { + return err + } + time.Sleep(100 * time.Millisecond) + } + return err +} + func replaceDBName(dsn, newDBName string) string { re := regexp.MustCompile(`(?P
[:/@])(?P[^/?]+)(?P\?|$)`)
 	return re.ReplaceAllString(dsn, `${pre}`+newDBName+`${post}`)
diff --git a/management/server/testutil/store.go b/management/server/testutil/store.go
index 07699e2c3..3854c1173 100644
--- a/management/server/testutil/store.go
+++ b/management/server/testutil/store.go
@@ -37,6 +37,18 @@ func CreateMysqlTestContainer() (func(), string, error) {
 		mysql.WithDatabase("testing"),
 		mysql.WithUsername("root"),
 		mysql.WithPassword("testing"),
+		// Every test creates and drops a database with about 40 tables, so with
+		// the server defaults the run is dominated by durability work: each
+		// CREATE TABLE fsyncs the redo log, the binary log and the doublewrite
+		// buffer. None of it protects anything in a container that is discarded
+		// after the run. Tables stay in per-table files on purpose: in the shared
+		// system tablespace the cost of every CREATE and DROP grew with the number
+		// of databases the run had already created.
+		testcontainers.WithCmd("mysqld",
+			"--innodb-flush-log-at-trx-commit=0",
+			"--innodb-doublewrite=OFF",
+			"--skip-log-bin",
+		),
 		testcontainers.WithWaitStrategy(
 			wait.ForLog("/usr/sbin/mysqld: ready for connections").
 				WithOccurrence(1).WithStartupTimeout(15*time.Second).WithPollInterval(100*time.Millisecond),
diff --git a/tools/gotestsummary/main.go b/tools/gotestsummary/main.go
new file mode 100644
index 000000000..2a5487fc6
--- /dev/null
+++ b/tools/gotestsummary/main.go
@@ -0,0 +1,402 @@
+// Package main turns a `go test -json` stream into a readable CI log.
+//
+// It prints one line per top-level test as it finishes, the captured output of
+// every failed test, the head of a package-level panic (which is where Go
+// reports "test timed out" and the list of still-running tests), and ends with
+// the per-package durations and the slowest tests. The exit code is always zero
+// unless the input cannot be read; the `go test` exit code is what CI should act
+// on, so run the two with `set -o pipefail`.
+//
+// Usage:
+//
+//	go test -json ./... | go run ./tools/gotestsummary
+//	go run ./tools/gotestsummary -slowest 60 test-output.jsonl
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"sort"
+	"strings"
+	"time"
+)
+
+const (
+	modulePrefix = "github.com/netbirdio/netbird/"
+
+	// failedTestOutputLines bounds how much captured output a single failed
+	// test may print, so one noisy failure cannot flood the job log.
+	failedTestOutputLines = 200
+	// panicHeadLines is enough for the "panic: test timed out" header, the
+	// "running tests:" list, and the first few goroutines of the dump.
+	panicHeadLines = 150
+	// bufferedOutputLines bounds the per-test output kept in memory while the
+	// test runs; only the tail is kept once the cap is reached.
+	bufferedOutputLines = 400
+)
+
+type event struct {
+	Action  string  `json:"Action"`
+	Package string  `json:"Package"`
+	Test    string  `json:"Test"`
+	Output  string  `json:"Output"`
+	Elapsed float64 `json:"Elapsed"`
+	// ImportPath is set instead of Package on build events. It carries a
+	// " [pkg.test]" suffix naming the test binary the package was compiled
+	// for, and the same package can be built for several binaries at once.
+	ImportPath string `json:"ImportPath"`
+	// FailedBuild names the ImportPath whose build failure made the package
+	// fail; go test reports the package fail event after the build-fail one.
+	FailedBuild string `json:"FailedBuild"`
+}
+
+type testKey struct {
+	pkg, name string
+}
+
+type testResult struct {
+	pkg, name string
+	action    string
+	elapsed   time.Duration
+}
+
+type packageResult struct {
+	pkg     string
+	action  string
+	elapsed time.Duration
+}
+
+type summarizer struct {
+	out io.Writer
+
+	output  map[testKey][]string
+	dropped map[testKey]int
+	// pkgOutput keeps what a package printed outside any test, which is where
+	// compiler diagnostics of a failed build end up.
+	pkgOutput map[string][]string
+	// failedBuilds holds the ImportPaths whose build failed and has not been
+	// reported through a package fail event yet.
+	failedBuilds map[string]bool
+	tests        []testResult
+	packages     []packageResult
+	// panics holds the head of a panic per package. Package streams interleave
+	// in a go test -json run, so one package's dump must not swallow another's
+	// output.
+	panics map[string][]string
+}
+
+func newSummarizer(out io.Writer) *summarizer {
+	return &summarizer{
+		out:          out,
+		output:       make(map[testKey][]string),
+		dropped:      make(map[testKey]int),
+		pkgOutput:    make(map[string][]string),
+		failedBuilds: make(map[string]bool),
+		panics:       make(map[string][]string),
+	}
+}
+
+func main() {
+	slowest := flag.Int("slowest", 40, "number of slowest top-level tests to list")
+	flag.Parse()
+
+	if err := run(flag.Arg(0), *slowest); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(path string, slowest int) error {
+	in := os.Stdin
+	if path != "" {
+		f, err := os.Open(path)
+		if err != nil {
+			return fmt.Errorf("open %s: %w", path, err)
+		}
+		defer f.Close()
+		in = f
+	}
+
+	s := newSummarizer(os.Stdout)
+	if err := s.consume(in); err != nil {
+		return fmt.Errorf("read input: %w", err)
+	}
+	s.printSummary(slowest)
+	return nil
+}
+
+func (s *summarizer) consume(r io.Reader) error {
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 1024*1024), 16*1024*1024)
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		var ev event
+		if err := json.Unmarshal(line, &ev); err != nil {
+			// Build errors and other non-JSON lines are passed through untouched.
+			fmt.Fprintln(s.out, string(line))
+			continue
+		}
+		s.handle(ev)
+	}
+	return scanner.Err()
+}
+
+func (s *summarizer) handle(ev event) {
+	if ev.Package == "" {
+		// Keep the full path as the key so concurrent builds of one package for
+		// different test binaries do not share, and delete, each other's state.
+		ev.Package = ev.ImportPath
+	}
+	key := testKey{pkg: ev.Package, name: ev.Test}
+	switch ev.Action {
+	case "run":
+		// Register the test even before it prints anything, so a test that
+		// hangs silently still shows up as unfinished.
+		if ev.Test != "" {
+			if _, ok := s.output[key]; !ok {
+				s.output[key] = []string{}
+			}
+		}
+	case "output":
+		s.handleOutput(key, strings.TrimRight(ev.Output, "\n"))
+	case "build-output":
+		// Compiler output may carry several lines per event and is never test
+		// output, so it skips the panic detection.
+		for _, line := range strings.Split(strings.TrimRight(ev.Output, "\n"), "\n") {
+			s.pkgOutput[key.pkg] = appendBounded(s.pkgOutput[key.pkg], line)
+		}
+	case "build-fail":
+		// The package fail event that follows carries FailedBuild and reports
+		// the compiler output; this only remembers the build in case it never
+		// comes.
+		s.failedBuilds[key.pkg] = true
+	case "pass", "fail", "skip":
+		if ev.Test == "" {
+			s.handlePackageResult(ev)
+			return
+		}
+		s.handleTestResult(key, ev)
+	}
+}
+
+func (s *summarizer) handleOutput(key testKey, line string) {
+	if strings.HasPrefix(line, "panic: ") || strings.HasPrefix(line, "fatal error: ") {
+		if _, ok := s.panics[key.pkg]; !ok {
+			s.panics[key.pkg] = []string{}
+		}
+	}
+	if head, ok := s.panics[key.pkg]; ok {
+		// The goroutine dump that follows a panic is kept in the panic head only;
+		// letting it flood the per-test buffers would hide the test's own output.
+		if len(head) < panicHeadLines {
+			s.panics[key.pkg] = append(head, line)
+		}
+		return
+	}
+
+	if key.name == "" {
+		s.pkgOutput[key.pkg] = appendBounded(s.pkgOutput[key.pkg], line)
+		return
+	}
+	if len(s.output[key]) >= bufferedOutputLines {
+		s.dropped[key]++
+	}
+	s.output[key] = appendBounded(s.output[key], line)
+}
+
+// appendBounded keeps the most recent bufferedOutputLines lines.
+func appendBounded(buf []string, line string) []string {
+	if len(buf) >= bufferedOutputLines {
+		buf = buf[1:]
+	}
+	return append(buf, line)
+}
+
+func (s *summarizer) handleTestResult(key testKey, ev event) {
+	elapsed := time.Duration(ev.Elapsed * float64(time.Second))
+	s.tests = append(s.tests, testResult{
+		pkg:     ev.Package,
+		name:    ev.Test,
+		action:  ev.Action,
+		elapsed: elapsed,
+	})
+
+	if !strings.Contains(ev.Test, "/") || ev.Action == "fail" {
+		fmt.Fprintf(s.out, "--- %s: %s.%s (%s)\n", strings.ToUpper(ev.Action), shortPkg(ev.Package), ev.Test, elapsed.Round(time.Millisecond))
+	}
+	if ev.Action == "fail" {
+		s.printTestOutput(key)
+	}
+	delete(s.output, key)
+	delete(s.dropped, key)
+}
+
+func (s *summarizer) printTestOutput(key testKey) {
+	lines := s.output[key]
+	if len(lines) == 0 {
+		return
+	}
+	skipped := s.dropped[key]
+	if len(lines) > failedTestOutputLines {
+		skipped += len(lines) - failedTestOutputLines
+		lines = lines[len(lines)-failedTestOutputLines:]
+	}
+	if skipped > 0 {
+		fmt.Fprintf(s.out, "    ... %d earlier output lines omitted ...\n", skipped)
+	}
+	for _, l := range lines {
+		fmt.Fprintf(s.out, "    %s\n", l)
+	}
+}
+
+func (s *summarizer) handlePackageResult(ev event) {
+	elapsed := time.Duration(ev.Elapsed * float64(time.Second))
+	s.packages = append(s.packages, packageResult{pkg: ev.Package, action: ev.Action, elapsed: elapsed})
+
+	label := "ok  "
+	switch ev.Action {
+	case "fail", "build-fail":
+		label = "FAIL"
+	case "skip":
+		label = "skip"
+	}
+	fmt.Fprintf(s.out, "%s %s %s\n", label, shortPkg(ev.Package), elapsed.Round(time.Millisecond))
+
+	if label == "FAIL" {
+		if ev.FailedBuild != "" {
+			// Several test binaries can share one failed dependency, so its
+			// output stays available for the next package that names it.
+			s.printPackageOutput(ev.FailedBuild, "build output of %s")
+			delete(s.failedBuilds, ev.FailedBuild)
+		}
+		s.printPackageOutput(ev.Package, "output of %s outside tests")
+		s.printUnfinished(ev.Package)
+		s.printPanicHead(ev.Package)
+	}
+	delete(s.pkgOutput, ev.Package)
+	delete(s.panics, ev.Package)
+}
+
+// printUnclaimedBuildFailures reports the failed builds no package fail event
+// accounted for, so a compiler error never disappears from the log.
+func (s *summarizer) printUnclaimedBuildFailures() {
+	var builds []string
+	for b := range s.failedBuilds {
+		builds = append(builds, b)
+	}
+	sort.Strings(builds)
+	for _, b := range builds {
+		fmt.Fprintf(s.out, "FAIL %s [build failed]\n", shortPkg(b))
+		s.printPackageOutput(b, "build output of %s")
+	}
+}
+
+// printPackageOutput shows what a failed package printed outside its tests,
+// or the compiler errors of a failed build, under the given header.
+func (s *summarizer) printPackageOutput(pkg, header string) {
+	lines := s.pkgOutput[pkg]
+	if len(lines) == 0 {
+		return
+	}
+	if len(lines) > failedTestOutputLines {
+		lines = lines[len(lines)-failedTestOutputLines:]
+	}
+	fmt.Fprintf(s.out, "\n==== "+header+" ====\n", shortPkg(pkg))
+	for _, l := range lines {
+		fmt.Fprintf(s.out, "    %s\n", l)
+	}
+}
+
+func (s *summarizer) printPanicHead(pkg string) {
+	head := s.panics[pkg]
+	if len(head) == 0 {
+		return
+	}
+	fmt.Fprintf(s.out, "\n==== panic in %s (first %d lines) ====\n", shortPkg(pkg), len(head))
+	for _, l := range head {
+		fmt.Fprintln(s.out, l)
+	}
+	fmt.Fprintln(s.out, "==== end of panic head ====")
+	fmt.Fprintln(s.out)
+}
+
+// printUnfinished names the tests of a failed package that never reported a
+// result, which is what a timeout leaves behind, and shows their last output.
+func (s *summarizer) printUnfinished(pkg string) {
+	var keys []testKey
+	for key := range s.output {
+		if key.pkg == pkg && key.name != "" {
+			keys = append(keys, key)
+		}
+	}
+	if len(keys) == 0 {
+		return
+	}
+	sort.Slice(keys, func(i, j int) bool { return keys[i].name < keys[j].name })
+	fmt.Fprintf(s.out, "\n==== tests in %s that did not finish (%d) ====\n", shortPkg(pkg), len(keys))
+	for _, key := range keys {
+		fmt.Fprintf(s.out, "--- UNFINISHED: %s.%s\n", shortPkg(key.pkg), key.name)
+		s.printTestOutput(key)
+		delete(s.output, key)
+		delete(s.dropped, key)
+	}
+}
+
+func (s *summarizer) printSummary(slowest int) {
+	s.printUnclaimedBuildFailures()
+
+	fmt.Fprintln(s.out)
+	fmt.Fprintln(s.out, "==== package durations ====")
+	sort.Slice(s.packages, func(i, j int) bool { return s.packages[i].elapsed > s.packages[j].elapsed })
+	for _, p := range s.packages {
+		fmt.Fprintf(s.out, "%9s  %-4s  %s\n", p.elapsed.Round(time.Millisecond), p.action, shortPkg(p.pkg))
+	}
+
+	var failed []testResult
+	for _, t := range s.tests {
+		if t.action == "fail" {
+			failed = append(failed, t)
+		}
+	}
+	if len(failed) > 0 {
+		fmt.Fprintln(s.out)
+		fmt.Fprintf(s.out, "==== failed tests (%d) ====\n", len(failed))
+		for _, t := range failed {
+			fmt.Fprintf(s.out, "%9s  %s.%s\n", t.elapsed.Round(time.Millisecond), shortPkg(t.pkg), t.name)
+		}
+	}
+
+	s.printSlowest("slowest top-level tests", slowest, func(t testResult) bool { return !strings.Contains(t.name, "/") })
+	s.printSlowest("slowest subtests", slowest/2, func(t testResult) bool { return strings.Contains(t.name, "/") })
+}
+
+func (s *summarizer) printSlowest(title string, limit int, keep func(testResult) bool) {
+	var tests []testResult
+	for _, t := range s.tests {
+		if keep(t) {
+			tests = append(tests, t)
+		}
+	}
+	if len(tests) == 0 || limit <= 0 {
+		return
+	}
+	sort.Slice(tests, func(i, j int) bool { return tests[i].elapsed > tests[j].elapsed })
+	if len(tests) > limit {
+		tests = tests[:limit]
+	}
+
+	fmt.Fprintln(s.out)
+	fmt.Fprintf(s.out, "==== %s (%d) ====\n", title, len(tests))
+	for _, t := range tests {
+		fmt.Fprintf(s.out, "%9s  %-4s  %s.%s\n", t.elapsed.Round(time.Millisecond), t.action, shortPkg(t.pkg), t.name)
+	}
+}
+
+func shortPkg(pkg string) string {
+	pkg, _, _ = strings.Cut(pkg, " [")
+	return strings.TrimPrefix(pkg, modulePrefix)
+}
diff --git a/tools/gotestsummary/main_test.go b/tools/gotestsummary/main_test.go
new file mode 100644
index 000000000..d5b51e132
--- /dev/null
+++ b/tools/gotestsummary/main_test.go
@@ -0,0 +1,180 @@
+package main
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func feed(t *testing.T, events string) string {
+	t.Helper()
+	var out bytes.Buffer
+	s := newSummarizer(&out)
+	if err := s.consume(strings.NewReader(events)); err != nil {
+		t.Fatalf("consume: %v", err)
+	}
+	s.printSummary(10)
+	return out.String()
+}
+
+func TestTimeoutReportsUnfinishedTestsAndPanicHead(t *testing.T) {
+	events := `
+{"Action":"run","Package":"a","Test":"TestHang"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"=== RUN   TestHang\n"}
+{"Action":"run","Package":"a","Test":"TestSilent"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"panic: test timed out after 1s\n"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"\trunning tests:\n"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"\t\tTestHang (1s)\n"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"goroutine 7 [running]:\n"}
+{"Action":"fail","Package":"a","Elapsed":1.0}
+`
+	got := feed(t, events)
+	for _, want := range []string{
+		"--- UNFINISHED: a.TestHang",
+		"--- UNFINISHED: a.TestSilent",
+		"==== panic in a (first 4 lines) ====",
+		"\t\tTestHang (1s)",
+		"    === RUN   TestHang",
+	} {
+		if !strings.Contains(got, want) {
+			t.Errorf("output lacks %q:\n%s", want, got)
+		}
+	}
+	if strings.Contains(got, "    goroutine 7 [running]:") {
+		t.Errorf("goroutine dump leaked into the test's own output:\n%s", got)
+	}
+}
+
+func TestPanicInOnePackageKeepsOtherPackageOutput(t *testing.T) {
+	events := `
+{"Action":"run","Package":"a","Test":"TestHang"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"panic: test timed out after 1s\n"}
+{"Action":"run","Package":"b","Test":"TestOther"}
+{"Action":"output","Package":"b","Test":"TestOther","Output":"    other_test.go:9: expected 1, got 2\n"}
+{"Action":"output","Package":"a","Test":"TestHang","Output":"goroutine 7 [running]:\n"}
+{"Action":"fail","Package":"b","Test":"TestOther","Elapsed":0.01}
+{"Action":"fail","Package":"b","Elapsed":0.02}
+{"Action":"fail","Package":"a","Elapsed":1.0}
+`
+	got := feed(t, events)
+	if !strings.Contains(got, "    other_test.go:9: expected 1, got 2") {
+		t.Errorf("other package's output was swallowed by the panic head:\n%s", got)
+	}
+	if strings.Contains(got, "panic in b") {
+		t.Errorf("panic head attributed to the wrong package:\n%s", got)
+	}
+	if !strings.Contains(got, "==== panic in a (first 2 lines) ====") {
+		t.Errorf("panic head missing for package a:\n%s", got)
+	}
+}
+
+func TestBuildFailureShowsCompilerOutput(t *testing.T) {
+	// The event sequence go test emits for a build failure: the build events
+	// name the test binary, then the package itself fails with FailedBuild.
+	events := `
+{"Action":"build-output","ImportPath":"a [a.test]","Output":"# a [a.test]\na_test.go:7:2: undefined: nope\na_test.go:9:2: undefined: nope2\n"}
+{"Action":"build-fail","ImportPath":"a [a.test]"}
+{"Action":"start","Package":"a"}
+{"Action":"output","Package":"a","Output":"FAIL\ta [build failed]\n"}
+{"Action":"fail","Package":"a","Elapsed":0,"FailedBuild":"a [a.test]"}
+`
+	got := feed(t, events)
+	for _, want := range []string{
+		"==== build output of a ====",
+		"    a_test.go:7:2: undefined: nope\n    a_test.go:9:2: undefined: nope2",
+		"FAIL\ta [build failed]",
+	} {
+		if !strings.Contains(got, want) {
+			t.Errorf("output lacks %q:\n%s", want, got)
+		}
+	}
+	if n := strings.Count(got, "FAIL a 0s"); n != 1 {
+		t.Errorf("expected one FAIL line for the package, got %d:\n%s", n, got)
+	}
+	if n := strings.Count(got, "undefined: nope2"); n != 1 {
+		t.Errorf("expected the compiler output once, got %d:\n%s", n, got)
+	}
+}
+
+func TestFailedDependencyOutputIsShownForEveryImporter(t *testing.T) {
+	events := `
+{"Action":"build-output","ImportPath":"m/x","Output":"# m/x\nx.go:3:11: undefined: y\n"}
+{"Action":"build-fail","ImportPath":"m/x"}
+{"Action":"start","Package":"m/a"}
+{"Action":"output","Package":"m/a","Output":"FAIL\tm/a [build failed]\n"}
+{"Action":"fail","Package":"m/a","Elapsed":0,"FailedBuild":"m/x"}
+{"Action":"start","Package":"m/b"}
+{"Action":"output","Package":"m/b","Output":"FAIL\tm/b [build failed]\n"}
+{"Action":"fail","Package":"m/b","Elapsed":0,"FailedBuild":"m/x"}
+`
+	got := feed(t, events)
+	if n := strings.Count(got, "x.go:3:11: undefined: y"); n != 2 {
+		t.Errorf("expected the dependency's compiler output under both packages, got %d:\n%s", n, got)
+	}
+	if strings.Contains(got, "FAIL m/x") {
+		t.Errorf("the dependency must not be reported as a package of its own:\n%s", got)
+	}
+}
+
+func TestBuildFailureWithoutPackageEventIsStillReported(t *testing.T) {
+	events := `
+{"Action":"build-output","ImportPath":"a [a.test]","Output":"a_test.go:7:2: undefined: nope\n"}
+{"Action":"build-fail","ImportPath":"a [a.test]"}
+`
+	got := feed(t, events)
+	for _, want := range []string{"FAIL a [build failed]", "==== build output of a ====", "undefined: nope"} {
+		if !strings.Contains(got, want) {
+			t.Errorf("output lacks %q:\n%s", want, got)
+		}
+	}
+}
+
+func TestCompilerPanicIsBuildOutputNotTestPanic(t *testing.T) {
+	events := `
+{"Action":"build-output","ImportPath":"a [a.test]","Output":"# a [a.test]\npanic: internal compiler error\n\ngoroutine 1 [running]:\n"}
+{"Action":"build-fail","ImportPath":"a [a.test]"}
+{"Action":"start","Package":"a"}
+{"Action":"output","Package":"a","Output":"FAIL\ta [build failed]\n"}
+{"Action":"fail","Package":"a","Elapsed":0,"FailedBuild":"a [a.test]"}
+`
+	got := feed(t, events)
+	if !strings.Contains(got, "==== build output of a ====\n    # a [a.test]\n    panic: internal compiler error") {
+		t.Errorf("compiler diagnostic missing from the build output block:\n%s", got)
+	}
+	if strings.Contains(got, "==== panic in") {
+		t.Errorf("compiler output must not be reported as a test panic:\n%s", got)
+	}
+}
+
+func TestBuildVariantsOfOnePackageKeepSeparateOutput(t *testing.T) {
+	events := `
+{"Action":"build-output","ImportPath":"a [a.test]","Output":"a.go:1:1: broken for a.test\n"}
+{"Action":"build-output","ImportPath":"a [b.test]","Output":"a.go:1:1: broken for b.test\n"}
+{"Action":"build-fail","ImportPath":"a [a.test]"}
+{"Action":"build-fail","ImportPath":"a [b.test]"}
+{"Action":"start","Package":"a"}
+{"Action":"fail","Package":"a","Elapsed":0,"FailedBuild":"a [a.test]"}
+{"Action":"start","Package":"b"}
+{"Action":"fail","Package":"b","Elapsed":0,"FailedBuild":"a [b.test]"}
+`
+	got := feed(t, events)
+	if strings.Count(got, "==== build output of a ====") != 2 {
+		t.Errorf("expected one output block per build variant:\n%s", got)
+	}
+	for _, want := range []string{"broken for a.test", "broken for b.test"} {
+		if strings.Count(got, want) != 1 {
+			t.Errorf("expected %q exactly once:\n%s", want, got)
+		}
+	}
+}
+
+func TestPassingPackageOutputIsNotPrinted(t *testing.T) {
+	events := `
+{"Action":"output","Package":"a","Output":"level=info msg=\"noise between tests\"\n"}
+{"Action":"pass","Package":"a","Elapsed":0.5}
+`
+	got := feed(t, events)
+	if strings.Contains(got, "noise between tests") {
+		t.Errorf("package output of a passing package should stay quiet:\n%s", got)
+	}
+}

From ea294e1d46c2fd3f51cbe92acf1e2551a4e14ba9 Mon Sep 17 00:00:00 2001
From: Maycon Santos 
Date: Mon, 14 Sep 2026 22:20:25 +0200
Subject: [PATCH 3/9] [management] Refuse to pin an agent network gateway onto
 another account's host (#7519)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An agent network bootstrap stores its cluster as proxy_address, which selects the proxy that serves the endpoint. An account-scoped proxy only receives its own account's mappings, so a pin onto a host another account's proxy declares can never be served, and the endpoint is immutable — a dead gateway until the account deletes its settings. Nothing refused that pin; the domain unique index only arbitrates between endpoints.

Both bootstrap paths now refuse, before the insert, a host that another account's proxy declares, a host another account has labeled pins beneath (self-addressed path), or a hostname that is another account's endpoint (labeled path).

Shared clusters are unaffected: shared proxies are never foreign, and labeled pins under one cluster are never asked about, so any number of accounts still pin beneath eu.proxy.netbird.io. Registration is deliberately unchanged — refusing a proxy for another account's pin would let a pin lock a tenant out after the reaper drops its rows.
---
 .../internals/modules/agentnetwork/manager.go |  54 +++++
 .../agentnetwork/settings_bootstrap_test.go   | 186 +++++++++++++++++-
 management/server/store/sql_store.go          |  19 ++
 .../server/store/sql_store_agentnetwork.go    |  30 +++
 management/server/store/store.go              |   3 +
 management/server/store/store_mock.go         |  45 +++++
 6 files changed, 336 insertions(+), 1 deletion(-)

diff --git a/management/internals/modules/agentnetwork/manager.go b/management/internals/modules/agentnetwork/manager.go
index efcc944be..d4524a218 100644
--- a/management/internals/modules/agentnetwork/manager.go
+++ b/management/internals/modules/agentnetwork/manager.go
@@ -1036,6 +1036,15 @@ func (m *managerImpl) bootstrapSelfAddressed(ctx context.Context, settings *type
 	if err != nil {
 		return status.Errorf(status.InvalidArgument, "invalid endpoint: %s", err)
 	}
+	if err := m.requireHostNotForeign(ctx, settings.AccountID, hostname); err != nil {
+		return err
+	}
+	// Another account's labeled pin beneath this hostname makes it their
+	// cluster: a proxy serving them there would never serve this endpoint.
+	// The domain unique index already arbitrates two endpoints on one name.
+	if err := m.requireNotClaimedByOtherAccount(ctx, settings.AccountID, hostname, m.store.HasGatewayClusterPinnedByOtherAccount); err != nil {
+		return err
+	}
 
 	settings.Domain = hostname
 	settings.ProxyAddress = hostname
@@ -1065,6 +1074,16 @@ func (m *managerImpl) bootstrapLabeled(ctx context.Context, settings *types.Sett
 	if err != nil {
 		return status.Errorf(status.InvalidArgument, "invalid proxy_address: %s", err)
 	}
+	if err := m.requireHostNotForeign(ctx, settings.AccountID, parent); err != nil {
+		return err
+	}
+	// Another account's endpoint at this exact hostname means the proxy that
+	// declares it is theirs, so nothing would serve a label beneath it. Other
+	// accounts' labeled pins under the same cluster are not asked about: a
+	// shared cluster carries many of them by design.
+	if err := m.requireNotClaimedByOtherAccount(ctx, settings.AccountID, parent, m.store.HasGatewayEndpointByOtherAccount); err != nil {
+		return err
+	}
 
 	for attempt := 1; attempt <= maxDomainAllocationAttempts; attempt++ {
 		label := labelgen.PickTuple()
@@ -1111,6 +1130,41 @@ func (m *managerImpl) bootstrapLabeled(ctx context.Context, settings *types.Sett
 	return fmt.Errorf("allocate agent network endpoint for account %s: %d attempts exhausted", settings.AccountID, maxDomainAllocationAttempts)
 }
 
+// requireHostNotForeign refuses to pin the account's gateway onto a host that
+// another account's proxy declares. The pin's proxy_address is what selects
+// the proxy that serves the endpoint, and an account-scoped proxy only ever
+// receives its own account's mappings, so such a pin could never be served —
+// and the endpoint it assigns is immutable. Shared proxies are not foreign, and
+// a host no proxy has declared stays pinnable: claiming the address before the
+// proxy's first connection is the documented order.
+func (m *managerImpl) requireHostNotForeign(ctx context.Context, accountID, host string) error {
+	foreign, err := m.store.HasForeignAccountProxyAtHost(ctx, host, accountID)
+	if err != nil {
+		return fmt.Errorf("check proxy host ownership: %w", err)
+	}
+	if foreign {
+		return errHostNotAvailable(host)
+	}
+	return nil
+}
+
+// requireNotClaimedByOtherAccount refuses the pin when another account's
+// gateway settings already claim the host in the shape claimed answers for.
+func (m *managerImpl) requireNotClaimedByOtherAccount(ctx context.Context, accountID, host string, claimed func(context.Context, string, string) (bool, error)) error {
+	taken, err := claimed(ctx, host, accountID)
+	if err != nil {
+		return fmt.Errorf("check agent network gateway claims at host: %w", err)
+	}
+	if taken {
+		return errHostNotAvailable(host)
+	}
+	return nil
+}
+
+func errHostNotAvailable(host string) error {
+	return status.Errorf(status.InvalidArgument, "proxy cluster %s is not available to this account", host)
+}
+
 // isUniqueConstraintError reports whether err is a database unique-constraint
 // violation, matched on the driver message because CreateAgentNetworkSettings
 // deliberately returns the driver error unwrapped.
diff --git a/management/internals/modules/agentnetwork/settings_bootstrap_test.go b/management/internals/modules/agentnetwork/settings_bootstrap_test.go
index fea62353e..2e4be721b 100644
--- a/management/internals/modules/agentnetwork/settings_bootstrap_test.go
+++ b/management/internals/modules/agentnetwork/settings_bootstrap_test.go
@@ -5,12 +5,14 @@ import (
 	"runtime"
 	"strings"
 	"testing"
+	"time"
 
-	"go.uber.org/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
 	"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
+	"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
 	"github.com/netbirdio/netbird/management/server/account"
 	"github.com/netbirdio/netbird/management/server/permissions"
 	"github.com/netbirdio/netbird/management/server/permissions/modules"
@@ -70,6 +72,50 @@ func (f *bootstrapFixture) createSettings(ctx context.Context, accountID, userID
 	return f.manager.CreateSettings(ctx, userID, types.DefaultSettings(accountID), proxyAddress, endpoint)
 }
 
+func ptrTo[T any](v T) *T { return &v }
+
+// seedProxy registers a proxy in clusterAddr, heartbeating now, so the labeled
+// bootstrap path has a real cluster to validate against. accountID empty makes
+// it a shared (NetBird-operated) cluster; private mirrors the capability an
+// embedded `netbird proxy` reports, nil an unreported one.
+func (f *bootstrapFixture) seedProxy(t *testing.T, proxyID, accountID, clusterAddr string, private *bool) {
+	t.Helper()
+	f.seedProxyAt(t, proxyID, accountID, clusterAddr, private, time.Now().UTC())
+}
+
+// seedProxyAt is seedProxy with an explicit last-seen, for cases that need a
+// proxy whose heartbeat has aged past the active window while its row (and so
+// its cluster) is still on record.
+func (f *bootstrapFixture) seedProxyAt(t *testing.T, proxyID, accountID, clusterAddr string, private *bool, lastSeen time.Time) {
+	t.Helper()
+	p := &proxy.Proxy{
+		ID:             proxyID,
+		ClusterAddress: clusterAddr,
+		Status:         proxy.StatusConnected,
+		LastSeen:       lastSeen,
+		Capabilities:   proxy.Capabilities{Private: private},
+	}
+	if accountID != "" {
+		p.AccountID = &accountID
+	}
+	require.NoError(t, f.store.SaveProxy(context.Background(), p), "seeding a proxy must succeed")
+}
+
+// requireForeignClusterRefusal asserts the refusal a pin onto another
+// account's host gets, and that it left no row behind.
+func (f *bootstrapFixture) requireForeignClusterRefusal(t *testing.T, err error, accountID string) {
+	t.Helper()
+	require.Error(t, err, "another account's host must be refused")
+	var sErr *status.Error
+	require.ErrorAs(t, err, &sErr)
+	assert.Equal(t, status.InvalidArgument, sErr.Type(), "rejection must be a validation error")
+	assert.Contains(t, err.Error(), "not available to this account",
+		"the error must say the host is not the account's to use")
+
+	_, err = f.store.GetAgentNetworkSettings(context.Background(), store.LockingStrengthNone, accountID)
+	assert.Error(t, err, "no row may be left behind by a rejected bootstrap")
+}
+
 // TestCreateSettingsRequiresPermission pins the gate: bootstrap assigns the
 // account's immutable endpoint, a settings write requiring the settings
 // Create permission — and a denial leaves no row behind.
@@ -230,3 +276,141 @@ func TestCreateProviderHasNoSettingsSideEffects(t *testing.T) {
 	_, err = f.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, "account1")
 	assert.Error(t, err, "provider create must not conjure a settings row")
 }
+
+// TestCreateSettingsRejectsForeignCluster pins tenant consistency on the pin:
+// an account may not pin its gateway onto a host another account's proxy
+// declares. That proxy only ever receives its own account's mappings, so the
+// pin could never be served, and the endpoint it assigns is immutable.
+// Ownership is decided on the proxy rows, not on heartbeat freshness — a
+// cluster whose proxies are merely offline is still somebody's — and on the
+// normalised host, since proxies declare their address as the operator
+// spelled it.
+func TestCreateSettingsRejectsForeignCluster(t *testing.T) {
+	ctx := context.Background()
+
+	cases := map[string]struct {
+		spelling string
+		lastSeen time.Time
+	}{
+		"live":            {"byop.account2.example.com", time.Now().UTC()},
+		"offline":         {"byop.account2.example.com", time.Now().UTC().Add(-time.Hour)},
+		"spelled in caps": {"BYOP.Account2.Example.com", time.Now().UTC()},
+	}
+	for name, tc := range cases {
+		t.Run("labeled "+name, func(t *testing.T) {
+			f := newBootstrapFixture(t)
+			f.seedProxyAt(t, "proxy1", "account2", tc.spelling, ptrTo(true), tc.lastSeen)
+			f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+
+			_, err := f.createSettings(ctx, "account1", "user1", "byop.account2.example.com", "")
+			f.requireForeignClusterRefusal(t, err, "account1")
+		})
+		t.Run("self-addressed "+name, func(t *testing.T) {
+			f := newBootstrapFixture(t)
+			f.seedProxyAt(t, "proxy1", "account2", tc.spelling, ptrTo(true), tc.lastSeen)
+			f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+
+			_, err := f.createSettings(ctx, "account1", "user1", "", "byop.account2.example.com")
+			f.requireForeignClusterRefusal(t, err, "account1")
+		})
+	}
+}
+
+// TestCreateSettingsSharedClusterStaysPinnable pins the constraint the
+// ownership check must respect: a shared (NetBird-operated) cluster is not
+// anybody's, so any number of accounts pin their gateways to it — including
+// an account that also runs a proxy of its own elsewhere.
+func TestCreateSettingsSharedClusterStaysPinnable(t *testing.T) {
+	ctx := context.Background()
+	f := newBootstrapFixture(t)
+	f.seedProxy(t, "shared", "", "eu.proxy.netbird.io", ptrTo(true))
+	f.seedProxy(t, "own", "account1", "byop.account1.example.com", ptrTo(true))
+
+	for _, account := range []string{"account1", "account2"} {
+		f.expectPermission(account, "user", modules.AgentNetworkSettings, operations.Create, true)
+		created, err := f.createSettings(ctx, account, "user", "eu.proxy.netbird.io", "")
+		require.NoError(t, err, "a shared cluster must stay pinnable by %s", account)
+		assert.Equal(t, "eu.proxy.netbird.io", created.ProxyAddress)
+	}
+}
+
+// TestCreateSettingsOwnClusterIsPinnable is the BYOP order in both directions:
+// the account's own proxy is not a competing claim, whether the pin is labeled
+// beneath its cluster or self-addressed onto the very host it declares.
+func TestCreateSettingsOwnClusterIsPinnable(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("labeled", func(t *testing.T) {
+		f := newBootstrapFixture(t)
+		f.seedProxy(t, "own", "account1", "byop.account1.example.com", ptrTo(true))
+		f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+
+		created, err := f.createSettings(ctx, "account1", "user1", "byop.account1.example.com", "")
+		require.NoError(t, err, "the account's own cluster must be pinnable")
+		assert.True(t, strings.HasSuffix(created.Domain, ".byop.account1.example.com"))
+	})
+	t.Run("self-addressed", func(t *testing.T) {
+		f := newBootstrapFixture(t)
+		f.seedProxy(t, "own", "account1", "gw.account1.example.com", ptrTo(true))
+		f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+
+		created, err := f.createSettings(ctx, "account1", "user1", "", "gw.account1.example.com")
+		require.NoError(t, err, "the host the account's own proxy declares must be pinnable")
+		assert.Equal(t, "gw.account1.example.com", created.ProxyAddress)
+	})
+}
+
+// TestCreateSettingsUnknownHostIsPinnable pins the address-first order: a host
+// no proxy has ever declared is nobody's, so the pin goes through and the
+// proxy is deployed after.
+func TestCreateSettingsUnknownHostIsPinnable(t *testing.T) {
+	ctx := context.Background()
+	f := newBootstrapFixture(t)
+	f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+
+	created, err := f.createSettings(ctx, "account1", "user1", "future.example.com", "")
+	require.NoError(t, err, "a host no proxy has declared must stay pinnable")
+	assert.Equal(t, "future.example.com", created.ProxyAddress)
+}
+
+// TestCreateSettingsRejectsHostAnotherAccountPinned covers claims made by pins
+// rather than proxies, which the proxy-row check cannot see. A labeled pin
+// beneath a host makes that host the other account's cluster, so a
+// self-addressed endpoint on it would never be served; a self-addressed
+// endpoint on a host makes the proxy declaring it theirs, so a label beneath
+// it would never be served either. Neither is a shared-cluster shape: many
+// labeled pins under one cluster are asked about in neither direction.
+func TestCreateSettingsRejectsHostAnotherAccountPinned(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("self-addressed onto another account's cluster", func(t *testing.T) {
+		f := newBootstrapFixture(t)
+		f.expectPermission("account2", "user2", modules.AgentNetworkSettings, operations.Create, true)
+		_, err := f.createSettings(ctx, "account2", "user2", "gw.example.com", "")
+		require.NoError(t, err, "account2's labeled pin beneath the host must go through first")
+
+		f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+		_, err = f.createSettings(ctx, "account1", "user1", "", "gw.example.com")
+		f.requireForeignClusterRefusal(t, err, "account1")
+	})
+
+	t.Run("labeled beneath another account's endpoint", func(t *testing.T) {
+		f := newBootstrapFixture(t)
+		f.expectPermission("account2", "user2", modules.AgentNetworkSettings, operations.Create, true)
+		_, err := f.createSettings(ctx, "account2", "user2", "", "gw.example.com")
+		require.NoError(t, err, "account2's self-addressed endpoint must go through first")
+
+		f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
+		_, err = f.createSettings(ctx, "account1", "user1", "gw.example.com", "")
+		f.requireForeignClusterRefusal(t, err, "account1")
+	})
+
+	t.Run("labeled beside another account's labeled pin stays allowed", func(t *testing.T) {
+		f := newBootstrapFixture(t)
+		for _, account := range []string{"account1", "account2"} {
+			f.expectPermission(account, "user", modules.AgentNetworkSettings, operations.Create, true)
+			_, err := f.createSettings(ctx, account, "user", "eu.proxy.netbird.io", "")
+			require.NoError(t, err, "labeled pins under one cluster are the shared-cluster shape and must not refuse each other")
+		}
+	})
+}
diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go
index d2d197a03..1e99251b2 100644
--- a/management/server/store/sql_store.go
+++ b/management/server/store/sql_store.go
@@ -6471,6 +6471,25 @@ func (s *SqlStore) IsClusterAddressConflicting(ctx context.Context, clusterAddre
 	return count > 0, nil
 }
 
+// HasForeignAccountProxyAtHost reports whether a proxy owned by a different
+// account declares this host. Shared proxies (account_id IS NULL) are not
+// foreign: a shared cluster is what most accounts pin their agent network
+// gateway to. The match folds case because proxies declare their address as
+// the operator spelled it while the caller's host is normalised; that costs a
+// scan of the proxies table, taken once per account when its gateway is
+// bootstrapped, not on the per-connect path IsClusterAddressConflicting serves.
+func (s *SqlStore) HasForeignAccountProxyAtHost(ctx context.Context, host, accountID string) (bool, error) {
+	var count int64
+	result := s.db.
+		Model(&proxy.Proxy{}).
+		Where("LOWER(cluster_address) = LOWER(?) AND account_id IS NOT NULL AND account_id != ?", host, accountID).
+		Count(&count)
+	if result.Error != nil {
+		return false, status.Errorf(status.Internal, "check proxy host ownership: %v", result.Error)
+	}
+	return count > 0, nil
+}
+
 func (s *SqlStore) DeleteAccountCluster(ctx context.Context, clusterAddress, accountID string) error {
 	result := s.db.
 		Where("cluster_address = ? AND account_id = ?", clusterAddress, accountID).
diff --git a/management/server/store/sql_store_agentnetwork.go b/management/server/store/sql_store_agentnetwork.go
index e75e36320..8a92f7147 100644
--- a/management/server/store/sql_store_agentnetwork.go
+++ b/management/server/store/sql_store_agentnetwork.go
@@ -315,6 +315,36 @@ func (s *SqlStore) GetAllAgentNetworkSettings(ctx context.Context, lockStrength
 	return settings, nil
 }
 
+// HasGatewayClusterPinnedByOtherAccount reports whether another account has a
+// labeled agent network gateway pinned beneath host, making host its cluster.
+// A self-addressed endpoint on the very same hostname is not counted: that
+// collision is the domain unique index's to refuse, as a conflict. Case-folded,
+// since a settings row written before hostnames were normalised may carry
+// capitals; one row per account, so the scan is cheap.
+func (s *SqlStore) HasGatewayClusterPinnedByOtherAccount(ctx context.Context, host, accountID string) (bool, error) {
+	return s.countGatewayRowsByOtherAccount(ctx, "LOWER(proxy_address) = LOWER(?) AND LOWER(domain) <> LOWER(proxy_address)", host, accountID)
+}
+
+// HasGatewayEndpointByOtherAccount reports whether host is another account's
+// agent network endpoint hostname (domain). Case-folded for the same reason as
+// HasGatewayClusterPinnedByOtherAccount.
+func (s *SqlStore) HasGatewayEndpointByOtherAccount(ctx context.Context, host, accountID string) (bool, error) {
+	return s.countGatewayRowsByOtherAccount(ctx, "LOWER(domain) = LOWER(?)", host, accountID)
+}
+
+func (s *SqlStore) countGatewayRowsByOtherAccount(ctx context.Context, predicate, host, accountID string) (bool, error) {
+	var count int64
+	result := s.db.
+		Model(&agentNetworkTypes.Settings{}).
+		Where(predicate+" AND account_id != ?", host, accountID).
+		Count(&count)
+	if result.Error != nil {
+		log.WithContext(ctx).Errorf("failed to check agent network gateway claims at host: %v", result.Error)
+		return false, status.Errorf(status.Internal, "check agent network gateway claims at host")
+	}
+	return count > 0, nil
+}
+
 // GetAgentNetworkSettingsByProxyAddress returns every Settings row whose
 // gateway is served by the proxy declaring the given cluster address. Used by
 // cluster-scoped synthesis to find the accounts a shared proxy serves.
diff --git a/management/server/store/store.go b/management/server/store/store.go
index 7a92de0a6..97da95b4c 100644
--- a/management/server/store/store.go
+++ b/management/server/store/store.go
@@ -342,6 +342,9 @@ type Store interface {
 	CountProxiesByAccountID(ctx context.Context, accountID string) (int64, error)
 	IsClusterAddressConflicting(ctx context.Context, clusterAddress, accountID string) (bool, error)
 	HasActiveProxyAtClusterAddress(ctx context.Context, clusterAddress string) (bool, error)
+	HasForeignAccountProxyAtHost(ctx context.Context, host, accountID string) (bool, error)
+	HasGatewayClusterPinnedByOtherAccount(ctx context.Context, host, accountID string) (bool, error)
+	HasGatewayEndpointByOtherAccount(ctx context.Context, host, accountID string) (bool, error)
 	DeleteAccountCluster(ctx context.Context, clusterAddress, accountID string) error
 
 	GetCustomDomainsCounts(ctx context.Context) (total int64, validated int64, err error)
diff --git a/management/server/store/store_mock.go b/management/server/store/store_mock.go
index 04f79d30a..399a07a19 100644
--- a/management/server/store/store_mock.go
+++ b/management/server/store/store_mock.go
@@ -3065,6 +3065,51 @@ func (mr *MockStoreMockRecorder) HasActiveProxyAtClusterAddress(ctx, clusterAddr
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasActiveProxyAtClusterAddress", reflect.TypeOf((*MockStore)(nil).HasActiveProxyAtClusterAddress), ctx, clusterAddress)
 }
 
+// HasForeignAccountProxyAtHost mocks base method.
+func (m *MockStore) HasForeignAccountProxyAtHost(ctx context.Context, host, accountID string) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "HasForeignAccountProxyAtHost", ctx, host, accountID)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// HasForeignAccountProxyAtHost indicates an expected call of HasForeignAccountProxyAtHost.
+func (mr *MockStoreMockRecorder) HasForeignAccountProxyAtHost(ctx, host, accountID any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasForeignAccountProxyAtHost", reflect.TypeOf((*MockStore)(nil).HasForeignAccountProxyAtHost), ctx, host, accountID)
+}
+
+// HasGatewayClusterPinnedByOtherAccount mocks base method.
+func (m *MockStore) HasGatewayClusterPinnedByOtherAccount(ctx context.Context, host, accountID string) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "HasGatewayClusterPinnedByOtherAccount", ctx, host, accountID)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// HasGatewayClusterPinnedByOtherAccount indicates an expected call of HasGatewayClusterPinnedByOtherAccount.
+func (mr *MockStoreMockRecorder) HasGatewayClusterPinnedByOtherAccount(ctx, host, accountID any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasGatewayClusterPinnedByOtherAccount", reflect.TypeOf((*MockStore)(nil).HasGatewayClusterPinnedByOtherAccount), ctx, host, accountID)
+}
+
+// HasGatewayEndpointByOtherAccount mocks base method.
+func (m *MockStore) HasGatewayEndpointByOtherAccount(ctx context.Context, host, accountID string) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "HasGatewayEndpointByOtherAccount", ctx, host, accountID)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// HasGatewayEndpointByOtherAccount indicates an expected call of HasGatewayEndpointByOtherAccount.
+func (mr *MockStoreMockRecorder) HasGatewayEndpointByOtherAccount(ctx, host, accountID any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasGatewayEndpointByOtherAccount", reflect.TypeOf((*MockStore)(nil).HasGatewayEndpointByOtherAccount), ctx, host, accountID)
+}
+
 // IncrementAgentNetworkConsumption mocks base method.
 func (m *MockStore) IncrementAgentNetworkConsumption(ctx context.Context, accountID string, kind types.ConsumptionDimension, dimID string, windowSeconds int64, windowStart time.Time, tokensIn, tokensOut int64, costUSD float64) error {
 	m.ctrl.T.Helper()

From 7ce6a63dcba9e67670b9e6e0cfbebc2e92027af6 Mon Sep 17 00:00:00 2001
From: Maycon Santos 
Date: Tue, 15 Sep 2026 08:42:21 +0200
Subject: [PATCH 4/9] [management] Validate the proxy cluster an agent network
 bootstraps onto (#7402)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The agent network gateway service is private: agents reach it over the WireGuard tunnel, authorised by peer identity, with the cluster as its only target. Only a reverse proxy cluster with private capabilities can serve that, reported per cluster as the `private` capability — the same `supports_private` flag the dashboard gates NetBird-only services on.

A bootstrap could pin an account to a cluster without private capabilities, leaving an immutable dead gateway. Both bootstrap paths now validate the picked cluster: one the account can see must have a connected proxy reporting the capability. Shared and account-owned clusters qualify alike. Known-ness comes from proxy rows, not heartbeat freshness, so a cluster without the capability stays refused while merely offline. A hostname no proxy has declared stays pinnable (address-first). Identity is compared case-insensitively over the account's cluster list.
---
 .../settings_cluster_validation_test.go       | 178 ++++++++++++++++++
 .../agentnetwork/handlers/handlers_test.go    |  29 +++
 .../internals/modules/agentnetwork/manager.go | 100 ++++++++++
 .../agentnetwork/settings_bootstrap_test.go   | 149 ++++++++++++++-
 .../agentnetwork_budgetrule_realstack_test.go |   1 +
 .../server/agentnetwork_realstack_test.go     |  21 +++
 6 files changed, 477 insertions(+), 1 deletion(-)
 create mode 100644 e2e/agentnetwork/settings_cluster_validation_test.go

diff --git a/e2e/agentnetwork/settings_cluster_validation_test.go b/e2e/agentnetwork/settings_cluster_validation_test.go
new file mode 100644
index 000000000..82c84dc74
--- /dev/null
+++ b/e2e/agentnetwork/settings_cluster_validation_test.go
@@ -0,0 +1,178 @@
+//go:build e2e
+
+package agentnetwork
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/e2e/harness"
+	"github.com/netbirdio/netbird/shared/management/http/api"
+)
+
+// TestSettingsBootstrapValidatesProxyCluster covers the bootstrap-time check
+// on the picked cluster, end to end against a real proxy.
+//
+// The synthesised gateway service is always private: agents reach it over the
+// WireGuard tunnel and are authorised by their peer identity. Only a cluster
+// with private capabilities can serve that, and management reports it per
+// cluster as the `private` capability — the same supports_private flag the
+// dashboard reads to decide which clusters it may offer. The endpoint assigned at bootstrap is immutable, so pinning
+// to a cluster that cannot serve it has to be refused up front rather than
+// leaving the account with a dead gateway.
+//
+// One combined server and one cluster address, walked through three states:
+// a live centralised proxy (refused), that proxy stopped so nothing in the
+// cluster is live any more (still refused — the record of what the cluster is
+// outlives its heartbeats), and finally a private-capable proxy (accepted, the
+// capability being any-true across the cluster's live proxies). Same account,
+// same address, so nothing but the cluster's state accounts for the different
+// answers.
+func TestSettingsBootstrapValidatesProxyCluster(t *testing.T) {
+	ctx := context.Background()
+
+	fresh, err := harnessStartFresh(ctx, t)
+	require.NoError(t, err, "start dedicated combined server")
+
+	proxyToken, err := fresh.CreateProxyTokenCLI(ctx, "e2e-cluster-validation")
+	require.NoError(t, err, "mint proxy token via CLI")
+
+	const cluster = harness.AgentNetworkCluster
+
+	// A centralised proxy: connected and serving the cluster, but without
+	// private capabilities, so it cannot serve a private service.
+	central, err := harness.StartProxy(ctx, fresh, proxyToken, map[string]string{
+		"NB_PROXY_PRIVATE": "false",
+	})
+	require.NoError(t, err, "start centralised proxy")
+	// Terminated mid-test; the cleanup only covers an early failure.
+	t.Cleanup(func() { _ = central.Terminate(context.Background()) })
+
+	waitClusterPrivate(ctx, t, fresh, cluster, false)
+
+	_, err = fresh.CreateSettings(ctx, api.AgentNetworkSettingsCreateRequest{
+		ProxyAddress: ptr(cluster),
+	})
+	require.Error(t, err, "bootstrap onto a cluster without private capabilities must be refused")
+	requireClientError(t, err)
+	assert.Contains(t, err.Error(), "private capabilities",
+		"the refusal must name what the cluster is missing: %v", err)
+
+	after, err := fresh.GetSettings(ctx)
+	require.NoError(t, err, "settings must still read after a refused bootstrap")
+	assert.Empty(t, after.Endpoint, "a refused bootstrap must not assign an endpoint")
+	assert.Empty(t, after.ProxyAddress, "a refused bootstrap must not pin a cluster")
+
+	// Stopping the centralised proxy must not turn the refusal into an
+	// acceptance: the cluster's proxy rows outlive their heartbeats (only the
+	// hourly stale reaper removes them), so the cluster is still on record as
+	// one that cannot serve the gateway. Judging on liveness instead would
+	// make "wait for the proxy to go quiet" a way to pin the account's
+	// immutable endpoint to a cluster that can never serve it.
+	require.NoError(t, central.Terminate(ctx), "stop the centralised proxy")
+	waitClusterAbsent(ctx, t, fresh, cluster)
+
+	_, err = fresh.CreateSettings(ctx, api.AgentNetworkSettingsCreateRequest{
+		ProxyAddress: ptr(cluster),
+	})
+	require.Error(t, err, "an offline cluster without private capabilities on record must stay refused")
+	requireClientError(t, err)
+
+	// Add a private-capable proxy to the same cluster: now it can serve a private
+	// service, and the very same request must go through.
+	privateProxy, err := harness.StartProxy(ctx, fresh, proxyToken)
+	require.NoError(t, err, "start private-capable proxy")
+	t.Cleanup(func() { _ = privateProxy.Terminate(context.Background()) })
+
+	waitClusterPrivate(ctx, t, fresh, cluster, true)
+
+	bootstrapped, err := fresh.CreateSettings(ctx, api.AgentNetworkSettingsCreateRequest{
+		ProxyAddress: ptr(cluster),
+	})
+	require.NoError(t, err, "bootstrap onto a private-capable cluster must succeed")
+	assert.Equal(t, cluster, bootstrapped.ProxyAddress, "the pinned cluster is the requested one")
+	assert.True(t, strings.HasSuffix(bootstrapped.Endpoint, "."+cluster),
+		"the endpoint must hang one label beneath the cluster: %s", bootstrapped.Endpoint)
+}
+
+// waitClusterPrivate polls the domains endpoint — the list the dashboard picks
+// its bootstrap cluster from — until the free domain for clusterAddr reports
+// supports_private == want. A proxy's capabilities land when it registers, so
+// this is the barrier between starting a proxy and asserting on what
+// management thinks its cluster can do.
+func waitClusterPrivate(ctx context.Context, t *testing.T, c *harness.Combined, clusterAddr string, want bool) {
+	t.Helper()
+
+	deadline := time.Now().Add(90 * time.Second)
+	var last string
+	for time.Now().Before(deadline) {
+		domains, err := c.API().ReverseProxyDomains.List(ctx)
+		if err != nil {
+			last = "list domains: " + err.Error()
+		} else {
+			last = "cluster not listed"
+			for _, d := range domains {
+				if d.Domain != clusterAddr {
+					continue
+				}
+				if d.SupportsPrivate == nil {
+					last = "supports_private not reported yet"
+					break
+				}
+				if *d.SupportsPrivate == want {
+					return
+				}
+				last = "supports_private is not the expected value"
+				break
+			}
+		}
+		if !waitBeforeRetry(ctx, 2*time.Second) {
+			break
+		}
+	}
+	t.Fatalf("cluster %s never reported supports_private=%v: %s", clusterAddr, want, last)
+}
+
+// waitClusterAbsent polls the domains endpoint until clusterAddr is no longer
+// offered, i.e. management sees no live proxy in it. The free-domain list is
+// built from the active clusters, so this is how a proxy going away becomes
+// observable — while the cluster's rows, and so its capability record, remain.
+//
+// The budget has to clear the active window, not just the disconnect: a proxy
+// that closes its stream cleanly is marked disconnected at once, but one that
+// dies without that is only dropped when its last heartbeat ages past
+// proxyActiveThreshold (2 minutes), so a 90s deadline could fail the test on
+// the slow path alone.
+func waitClusterAbsent(ctx context.Context, t *testing.T, c *harness.Combined, clusterAddr string) {
+	t.Helper()
+
+	deadline := time.Now().Add(3 * time.Minute)
+	var last string
+	for time.Now().Before(deadline) {
+		domains, err := c.API().ReverseProxyDomains.List(ctx)
+		if err != nil {
+			last = "list domains: " + err.Error()
+		} else {
+			listed := false
+			for _, d := range domains {
+				if d.Domain == clusterAddr {
+					listed = true
+					break
+				}
+			}
+			if !listed {
+				return
+			}
+			last = "cluster still listed as active"
+		}
+		if !waitBeforeRetry(ctx, 2*time.Second) {
+			break
+		}
+	}
+	t.Fatalf("cluster %s never dropped out of the active list: %s", clusterAddr, last)
+}
diff --git a/management/internals/modules/agentnetwork/handlers/handlers_test.go b/management/internals/modules/agentnetwork/handlers/handlers_test.go
index 6d1be3562..2b2f3b8a3 100644
--- a/management/internals/modules/agentnetwork/handlers/handlers_test.go
+++ b/management/internals/modules/agentnetwork/handlers/handlers_test.go
@@ -9,6 +9,7 @@ import (
 	"runtime"
 	"strings"
 	"testing"
+	"time"
 
 	"go.uber.org/mock/gomock"
 	"github.com/gorilla/mux"
@@ -17,6 +18,7 @@ import (
 
 	"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
 	agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
+	rpproxy "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
 	"github.com/netbirdio/netbird/management/server/account"
 	nbcontext "github.com/netbirdio/netbird/management/server/context"
 	"github.com/netbirdio/netbird/management/server/permissions"
@@ -29,6 +31,9 @@ import (
 const (
 	testAccountID = "acc-1"
 	testUserID    = "user-bob"
+	// testClusterAddress is the shared proxy cluster the settings tests pin
+	// their gateway to; the fixture seeds a connected private-capable proxy for it.
+	testClusterAddress = "eu.proxy.netbird.io"
 )
 
 // agentNetworkHandlerFixture builds a real agentnetwork.Manager with
@@ -75,6 +80,12 @@ func newAgentNetworkHandlerFixture(t *testing.T) *agentNetworkHandlerFixture {
 	manager := agentnetwork.NewManager(st, perms, accounts, nil)
 	h := &handler{manager: manager}
 
+	// The labeled bootstrap validates its proxy_address against the live
+	// clusters, so seed the shared cluster these tests pin to as a real,
+	// private-capable one — the wire-shape assertions then run through the
+	// validated path rather than the "nothing connected yet" carve-out.
+	seedSharedPrivateCluster(t, st, testClusterAddress)
+
 	router := mux.NewRouter()
 	router.HandleFunc("/agent-network/providers", h.createProvider).Methods("POST")
 	router.HandleFunc("/agent-network/providers/{providerId}", h.getProvider).Methods("GET")
@@ -268,3 +279,21 @@ func TestConsumptionHandler_PopulatedAccountListsRows(t *testing.T) {
 	assert.Equal(t, groupRow.WindowStartUtc, userRow.WindowStartUtc,
 		"rows recorded in the same window must share the aligned window_start_utc")
 }
+
+// seedSharedPrivateCluster registers a connected, NetBird-operated proxy
+// with private capabilities (the `private` capability) so
+// clusterAddr is a cluster any account may pin its agent-network gateway to.
+func seedSharedPrivateCluster(t *testing.T, st store.Store, clusterAddr string) {
+	t.Helper()
+	private := true
+	now := time.Now().UTC()
+	require.NoError(t, st.SaveProxy(context.Background(), &rpproxy.Proxy{
+		ID:             "shared-proxy-" + clusterAddr,
+		SessionID:      "shared-session",
+		ClusterAddress: clusterAddr,
+		LastSeen:       now,
+		ConnectedAt:    &now,
+		Status:         rpproxy.StatusConnected,
+		Capabilities:   rpproxy.Capabilities{Private: &private},
+	}), "seeding the shared proxy cluster must succeed")
+}
diff --git a/management/internals/modules/agentnetwork/manager.go b/management/internals/modules/agentnetwork/manager.go
index d4524a218..d1a5ebd7b 100644
--- a/management/internals/modules/agentnetwork/manager.go
+++ b/management/internals/modules/agentnetwork/manager.go
@@ -1045,6 +1045,9 @@ func (m *managerImpl) bootstrapSelfAddressed(ctx context.Context, settings *type
 	if err := m.requireNotClaimedByOtherAccount(ctx, settings.AccountID, hostname, m.store.HasGatewayClusterPinnedByOtherAccount); err != nil {
 		return err
 	}
+	if err := m.validateGatewayCluster(ctx, settings.AccountID, hostname); err != nil {
+		return err
+	}
 
 	settings.Domain = hostname
 	settings.ProxyAddress = hostname
@@ -1063,6 +1066,99 @@ func (m *managerImpl) bootstrapSelfAddressed(ctx context.Context, settings *type
 	return nil
 }
 
+// validateGatewayCluster rejects a bootstrap pinned to a cluster that cannot
+// serve the account's gateway — a labeled endpoint beneath the cluster and a
+// self-addressed one on the very address a proxy declares alike, since the
+// service behind either is the same private one.
+//
+// The synthesised gateway service is unconditionally private
+// (buildAccountService): agents reach it over the WireGuard tunnel and are
+// authorised by ValidateTunnelPeer against the policies' source groups, and
+// its single target is the cluster itself with DirectUpstream. Only a cluster
+// with private capabilities can serve that. Management reports it per cluster
+// as the `private` capability, the same flag the dashboard renders as
+// supports_private when it gates NetBird-only services.
+//
+// Without this check the bootstrap happily pins to any cluster the caller
+// names, including one without private capabilities — and the endpoint it
+// allocates is immutable, so the account is left with a dead gateway that only
+// a DeleteSettings/re-bootstrap can undo.
+//
+// Whether management knows the cluster is decided on the proxy rows
+// themselves, never on how fresh their heartbeats are: a cluster's rows
+// outlive its proxies' liveness (only the stale-proxy reaper removes them), so
+// a cluster that exists stays judged as one. Judging on liveness instead would
+// make the same centralised cluster pass or fail depending on whether its
+// proxies happened to have heartbeated in the last couple of minutes.
+//
+// The single opening left is a cluster management holds no proxy row for at
+// all: pinning ahead of a proxy's first connection is a legitimate order — the
+// dedicated path claims an address the same way, before any proxy declares it.
+func (m *managerImpl) validateGatewayCluster(ctx context.Context, accountID, clusterAddr string) error {
+	declared, err := m.accountClusterSpellings(ctx, accountID, clusterAddr)
+	if err != nil {
+		return err
+	}
+	if len(declared) == 0 {
+		// No proxy has ever declared this address: an address-first pin.
+		return nil
+	}
+
+	// A cluster management knows has to prove it can serve the gateway, and
+	// only a live proxy reporting the capability proves that. Both an explicit false and an
+	// unreported capability (nothing live in the cluster, or proxies predating
+	// capability reporting) fail here: unusable and unproven are the same
+	// answer for a decision that cannot be revisited later.
+	//
+	// The capability is read per declared spelling and taken as any-true, the
+	// same way it aggregates over a cluster's proxies: the store matches
+	// cluster_address exactly, so a host two proxies spelled differently must
+	// not come back unproven just because it was asked about under one of them.
+	for _, address := range declared {
+		if private := m.store.GetClusterSupportsPrivate(ctx, address); private != nil && *private {
+			return nil
+		}
+	}
+
+	return status.Errorf(status.InvalidArgument,
+		"proxy cluster %s has no private capabilities: the agent network gateway requires a reverse proxy cluster "+
+			"with private capabilities", clusterAddr)
+}
+
+// accountClusterSpellings returns every proxy cluster address in the account's
+// view — its own (BYOP) clusters plus the shared ones — that names the same
+// host as clusterAddr. Empty means management holds no proxy row for that host
+// in this account's view.
+//
+// A proxy declares its cluster address as the operator spelled it, so identity
+// is compared on the normalised form rather than byte-equal — an in-memory pass
+// over the account's clusters, not a query. What comes back is the stored
+// spelling, because the capability lookup matches cluster_address exactly and
+// would silently find nothing under a spelling the store never held. The
+// cluster listing is not gated on heartbeats, so this answer does not change
+// while a cluster's proxies are merely offline.
+func (m *managerImpl) accountClusterSpellings(ctx context.Context, accountID, clusterAddr string) ([]string, error) {
+	clusters, err := m.store.GetProxyClusters(ctx, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("list proxy clusters: %w", err)
+	}
+
+	var spellings []string
+	for _, cluster := range clusters {
+		normalized, err := types.NormalizeHostname(cluster.Address)
+		if err != nil {
+			// An address declared in a shape we cannot normalise is not one an
+			// endpoint can be allocated beneath.
+			log.WithContext(ctx).Debugf("skipping unusable proxy cluster address %q: %s", cluster.Address, err)
+			continue
+		}
+		if normalized == clusterAddr {
+			spellings = append(spellings, cluster.Address)
+		}
+	}
+	return spellings, nil
+}
+
 // bootstrapLabeled allocates a labeled endpoint one label beneath the given
 // cluster address: Domain =