From 60980b5eb6d2522ec816f889480430aeed6e4b7f Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 13 Aug 2026 17:59:14 +0100 Subject: [PATCH] R76: dcr singleflight, L1 no-resurrect, TLS loopback --- dynamic_client_registration.go | 26 ++++++- internal/cache/backends/hybrid.go | 36 +++++++--- internal/cache/backends/review_r76_test.go | 84 ++++++++++++++++++++++ main.go | 8 ++- review_r76_test.go | 81 +++++++++++++++++++++ review_r76_tls_test.go | 25 +++++++ types.go | 1 + 7 files changed, 249 insertions(+), 12 deletions(-) create mode 100644 internal/cache/backends/review_r76_test.go create mode 100644 review_r76_test.go create mode 100644 review_r76_tls_test.go diff --git a/dynamic_client_registration.go b/dynamic_client_registration.go index 1457075..a37151a 100644 --- a/dynamic_client_registration.go +++ b/dynamic_client_registration.go @@ -7,7 +7,9 @@ import ( "encoding/json" "fmt" "io" + "net" "net/http" + "net/url" "os" "strings" "sync" @@ -141,8 +143,10 @@ func (r *DynamicClientRegistrar) RegisterClient(ctx context.Context, registratio // Validate the endpoint URL if !strings.HasPrefix(endpoint, "https://") { - // Allow http only for localhost/development - if !strings.HasPrefix(endpoint, "http://localhost") && !strings.HasPrefix(endpoint, "http://127.0.0.1") { + // Allow http only for loopback/development endpoints. Use a + // host-boundary check (not a prefix) so e.g. + // http://localhost.evil.com isn't treated as localhost. + if !isLoopbackRegistrationEndpoint(endpoint) { return nil, fmt.Errorf("registration endpoint must use HTTPS for security") } r.logger.Infof("Warning: using insecure HTTP for registration endpoint (development only): %s", endpoint) @@ -410,3 +414,21 @@ func (r *DynamicClientRegistrar) loadCredentials() (*ClientRegistrationResponse, return &resp, nil } + +// isLoopbackRegistrationEndpoint reports whether the (plaintext) registration +// endpoint targets a loopback host, matching the hostname boundary (not a +// string prefix) so http://localhost.evil.com is not treated as localhost. +func isLoopbackRegistrationEndpoint(endpoint string) bool { + u, err := url.Parse(endpoint) + if err != nil || !u.IsAbs() { + return false + } + host := u.Hostname() + if host == "localhost" { + return true + } + if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() { + return true + } + return false +} diff --git a/internal/cache/backends/hybrid.go b/internal/cache/backends/hybrid.go index 7dfc8cc..4a68f2f 100644 --- a/internal/cache/backends/hybrid.go +++ b/internal/cache/backends/hybrid.go @@ -561,9 +561,11 @@ func (h *HybridBackend) l1BackfillWorker() { for len(h.l1BackfillBuffer) > 0 { select { case item := <-h.l1BackfillBuffer: - writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) - _ = h.primary.Set(writeCtx, item.key, item.value, item.ttl) - cancel() + if h.keyStillInSecondary(item.key) { + writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + _ = h.primary.Set(writeCtx, item.key, item.value, item.ttl) + cancel() + } default: return } @@ -574,17 +576,33 @@ func (h *HybridBackend) l1BackfillWorker() { if !ok { return } - writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) - if err := h.primary.Set(writeCtx, item.key, item.value, item.ttl); err != nil { - h.logger.Debugf("Failed to populate L1 cache from L2 for key %s: %v", redactKey(item.key), err) - } else { - h.logger.Debugf("Populated L1 cache from L2 for key: %s", redactKey(item.key)) + if h.keyStillInSecondary(item.key) { + writeCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + if err := h.primary.Set(writeCtx, item.key, item.value, item.ttl); err != nil { + h.logger.Debugf("Failed to populate L1 cache from L2 for key %s: %v", redactKey(item.key), err) + } else { + h.logger.Debugf("Populated L1 cache from L2 for key: %s", redactKey(item.key)) + } + cancel() } - cancel() } } } +// keyStillInSecondary reports whether the key still exists in the L2 cache. +// Used by the L1 backfill worker: the value was captured from L2 at read +// time, and a Delete may have run since. Writing it into L1 without this +// check would resurrect a key the caller just removed from both tiers. +func (h *HybridBackend) keyStillInSecondary(key string) bool { + if h.fallbackMode.Load() { + return true // No L2 available to check; keep behavior unchanged. + } + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + exists, err := h.secondary.Exists(ctx, key) + return err != nil || exists +} + // asyncWriteWorker processes asynchronous writes to L2 func (h *HybridBackend) asyncWriteWorker() { defer h.wg.Done() diff --git a/internal/cache/backends/review_r76_test.go b/internal/cache/backends/review_r76_test.go new file mode 100644 index 0000000..30c653f --- /dev/null +++ b/internal/cache/backends/review_r76_test.go @@ -0,0 +1,84 @@ +//go:build !yaegi + +package backends + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// gatedSetBackend wraps a mockBackend whose Set blocks until release is +// closed (once). Lets the test park the backfill worker mid-flight. +type gatedSetBackend struct { + *mockBackend + startOnce sync.Once + started chan struct{} + release chan struct{} +} + +func (b *gatedSetBackend) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { + b.startOnce.Do(func() { close(b.started) }) + <-b.release + return b.mockBackend.Set(ctx, key, value, ttl) +} + +// TestHybridBackend_L1Backfill_NoResurrectAfterDelete regresses the L1 +// backfill worker resurrecting a key that was deleted after the L2 read +// (Get) captured its value. The worker must re-check L2 and not write a +// value into L1 for a key that is gone from L2. +func TestHybridBackend_L1Backfill_NoResurrectAfterDelete(t *testing.T) { + const key = "k" + + inner := newMockBackend() + gated := &gatedSetBackend{mockBackend: inner, started: make(chan struct{}), release: make(chan struct{})} + secondary := newMockBackend() + + hybrid, err := NewHybridBackend(&HybridConfig{ + Primary: gated, + Secondary: secondary, + AsyncBufferSize: 64, + }) + require.NoError(t, err) + defer hybrid.Close() + + ctx := context.Background() + + // Park the worker on a filler backfill first so `key` (enqueued after + // the filler) is processed only once we delete it. + require.NoError(t, secondary.Set(ctx, "filler", []byte("x"), time.Minute)) + _, _, exists, err := hybrid.Get(ctx, "filler") + require.NoError(t, err) + require.True(t, exists) + + select { + case <-gated.started: + case <-time.After(2 * time.Second): + t.Fatal("worker never parked on filler Set") + } + + // Now trigger a backfill for `key` (this appends it behind the parked + // filler), then delete it before the worker can reach it. + require.NoError(t, secondary.Set(ctx, key, []byte("v"), time.Minute)) + _, _, exists, err = hybrid.Get(ctx, key) + require.NoError(t, err) + require.True(t, exists) + _, err = hybrid.Delete(ctx, key) + require.NoError(t, err) + + close(gated.release) + + // Wait for the worker to drain, failing if the deleted key ever appears + // in L1 (resurrection). On the fixed code it re-checks L2 and skips; + // on the buggy code it writes the stale value shortly after release. + deadline := time.Now().Add(1 * time.Second) + for time.Now().Before(deadline) { + if _, _, ok, _ := inner.Get(ctx, key); ok { + t.Fatal("deleted key was resurrected into L1 by the backfill worker") + } + time.Sleep(5 * time.Millisecond) + } +} diff --git a/main.go b/main.go index e9b95c2..e20c1b1 100644 --- a/main.go +++ b/main.go @@ -678,10 +678,16 @@ func (t *TraefikOidc) updateMetadataEndpoints(metadata *ProviderMetadata) { t.logger.Debugf("Dynamic client registration endpoint discovered: %s", registrationURL) } - // Perform Dynamic Client Registration if enabled and ClientID is not set + // Perform Dynamic Client Registration if enabled and ClientID is not set. + // dcrMu serializes the check and the registration so the concurrent + // metadata-refresh goroutine can't pass the same gate and register a + // second client (and so the clientID read here can't race the write in + // performDynamicClientRegistration). + t.dcrMu.Lock() if t.dcrConfig != nil && t.dcrConfig.Enabled && t.clientID == "" { t.performDynamicClientRegistration() } + t.dcrMu.Unlock() } // performDynamicClientRegistration performs automatic client registration with the OIDC provider diff --git a/review_r76_test.go b/review_r76_test.go new file mode 100644 index 0000000..920934a --- /dev/null +++ b/review_r76_test.go @@ -0,0 +1,81 @@ +package traefikoidc + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync" + + "testing" + "time" +) + +// TestDCR_ConcurrentGate_RegistersOnce regresses the data race and double +// dynamic client registration when the init goroutine and the metadata +// refresh goroutine both evaluate the DCR gate concurrently. The dcrMu guard +// must allow exactly one RegistrationRequest. +func TestDCR_ConcurrentGate_RegistersOnce(t *testing.T) { + var mu sync.Mutex + var regs int + entered := make(chan struct{}) + release := make(chan struct{}) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + regs++ + mu.Unlock() + select { + case <-entered: + default: + close(entered) + } + <-release + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"client_id":"reg-client","client_secret":"s","client_secret_expires_at":0}`) + })) + defer srv.Close() + + logger := NewLogger("error") + dcr := &DynamicClientRegistrationConfig{Enabled: true, RegistrationEndpoint: srv.URL, ClientMetadata: &ClientRegistrationMetadata{RedirectURIs: []string{"https://app.example.com/callback"}}} + oidc := &TraefikOidc{ + logger: logger, + providerURL: "https://provider.example.com", + dcrConfig: dcr, + ctx: context.Background(), + httpClient: srv.Client(), + dynamicClientRegistrar: NewDynamicClientRegistrar(srv.Client(), logger, dcr, "https://provider.example.com"), + } + + md := &ProviderMetadata{ + Issuer: "https://provider.example.com", + AuthURL: "https://provider.example.com/auth", + TokenURL: "https://provider.example.com/token", + JWKSURL: "https://provider.example.com/jwks", + RegistrationURL: srv.URL, + } + + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); oidc.updateMetadataEndpoints(md) }() + + // Wait until the first registration has reached the server (and is + // blocked), then let a second goroutine evaluate the gate concurrently. + <-entered + + wg.Add(1) + go func() { defer wg.Done(); oidc.updateMetadataEndpoints(md) }() + + // On the unfixed code the second goroutine sees clientID=="" and + // registers a second client; on the fix it blocks on dcrMu and skips. + time.Sleep(500 * time.Millisecond) + close(release) + wg.Wait() + + mu.Lock() + n := regs + mu.Unlock() + if n != 1 { + t.Fatalf("want exactly 1 client registration under concurrent gate eval, got %d", n) + } +} diff --git a/review_r76_tls_test.go b/review_r76_tls_test.go new file mode 100644 index 0000000..3d86b4f --- /dev/null +++ b/review_r76_tls_test.go @@ -0,0 +1,25 @@ +package traefikoidc + +import "testing" + +// TestIsLoopbackRegistrationEndpoint_HostBoundary checks the HTTP-allowed +// gate is host-bounded: http://localhost and http://127.0.0.1 pass, but +// http://localhost.evil.com (a remote host with an unfortunate name) must +// not be mistaken for loopback. +func TestIsLoopbackRegistrationEndpoint_HostBoundary(t *testing.T) { + cases := []struct { + in string + want bool + }{ + {"http://localhost/reg", true}, + {"http://127.0.0.1:8080/reg", true}, + {"http://localhost.evil.com/reg", false}, + {"http://127.0.0.1.evil.com/reg", false}, + {"http://provider.example.com/reg", false}, + } + for _, c := range cases { + if got := isLoopbackRegistrationEndpoint(c.in); got != c.want { + t.Fatalf("isLoopbackRegistrationEndpoint(%q) = %v, want %v", c.in, got, c.want) + } + } +} diff --git a/types.go b/types.go index 2cc5430..1e94984 100644 --- a/types.go +++ b/types.go @@ -179,6 +179,7 @@ type TraefikOidc struct { refreshGracePeriod time.Duration maxRefreshTokenAge time.Duration metadataMu sync.RWMutex + dcrMu sync.Mutex shutdownOnce sync.Once sessionInvalidationCache CacheInterface refreshResultCache CacheInterface