From 50fbb3d6efc2d42e574e9a8b208d07fb08067382 Mon Sep 17 00:00:00 2001 From: Ryanmello07 <67509637+Ryanmello07@users.noreply.github.com> Date: Wed, 12 Aug 2026 05:41:55 +0100 Subject: [PATCH 1/5] feat(blackhole): hourly fleet-wide check that any traffic gets through The full pass spends minutes per provider -- ~131 destinations, geolocation consensus, optionally bandwidth -- so it sweeps a fleet over hours to days. In that window a provider that silently stops forwarding keeps its last passing measurement, and every consumer of that measurement keeps believing it. On beta that meant 98.6% of advertised providers were judged on evidence over six hours old, and twelve sampled from the stalest cohort answered ok=0/131 while still being handed to clients. Nothing about being dark looks different from outside: the provider stays connected and goes on accepting clients. Only asking it to carry something finds out. So this is the cheapest question that still means something -- did ANY traffic get through -- asked of every provider hourly, on its own loop. - egresshealth.Blackhole draws up to 3 connectivity destinations and passes on the FIRST success. Three, not one, so a single destination having a bad minute cannot condemn a provider; ANY, not all, because a provider reaching some destinations is degraded rather than dark, and degradation is Check's department. It reuses fetch, so the table's headers, body caps and Verify contracts all apply -- a captive portal answering 200 with its own body fails here exactly as it fails a full run, which is what makes "something got through" mean anything. - The sample is drawn fresh per run for the same anti-gaming reason the full check samples: a provider that knew the three addresses could carry those and blackhole everything else. - A tunnel that cannot be opened counts as dark, under its own failure class. From a client's point of view a provider it cannot build a circuit through is exactly as useless as one that carries nothing, and not advertising those is the whole point. - BlackholeHosts feeds the startup confinement self-check. If the prober could reach these directly, a provider carrying nothing would still be recorded ok -- the check would confirm the prober's own connectivity, remove nothing forever, and look healthy doing it. - The sweeper stops on ErrBlackholeUnsupported and ErrUnauthorized, and only those. An old server will never grow the endpoint mid-run, and a rejected secret is a broken deployment; everything else is retried next tick, so a momentary server outage cannot silently end blackhole detection for the life of the process. Results are submitted as one batch per pass: a sweep produces hundreds of one-bit answers and a request each would spend more on http than on the checks. Defaults: -blackhole-interval 1h, -blackhole-limit 500, -blackhole-concurrency 32. Concurrency is higher than the full pass because a check is one round trip through the tunnel rather than a 131-destination sweep. 0 disables the sweep. --- cmd/egress-prober/blackhole.go | 196 +++++++++++++++++++++++++++++++++ cmd/egress-prober/main.go | 17 ++- egresshealth/blackhole.go | 146 ++++++++++++++++++++++++ egresshealth/blackhole_test.go | 130 ++++++++++++++++++++++ ingest/blackhole.go | 166 ++++++++++++++++++++++++++++ 5 files changed, 654 insertions(+), 1 deletion(-) create mode 100644 cmd/egress-prober/blackhole.go create mode 100644 egresshealth/blackhole.go create mode 100644 egresshealth/blackhole_test.go create mode 100644 ingest/blackhole.go diff --git a/cmd/egress-prober/blackhole.go b/cmd/egress-prober/blackhole.go new file mode 100644 index 0000000..6690fcc --- /dev/null +++ b/cmd/egress-prober/blackhole.go @@ -0,0 +1,196 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "sync" + "time" + + "github.com/urnetwork/connect" + + "github.com/urnetwork/operator-proxy/egresshealth" + "github.com/urnetwork/operator-proxy/ingest" + "github.com/urnetwork/operator-proxy/providertunnel" +) + +// blackholeSweeper runs the cheap liveness check across the whole fleet on its +// own cadence, independently of the geolocation/health pass. +// +// The two must not share a loop. The full pass spends minutes per provider and +// sweeps the fleet over hours to days; in that window a provider that silently +// stops forwarding keeps its last passing measurement and stays in the public +// list. This exists to close that window, which only works if it runs on its +// own much shorter one. +type blackholeSweeper struct { + operator *ingest.Client + tunnelCfg providertunnel.Config + pins *pinSet + timeout time.Duration + concurrency int + limit int +} + +// blackholeResult carries one provider's outcome out of the worker pool. +type blackholeResult struct { + check ingest.BlackholeCheck + dark bool + tunnel bool + details string +} + +// sweep runs one pass: ask what is due, check each, report the batch. +// +// Returns the number checked, so the caller can tell "the fleet is covered" +// from "the queue handed us nothing", which look identical in a log line. +func (s *blackholeSweeper) sweep(ctx context.Context) (checked int, err error) { + clientIds, err := s.operator.BlackholeDue(ctx, s.limit) + if err != nil { + return 0, err + } + if len(clientIds) == 0 { + return 0, nil + } + + results := make([]blackholeResult, len(clientIds)) + + sem := make(chan struct{}, s.concurrency) + var wg sync.WaitGroup + for i, clientId := range clientIds { + if ctx.Err() != nil { + break + } + wg.Add(1) + go func(i int, clientId string) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + results[i] = s.checkOne(ctx, clientId) + }(i, clientId) + } + wg.Wait() + + checks := make([]ingest.BlackholeCheck, 0, len(results)) + var dark, tunnelFailed int + for _, r := range results { + if r.check.ClientId == "" { + // never ran: the pass was cancelled before this slot started + continue + } + checks = append(checks, r.check) + if r.dark { + dark++ + if r.tunnel { + tunnelFailed++ + } + log.Printf("blackhole: provider=%s DARK %s", r.check.ClientId, r.details) + } + } + + if len(checks) == 0 { + return 0, nil + } + if err := s.operator.SubmitBlackholeChecks(ctx, checks); err != nil { + // the whole batch is lost, not part of it -- the server validates before + // writing -- so say how much + return 0, fmt.Errorf("submitting %d checks: %w", len(checks), err) + } + + log.Printf("blackhole: pass checked=%d dark=%d (tunnel_failed=%d) ok=%d", + len(checks), dark, tunnelFailed, len(checks)-dark) + return len(checks), nil +} + +// checkOne opens a tunnel through one provider and asks whether anything gets +// through. +// +// A tunnel that cannot be opened counts as dark, and that is a deliberate +// choice rather than an oversight: from a client's point of view a provider it +// cannot establish a circuit through is exactly as useless as one that carries +// nothing, and the whole purpose of this signal is to stop advertising +// providers a client cannot use. It is recorded under its own failure class so +// the two remain distinguishable in the data. +func (s *blackholeSweeper) checkOne(ctx context.Context, clientId string) blackholeResult { + checkedAt := time.Now().UTC() + + id, err := connect.ParseId(clientId) + if err != nil { + return blackholeResult{ + check: ingest.BlackholeCheck{ClientId: clientId, OK: false, Failure: "bad_client_id", CheckedAt: checkedAt}, + dark: true, + details: err.Error(), + } + } + + cfg := s.tunnelCfg + cfg.Pins = s.pins.get() + t, err := providertunnel.Open(ctx, cfg, id) + if err != nil { + return blackholeResult{ + check: ingest.BlackholeCheck{ClientId: clientId, OK: false, Failure: "tunnel_failed", CheckedAt: checkedAt}, + dark: true, + tunnel: true, + details: err.Error(), + } + } + defer t.Close() + + // Only the blackhole destinations are allowed through this client. The full + // pass allows the whole egress-health table and the bandwidth targets; this + // check reaches three connectivity endpoints and nothing else, so the + // allowlist says exactly that. + client := t.HTTPClientForHosts(s.timeout, egresshealth.BlackholeHosts()) + + res := egresshealth.Blackhole(ctx, client, egresshealth.Options{PerRequestTimeout: s.timeout}) + + check := ingest.BlackholeCheck{ClientId: clientId, OK: res.OK, CheckedAt: checkedAt} + if !res.OK { + check.Failure = res.Failure + } + + details := "" + if !res.OK { + for _, r := range res.Results { + details += r.Name + "=" + r.Err + " " + } + } + return blackholeResult{check: check, dark: !res.OK, details: details} +} + +// run sweeps on the interval until the context ends. +// +// A pass that errors is logged and retried on the next tick rather than +// stopping the sweeper: the server being briefly unreachable must not silently +// end blackhole detection for the life of the process. ErrBlackholeUnsupported +// is the one exception -- an older server will never grow the endpoint mid-run, +// so it says so once and stops instead of logging the same 404 hourly. +func (s *blackholeSweeper) run(ctx context.Context, interval time.Duration) { + for { + start := time.Now() + checked, err := s.sweep(ctx) + switch { + case err == nil: + if checked == 0 { + log.Printf("blackhole: pass found nothing due") + } + case errors.Is(err, ingest.ErrBlackholeUnsupported): + log.Printf("blackhole: the server does not implement the blackhole endpoints; sweeping is disabled") + return + case errors.Is(err, ingest.ErrUnauthorized): + // same posture as the credential self-check: a rejected secret is a + // broken deployment, and retrying hourly would hide it behind a + // sweep that never records anything + log.Printf("blackhole: the server rejected the operator secret; sweeping is disabled. Fix -operator-secret and restart.") + return + default: + log.Printf("blackhole: pass failed after %s: %s", time.Since(start).Round(time.Second), err) + } + + select { + case <-ctx.Done(): + return + case <-time.After(interval): + } + } +} diff --git a/cmd/egress-prober/main.go b/cmd/egress-prober/main.go index 05b70ac..33fdc41 100644 --- a/cmd/egress-prober/main.go +++ b/cmd/egress-prober/main.go @@ -67,6 +67,9 @@ func main() { concurrency := flag.Int("concurrency", 4, "max simultaneous provider tunnels") cacheTTL := flag.Duration("cache-ttl", 24*time.Hour, "do not re-probe a provider within this window. Only applies to the enumeration fallback used against a server with no due endpoint; when the server supplies the due list it owns the schedule") interval := flag.Duration("interval", time.Hour, "sleep between passes; 0 runs a single pass and exits") + blackholeInterval := flag.Duration("blackhole-interval", time.Hour, "how often to sweep the WHOLE fleet with the cheap blackhole check (did any traffic get through). Separate from -interval on purpose: the full pass sweeps a fleet over hours to days, and a provider that goes dark keeps its last passing measurement for that whole window. 0 disables the sweep") + blackholeLimit := flag.Int("blackhole-limit", 500, "providers per blackhole sweep request; the server clamps it to its own maximum") + blackholeConcurrency := flag.Int("blackhole-concurrency", 32, "simultaneous blackhole checks. Higher than -concurrency because a check is one round trip through the tunnel rather than a ~131 destination sweep") probeTimeout := flag.Duration("probe-timeout", 60*time.Second, "per-provider probe timeout, and the per-source deadline within a probe") skipConfinementCheck := flag.Bool("skip-confinement-check", false, "DANGEROUS: start even if this host can reach a geolocation api directly. Only for a one-shot manual probe on a host you know is not the operator's; a direct lookup records the OPERATOR's location for the provider and exposes the operator's address to the api") confinementTimeout := flag.Duration("confinement-timeout", 3*time.Second, "per-address deadline for the startup confinement self-check; a timeout counts as blocked. Must be at least "+confinement.MinTimeout.String()) @@ -240,7 +243,7 @@ func main() { if *skipConfinementCheck { log.Printf("egress-prober: WARNING -skip-confinement-check is set: the startup confinement self-check is DISABLED.") log.Printf("egress-prober: WARNING if this host can reach a geolocation api directly, a probe that fails to tunnel records the OPERATOR's own location for the provider and exposes the operator's address to third-party apis. Do not set this on the operator's deployment.") - } else if err := checkConfinement(ctx, (&net.Dialer{}).DialContext, net.DefaultResolver.LookupHost, confinementAddrs, *confinementTimeout, bandwidthProbeHosts(*skipBandwidth, *bandwidthCDNURL)...); err != nil { + } else if err := checkConfinement(ctx, (&net.Dialer{}).DialContext, net.DefaultResolver.LookupHost, confinementAddrs, *confinementTimeout, append(bandwidthProbeHosts(*skipBandwidth, *bandwidthCDNURL), egresshealth.BlackholeHosts()...)...); err != nil { log.Printf("egress-prober: confinement self-check failed: %s", err) // ErrNoEvidence is not a claim that this host is unconfined -- it is // the check saying it could not find out -- so the "go and confine it" @@ -344,6 +347,18 @@ func main() { *cacheTTL, ) + if 0 < *blackholeInterval { + sweeper := &blackholeSweeper{ + operator: operator, + tunnelCfg: tunnelCfg, + pins: pins, + timeout: *probeTimeout, + concurrency: *blackholeConcurrency, + limit: *blackholeLimit, + } + go sweeper.run(ctx, *blackholeInterval) + } + // I3: a single-shot run (-interval 0) is the mode the README recommends // for external cron/systemd scheduling, which decides success or // failure purely from the exit code -- so this process must not report diff --git a/egresshealth/blackhole.go b/egresshealth/blackhole.go new file mode 100644 index 0000000..d7acdfe --- /dev/null +++ b/egresshealth/blackhole.go @@ -0,0 +1,146 @@ +package egresshealth + +import ( + "context" + "math/rand" + "net/http" + "net/url" +) + +// BlackholeSampleSize is how many destinations one blackhole check asks for. +// +// Three, not one: a single destination conflates "this provider carries no +// traffic" with "this destination is having a bad minute", and the check's +// whole job is to be trusted enough to remove a provider from the public list. +// Three drawn from different operators makes a false positive require three +// independent failures at once. +// +// Not more than three, because this runs hourly against the entire fleet and +// every added destination multiplies by the population. The rich picture is +// what Check is for; this answers one bit. +const BlackholeSampleSize = 3 + +// BlackholeResult is the outcome of one provider's check. +type BlackholeResult struct { + // OK is true when at least one destination answered correctly. ANY, not + // all: the question is whether the provider carries traffic at all, and a + // provider that reaches two of three is degraded, not dark. Degradation is + // Check's department -- treating it as a blackhole here would remove + // working providers on a signal that cannot tell the two apart. + OK bool + // Failure is "" when OK, otherwise a short class suitable for the server's + // varchar(64): all_destinations_failed. + Failure string + // Results is every destination tried, for logging. A caller that reports + // only the bit throws away the reason. + Results []CheckResult +} + +// FailureAllDestinationsFailed is the only failure class this check itself +// produces. A tunnel that could not be opened never reaches here, and is the +// caller's to classify. +const FailureAllDestinationsFailed = "all_destinations_failed" + +// Blackhole answers one question about a provider: did ANY traffic get through. +// +// It exists beside Check rather than inside it because they answer different +// questions on different cadences. Check samples ~131 destinations across four +// classes to describe HOW a provider is failing, and is expensive enough that +// sweeping a fleet with it takes hours to days. In that window a provider that +// silently stops forwarding keeps its last passing measurement, and every +// consumer of that measurement keeps believing it -- while the provider stays +// connected and goes on accepting clients, because nothing about being dark +// looks different from the outside. +// +// So this is deliberately the cheapest useful check: a small fixed sample, one +// round trip each, pass on the first success. It reuses fetch, and therefore +// the table's headers, body caps and Verify contracts -- a captive portal that +// answers 200 with its own body fails here exactly as it fails a full run, +// which is the property that makes "something got through" mean anything. +// +// Only the connectivity class is drawn. Those destinations exist to answer +// "is there internet", they are operated by several independent parties, they +// return a few hundred bytes at most, and they are the least likely in the +// table to be blocked for a reason that has nothing to do with the provider. +func Blackhole(ctx context.Context, client *http.Client, opts Options) *BlackholeResult { + return blackhole(ctx, client, Destinations(), opts) +} + +// blackhole is the testable form: the destination table is injected. +func blackhole(ctx context.Context, client *http.Client, dests []Destination, opts Options) *BlackholeResult { + timeout := opts.PerRequestTimeout + if timeout <= 0 { + timeout = DefaultPerRequestTimeout + } + + sample := blackholeSample(dests, opts.rng()) + + result := &BlackholeResult{Failure: FailureAllDestinationsFailed} + for _, d := range sample { + if ctx.Err() != nil { + break + } + cr := fetch(ctx, client, d, timeout) + result.Results = append(result.Results, cr) + if cr.OK { + // stop on the first success: the question is answered, and every + // further request is spend on a provider already known to work. + // Sequential rather than concurrent for the same reason -- the + // common case costs exactly one round trip. + result.OK = true + result.Failure = "" + return result + } + } + + return result +} + +// blackholeSample draws up to BlackholeSampleSize connectivity destinations. +// +// Drawn fresh per run rather than fixed, for the same anti-gaming reason the +// full check samples: a provider that knew the three addresses could carry +// those and blackhole everything else. +func blackholeSample(dests []Destination, r *rand.Rand) []Destination { + candidates := []Destination{} + for _, d := range dests { + if d.Class == ClassConnectivity { + candidates = append(candidates, d) + } + } + if len(candidates) == 0 { + return nil + } + + r.Shuffle(len(candidates), func(i, j int) { + candidates[i], candidates[j] = candidates[j], candidates[i] + }) + return candidates[:min(BlackholeSampleSize, len(candidates))] +} + +// BlackholeHosts is the set of hosts a blackhole check can dial, for the +// startup confinement self-check. +// +// The check is only meaningful if these are unreachable EXCEPT through a +// provider tunnel. If the prober could reach them directly, a provider that +// carries nothing would still be recorded as ok -- the check would confirm the +// prober's own connectivity and remove nothing, forever, while looking healthy. +func BlackholeHosts() []string { + seen := map[string]bool{} + hosts := []string{} + for _, d := range Destinations() { + if d.Class != ClassConnectivity { + continue + } + u, err := url.Parse(d.URL) + if err != nil { + continue + } + h := u.Hostname() + if h != "" && !seen[h] { + seen[h] = true + hosts = append(hosts, h) + } + } + return hosts +} diff --git a/egresshealth/blackhole_test.go b/egresshealth/blackhole_test.go new file mode 100644 index 0000000..a7df1d6 --- /dev/null +++ b/egresshealth/blackhole_test.go @@ -0,0 +1,130 @@ +package egresshealth + +import ( + "context" + "math/rand" + "net/http" + "net/http/httptest" + "testing" +) + +// stubDests builds a connectivity table pointed at one server, plus a +// non-connectivity entry that must never be drawn. +func stubDests(url string, n int) []Destination { + dests := []Destination{{ + Name: "not-connectivity", Class: ClassCDN, URL: url + "/cdn", + Expect: ExpectStatus, Status: http.StatusNoContent, + }} + for i := 0; i < n; i++ { + dests = append(dests, Destination{ + Name: "conn-" + string(rune('a'+i)), + Class: ClassConnectivity, + URL: url + "/conn", + // the real connectivity entries are 204 probes + Expect: ExpectStatus, Status: http.StatusNoContent, + }) + } + return dests +} + +// A provider that carries traffic passes on the FIRST success, without paying +// for the rest of the sample. +func TestBlackholePassesOnFirstSuccess(t *testing.T) { + var requests int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + res := blackhole(context.Background(), srv.Client(), stubDests(srv.URL, 3), + Options{Rand: rand.New(rand.NewSource(1))}) + + if !res.OK { + t.Fatalf("OK = false, want true: every destination answered correctly") + } + if res.Failure != "" { + t.Errorf("Failure = %q, want empty on success", res.Failure) + } + if requests != 1 { + t.Errorf("made %d requests, want 1: the question is answered by the first success, "+ + "and this runs hourly against the whole fleet", requests) + } +} + +// A blackhole fails only when EVERY drawn destination fails. +func TestBlackholeFailsOnlyWhenAllFail(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // a captive-portal shaped answer: 200 with a body where 204 was required + w.WriteHeader(http.StatusOK) + w.Write([]byte("hijacked")) + })) + defer srv.Close() + + res := blackhole(context.Background(), srv.Client(), stubDests(srv.URL, 3), + Options{Rand: rand.New(rand.NewSource(1))}) + + if res.OK { + t.Fatalf("OK = true, want false: no destination met its contract") + } + if res.Failure != FailureAllDestinationsFailed { + t.Errorf("Failure = %q, want %q", res.Failure, FailureAllDestinationsFailed) + } + if len(res.Results) != BlackholeSampleSize { + t.Errorf("tried %d destinations, want the full sample of %d before declaring a blackhole", + len(res.Results), BlackholeSampleSize) + } +} + +// One reachable destination among failures is NOT a blackhole. A provider +// reaching some destinations is degraded, which is Check's department -- calling +// it dark here would remove working providers on a signal that cannot tell the +// two apart. +func TestBlackholePartialReachabilityIsNotABlackhole(t *testing.T) { + var n int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n++ + if n < 3 { + w.WriteHeader(http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + res := blackhole(context.Background(), srv.Client(), stubDests(srv.URL, 3), + Options{Rand: rand.New(rand.NewSource(1))}) + + if !res.OK { + t.Errorf("OK = false, want true: the third destination answered, so traffic is getting through") + } +} + +// Only the connectivity class is ever drawn. +func TestBlackholeSampleDrawsConnectivityOnly(t *testing.T) { + sample := blackholeSample(stubDests("http://x", 5), rand.New(rand.NewSource(7))) + + if len(sample) != BlackholeSampleSize { + t.Fatalf("drew %d, want %d", len(sample), BlackholeSampleSize) + } + for _, d := range sample { + if d.Class != ClassConnectivity { + t.Errorf("drew %s from class %q, want %q only", d.Name, d.Class, ClassConnectivity) + } + } +} + +// The real table must actually contain connectivity destinations, or the check +// silently degrades to "no sample, therefore a blackhole" and would condemn the +// entire fleet. +func TestBlackholeRealTableHasConnectivityDestinations(t *testing.T) { + sample := blackholeSample(Destinations(), rand.New(rand.NewSource(1))) + if len(sample) == 0 { + t.Fatal("the real destination table drew no connectivity destinations: " + + "every provider would be recorded as a blackhole") + } + if hosts := BlackholeHosts(); len(hosts) == 0 { + t.Error("BlackholeHosts() is empty: the confinement self-check would not cover " + + "the addresses this check dials, so a prober that could reach them directly would record every provider as ok") + } +} diff --git a/ingest/blackhole.go b/ingest/blackhole.go new file mode 100644 index 0000000..bab2c53 --- /dev/null +++ b/ingest/blackhole.go @@ -0,0 +1,166 @@ +package ingest + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// ErrBlackholeUnsupported reports that the server has no blackhole endpoints +// (404), for the same reason ErrDueUnsupported exists: the prober still works +// against a server that has not deployed them. The sweep is simply skipped, and +// says so once rather than every pass. +var ErrBlackholeUnsupported = errors.New("ingest: the server does not implement the provider-blackhole endpoints") + +// MaxBlackholeFailureLen is the width of the server's failure column +// (varchar(64)). The server REJECTS an oversized value rather than truncating +// it, and a rejected batch is a lost sweep, so truncate before sending. +const MaxBlackholeFailureLen = 64 + +// blackholeDueURL resolves the due endpoint from ServerURL. +func (c *Client) blackholeDueURL() string { + return strings.TrimRight(c.ServerURL, "/") + "/network/provider-blackhole-due" +} + +// BlackholeDue asks which providers to check next: never checked first, then +// least recently checked. +// +// Unlike Due there is no attempt backoff on the server side, by design: a +// provider that failed last hour must be re-checked this hour, because that is +// how it returns to the public list once it recovers. +func (c *Client) BlackholeDue(ctx context.Context, limit int) ([]string, error) { + if limit < 1 { + return nil, fmt.Errorf("ingest: blackhole due limit must be positive (got %d)", limit) + } + if 1 < c.ShardCount && (c.ShardIndex < 0 || c.ShardCount <= c.ShardIndex) { + return nil, fmt.Errorf( + "ingest: shard index %d is out of range for shard count %d", + c.ShardIndex, c.ShardCount, + ) + } + + u, err := url.Parse(c.blackholeDueURL()) + if err != nil { + return nil, err + } + q := u.Query() + q.Set("limit", strconv.Itoa(limit)) + if 1 < c.ShardCount { + q.Set("shard_count", strconv.Itoa(c.ShardCount)) + q.Set("shard_index", strconv.Itoa(c.ShardIndex)) + } + u.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return nil, err + } + req.Header.Set("X-UR-Operator-Secret", c.OperatorSecret) + + resp, err := c.httpClient().Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + switch resp.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return nil, ErrBlackholeUnsupported + case http.StatusUnauthorized: + return nil, ErrUnauthorized + default: + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("%w: status %d: %s", ErrRejected, resp.StatusCode, strings.TrimSpace(string(msg))) + } + + var out dueResult + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, err + } + return out.ClientIds, nil +} + +// BlackholeCheck is one provider's result, as submitted. +type BlackholeCheck struct { + ClientId string `json:"client_id"` + OK bool `json:"ok"` + Failure string `json:"failure,omitempty"` + CheckedAt time.Time `json:"checked_at"` +} + +type blackholeChecksBody struct { + Checks []BlackholeCheck `json:"checks"` +} + +// SubmitBlackholeChecks reports a whole pass in one request. +// +// Batched because a sweep produces hundreds of one-bit answers and a request +// each would spend more on http than on the checks themselves. +// +// The server validates the entire batch before writing any of it, so a single +// malformed entry loses the whole pass. Everything that can be fixed locally is +// fixed here rather than sent and rejected: a zero CheckedAt is refused, and an +// over-long failure class is truncated to the column width. +func (c *Client) SubmitBlackholeChecks(ctx context.Context, checks []BlackholeCheck) error { + if len(checks) == 0 { + return nil + } + + body := blackholeChecksBody{Checks: make([]BlackholeCheck, 0, len(checks))} + for _, check := range checks { + if check.CheckedAt.IsZero() { + // never fabricated: an "as of now" timestamp would defeat the + // server's freshness bound and could pin a stale verdict + return fmt.Errorf("ingest: blackhole check for %s has a zero CheckedAt", check.ClientId) + } + if !check.OK && strings.TrimSpace(check.Failure) == "" { + // the server rejects this, and a rejected batch is a lost sweep + return fmt.Errorf("ingest: failed blackhole check for %s names no failure class", check.ClientId) + } + if check.OK { + check.Failure = "" + } + check.Failure = truncateUTF8(check.Failure, MaxBlackholeFailureLen) + body.Checks = append(body.Checks, check) + } + + buf, err := json.Marshal(body) + if err != nil { + return err + } + + url := strings.TrimRight(c.ServerURL, "/") + "/network/provider-blackhole-checks" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(buf)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-UR-Operator-Secret", c.OperatorSecret) + + resp, err := c.httpClient().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + switch resp.StatusCode { + case http.StatusOK: + return nil + case http.StatusNotFound: + return ErrBlackholeUnsupported + case http.StatusUnauthorized: + return fmt.Errorf("%w: %w", ErrRejected, ErrUnauthorized) + default: + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("%w: status %d: %s", ErrRejected, resp.StatusCode, strings.TrimSpace(string(msg))) + } +} From 585fd419cf8ed9a467c9a31db92c158445e65789 Mon Sep 17 00:00:00 2001 From: Ryanmello07 <67509637+Ryanmello07@users.noreply.github.com> Date: Wed, 12 Aug 2026 05:49:17 +0100 Subject: [PATCH 2/5] fix(blackhole): give the sweep its own short timeout The sweeper reused -probe-timeout (2m20s on beta) as its per-request deadline, which defeats the point of the check. A dark provider burns the full timeout on every destination, so a 500-provider batch at concurrency 32 takes ~37 minutes rather than ~4 -- on a fleet that is mostly dark, which is exactly the fleet this exists to detect, the cheap loop runs as slowly as the full pass it was meant to complement. 15s: a provider that carries traffic answers a 204 probe in well under that, and one that does not must fail fast for the sweep to stay hourly. --- cmd/egress-prober/main.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cmd/egress-prober/main.go b/cmd/egress-prober/main.go index 33fdc41..1fa8ca6 100644 --- a/cmd/egress-prober/main.go +++ b/cmd/egress-prober/main.go @@ -70,6 +70,7 @@ func main() { blackholeInterval := flag.Duration("blackhole-interval", time.Hour, "how often to sweep the WHOLE fleet with the cheap blackhole check (did any traffic get through). Separate from -interval on purpose: the full pass sweeps a fleet over hours to days, and a provider that goes dark keeps its last passing measurement for that whole window. 0 disables the sweep") blackholeLimit := flag.Int("blackhole-limit", 500, "providers per blackhole sweep request; the server clamps it to its own maximum") blackholeConcurrency := flag.Int("blackhole-concurrency", 32, "simultaneous blackhole checks. Higher than -concurrency because a check is one round trip through the tunnel rather than a ~131 destination sweep") + blackholeTimeout := flag.Duration("blackhole-timeout", 15*time.Second, "per-request deadline for a blackhole check. Deliberately far shorter than -probe-timeout: this check asks one question and a dark provider must fail it FAST, or the sweep costs the full timeout on every dead provider and stops being the cheap loop it exists to be -- at 2m20s a 500-provider batch at concurrency 32 takes ~37 minutes instead of ~4") probeTimeout := flag.Duration("probe-timeout", 60*time.Second, "per-provider probe timeout, and the per-source deadline within a probe") skipConfinementCheck := flag.Bool("skip-confinement-check", false, "DANGEROUS: start even if this host can reach a geolocation api directly. Only for a one-shot manual probe on a host you know is not the operator's; a direct lookup records the OPERATOR's location for the provider and exposes the operator's address to the api") confinementTimeout := flag.Duration("confinement-timeout", 3*time.Second, "per-address deadline for the startup confinement self-check; a timeout counts as blocked. Must be at least "+confinement.MinTimeout.String()) @@ -352,7 +353,7 @@ func main() { operator: operator, tunnelCfg: tunnelCfg, pins: pins, - timeout: *probeTimeout, + timeout: *blackholeTimeout, concurrency: *blackholeConcurrency, limit: *blackholeLimit, } From 08968d58c45e93b5fe7af080aa83e1c7e2ac30be Mon Sep 17 00:00:00 2001 From: Ryanmello07 <67509637+Ryanmello07@users.noreply.github.com> Date: Wed, 12 Aug 2026 05:56:02 +0100 Subject: [PATCH 3/5] fix(blackhole): drain the queue each pass, not one batch A pass took one batch and slept for the interval. -blackhole-limit is the server's per-request ceiling, not the size of the fleet: at 500 per request against ~2,700 eligible providers that covers under a fifth of them per hour, so the oldest evidence ages out faster than the sweep reaches it -- and the requirement is that the whole fleet is checked every interval. Now a pass keeps requesting until the queue returns nothing, bounded at 40 rounds so a server that keeps handing back work cannot hold a pass open forever and starve the interval. --- cmd/egress-prober/blackhole.go | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/cmd/egress-prober/blackhole.go b/cmd/egress-prober/blackhole.go index 6690fcc..fa3ba4a 100644 --- a/cmd/egress-prober/blackhole.go +++ b/cmd/egress-prober/blackhole.go @@ -32,6 +32,12 @@ type blackholeSweeper struct { limit int } +// maxBlackholeRounds bounds one sweep's batches. 40 rounds x the server's 5000 +// ceiling is far above any real fleet, so it never truncates a legitimate +// sweep; it exists so a server that keeps handing back work cannot hold a pass +// open indefinitely and starve the interval. +const maxBlackholeRounds = 40 + // blackholeResult carries one provider's outcome out of the worker pool. type blackholeResult struct { check ingest.BlackholeCheck @@ -168,11 +174,29 @@ func (s *blackholeSweeper) checkOne(ctx context.Context, clientId string) blackh func (s *blackholeSweeper) run(ctx context.Context, interval time.Duration) { for { start := time.Now() - checked, err := s.sweep(ctx) + // Drain the queue, do not take one batch and sleep. The requirement is + // that the WHOLE fleet is checked every interval, and the batch size is the + // server's per-request ceiling, not the size of the fleet: at 500 per + // request against ~2,700 eligible providers, one batch per hour covers + // under a fifth of them and the oldest evidence would age out faster + // than the sweep reaches it. Rounds are bounded so a server that keeps + // returning work cannot hold a pass open forever. + total, err := 0, error(nil) + for round := 0; round < maxBlackholeRounds; round++ { + var checked int + checked, err = s.sweep(ctx) + total += checked + if err != nil || checked == 0 || ctx.Err() != nil { + break + } + } + checked := total switch { case err == nil: if checked == 0 { log.Printf("blackhole: pass found nothing due") + } else { + log.Printf("blackhole: sweep complete: %d checked in %s", checked, time.Since(start).Round(time.Second)) } case errors.Is(err, ingest.ErrBlackholeUnsupported): log.Printf("blackhole: the server does not implement the blackhole endpoints; sweeping is disabled") From 42b30041e997c4f95923c6cd557d329973ebe122 Mon Sep 17 00:00:00 2001 From: Ryanmello07 <67509637+Ryanmello07@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:02:22 +0100 Subject: [PATCH 4/5] feat(credential): let the prober fetch its own jwt from the server The prober required UR_PROBER_BY_JWT to be provisioned and placed by hand. The server now mints the prober's network client identity in a bootstrap task and serves it at GET /network/prober-credential, authenticated with the same X-UR-Operator-Secret the other operator endpoints use -- so the last hand-placed secret can go. ingest.Client.ProberCredential fetches it, with three disjoint outcomes so a caller cannot take the wrong branch: 404 is ErrCredentialNotReady ("not yet", keep asking), 401 is a bare ErrUnauthorized (a wrong secret must be loud and must never be retried forever), everything else is ErrCredentialUnavailable (transient, retry). A 200 carrying no usable by_client_jwt is refused rather than returned empty -- decodable is not usable, the same lesson GeolocationPins learned about a nil pin map. cmd/egress-prober fills an empty -by-jwt from it, with envFallback's precedence: an explicitly supplied jwt wins and the endpoint is never contacted, so the running deployment is untouched and gains no dependency on it. The fetch sits above parseByJwtClientId, so a fetched credential takes the identical path a supplied one does and a token this process cannot use fails at startup instead of becoming a silent outage. The bootstrap task runs every 6h, so a 404 is a wait on a backoff capped at 5m, ended only by ctx -- an interrupt there exits zero, since a prober interrupted while waiting is a shutdown, not a broken deployment. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01QtgqtCmKJRXdsQ5ktiqwkg --- README.md | 40 ++- cmd/egress-prober/credential_test.go | 430 +++++++++++++++++++++++++++ cmd/egress-prober/main.go | 158 +++++++++- ingest/credential.go | 126 ++++++++ ingest/credential_test.go | 237 +++++++++++++++ 5 files changed, 967 insertions(+), 24 deletions(-) create mode 100644 cmd/egress-prober/credential_test.go create mode 100644 ingest/credential.go create mode 100644 ingest/credential_test.go diff --git a/README.md b/README.md index 5f687d3..d56ae4d 100644 --- a/README.md +++ b/README.md @@ -45,26 +45,44 @@ go build ./cmd/egress-prober ./egress-prober \ -api-url https://api.example.net \ -platform-url wss://connect.example.net \ - -by-jwt "$UR_PROBER_BY_JWT" \ -operator-secret "$UR_OPERATOR_SECRET" \ -concurrency 4 \ -cache-ttl 24h \ -interval 1h ``` +That run fetches the prober's own identity from the server. To supply one you +provisioned yourself instead, add `-by-jwt "$UR_PROBER_BY_JWT"` (or export +`UR_PROBER_BY_JWT`); it takes precedence and the fetch is skipped entirely. + `-by-jwt` and `-operator-secret` may also be supplied via the `UR_PROBER_BY_JWT` and `UR_OPERATOR_SECRET` environment variables instead of flags, which is the recommended way to run this under systemd (keeps secrets -out of `ps`/shell history). All four of `-api-url`, `-platform-url`, `-by-jwt` -and `-operator-secret` are required; the prober exits immediately with a -message naming the missing flag(s) if any are absent, rather than starting in -a broken state. - -The prober needs its own network client identity (`-by-jwt`), provisioned like -any other client. `-operator-secret` must match `ingest_secret` in the server's -`provider_egress.yml` vault resource — it authenticates the pin fetch as well as -ingest, so a wrong secret now stops the prober at startup rather than only -having its submissions rejected. +out of `ps`/shell history). `-api-url`, `-platform-url` and `-operator-secret` +are required; the prober exits immediately with a message naming the missing +flag(s) if any are absent, rather than starting in a broken state. + +The prober needs its own network client identity (`-by-jwt`). **Leave it empty +and the prober fetches one for itself** from the server's +`/network/prober-credential` endpoint, authenticating with `-operator-secret` — +no hand-provisioned identity, and one less secret to place. The server mints +that identity in a bootstrap task which runs every 6h, so a prober brought up +alongside a fresh deployment may start before its credential exists: it waits +for it, logging one line per attempt on a backoff capped at 5 minutes, rather +than exiting into a restart loop. The wait has no deadline of its own — impose +one with the supervisor's start timeout if a deployment wants it. + +An explicitly supplied `-by-jwt` (or `UR_PROBER_BY_JWT`) always wins and the +endpoint is never contacted, so an existing deployment that provisions the +identity by hand is unaffected and acquires no dependency on it. Either way the +jwt goes through the same startup check, so a credential the process cannot use +stops it at startup instead of leaving a prober that looks healthy and probes +nothing. + +`-operator-secret` must match `ingest_secret` in the server's +`provider_egress.yml` vault resource — it authenticates the credential fetch and +the pin fetch as well as ingest, so a wrong secret stops the prober at startup +rather than only having its submissions rejected. The server must have observed the geolocation certificate pins before the prober can start: it fetches them at startup and **refuses to run without a diff --git a/cmd/egress-prober/credential_test.go b/cmd/egress-prober/credential_test.go new file mode 100644 index 0000000..77b249d --- /dev/null +++ b/cmd/egress-prober/credential_test.go @@ -0,0 +1,430 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/urnetwork/operator-proxy/ingest" +) + +// --------------------------------------------------------------------------- +// fetchByJwtIfEmpty: the precedence and the wait. +// --------------------------------------------------------------------------- + +type credentialResult struct { + cred *ingest.ProberCredential + err error +} + +// stubCredentialFetcher hands back one result per call, repeating the last one +// forever, and counts the calls. The count is the assertion that matters in +// most of these tests: whether the server was asked at all. +type stubCredentialFetcher struct { + results []credentialResult + calls int +} + +func (s *stubCredentialFetcher) ProberCredential(ctx context.Context) (*ingest.ProberCredential, error) { + s.calls++ + if len(s.results) == 0 { + return nil, errors.New("stub: no result configured") + } + r := s.results[0] + if 1 < len(s.results) { + s.results = s.results[1:] + } + return r.cred, r.err +} + +func okCredential(jwt string) credentialResult { + return credentialResult{cred: &ingest.ProberCredential{ByClientJwt: jwt, ClientId: "019f8835-158d-6fd8-e9dd-fd0e4c6d6792"}} +} + +// TestFetchByJwtIfEmptyLeavesAnExplicitJwtAlone is the regression test for the +// existing deployment, which supplies UR_PROBER_BY_JWT today. +// +// The assertion is on the CALL COUNT, not only on the returned value. A +// fetch-then-prefer-the-explicit-one implementation would leave the right +// string in place and still be broken: against a server whose bootstrap task +// has not run, it would sit in the 404 wait forever for a prober that never +// needed a credential at all -- turning a working deployment into one that +// never starts. Asserting the value alone would not catch that. +func TestFetchByJwtIfEmptyLeavesAnExplicitJwtAlone(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{okCredential("fetched-jwt")}} + byJwt := "explicitly-supplied-jwt" + + if err := fetchByJwtIfEmpty(context.Background(), &byJwt, f, time.Millisecond, 2*time.Millisecond); err != nil { + t.Fatalf("fetchByJwtIfEmpty err = %v", err) + } + if byJwt != "explicitly-supplied-jwt" { + t.Errorf("byJwt = %q, want the explicitly supplied jwt to survive untouched", byJwt) + } + if f.calls != 0 { + t.Fatalf("the server was asked for a credential %d time(s) even though a jwt was supplied; startup would then depend on an endpoint this deployment does not need", f.calls) + } +} + +func TestFetchByJwtIfEmptyFetchesWhenEmpty(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{okCredential("fetched-jwt")}} + byJwt := "" + + if err := fetchByJwtIfEmpty(context.Background(), &byJwt, f, time.Millisecond, 2*time.Millisecond); err != nil { + t.Fatalf("fetchByJwtIfEmpty err = %v", err) + } + if byJwt != "fetched-jwt" { + t.Errorf("byJwt = %q, want the jwt the server handed over", byJwt) + } + if f.calls != 1 { + t.Errorf("calls = %d, want exactly 1", f.calls) + } +} + +// TestFetchByJwtIfEmptyKeepsPollingUntilTheCredentialExists: the bootstrap task +// runs every 6h, so a prober started alongside a fresh deployment arrives +// before its credential does. That must be a wait, not an exit. +func TestFetchByJwtIfEmptyKeepsPollingUntilTheCredentialExists(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{ + {err: ingest.ErrCredentialNotReady}, + {err: ingest.ErrCredentialNotReady}, + okCredential("fetched-after-the-wait"), + }} + byJwt := "" + + if err := fetchByJwtIfEmpty(context.Background(), &byJwt, f, time.Millisecond, 2*time.Millisecond); err != nil { + t.Fatalf("fetchByJwtIfEmpty err = %v; a 404 means \"not yet\", and exiting on it would crash-loop a prober started before the bootstrap task", err) + } + if byJwt != "fetched-after-the-wait" { + t.Errorf("byJwt = %q, want the credential that appeared on the third ask", byJwt) + } + if f.calls != 3 { + t.Errorf("calls = %d, want 3 (two 404s then the credential)", f.calls) + } +} + +// TestFetchByJwtIfEmptyStopsOnUnauthorized: a wrong operator secret is a +// broken deployment, and it will be just as wrong on the thousandth ask. It +// must surface once, immediately -- the posture blackhole.go already takes on +// the same error. +func TestFetchByJwtIfEmptyStopsOnUnauthorized(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{{err: ingest.ErrUnauthorized}}} + byJwt := "" + + err := fetchByJwtIfEmpty(context.Background(), &byJwt, f, time.Millisecond, 2*time.Millisecond) + if err == nil { + t.Fatal("fetchByJwtIfEmpty returned nil on a 401; a rejected operator secret must be loud") + } + if !errors.Is(err, ingest.ErrUnauthorized) { + t.Errorf("err = %v, want it to wrap ingest.ErrUnauthorized", err) + } + if !strings.Contains(err.Error(), "-operator-secret") { + t.Errorf("err = %v, want it to name -operator-secret so the operator knows what to fix", err) + } + if f.calls != 1 { + t.Errorf("calls = %d, want exactly 1: a rejected secret must never be retried", f.calls) + } + if byJwt != "" { + t.Errorf("byJwt = %q, want it left empty", byJwt) + } +} + +// TestFetchByJwtIfEmptyRetriesEverythingElse: a 500 or an unreachable api is +// transient. The prober may well come up before the server does. +func TestFetchByJwtIfEmptyRetriesEverythingElse(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{ + {err: ingest.ErrCredentialUnavailable}, + okCredential("fetched-after-the-blip"), + }} + byJwt := "" + + if err := fetchByJwtIfEmpty(context.Background(), &byJwt, f, time.Millisecond, 2*time.Millisecond); err != nil { + t.Fatalf("fetchByJwtIfEmpty err = %v; a transient failure must be retried", err) + } + if byJwt != "fetched-after-the-blip" { + t.Errorf("byJwt = %q, want the credential fetched after the retry", byJwt) + } + if f.calls != 2 { + t.Errorf("calls = %d, want 2", f.calls) + } +} + +// TestFetchByJwtIfEmptyStopsWhenTheContextEnds: the wait is unbounded in time, +// so ctx is the only thing that ends it. A loop watching only its timer would +// swallow SIGTERM for the length of a backoff and, worse, could never be shut +// down while the server keeps answering 404 -- the same +// interrupt-during-a-wait shape this codebase already fixed once. +func TestFetchByJwtIfEmptyStopsWhenTheContextEnds(t *testing.T) { + f := &stubCredentialFetcher{results: []credentialResult{{err: ingest.ErrCredentialNotReady}}} + byJwt := "" + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := fetchByJwtIfEmpty(ctx, &byJwt, f, time.Hour, time.Hour) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled so an interrupted wait is not reported as a broken deployment", err) + } + if 1 < f.calls { + t.Errorf("calls = %d, want the loop to stop at the first cancellation check", f.calls) + } +} + +// TestNextBackoffDoublesAndCaps pins the schedule without sleeping through it. +// The cap is the "bounded" half of bounded backoff: without it a prober that +// waited through a full 6h bootstrap cycle would end up asking once a day. +func TestNextBackoffDoublesAndCaps(t *testing.T) { + max := 5 * time.Minute + got := []time.Duration{} + cur := 30 * time.Second + for i := 0; i < 6; i++ { + got = append(got, cur) + cur = nextBackoff(cur, max) + } + want := []time.Duration{ + 30 * time.Second, + time.Minute, + 2 * time.Minute, + 4 * time.Minute, + 5 * time.Minute, // capped, not 8m + 5 * time.Minute, + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("backoff schedule = %v, want %v", got, want) + } + } + if n := nextBackoff(10*time.Minute, max); n != max { + t.Errorf("nextBackoff(10m, 5m) = %s, want the cap %s", n, max) + } +} + +// The constants the binary actually runs with have to satisfy the same +// property the loop's own guard assumes: a positive interval that never +// exceeds the cap. A zero here would be a busy loop against the server. +func TestCredentialPollConstantsAreSane(t *testing.T) { + if credentialPollInitial <= 0 { + t.Errorf("credentialPollInitial = %s, must be positive or the poll becomes a busy loop", credentialPollInitial) + } + if credentialPollMax < credentialPollInitial { + t.Errorf("credentialPollMax %s < credentialPollInitial %s", credentialPollMax, credentialPollInitial) + } +} + +// --------------------------------------------------------------------------- +// Startup, end to end: the fetched jwt goes through the same checks. +// --------------------------------------------------------------------------- + +// credentialStub serves the prober-credential endpoint and counts what was +// asked for. Everything else 404s, which for the pin endpoint means startup +// stops there -- the marker these tests use for "the prober got past the +// credential stage". +type credentialStub struct { + mu sync.Mutex + credentialCalls int + pinCalls int + status int + body string +} + +func (s *credentialStub) counts() (credential int, pin int) { + s.mu.Lock() + defer s.mu.Unlock() + return s.credentialCalls, s.pinCalls +} + +func (s *credentialStub) server(t *testing.T) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + switch r.URL.Path { + case "/network/prober-credential": + s.credentialCalls++ + case "/network/geolocation-source-pins": + s.pinCalls++ + } + s.mu.Unlock() + + if r.URL.Path != "/network/prober-credential" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(s.status) + _, _ = w.Write([]byte(s.body)) + })) +} + +// runProberWithoutJwt runs the built binary with NO UR_PROBER_BY_JWT. +// +// It strips the variable from the inherited environment rather than merely not +// setting it: a developer or CI runner with UR_PROBER_BY_JWT exported would +// otherwise silently supply the very thing these tests exist to prove is +// fetched, and every one of them would pass without exercising anything. +func runProberWithoutJwt(t *testing.T, args ...string) (string, int) { + t.Helper() + // Insurance, not a deadline the tests rely on: every stub below answers + // immediately, so a run that hangs means the fetch was reached when it + // should not have been. Killing it turns that into a failure rather than a + // 20-minute package timeout. + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + cmd := exec.CommandContext(ctx, buildProber(t), args...) + env := make([]string, 0, len(os.Environ())+1) + for _, kv := range os.Environ() { + if strings.HasPrefix(kv, "UR_PROBER_BY_JWT=") { + continue + } + env = append(env, kv) + } + cmd.Env = append(env, "UR_OPERATOR_SECRET="+testOperatorSecret) + + out, err := cmd.CombinedOutput() + var exitErr *exec.ExitError + switch { + case err == nil: + return string(out), 0 + case errors.As(err, &exitErr): + return string(out), exitErr.ExitCode() + default: + t.Fatalf("running the prober: %s", err) + return "", -1 + } +} + +// TestProberStartsWithNoJwtAndFetchesOne is the end-to-end proof of the +// feature: with UR_PROBER_BY_JWT absent the process no longer refuses to +// start, it asks the server instead, and the credential it gets back carries +// it into the rest of startup. +// +// "Carries it into the rest of startup" is asserted as the pin fetch having +// been reached, because that is the next thing a running prober does. Without +// it the test would pass against an implementation that fetched the jwt and +// then dropped it. +func TestProberStartsWithNoJwtAndFetchesOne(t *testing.T) { + jwt := testByJwt(t) + stub := &credentialStub{ + status: http.StatusOK, + body: `{"by_client_jwt":"` + jwt + `","client_id":"019f8835-158d-6fd8-e9dd-fd0e4c6d6792"}`, + } + srv := stub.server(t) + defer srv.Close() + + out, code := runProberWithoutJwt(t, + "-api-url", srv.URL, + "-platform-url", "ws://127.0.0.1:1", + "-interval", "0", + "-skip-confinement-check", + "-skip-bandwidth", + ) + + if strings.Contains(out, "missing required flag") { + t.Fatalf("the prober still refuses to start without -by-jwt; the whole point is that an empty one is fetched.\n--- output ---\n%s", out) + } + credentialCalls, pinCalls := stub.counts() + if credentialCalls != 1 { + t.Errorf("the credential endpoint was called %d time(s), want exactly 1.\n--- output ---\n%s", credentialCalls, out) + } + if pinCalls == 0 { + t.Errorf("startup never reached the pin fetch, so the fetched jwt was not actually carried forward.\n--- output ---\n%s", out) + } + if !strings.Contains(out, "019f8835-158d-6fd8-e9dd-fd0e4c6d6792") { + t.Errorf("the prober did not report which client id it got; an operator has to be able to match it against the server's record.\n--- output ---\n%s", out) + } + // The jwt is a credential and this output is journald's. The client id is + // logged instead, on purpose. + if strings.Contains(out, jwt) { + t.Errorf("the fetched jwt was printed verbatim.\n--- output ---\n%s", out) + } + // The stub has no pin endpoint, so startup stops there -- the existing + // fail-closed behaviour, unchanged by this feature. + if code == 0 { + t.Errorf("exited 0 with no pin set.\n--- output ---\n%s", out) + } + assertNoSecrets(t, "the credential fetch", out) +} + +// TestProberRunsAFetchedJwtThroughTheStartupCheck is requirement 4: a +// credential the server hands over is not trusted because the server handed it +// over. It goes through parseByJwtClientId exactly as a hand-placed one does, +// so a token this process cannot use stops it here, loudly, instead of +// becoming an outage that looks like a healthy prober doing nothing. +// +// The assertion that the pin endpoint was never reached is what proves the +// check runs BEFORE the rest of startup rather than somewhere after it. +func TestProberRunsAFetchedJwtThroughTheStartupCheck(t *testing.T) { + stub := &credentialStub{ + status: http.StatusOK, + body: `{"by_client_jwt":"this-is-not-a-jwt","client_id":"019f8835-158d-6fd8-e9dd-fd0e4c6d6792"}`, + } + srv := stub.server(t) + defer srv.Close() + + out, code := runProberWithoutJwt(t, + "-api-url", srv.URL, + "-platform-url", "ws://127.0.0.1:1", + "-interval", "0", + "-skip-confinement-check", + "-skip-bandwidth", + ) + + if code == 0 { + t.Errorf("exited 0 with a fetched jwt it cannot parse; a credential the server serves but this process cannot use must fail loudly.\n--- output ---\n%s", out) + } + if !strings.Contains(out, "parse by-jwt client id") { + t.Errorf("the fetched jwt did not go through the same startup credential check a supplied one does.\n--- output ---\n%s", out) + } + credentialCalls, pinCalls := stub.counts() + if credentialCalls != 1 { + t.Errorf("the credential endpoint was called %d time(s), want exactly 1", credentialCalls) + } + if pinCalls != 0 { + t.Errorf("startup continued to the pin fetch after being handed an unusable jwt; the check must gate everything that follows it.\n--- output ---\n%s", out) + } + assertNoSecrets(t, "the unusable fetched credential", out) +} + +// TestProberDoesNotFetchWhenAJwtIsSupplied is the deployment-safety test at the +// process level: with UR_PROBER_BY_JWT set, exactly as the running deployment +// sets it today, the new endpoint is never contacted. A prober configured the +// old way must not acquire a new dependency on a server endpoint that may not +// be deployed, or on a bootstrap task that may not have run. +func TestProberDoesNotFetchWhenAJwtIsSupplied(t *testing.T) { + stub := &credentialStub{ + status: http.StatusOK, + body: `{"by_client_jwt":"` + testByJwt(t) + `","client_id":"019f8835-158d-6fd8-e9dd-fd0e4c6d6792"}`, + } + srv := stub.server(t) + defer srv.Close() + + out, code := runProberWithJwt(t, testByJwt(t), + "-api-url", srv.URL, + "-platform-url", "ws://127.0.0.1:1", + "-interval", "0", + "-skip-confinement-check", + "-skip-bandwidth", + ) + + credentialCalls, pinCalls := stub.counts() + if credentialCalls != 0 { + t.Errorf("the prober asked the server for a credential %d time(s) despite being given one; startup would then depend on an endpoint this deployment does not need.\n--- output ---\n%s", credentialCalls, out) + } + if strings.Contains(out, "asking the server for the prober credential") { + t.Errorf("the prober announced a credential fetch even though a jwt was supplied.\n--- output ---\n%s", out) + } + // It must still get on with the run it always did: straight to the pin + // fetch, which this stub refuses, so it stops there as before. + if pinCalls == 0 { + t.Errorf("startup never reached the pin fetch with a supplied jwt; the existing deployment path is broken.\n--- output ---\n%s", out) + } + if code == 0 { + t.Errorf("exited 0 with no pin set.\n--- output ---\n%s", out) + } +} diff --git a/cmd/egress-prober/main.go b/cmd/egress-prober/main.go index 1fa8ca6..63696fe 100644 --- a/cmd/egress-prober/main.go +++ b/cmd/egress-prober/main.go @@ -62,7 +62,7 @@ func main() { // echoes both secrets verbatim to stderr, into journald or a CI log. That // would invert the README's own advice, which presents these env vars as // the way to keep secrets out of logs and ps. - byJwt := flag.String("by-jwt", "", "the prober's network client jwt; prefer the UR_PROBER_BY_JWT env var, which keeps it out of ps (required)") + byJwt := flag.String("by-jwt", "", "the prober's network client jwt; prefer the UR_PROBER_BY_JWT env var, which keeps it out of ps. Leave it EMPTY to fetch it from the server's /network/prober-credential endpoint using -operator-secret, which is the unattended mode: the server mints the prober's identity in a bootstrap task and this waits for it. An explicitly supplied value always wins and is never overwritten") operatorSecret := flag.String("operator-secret", "", "ingest secret, must match ingest_secret in provider_egress.yml; prefer the UR_OPERATOR_SECRET env var, which keeps it out of ps (required)") concurrency := flag.Int("concurrency", 4, "max simultaneous provider tunnels") cacheTTL := flag.Duration("cache-ttl", 24*time.Hour, "do not re-probe a provider within this window. Only applies to the enumeration fallback used against a server with no due endpoint; when the server supplies the due list it owns the schedule") @@ -101,9 +101,10 @@ func main() { if *platformURL == "" { missing = append(missing, "-platform-url") } - if *byJwt == "" { - missing = append(missing, "-by-jwt (or UR_PROBER_BY_JWT)") - } + // -by-jwt is deliberately NOT in this list any more: an empty one is + // fetched from the server below (see fetchByJwtIfEmpty), which is the + // whole point of the prober-credential endpoint. -operator-secret stays + // required precisely because that fetch authenticates with it. if *operatorSecret == "" { missing = append(missing, "-operator-secret (or UR_OPERATOR_SECRET)") } @@ -255,6 +256,44 @@ func main() { os.Exit(1) } + // Built here rather than after the jwt because the credential fetch below + // needs it. Nothing in it depends on the jwt: it authenticates with the + // operator secret alone, which is what makes fetching the jwt possible at + // all. + operator := &ingest.Client{ + ServerURL: *apiURL, + OperatorSecret: *operatorSecret, + DueURL: *dueURL, + ShardIndex: *shardIndex, + ShardCount: *shardCount, + HTTP: &http.Client{Timeout: 30 * time.Second}, + } + + // The jwt, if the deployment did not supply one. This sits ABOVE + // parseByJwtClientId on purpose: a fetched jwt then travels the identical + // path an explicitly supplied one does -- same parse, same client id, same + // tunnel config -- so a credential the server hands over but that this + // process cannot use still fails loudly, right here, instead of becoming a + // fleet-wide outage that looks like nothing at all. + // + // It also sits below the confinement self-check, which must stay the first + // thing that touches the network. The operator's own server is the one + // direct call the prober is allowed to make, so this is the earliest point + // at which it may run. + switch err := fetchByJwtIfEmpty(ctx, byJwt, operator, credentialPollInitial, credentialPollMax); { + case err == nil: + case ctx.Err() != nil: + // Interrupted while waiting for the server's bootstrap task. That is a + // shutdown, not a broken deployment, and the same reasoning as the + // pass-result exit codes below applies: it must not exit non-zero and + // blame a configuration that is fine. + log.Printf("egress-prober: interrupted while waiting for the prober credential (%v); nothing was probed", ctx.Err()) + return + default: + log.Printf("egress-prober: %s", err) + os.Exit(1) + } + clientId, err := parseByJwtClientId(*byJwt) if err != nil { log.Fatalf("parse by-jwt client id: %s", err) @@ -273,15 +312,6 @@ func main() { Version: "0.0.0", } - operator := &ingest.Client{ - ServerURL: *apiURL, - OperatorSecret: *operatorSecret, - DueURL: *dueURL, - ShardIndex: *shardIndex, - ShardCount: *shardCount, - HTTP: &http.Client{Timeout: 30 * time.Second}, - } - // The startup fetch. This is a separate call site from the refresh in the // pass loop below, and the difference between them is the whole fail-closed // property: THIS one exits, the other one keeps the last good set. Folding @@ -848,6 +878,108 @@ func checkConfinement(ctx context.Context, dial confinement.DialFunc, lookup con return nil } +// credentialFetcher is the server's prober-credential endpoint, injected so +// fetchByJwtIfEmpty is testable without one. +type credentialFetcher interface { + ProberCredential(ctx context.Context) (*ingest.ProberCredential, error) +} + +// credentialPollInitial and credentialPollMax bound the wait for a credential +// the server has not minted yet. +// +// The server's bootstrap task runs every 6h, so a prober brought up alongside +// a fresh deployment can legitimately be hours early. That is a WAIT, not a +// crash: exiting would put a supervised process into a restart loop that +// re-runs the confinement self-check and re-fetches on every restart, and +// which reads in journald as a broken prober rather than as one patiently +// doing the right thing. Starting at 30s keeps the pickup prompt when the task +// runs minutes later; capping at 5m keeps a six-hour wait to ~75 requests. +const ( + credentialPollInitial = 30 * time.Second + credentialPollMax = 5 * time.Minute +) + +// nextBackoff doubles cur without exceeding max. Split out from the loop so the +// schedule is testable without sleeping through it. +func nextBackoff(cur, max time.Duration) time.Duration { + if max <= cur { + return max + } + if doubled := cur * 2; doubled < max { + return doubled + } + return max +} + +// fetchByJwtIfEmpty fills *byJwt from the server's prober-credential endpoint +// when it is empty, waiting for the server's bootstrap task if it has to. +// +// The precedence is envFallback's, deliberately: an explicitly supplied value +// is left alone and the server is not even asked. That is what keeps the +// existing deployment -- which supplies UR_PROBER_BY_JWT today -- working +// exactly as it does now, and it is why this cannot be written as "fetch, then +// prefer the explicit one": that would still block startup on a server which +// has no credential yet, for a prober that never needed one. +// +// The three outcomes of the fetch map to three different behaviours, and +// keeping them apart is the whole point: +// +// - not ready (404): the expected state before the bootstrap task has run. +// Log it as a wait and ask again, forever, on a capped backoff. There is no +// total deadline: "wait rather than crash-loop" has no useful upper bound +// here, and a supervisor's own start timeout is the right place to impose +// one if a deployment wants it. ctx is what ends the wait. +// - unauthorized (401): a wrong -operator-secret. Fatal and immediate. A +// secret the server rejects will be rejected on every retry, so polling +// would turn a two-minute fix into an outage nobody is paged for. +// - anything else: transient. Retry on the same backoff, but log the actual +// error each time rather than the reassuring "not ready" line, because +// these are the ones that might need a human. +func fetchByJwtIfEmpty(ctx context.Context, byJwt *string, f credentialFetcher, initial, max time.Duration) error { + if *byJwt != "" { + return nil + } + + // Guards against a zero or negative interval turning the loop below into a + // busy wait against the server (time.After fires immediately, and doubling + // zero stays zero). + if initial <= 0 { + initial = time.Second + } + if max < initial { + max = initial + } + + log.Printf("egress-prober: no -by-jwt (or UR_PROBER_BY_JWT) was supplied; asking the server for the prober credential") + + backoff := initial + for { + cred, err := f.ProberCredential(ctx) + switch { + case err == nil: + *byJwt = cred.ByClientJwt + // The client id, never the jwt: this line goes to journald, and + // the jwt is a credential. The id is what an operator needs to + // match the prober against the server's record of it. + log.Printf("egress-prober: got the prober credential from the server for client %s", cred.ClientId) + return nil + case errors.Is(err, ingest.ErrUnauthorized): + return fmt.Errorf("the server rejected the operator secret when asked for the prober credential; check -operator-secret against ingest_secret in the server's provider_egress.yml: %w", err) + case errors.Is(err, ingest.ErrCredentialNotReady): + log.Printf("egress-prober: the server has not minted the prober credential yet; asking again in %s (its bootstrap task runs every 6h)", backoff) + default: + log.Printf("egress-prober: could not get the prober credential: %s; asking again in %s", err, backoff) + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + backoff = nextBackoff(backoff, max) + } +} + // envFallback fills *value from the named environment variable when the flag // was not given. Reading the environment here, rather than through the flag's // default, is what keeps the value out of flag.Usage() output. diff --git a/ingest/credential.go b/ingest/credential.go new file mode 100644 index 0000000..4cf5454 --- /dev/null +++ b/ingest/credential.go @@ -0,0 +1,126 @@ +package ingest + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" +) + +// ErrCredentialNotReady reports that the server has not minted the prober's +// network client credential yet (404). +// +// This is a WAIT, and it is deliberately its own sentinel rather than a member +// of either family already in this package: +// +// - It is NOT ErrDueUnsupported/ErrAttemptUnsupported ("this server is older, +// carry on without the feature"). There is nothing to carry on without: no +// jwt means no tunnel, so there is no degraded mode to fall back to. +// - It is NOT ErrCredentialUnavailable ("something went wrong, try again"). +// A 404 is the server working correctly and saying "not yet" -- its +// bootstrap task runs every 6h, so the first prober to start legitimately +// arrives before the credential exists. Reporting that as a failure would +// train an operator to ignore the one line that means a real fault. +// +// The caller polls on this and on the retryable sentinel below, and on nothing +// else. +var ErrCredentialNotReady = errors.New("ingest: the server has not minted the prober credential yet") + +// ErrCredentialUnavailable reports that the prober credential could not be +// fetched for any reason that is not a 404 and not a rejected operator secret: +// an unreachable server, a 5xx, an undecodable body, or a 200 that carries no +// usable jwt. +// +// Every one of those is retryable, which is why they share one sentinel: the +// caller's response to all of them is the same backoff it uses for a 404. What +// must NOT share it is the 401 -- see ErrUnauthorized, returned bare below, so +// that a caller which retries on ErrCredentialUnavailable cannot end up +// retrying a wrong secret forever. +var ErrCredentialUnavailable = errors.New("ingest: could not get the prober credential from the server") + +// ProberCredential is the server's answer from GET /network/prober-credential: +// the network client jwt the prober authenticates its tunnels with, and the +// client id that jwt belongs to. +// +// The field tags are the fixed contract of that endpoint's result. Note +// by_client_jwt, not by_jwt: the prober's own flag and env var are named +// -by-jwt / UR_PROBER_BY_JWT, so the wire name and the local name differ by one +// word, and getting it wrong yields a 200 that decodes cleanly into an empty +// string. That is why the method refuses an empty jwt below rather than +// returning it -- a silent empty would surface much later as an unparseable +// jwt or a refused tunnel, far from the typo that caused it. +type ProberCredential struct { + ByClientJwt string `json:"by_client_jwt"` + ClientId string `json:"client_id"` +} + +// ProberCredential fetches the prober's own network client credential. +// +// It authenticates with X-UR-Operator-Secret, the same header and the same +// secret as Due, Submit, ReportAttempt and GeolocationPins: one secret, one +// mechanism, one thing for a deployment to get right. That is what makes an +// unattended prober possible at all -- the operator secret is already in the +// deployment, so the jwt no longer has to be provisioned by hand and pasted +// into the environment. +// +// Four outcomes, deliberately disjoint under errors.Is: +// +// 200 -> the credential +// 404 -> ErrCredentialNotReady (wait and ask again) +// 401 -> ErrUnauthorized (stop; the deployment is misconfigured) +// else -> ErrCredentialUnavailable (retry) +func (c *Client) ProberCredential(ctx context.Context) (*ProberCredential, error) { + url := strings.TrimRight(c.ServerURL, "/") + "/network/prober-credential" + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("%w: %w", ErrCredentialUnavailable, err) + } + req.Header.Set("X-UR-Operator-Secret", c.OperatorSecret) + + resp, err := c.httpClient().Do(req) + if err != nil { + // %w, not %s, for the same reason GeolocationPins does it: a caller + // triaging a shutdown needs errors.Is(err, context.Canceled) to + // survive this wrapping. The startup poll built on this can sit for + // hours waiting on the bootstrap task, so an interrupt landing + // mid-request is the ordinary case here rather than a corner one. + return nil, fmt.Errorf("%w: %w", ErrCredentialUnavailable, err) + } + defer resp.Body.Close() + + switch resp.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return nil, ErrCredentialNotReady + case http.StatusUnauthorized: + // Bare, exactly as Due and ReportAttempt return it. Wrapping it in + // ErrCredentialUnavailable as well (which is what GeolocationPins + // does) would be wrong HERE specifically: this error is the one the + // caller branches on, and a 401 that also matched the retryable + // sentinel would be retried forever by a caller that happened to test + // the retryable case first -- the silent misconfiguration + // ErrUnauthorized exists to make loud. + return nil, ErrUnauthorized + default: + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("%w: status %d: %s", ErrCredentialUnavailable, resp.StatusCode, strings.TrimSpace(string(msg))) + } + + var cred ProberCredential + if err := json.NewDecoder(resp.Body).Decode(&cred); err != nil { + return nil, fmt.Errorf("%w: decoding the response: %s", ErrCredentialUnavailable, err) + } + // A body of `null`, of `{}`, or one keyed by anything other than + // by_client_jwt all decode without error into a zero-value struct, so + // without this check every one of them would read as a successful fetch of + // an empty jwt. Same lesson as GeolocationPins refusing a nil pin map: + // decodable is not the same as usable, and the gap has to close here + // rather than in the caller. + if strings.TrimSpace(cred.ByClientJwt) == "" { + return nil, fmt.Errorf("%w: the server answered 200 with no by_client_jwt", ErrCredentialUnavailable) + } + return &cred, nil +} diff --git a/ingest/credential_test.go b/ingest/credential_test.go new file mode 100644 index 0000000..c79a6fd --- /dev/null +++ b/ingest/credential_test.go @@ -0,0 +1,237 @@ +package ingest + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +// credentialServer mirrors pinServer: it serves one status and one body at the +// credential path, 404s everything else, and records the operator secret it +// was sent. +func credentialServer(t *testing.T, status int, body string, gotSecret *string, calls *int) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/network/prober-credential" { + http.NotFound(w, r) + return + } + if calls != nil { + *calls++ + } + if gotSecret != nil { + *gotSecret = r.Header.Get("X-UR-Operator-Secret") + } + if r.Method != http.MethodGet { + t.Errorf("credential request method = %s, want GET", r.Method) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + })) +} + +// TestProberCredentialDecodesTheServedCredential is the happy path, and it is +// also what holds the WIRE FIELD NAME. The body here is written from the +// server's documented result shape, not from the Go struct: by_client_jwt and +// client_id. A struct tag derived from the local name instead (-by-jwt / +// UR_PROBER_BY_JWT, so "by_jwt" is the natural typo) would decode this body +// into an empty string, which is why the assertion is on the value rather than +// only on err == nil. +func TestProberCredentialDecodesTheServedCredential(t *testing.T) { + var secret string + srv := credentialServer(t, http.StatusOK, + `{"by_client_jwt":"eyJhbGciOiJIUzI1NiJ9.payload.sig","client_id":"019f8835-158d-6fd8-e9dd-fd0e4c6d6792"}`, + &secret, nil) + defer srv.Close() + + c := &Client{ServerURL: srv.URL, OperatorSecret: "s3cret"} + cred, err := c.ProberCredential(context.Background()) + if err != nil { + t.Fatalf("ProberCredential err = %v", err) + } + if secret != "s3cret" { + t.Errorf("operator secret header = %q, want the configured secret; this endpoint authenticates the same way as the other operator endpoints", secret) + } + if cred.ByClientJwt != "eyJhbGciOiJIUzI1NiJ9.payload.sig" { + t.Errorf("ByClientJwt = %q, want the served by_client_jwt", cred.ByClientJwt) + } + if cred.ClientId != "019f8835-158d-6fd8-e9dd-fd0e4c6d6792" { + t.Errorf("ClientId = %q, want the served client_id", cred.ClientId) + } +} + +// TestProberCredentialTrailingSlashServerURL: the other methods all build +// their url with strings.TrimRight(ServerURL, "/"), and an -api-url written +// with a trailing slash is an ordinary way to configure a deployment. +func TestProberCredentialTrailingSlashServerURL(t *testing.T) { + srv := credentialServer(t, http.StatusOK, `{"by_client_jwt":"j","client_id":"c"}`, nil, nil) + defer srv.Close() + + c := &Client{ServerURL: srv.URL + "/", OperatorSecret: "s3cret"} + if _, err := c.ProberCredential(context.Background()); err != nil { + t.Fatalf("ProberCredential err = %v; a trailing slash on ServerURL must not produce a double slash the server 404s", err) + } +} + +// TestProberCredentialNotReadyOn404 is the requirement that keeps a prober +// started before the server's 6h bootstrap task alive. +// +// The assertion that matters is not just "an error": it is that the error is +// ErrCredentialNotReady and NOTHING ELSE. A caller distinguishes three +// outcomes by errors.Is, so if 404 also matched the retryable sentinel or the +// unauthorized one, a correctly-written caller could still take the wrong +// branch. pins_test.go asserts the same kind of non-overlap for the same +// reason. +func TestProberCredentialNotReadyOn404(t *testing.T) { + srv := credentialServer(t, http.StatusNotFound, "", nil, nil) + defer srv.Close() + + c := &Client{ServerURL: srv.URL, OperatorSecret: "s3cret"} + cred, err := c.ProberCredential(context.Background()) + if cred != nil { + t.Errorf("ProberCredential returned a credential %+v alongside the 404", cred) + } + if !errors.Is(err, ErrCredentialNotReady) { + t.Fatalf("err = %v, want ErrCredentialNotReady so the prober waits for the bootstrap task instead of exiting", err) + } + if errors.Is(err, ErrUnauthorized) { + t.Error("the 404 also matches ErrUnauthorized; a not-yet-bootstrapped server would be reported as a misconfigured secret and the prober would exit instead of waiting") + } + if errors.Is(err, ErrCredentialUnavailable) { + t.Error("the 404 also matches ErrCredentialUnavailable; \"not ready\" would be logged as a fetch failure, hiding the one message that means a real fault") + } + // The two "older server, carry on without it" sentinels. There is no + // carrying on without a jwt, so matching either would be a fail-open path. + if errors.Is(err, ErrDueUnsupported) || errors.Is(err, ErrAttemptUnsupported) { + t.Error("the 404 matches a sentinel that means \"degrade and continue\"; there is no degraded mode without a jwt") + } +} + +// TestProberCredentialUnauthorizedOn401: a wrong operator secret must be loud. +// It must NOT match the retryable sentinel, or a caller that retries on +// ErrCredentialUnavailable would sit re-asking a server that will never say +// yes -- the silent misconfiguration this error exists to prevent. +func TestProberCredentialUnauthorizedOn401(t *testing.T) { + srv := credentialServer(t, http.StatusUnauthorized, "", nil, nil) + defer srv.Close() + + c := &Client{ServerURL: srv.URL, OperatorSecret: "wrong"} + cred, err := c.ProberCredential(context.Background()) + if cred != nil { + t.Errorf("ProberCredential returned a credential %+v alongside the 401", cred) + } + if !errors.Is(err, ErrUnauthorized) { + t.Fatalf("err = %v, want ErrUnauthorized so the operator is told which secret to check", err) + } + if errors.Is(err, ErrCredentialNotReady) { + t.Error("the 401 also matches ErrCredentialNotReady; a wrong secret would be polled forever as though the bootstrap task had not run") + } + if errors.Is(err, ErrCredentialUnavailable) { + t.Error("the 401 also matches ErrCredentialUnavailable, which the caller retries; a wrong secret must never be retried forever") + } +} + +// TestProberCredentialRejectsAnUnusableBody covers every 200 that is not a +// usable credential. +// +// The `{"by_jwt":...}` case is the one worth naming: it is what a wrong struct +// tag looks like from the wire side, it decodes without error, and without the +// emptiness check it would be returned as a successful fetch of "". The prober +// would then fail far away from here, at parseByJwtClientId or at the tunnel, +// with an error that says nothing about the real cause. +func TestProberCredentialRejectsAnUnusableBody(t *testing.T) { + for _, tc := range []struct { + name string + body string + }{ + {name: "not json", body: `{"by_client_jwt":`}, + {name: "not an object", body: `["nope"]`}, + {name: "null", body: `null`}, + {name: "empty object", body: `{}`}, + {name: "empty jwt", body: `{"by_client_jwt":"","client_id":"c"}`}, + {name: "blank jwt", body: `{"by_client_jwt":" ","client_id":"c"}`}, + {name: "wrong field name (by_jwt)", body: `{"by_jwt":"a.b.c","client_id":"c"}`}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := credentialServer(t, http.StatusOK, tc.body, nil, nil) + defer srv.Close() + + c := &Client{ServerURL: srv.URL, OperatorSecret: "s3cret"} + cred, err := c.ProberCredential(context.Background()) + if err == nil { + t.Fatalf("body %s returned credential %+v and no error; a 200 that carries no usable jwt must not read as a successful fetch", tc.body, cred) + } + if cred != nil { + t.Errorf("body %s returned both an error and a credential %+v", tc.body, cred) + } + if !errors.Is(err, ErrCredentialUnavailable) { + t.Errorf("body %s: err = %v, want it to wrap ErrCredentialUnavailable so the caller retries", tc.body, err) + } + if errors.Is(err, ErrCredentialNotReady) { + t.Errorf("body %s: err matches ErrCredentialNotReady, but the server answered 200; a broken body is not \"not yet\"", tc.body) + } + }) + } +} + +// TestProberCredentialRetryableOnServerErrors: everything that is not 200, 404 +// or 401 is a transient server fault the prober should keep asking through. +func TestProberCredentialRetryableOnServerErrors(t *testing.T) { + for _, status := range []int{ + http.StatusInternalServerError, + http.StatusBadGateway, + http.StatusServiceUnavailable, + http.StatusTooManyRequests, + http.StatusForbidden, + http.StatusNoContent, + } { + srv := credentialServer(t, status, "upstream is having a moment", nil, nil) + c := &Client{ServerURL: srv.URL, OperatorSecret: "s3cret"} + _, err := c.ProberCredential(context.Background()) + srv.Close() + + if !errors.Is(err, ErrCredentialUnavailable) { + t.Errorf("status %d: err = %v, want ErrCredentialUnavailable", status, err) + } + if errors.Is(err, ErrUnauthorized) || errors.Is(err, ErrCredentialNotReady) { + t.Errorf("status %d: err = %v matches a sentinel with different handling", status, err) + } + } +} + +// An unreachable server is retryable too: the prober may well start before the +// api does. +func TestProberCredentialFailsOnAnUnreachableServer(t *testing.T) { + c := &Client{ServerURL: "http://127.0.0.1:1", OperatorSecret: "s3cret"} + cred, err := c.ProberCredential(context.Background()) + if err == nil { + t.Fatalf("an unreachable server returned credential %+v and no error", cred) + } + if !errors.Is(err, ErrCredentialUnavailable) { + t.Errorf("err = %v, want ErrCredentialUnavailable", err) + } +} + +// TestProberCredentialPreservesTheTransportCause: the poll loop built on this +// can sit for hours, so an interrupt landing mid-request is ordinary. Wrapped +// with %s the cancellation would be invisible to errors.Is and a SIGTERM +// during the wait would be reported as a broken deployment. +func TestProberCredentialPreservesTheTransportCause(t *testing.T) { + srv := credentialServer(t, http.StatusOK, `{"by_client_jwt":"j","client_id":"c"}`, nil, nil) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + c := &Client{ServerURL: srv.URL, OperatorSecret: "s3cret"} + _, err := c.ProberCredential(ctx) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want it to wrap context.Canceled so an interrupted wait is distinguishable from an unreachable server", err) + } + if !errors.Is(err, ErrCredentialUnavailable) { + t.Errorf("err = %v, want it to also carry ErrCredentialUnavailable", err) + } +} From d79b20a325533da17fe9e6e8b90727805b2fc29a Mon Sep 17 00:00:00 2001 From: Ryanmello07 <67509637+Ryanmello07@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:08:09 +0100 Subject: [PATCH 5/5] test(credential): hold the decode-error branch with a type-mismatch body encoding/json populates the fields it could before returning an UnmarshalTypeError, so a body whose client_id is not a string yields a usable-looking jwt alongside the error. Every other unusable body is backstopped by the emptiness check; this is the only case that fails if the decode error is ignored. --- ingest/credential_test.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/ingest/credential_test.go b/ingest/credential_test.go index c79a6fd..eb52f31 100644 --- a/ingest/credential_test.go +++ b/ingest/credential_test.go @@ -149,6 +149,13 @@ func TestProberCredentialRejectsAnUnusableBody(t *testing.T) { }{ {name: "not json", body: `{"by_client_jwt":`}, {name: "not an object", body: `["nope"]`}, + // The one case the emptiness check cannot backstop, and so the only + // one that actually holds the decode-error branch: encoding/json + // reports a type mismatch but still populates the fields it could, so + // a usable-looking jwt arrives alongside the error. Ignoring the error + // here would return a plausible credential from a body the server did + // not mean to send. + {name: "client_id is not a string", body: `{"by_client_jwt":"a.b.c","client_id":5}`}, {name: "null", body: `null`}, {name: "empty object", body: `{}`}, {name: "empty jwt", body: `{"by_client_jwt":"","client_id":"c"}`},