test(ingestion): assert the bare token is absent from both auth log lines
PR #181 review (comment 15483): - TestBuildMuxLogsARESTAuthRejection asserted only that the "Bearer <tok>" header form was absent, which a field-logging chassis could never emit. Replaced by TestBuildMuxRejectionLogLineOmitsTheBarePresentedToken (a sentinel wrong token; neither it nor the static token may appear) and TestBuildMuxAcceptedLogLineOmitsTheBareStaticToken. A local chassis v0.7.1 copy logging rawToken as token_fp (deny, then accept) turns each red. - labelRESTPattern doc: the label bound holds for refused requests only; the authenticated-404 raw-path gap is hyperguild#185. - Remove internal/startupretry: nothing imports it since the chassis bump. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -79,8 +79,12 @@ func mountREST(mux *http.ServeMux, h *api.Handler, auth func(http.Handler) http.
|
|||||||
// labelRESTPattern sets the request's Pattern to the REST route it addresses
|
// labelRESTPattern sets the request's Pattern to the REST route it addresses
|
||||||
// before auth runs, so the metrics middleware (which reads Pattern from the
|
// before auth runs, so the metrics middleware (which reads Pattern from the
|
||||||
// same *http.Request) labels a refused request by its route rather than by
|
// same *http.Request) labels a refused request by its route rather than by
|
||||||
// the "/" mount (hyperguild#155). A path no route matches keeps "/", so the
|
// the "/" mount (hyperguild#155). A refused request to a path no route
|
||||||
// label set stays bounded by the registered routes.
|
// matches keeps "/", so for refused requests the label set stays bounded by
|
||||||
|
// the registered routes. That bound does not hold for authenticated
|
||||||
|
// requests: an unregistered path reaches the REST mux, which 404s with an
|
||||||
|
// empty Pattern, and the metrics middleware falls back to the raw URL path
|
||||||
|
// (hyperguild#185).
|
||||||
func labelRESTPattern(rest *http.ServeMux, next http.Handler) http.Handler {
|
func labelRESTPattern(rest *http.ServeMux, next http.Handler) http.Handler {
|
||||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if _, pattern := rest.Handler(r); pattern != "" {
|
if _, pattern := rest.Handler(r); pattern != "" {
|
||||||
|
|||||||
@@ -13,20 +13,41 @@ import (
|
|||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// wrongTokSentinel is a presented credential distinctive enough that any
|
||||||
|
// occurrence in a log line is the raw token, not a coincidence.
|
||||||
|
const wrongTokSentinel = "wrong-tok-sentinel-q7zx"
|
||||||
|
|
||||||
// hyperguild#155: since #149 every REST route sits behind BearerMiddleware,
|
// hyperguild#155: since #149 every REST route sits behind BearerMiddleware,
|
||||||
// and chassis v0.2.0 refused without logging, so an unknown caller got
|
// and chassis v0.2.0 refused without logging, so an unknown caller got
|
||||||
// silent 401s. The chassis now writes an `mcp auth rejected` line.
|
// silent 401s. The chassis now writes an `mcp auth rejected` line, and the
|
||||||
func TestBuildMuxLogsARESTAuthRejection(t *testing.T) {
|
// presented token appears nowhere in the log as a bare value (PR #181
|
||||||
|
// review: asserting on the "Bearer " header form could never fail, since
|
||||||
|
// the chassis logs fields, not headers).
|
||||||
|
func TestBuildMuxRejectionLogLineOmitsTheBarePresentedToken(t *testing.T) {
|
||||||
var buf bytes.Buffer
|
var buf bytes.Buffer
|
||||||
prev := slog.Default()
|
prev := slog.Default()
|
||||||
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
|
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
|
||||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||||
|
|
||||||
code, _ := do(t, testMux(t), "POST", "/query", "Bearer wrong")
|
code, _ := do(t, testMux(t), "POST", "/query", "Bearer "+wrongTokSentinel)
|
||||||
require.Equal(t, http.StatusUnauthorized, code)
|
require.Equal(t, http.StatusUnauthorized, code)
|
||||||
assert.Contains(t, buf.String(), "mcp auth rejected")
|
assert.Contains(t, buf.String(), "mcp auth rejected")
|
||||||
assert.Contains(t, buf.String(), "path=/query")
|
assert.Contains(t, buf.String(), "path=/query")
|
||||||
assert.NotContains(t, buf.String(), "Bearer wrong", "the credential itself is never logged")
|
assert.NotContains(t, buf.String(), wrongTokSentinel, "the rejected token is logged as a bare value")
|
||||||
|
assert.NotContains(t, buf.String(), routesTok, "the valid static token is logged")
|
||||||
|
}
|
||||||
|
|
||||||
|
// The `mcp auth accepted` line leaves out the valid static token.
|
||||||
|
func TestBuildMuxAcceptedLogLineOmitsTheBareStaticToken(t *testing.T) {
|
||||||
|
var buf bytes.Buffer
|
||||||
|
prev := slog.Default()
|
||||||
|
slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
|
||||||
|
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||||
|
|
||||||
|
code, _ := do(t, testMux(t), "GET", "/pending", "Bearer "+routesTok)
|
||||||
|
require.Equal(t, http.StatusOK, code)
|
||||||
|
assert.Contains(t, buf.String(), "mcp auth accepted")
|
||||||
|
assert.NotContains(t, buf.String(), routesTok, "the accepted static token is logged as a bare value")
|
||||||
}
|
}
|
||||||
|
|
||||||
// An authenticated REST request is labelled with the REST mux's pattern:
|
// An authenticated REST request is labelled with the REST mux's pattern:
|
||||||
|
|||||||
@@ -1,75 +0,0 @@
|
|||||||
// Package startupretry retries a startup dependency that may not be serving
|
|
||||||
// yet, instead of exiting on its first failure.
|
|
||||||
//
|
|
||||||
// It exists for the cold-boot ordering race (#83): on a cluster restart the
|
|
||||||
// ingestion pod can reach OIDC discovery before dex is answering, gets a 502
|
|
||||||
// from ingress, and dies — after having already connected postgres, backfilled
|
|
||||||
// the graph and ingested session batches, all of which the restart redoes.
|
|
||||||
// Nothing is misconfigured in that scenario; the pod simply started first.
|
|
||||||
//
|
|
||||||
// The budget is bounded on purpose. Retrying forever would turn a genuine
|
|
||||||
// misconfiguration (wrong issuer URL, bad audience) into a process that looks
|
|
||||||
// healthy and never serves, which is harder to diagnose than a crash loop.
|
|
||||||
package startupretry
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Policy bounds the retry: at most Attempts calls, with the delay doubling
|
|
||||||
// from Base and capped at Max.
|
|
||||||
type Policy struct {
|
|
||||||
Attempts int // total calls, including the first; < 1 means 1
|
|
||||||
Base time.Duration // delay before the second attempt
|
|
||||||
Max time.Duration // ceiling for the doubling delay
|
|
||||||
}
|
|
||||||
|
|
||||||
// Do calls fn until it succeeds, the attempt budget is spent, or ctx ends.
|
|
||||||
//
|
|
||||||
// The returned error wraps the last failure, so callers keep the diagnosis
|
|
||||||
// they would have had without the retry.
|
|
||||||
func Do[T any](ctx context.Context, p Policy, fn func(context.Context) (T, error)) (T, error) {
|
|
||||||
var zero T
|
|
||||||
|
|
||||||
attempts := p.Attempts
|
|
||||||
if attempts < 1 {
|
|
||||||
attempts = 1
|
|
||||||
}
|
|
||||||
|
|
||||||
delay := p.Base
|
|
||||||
var lastErr error
|
|
||||||
|
|
||||||
for i := range attempts {
|
|
||||||
if err := ctx.Err(); err != nil {
|
|
||||||
return zero, fmt.Errorf("startupretry: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
v, err := fn(ctx)
|
|
||||||
if err == nil {
|
|
||||||
return v, nil
|
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
|
|
||||||
if i == attempts-1 {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
if delay <= 0 {
|
|
||||||
delay = time.Second
|
|
||||||
}
|
|
||||||
if p.Max > 0 && delay > p.Max {
|
|
||||||
delay = p.Max
|
|
||||||
}
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
|
||||||
return zero, fmt.Errorf("startupretry: %w", ctx.Err())
|
|
||||||
case <-time.After(delay):
|
|
||||||
}
|
|
||||||
delay *= 2
|
|
||||||
}
|
|
||||||
|
|
||||||
return zero, fmt.Errorf("startupretry: gave up after %d attempts: %w", attempts, lastErr)
|
|
||||||
}
|
|
||||||
@@ -1,142 +0,0 @@
|
|||||||
// ingestion/internal/startupretry/startupretry_test.go
|
|
||||||
package startupretry
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
func fastPolicy(attempts int) Policy {
|
|
||||||
return Policy{Attempts: attempts, Base: time.Millisecond, Max: 2 * time.Millisecond}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDo_ReturnsImmediatelyOnFirstSuccess(t *testing.T) {
|
|
||||||
calls := 0
|
|
||||||
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
|
|
||||||
calls++
|
|
||||||
return "ready", nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Do: %v", err)
|
|
||||||
}
|
|
||||||
if got != "ready" {
|
|
||||||
t.Errorf("got %q, want %q", got, "ready")
|
|
||||||
}
|
|
||||||
if calls != 1 {
|
|
||||||
t.Errorf("calls = %d, want 1", calls)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDo_SurvivesTransientFailuresThenSucceeds(t *testing.T) {
|
|
||||||
// The observed cold-boot race: dex is not serving yet, ingress answers
|
|
||||||
// 502 for the first few attempts, then discovery works.
|
|
||||||
calls := 0
|
|
||||||
got, err := Do(context.Background(), fastPolicy(5), func(context.Context) (string, error) {
|
|
||||||
calls++
|
|
||||||
if calls < 4 {
|
|
||||||
return "", errors.New("oidc discovery: status 502")
|
|
||||||
}
|
|
||||||
return "ready", nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Do returned %v — a transient 502 must not be fatal", err)
|
|
||||||
}
|
|
||||||
if got != "ready" {
|
|
||||||
t.Errorf("got %q, want %q", got, "ready")
|
|
||||||
}
|
|
||||||
if calls != 4 {
|
|
||||||
t.Errorf("calls = %d, want 4", calls)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDo_GivesUpAfterTheAttemptBudgetAndReportsTheLastError(t *testing.T) {
|
|
||||||
// A genuinely broken issuer must still fail — retrying forever would turn
|
|
||||||
// a config error into a silent hang, which is worse than crash-looping.
|
|
||||||
last := errors.New("oidc discovery: status 404")
|
|
||||||
calls := 0
|
|
||||||
|
|
||||||
_, err := Do(context.Background(), fastPolicy(3), func(context.Context) (string, error) {
|
|
||||||
calls++
|
|
||||||
return "", last
|
|
||||||
})
|
|
||||||
|
|
||||||
if !errors.Is(err, last) {
|
|
||||||
t.Fatalf("err = %v, want it to wrap the last failure", err)
|
|
||||||
}
|
|
||||||
if calls != 3 {
|
|
||||||
t.Errorf("calls = %d, want 3 (the attempt budget)", calls)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDo_StopsPromptlyWhenTheContextIsCancelled(t *testing.T) {
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
calls := 0
|
|
||||||
|
|
||||||
done := make(chan struct{})
|
|
||||||
var err error
|
|
||||||
go func() {
|
|
||||||
defer close(done)
|
|
||||||
_, err = Do(ctx, Policy{Attempts: 100, Base: 50 * time.Millisecond, Max: time.Second},
|
|
||||||
func(context.Context) (string, error) {
|
|
||||||
calls++
|
|
||||||
if calls == 1 {
|
|
||||||
cancel()
|
|
||||||
}
|
|
||||||
return "", errors.New("still down")
|
|
||||||
})
|
|
||||||
}()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-done:
|
|
||||||
case <-time.After(2 * time.Second):
|
|
||||||
t.Fatal("Do did not return after context cancellation")
|
|
||||||
}
|
|
||||||
if !errors.Is(err, context.Canceled) {
|
|
||||||
t.Errorf("err = %v, want context.Canceled", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDo_BacksOffAndIsCappedByMax(t *testing.T) {
|
|
||||||
var delays []time.Duration
|
|
||||||
p := Policy{Attempts: 5, Base: 10 * time.Millisecond, Max: 25 * time.Millisecond}
|
|
||||||
|
|
||||||
last := time.Now()
|
|
||||||
_, _ = Do(context.Background(), p, func(context.Context) (string, error) {
|
|
||||||
now := time.Now()
|
|
||||||
delays = append(delays, now.Sub(last))
|
|
||||||
last = now
|
|
||||||
return "", errors.New("down")
|
|
||||||
})
|
|
||||||
|
|
||||||
// delays[0] is the first call (no wait); the rest are the sleeps.
|
|
||||||
if len(delays) != 5 {
|
|
||||||
t.Fatalf("attempts = %d, want 5", len(delays))
|
|
||||||
}
|
|
||||||
if delays[1] >= delays[2] {
|
|
||||||
t.Errorf("backoff is not growing: %v then %v", delays[1], delays[2])
|
|
||||||
}
|
|
||||||
for i, d := range delays[1:] {
|
|
||||||
if d > 2*p.Max {
|
|
||||||
t.Errorf("delay %d = %v, exceeds the Max cap %v with slack", i+1, d, p.Max)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestPolicy_ZeroValueStillMakesOneAttempt(t *testing.T) {
|
|
||||||
// A caller that forgets to configure the policy must not silently get a
|
|
||||||
// no-op that returns a nil validator and a nil error.
|
|
||||||
calls := 0
|
|
||||||
_, err := Do(context.Background(), Policy{}, func(context.Context) (string, error) {
|
|
||||||
calls++
|
|
||||||
return "", errors.New("down")
|
|
||||||
})
|
|
||||||
|
|
||||||
if calls != 1 {
|
|
||||||
t.Errorf("calls = %d, want 1", calls)
|
|
||||||
}
|
|
||||||
if err == nil {
|
|
||||||
t.Error("err = nil, want the failure surfaced")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user