Merge branch 'main' into poc/certificate-posture

This commit is contained in:
pascal
2026-09-16 23:31:07 +02:00
29 changed files with 3033 additions and 79 deletions
+10
View File
@@ -14,5 +14,15 @@ reviews:
- "!**/*.ts"
- "!**/*.js"
- "!**/*.svg"
pre_merge_checks:
custom_checks:
- name: "No attribution trailers"
mode: error
instructions: >-
Fail when the PR description or any commit message carries an
attribution trailer or footer: Co-Authored-By, Claude-Session,
Generated-By, or a "Generated with"/"Generated by" tool line.
Contributors own their contributions (AGENTS.md); ask for the
lines to be removed.
chat:
auto_reply: true
+26
View File
@@ -0,0 +1,26 @@
#!/bin/bash
# Refuses commit messages that carry attribution trailers. Contributors own
# their contributions (AGENTS.md, "No Co-Authored-By or tool-attribution
# trailers"); a trailer spreads that ownership onto a tool or a bystander.
msg_file="$1"
# Trailer keys in any casing, with any bullet or emoji in front.
trailers='^[^[:alnum:]]*(co-authored-by|claude-session|generated-by):'
# "Generated with/by" footers, including "Generated with <emoji> by".
footer='^[^[:alnum:]]*generated (with|by)( [^[:alnum:]]*by)? '
# A footer names a product, so a capitalized word must follow the phrase
# itself. Prose such as "generated by the protobuf compiler" stays legal.
tool='[Gg][Ee][Nn][Ee][Rr][Aa][Tt][Ee][Dd] ([Ww][Ii][Tt][Hh]|[Bb][Yy])( [^[:alnum:]]*[Bb][Yy])? [^[:alnum:]]*[A-Z]'
offending=$( {
grep -Ein "$trailers" "$msg_file"
grep -Ein "$footer" "$msg_file" | grep -E "$tool"
} | sort -un )
if [ -n "$offending" ]; then
echo "commit-msg: attribution trailers are not accepted in this repository:" >&2
printf '%s\n' "$offending" | sed 's/^/ /' >&2
echo "Remove them and commit again (see AGENTS.md)." >&2
exit 1
fi
+36 -3
View File
@@ -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'
+1 -1
View File
@@ -77,7 +77,7 @@ make lint # golangci-lint on files changed vs origin/main (also the p
make lint-all # full-repository lint, matches CI
make test-unit # host-safe unit tests, -tags devcert, no sudo
make test-privileged # privileged-tagged suite in a Docker container with NET_ADMIN
make setup-hooks # wire make lint into .githooks/pre-push
make setup-hooks # wire .githooks: pre-push runs make lint, commit-msg refuses attribution trailers
# Narrow runs
go test ./client/internal/dns/...
+4 -1
View File
@@ -1 +1,4 @@
See [AGENTS.md](AGENTS.md) for the agent guidelines in this repository.
The agent guidelines live in [AGENTS.md](AGENTS.md). It is imported here so
every session loads it in full rather than following a pointer.
@AGENTS.md
+2 -2
View File
@@ -23,8 +23,8 @@ lint-install: $(GOLANGCI_LINT)
# Setup git hooks for all developers
setup-hooks:
@git config core.hooksPath .githooks
@chmod +x .githooks/pre-push
@echo "✅ Git hooks configured! Pre-push will now run 'make lint'"
@chmod +x .githooks/pre-push .githooks/commit-msg
@echo "✅ Git hooks configured! Pre-push runs 'make lint'; commit-msg refuses attribution trailers"
# Host-safe unit tests: excludes the privileged-tagged tests (root / system-mutating).
# Runs as a normal user with no sudo and leaves host networking untouched.
+2 -2
View File
@@ -819,8 +819,8 @@ func (d *Status) SetSessionExpiresAt(deadline time.Time) {
// "none" would blank the UI at the exact moment it should say the session
// ended.
func (d *Status) GetSessionExpiresAt() time.Time {
d.mux.Lock()
defer d.mux.Unlock()
d.mux.RLock()
defer d.mux.RUnlock()
return d.sessionExpiresAt
}
+12 -4
View File
@@ -9,6 +9,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/e2e/harness"
"github.com/netbirdio/netbird/shared/management/http/api"
)
@@ -65,16 +66,24 @@ func configProvider(cfg api.AgentNetworkAgentConfig, name string) *api.AgentNetw
func TestAgentConfigAllowlistOfDeclaredModels(t *testing.T) {
ctx := context.Background()
// Saving a provider makes management verify the credential against the
// upstream, and a real vendor refuses the dummy key and the save with it.
// The providers point at the mock upstream instead: it resolves to a
// private address, which the check declines to dial and treats as
// unverifiable rather than as a failure, so the save goes through. The
// test is about the allowlist, not the upstream.
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
cases := []struct {
name string
catalogID string
upstream string
declared string
}{
{
name: "plain-declared-id",
catalogID: "openai_api",
upstream: "https://api.openai.com",
declared: "gpt-4o-mini",
},
{
@@ -83,7 +92,6 @@ func TestAgentConfigAllowlistOfDeclaredModels(t *testing.T) {
// picker copies it as-is.
name: "bedrock-declared-id",
catalogID: "bedrock_api",
upstream: "https://bedrock-runtime.eu-central-1.amazonaws.com",
declared: "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
},
}
@@ -101,7 +109,7 @@ func TestAgentConfigAllowlistOfDeclaredModels(t *testing.T) {
prov, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: providerName,
ProviderId: tc.catalogID,
UpstreamUrl: tc.upstream,
UpstreamUrl: vllm.URL,
ApiKey: ptr("sk-dummy-e2e-key"),
Enabled: ptr(true),
Models: &[]api.AgentNetworkProviderModel{{Id: tc.declared, InputPer1k: 0.001, OutputPer1k: 0.002}},
@@ -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)
}
@@ -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")
}
@@ -1036,6 +1036,18 @@ 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
}
if err := m.validateGatewayCluster(ctx, settings.AccountID, hostname); err != nil {
return err
}
settings.Domain = hostname
settings.ProxyAddress = hostname
@@ -1054,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 = <label>.<proxyAddress>, served by whichever proxy
// declares the parent. Labels are adjective-noun tuples; a candidate is
@@ -1065,6 +1170,20 @@ 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
}
if err := m.validateGatewayCluster(ctx, settings.AccountID, parent); err != nil {
return err
}
for attempt := 1; attempt <= maxDomainAllocationAttempts; attempt++ {
label := labelgen.PickTuple()
@@ -1111,6 +1230,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.
@@ -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,57 @@ 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
// proxy with private capabilities 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")
}
// seedPrivateCluster is the common case: a shared cluster with a connected
// proxy that has private capabilities, which is what a bootstrap requires.
func (f *bootstrapFixture) seedPrivateCluster(t *testing.T, clusterAddr string) {
t.Helper()
f.seedProxy(t, "proxy-"+clusterAddr, "", clusterAddr, ptrTo(true))
}
// 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.
@@ -94,6 +147,7 @@ func TestCreateSettingsRequiresPermission(t *testing.T) {
func TestCreateSettingsLabeled(t *testing.T) {
ctx := context.Background()
f := newBootstrapFixture(t)
f.seedPrivateCluster(t, "cluster1.example.com")
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
created, err := f.createSettings(ctx, "account1", "user1", "Cluster1.Example.com", "")
@@ -167,6 +221,7 @@ func TestCreateSettingsIdentityFieldValidation(t *testing.T) {
func TestCreateSettingsConflictsOnSecondBootstrap(t *testing.T) {
ctx := context.Background()
f := newBootstrapFixture(t)
f.seedPrivateCluster(t, "cluster1.example.com")
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
first, err := f.createSettings(ctx, "account1", "user1", "cluster1.example.com", "")
@@ -230,3 +285,279 @@ 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")
}
// TestCreateSettingsRejectsOfflineCluster is the guard against deciding on
// heartbeat freshness. A centralised cluster is refused while its proxies are
// live; the same cluster must stay refused once they stop heartbeating, which
// takes only a couple of minutes (proxyActiveThreshold). Judging on liveness
// would turn "wait for the proxy to go quiet" into a way to pin the account's
// immutable endpoint to a cluster that can never serve it.
func TestCreateSettingsRejectsOfflineCluster(t *testing.T) {
ctx := context.Background()
notPrivate := false
cases := map[string]*bool{
"centralised proxy gone quiet": &notPrivate,
// A cluster that could serve the gateway still has to have something
// live in it to prove so at bootstrap: refusing is the safe direction
// (reconnect the proxy and retry) where accepting is permanent.
"private proxy gone quiet": ptrTo(true),
}
for name, private := range cases {
t.Run(name, func(t *testing.T) {
f := newBootstrapFixture(t)
f.seedProxyAt(t, "proxy1", "", "offline.example.com", private,
time.Now().UTC().Add(-time.Hour))
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
_, err := f.createSettings(ctx, "account1", "user1", "offline.example.com", "")
require.Error(t, err, "a known cluster with nothing live in it must be rejected")
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(), "private capabilities",
"the error must say private capabilities are what is missing")
_, err = f.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, "account1")
assert.Error(t, err, "no row may be left behind by a rejected bootstrap")
})
}
}
// TestCreateSettingsRequiresPrivateCluster pins the capability gate: the
// synthesised gateway service is always private, so a live cluster whose
// proxies lack private capabilities cannot serve it and must not
// become the account's immutable endpoint.
func TestCreateSettingsRequiresPrivateCluster(t *testing.T) {
ctx := context.Background()
f := newBootstrapFixture(t)
notPrivate := false
f.seedProxy(t, "proxy1", "", "central.example.com", &notPrivate)
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
_, err := f.createSettings(ctx, "account1", "user1", "central.example.com", "")
require.Error(t, err, "a cluster without private capabilities must be rejected")
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(), "private capabilities", "the error must name what the cluster is missing")
_, err = f.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, "account1")
assert.Error(t, err, "no row may be left behind by a rejected bootstrap")
}
// TestCreateSettingsAcceptsOwnPrivateCluster pins the BYOP happy path: the
// account's own cluster with a connected private-capable proxy is a valid pin.
func TestCreateSettingsAcceptsOwnPrivateCluster(t *testing.T) {
ctx := context.Background()
f := newBootstrapFixture(t)
f.seedProxy(t, "proxy1", "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 private cluster must be accepted")
assert.Equal(t, "byop.account1.example.com", created.ProxyAddress)
}
// TestCreateSettingsMatchesClusterCasing pins that a cluster spelled with
// capitals in the store is still recognised as the same cluster the normalised
// proxy_address names, in both directions: a private cluster is accepted and a
// centralised one is refused, whatever the casing. The comparison is in memory
// over the account's cluster list; the capability lookup is still asked under
// the spelling the store actually holds, which is what an exact match needs.
func TestCreateSettingsMatchesClusterCasing(t *testing.T) {
ctx := context.Background()
t.Run("own private cluster is found", func(t *testing.T) {
f := newBootstrapFixture(t)
f.seedProxy(t, "proxy1", "", "EU.Proxy.Example.com", ptrTo(true))
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
created, err := f.createSettings(ctx, "account1", "user1", "eu.proxy.example.com", "")
require.NoError(t, err, "a private cluster declared with capitals must still be accepted")
assert.Equal(t, "eu.proxy.example.com", created.ProxyAddress)
})
t.Run("non-private cluster is still refused", func(t *testing.T) {
f := newBootstrapFixture(t)
f.seedProxy(t, "proxy1", "", "Central.Example.com", ptrTo(false))
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
_, err := f.createSettings(ctx, "account1", "user1", "central.example.com", "")
require.Error(t, err, "casing must not become a way past the capability check")
assert.Contains(t, err.Error(), "private capabilities")
})
}
// 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")
}
})
}
// TestCreateSettingsSelfAddressedRequiresPrivateCluster pins that the
// capability gate applies to a self-addressed endpoint too: the service behind
// it is the same private one, so a proxy that already declares the hostname
// must have private capabilities, whether the account's own or a shared cluster's. A
// hostname no proxy declares yet stays claimable (TestCreateSettingsSelfAddressed).
func TestCreateSettingsSelfAddressedRequiresPrivateCluster(t *testing.T) {
ctx := context.Background()
t.Run("centralised proxy at the hostname is refused", func(t *testing.T) {
f := newBootstrapFixture(t)
f.seedProxy(t, "central", "", "gw.example.com", ptrTo(false))
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
_, err := f.createSettings(ctx, "account1", "user1", "", "gw.example.com")
require.Error(t, err, "a self-addressed endpoint on a centralised proxy can never be served")
var sErr *status.Error
require.ErrorAs(t, err, &sErr)
assert.Equal(t, status.InvalidArgument, sErr.Type())
assert.Contains(t, err.Error(), "private capabilities")
_, err = f.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, "account1")
assert.Error(t, err, "no row may be left behind by a rejected bootstrap")
})
t.Run("private proxy at the hostname is accepted", func(t *testing.T) {
f := newBootstrapFixture(t)
f.seedProxy(t, "private", "", "gw.example.com", ptrTo(true))
f.expectPermission("account1", "user1", modules.AgentNetworkSettings, operations.Create, true)
created, err := f.createSettings(ctx, "account1", "user1", "", "gw.example.com")
require.NoError(t, err)
assert.Equal(t, "gw.example.com", created.ProxyAddress)
})
}
+7 -3
View File
@@ -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
@@ -89,6 +89,7 @@ func TestAgentNetwork_UpdateSettings_PreservesImmutableAndTogglesCollection(t *t
// Bootstrap is an explicit settings create; providers have no settings
// side effects anymore.
seedPrivateProxyCluster(t, am.Store, clusterAddr)
before, err := mgr.CreateSettings(ctx, adminUserID, agenttypes.DefaultSettings(accountID), clusterAddr, "")
require.NoError(t, err, "CreateSettings must bootstrap the row")
require.Equal(t, clusterAddr, before.ProxyAddress, "proxy address pinned at bootstrap")
@@ -13,6 +13,7 @@ import (
networkmap "github.com/netbirdio/netbird/management/internals/controllers/network_map"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
agenttypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
rpproxy "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
nbpeer "github.com/netbirdio/netbird/management/server/peer"
"github.com/netbirdio/netbird/management/server/permissions"
"github.com/netbirdio/netbird/management/server/store"
@@ -92,6 +93,7 @@ func TestAgentNetwork_ProviderCRUD_FansOutToProxyAndClientPeers(t *testing.T) {
// UpdateAccountPeers, which is the path under test.
agentMgr := agentnetwork.NewManager(am.Store, permissions.NewManager(am.Store), am, nil)
seedPrivateProxyCluster(t, am.Store, clusterAddr)
_, err = agentMgr.CreateSettings(ctx, adminUserID, agenttypes.DefaultSettings(accountID), clusterAddr, "")
require.NoError(t, err, "CreateSettings must bootstrap the endpoint")
// The bootstrap itself reconciles and queues updates on both channels;
@@ -222,3 +224,22 @@ func synthZoneRData(sync *nbproto.SyncResponse, clusterAddr, fqdn string) string
}
return ""
}
// seedPrivateProxyCluster registers a connected proxy with private capabilities in a
// netbird client for clusterAddr, matching what a real deployment looks like
// when the account bootstraps: the agent-network gateway service is always
// private, so its cluster has to be one that can serve private services.
func seedPrivateProxyCluster(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: "agent-net-proxy-" + clusterAddr,
SessionID: "agent-net-session",
ClusterAddress: clusterAddr,
LastSeen: now,
ConnectedAt: &now,
Status: rpproxy.StatusConnected,
Capabilities: rpproxy.Capabilities{Private: &private},
}), "seeding the proxy cluster must succeed")
}
+66 -22
View File
@@ -3157,7 +3157,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 {
@@ -3216,23 +3221,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
}
@@ -3244,11 +3246,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
@@ -3282,22 +3287,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
}
@@ -6449,6 +6474,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).
@@ -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.
+242 -14
View File
@@ -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"
@@ -340,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)
@@ -732,6 +737,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 +764,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 +788,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 +817,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 +866,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 +880,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 +1099,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 +1174,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<pre>[:/@])(?P<dbname>[^/?]+)(?P<post>\?|$)`)
return re.ReplaceAllString(dsn, `${pre}`+newDBName+`${post}`)
+45
View File
@@ -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()
+12
View File
@@ -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),
+4 -5
View File
@@ -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),
}
}
+60 -4
View File
@@ -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
+4 -6
View File
@@ -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{}),
+108
View File
@@ -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 == "" {
+455
View File
@@ -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
}
+587
View File
@@ -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}
}
+23 -11
View File
@@ -17,6 +17,9 @@ ARCH="$(uname -m)"
PACKAGE_MANAGER="bin"
INSTALL_DIR=""
SUDO=""
# curl protocol set for --proto / --proto-redir: https and nothing else, so no
# request and no redirect in a chain can fall back to plaintext.
PROTO_HTTPS="=https"
if command -v sudo > /dev/null && [ "$(id -u)" -ne 0 ]; then
@@ -25,6 +28,15 @@ elif command -v doas > /dev/null && [ "$(id -u)" -ne 0 ]; then
SUDO="doas"
fi
# Downloads are staged in a private directory instead of /tmp. Fixed names in a
# shared directory can collide with entries created there beforehand, and the
# paths staged here are consumed by the privileged install steps below.
NB_TMPDIR="$(mktemp -d "${TMPDIR:-/tmp}/netbird.XXXXXXXXXX")" || {
echo "Unable to create a temporary directory for the downloads"
exit 1
}
trap 'rm -rf "$NB_TMPDIR"' EXIT
if [ -z ${NETBIRD_RELEASE+x} ]; then
NETBIRD_RELEASE=latest
fi
@@ -73,10 +85,11 @@ download_release_binary() {
DOWNLOAD_URL="${BASE_URL}/${VERSION}/${BINARY_NAME}"
echo "Installing $1 from $DOWNLOAD_URL"
ARCHIVE_PATH="${NB_TMPDIR}/${BINARY_NAME}"
if [ -n "$GITHUB_TOKEN" ]; then
cd /tmp && curl -H "Authorization: token ${GITHUB_TOKEN}" -LO "$DOWNLOAD_URL"
curl -H "Authorization: token ${GITHUB_TOKEN}" -L --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" -o "$ARCHIVE_PATH" "$DOWNLOAD_URL"
else
cd /tmp && curl -LO "$DOWNLOAD_URL" || curl -LO --dns-servers 8.8.8.8 "$DOWNLOAD_URL"
curl -L --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" -o "$ARCHIVE_PATH" "$DOWNLOAD_URL" || curl -L --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" -o "$ARCHIVE_PATH" --dns-servers 8.8.8.8 "$DOWNLOAD_URL"
fi
@@ -89,12 +102,12 @@ download_release_binary() {
fi
# Unzip the app and move to INSTALL_DIR
unzip -q -o "$BINARY_NAME"
mv -v "netbird_ui_${OS_TYPE}/" "$INSTALL_DIR/" || mv -v "netbird_ui_${OS_TYPE}_${ARCH}/" "$INSTALL_DIR/"
unzip -q -o "$ARCHIVE_PATH" -d "$NB_TMPDIR"
mv -v "${NB_TMPDIR}/netbird_ui_${OS_TYPE}/" "$INSTALL_DIR/" || mv -v "${NB_TMPDIR}/netbird_ui_${OS_TYPE}_${ARCH}/" "$INSTALL_DIR/"
else
${SUDO} mkdir -p "$INSTALL_DIR"
tar -xzvf "$BINARY_NAME"
${SUDO} mv "${1%_"${BINARY_BASE_NAME}"}" "$INSTALL_DIR/"
tar -xzvf "$ARCHIVE_PATH" -C "$NB_TMPDIR"
${SUDO} mv "${NB_TMPDIR}/${1%_"${BINARY_BASE_NAME}"}" "$INSTALL_DIR/"
fi
}
@@ -110,7 +123,7 @@ add_apt_repo() {
/usr/share/keyrings/netbird-archive-keyring.gpg \
/usr/share/keyrings/wiretrustee-archive-keyring.gpg
curl -sSL https://pkgs.netbird.io/debian/public.key \
curl -sSL --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" https://pkgs.netbird.io/debian/public.key \
| ${SUDO} gpg --dearmor -o /usr/share/keyrings/netbird-archive-keyring.gpg
# Explicitly set the file permission
@@ -183,11 +196,10 @@ install_pkg() {
*) echo "Unsupported macOS arch: $(uname -m)" >&2; exit 1 ;;
esac
PKG_URL=$(curl -sIL -o /dev/null -w '%{url_effective}' "https://pkgs.netbird.io/macos/${ARCH}")
PKG_URL=$(curl -sIL --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" -o /dev/null -w '%{url_effective}' "https://pkgs.netbird.io/macos/${ARCH}")
echo "Downloading NetBird macOS installer from https://pkgs.netbird.io/macos/${ARCH}"
curl -fsSL -o /tmp/netbird.pkg "${PKG_URL}"
${SUDO} installer -pkg /tmp/netbird.pkg -target /
rm -f /tmp/netbird.pkg
curl -fsSL --proto "$PROTO_HTTPS" --proto-redir "$PROTO_HTTPS" -o "${NB_TMPDIR}/netbird.pkg" "${PKG_URL}"
${SUDO} installer -pkg "${NB_TMPDIR}/netbird.pkg" -target /
}
check_use_bin_variable() {
+402
View File
@@ -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)
}
+180
View File
@@ -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)
}
}