appview: stop serving service tokens past their expiry, and challenge the client when the hold rejects one

Seen in production on 2026-09-11: three cold pulls of a 22-layer image failed with BLOB_UNKNOWN for layers that exist. The hold had answered 403 "service token authentication failed: token has expired", and the same blobs served fine a minute later. Three things lined up. The registry middleware's validation cache kept a fetched service token for a flat 45 seconds regardless of its real remaining life, so a token fetched with 12 seconds left was still handed to the hold half a minute after it died. The registry JWT is stamped from the auth cache's expiry, which trailed the real exp by only 10 seconds, while distribution accepts a JWT for 60 seconds past its exp, so a client could hold an accepted JWT for most of a minute after the credential behind it was gone. And the hold's 403 was flattened to BLOB_UNKNOWN, so the client failed instead of re-authenticating. Now the validation cache bounds an entry by the token's exp minus a shared ServiceTokenSafetyMargin of 60 seconds, the same margin the auth cache and the JWT stamp use, chosen to equal distribution's leeway so the last instant a JWT is accepted is the service token's real exp. A PDS that grants less than the margin gets half its remaining life instead of an already-past deadline. When the hold rejects the service token as expired or missing, the appview drops both cached copies and returns a 401 challenge so Docker and crane re-run the token dance and retry; a genuine permission denial stays a 403, and a hold that is down still maps to blob unknown. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EvFJr4Dwz8p2NDAeXmgmBt
2026-09-29 21:45:33 +00:00 · 2026-09-11 19:26:51 -05:00
parent dcad0f8626
commit 9dbc53b670
11 changed files with 867 additions and 49 deletions
@@ -35,6 +35,21 @@ const pullerDIDKey contextKey = "puller.did"
 // hasPushScopeKey is the context key for storing whether the JWT has push scope
 const hasPushScopeKey contextKey = "token.has_push_scope"

+// validationCacheTTL is the longest a fetched service token is reused from the
+// validation cache. It covers a typical Docker push, whose many blob requests
+// would otherwise each race on OAuth/DPoP.
+//
+// It is a ceiling, not the actual lifetime: getOrFetch also clamps the entry to
+// the token's own exp minus auth.ServiceTokenSafetyMargin. Without that clamp a
+// token fetched with 12s of life left was still served for the full 45s, so the
+// hold saw an expired service token and answered 403 "token has expired" on
+// blobs that exist.
+const validationCacheTTL = 45 * time.Second
+
+// validationCacheErrorTTL is how long a failed fetch is remembered so
+// concurrent requests fast-fail instead of stampeding the PDS.
+const validationCacheErrorTTL = 5 * time.Second
+
 // validationCacheEntry stores a validated service token with expiration
 type validationCacheEntry struct {
 	serviceToken string
@@ -154,15 +169,14 @@ func (vc *validationCache) getOrFetch(ctx context.Context, cacheKey string, fetc
 	entry.inFlight = false

 	if err != nil {
-		// Cache errors for 5 seconds (fast-fail for subsequent requests)
+		// Cache errors briefly (fast-fail for subsequent requests)
 		entry.err = err
-		entry.validUntil = time.Now().Add(5 * time.Second)
+		entry.validUntil = time.Now().Add(validationCacheErrorTTL)
 		entry.serviceToken = ""
 	} else {
-		// Cache token for 45 seconds (covers typical Docker push operation)
 		entry.err = nil
 		entry.serviceToken = serviceToken
-		entry.validUntil = time.Now().Add(45 * time.Second)
+		entry.validUntil = tokenValidUntil(serviceToken)
 	}

 	// Signal completion to waiting goroutines
@@ -172,6 +186,65 @@ func (vc *validationCache) getOrFetch(ctx context.Context, cacheKey string, fetc
 	return serviceToken, err
 }

+// tokenValidUntil bounds a cached service token by its own exp claim, not just
+// by the flat validation-cache TTL.
+//
+// The cache used to pin any successful fetch for validationCacheTTL regardless
+// of how much life the token actually had. The token is minted with a fixed
+// absolute expiry, so a fetch that landed near the end of one (the auth cache
+// hands back a token until it is close to expiry, and the PDS may grant less
+// than asked) left the appview presenting a dead credential to the hold for the
+// rest of the 45s. The hold answered 403 "token has expired" and cold pulls
+// failed on layers that exist.
+//
+// A token whose exp cannot be parsed keeps the flat TTL: the appview cannot do
+// better than its previous behaviour for a token shape it does not understand,
+// and pkg/auth's cache applies the same fallback.
+func tokenValidUntil(serviceToken string) time.Time {
+	validUntil := time.Now().Add(validationCacheTTL)
+
+	exp, err := auth.ServiceTokenExpiry(serviceToken)
+	if err != nil {
+		slog.Warn("Service token exp unreadable, using flat validation cache TTL",
+			"component", "registry/middleware",
+			"error", err,
+			"ttl", validationCacheTTL)
+		return validUntil
+	}
+
+	// Same margin the auth cache and the registry JWT's exp use, so all three
+	// stop trusting the token at the same moment.
+	if safe := exp.Add(-auth.ServiceTokenSafetyMargin); safe.Before(validUntil) {
+		return safe
+	}
+	return validUntil
+}
+
+// invalidate expires the entry for cacheKey so the next getOrFetch re-mints.
+// Called when the hold rejects the token we handed it: the entry is stale by
+// definition and replaying it would fail the client's retry the same way.
+//
+// The entry is expired in place rather than deleted from the map because
+// concurrent goroutines already hold the pointer; an in-flight fetch is left
+// alone because it is about to store a fresh token anyway.
+func (vc *validationCache) invalidate(cacheKey string) {
+	vc.mu.RLock()
+	entry, exists := vc.entries[cacheKey]
+	vc.mu.RUnlock()
+
+	if !exists {
+		return
+	}
+
+	entry.mu.Lock()
+	if !entry.inFlight {
+		entry.serviceToken = ""
+		entry.err = nil
+		entry.validUntil = time.Time{}
+	}
+	entry.mu.Unlock()
+}
+
 // LabelChecker checks whether content has been taken down via ATProto labels.
 type LabelChecker interface {
 	IsTakenDown(did, repository string) (bool, error)
@@ -451,6 +524,9 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name
 	// IMPORTANT: Use PULLER's DID/PDS for service token, not owner's!
 	// The puller (authenticated user) needs to authenticate to the hold service.
 	var serviceToken string
+	// invalidateServiceToken is handed to the blob store so it can drop this
+	// token when the hold rejects it; nil unless we actually fetched one.
+	var invalidateServiceToken func()
 	authMethod, _ := ctx.Value(authMethodKey).(string)
 	pullerDID, _ := ctx.Value(pullerDIDKey).(string)
 	hasPushScope, _ := ctx.Value(hasPushScopeKey).(bool)
@@ -537,6 +613,16 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name
 				// Generic service token error
 				return nil, nr.authErrorMessage(fmt.Sprintf("Failed to obtain storage credentials: %v", fetchErr))
 			}
+
+			// Both caches have to go: the validation cache would otherwise
+			// replay the rejected token for the rest of its window, and
+			// pkg/auth's cache would hand the same one straight back to the
+			// refetch.
+			vc := nr.validationCache
+			invalidateServiceToken = func() {
+				vc.invalidate(cacheKey)
+				auth.InvalidateServiceToken(pullerDID, holdDID)
+			}
 		}
 	} else {
 		slog.Debug("Skipping service token fetch for unauthenticated request",
@@ -631,26 +717,27 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name
 	// 3. The refresher already caches sessions efficiently (in-memory + DB)
 	// 4. Caching the repository with a stale ATProtoClient causes refresh token errors
 	registryCtx := &storage.RegistryContext{
-		DID:                did,
-		Handle:             handle,
-		HoldDID:            holdDID,
-		HoldURL:            holdURL,
-		PDSEndpoint:        pdsEndpoint,
-		Repository:         repositoryName,
-		ServiceToken:       serviceToken, // Cached service token from puller's PDS
-		ATProtoClient:      atprotoClient,
-		AuthMethod:         authMethod,        // Auth method from JWT token
-		PullerDID:          pullerDID,         // Authenticated user making the request
-		PullerPDSEndpoint:  pullerPDSEndpoint, // Puller's PDS for service token refresh
-		HasPushScope:       hasPushScope,      // Whether JWT has push scope (for pull stats filtering)
-		Anonymous:          pullerDID == "",   // No puller identity: hold decides via captain.Public
-		AutoRemoveUntagged: prefs.AutoRemoveUntagged,
-		Database:           nr.database,
-		Authorizer:         nr.authorizer,
-		Refresher:          nr.refresher,
-		ReadmeFetcher:      nr.readmeFetcher,
-		WebhookDispatcher:  nr.webhookDispatcher,
-		ManifestRefChecker: nr.manifestRefChecker,
+		DID:                    did,
+		Handle:                 handle,
+		HoldDID:                holdDID,
+		HoldURL:                holdURL,
+		PDSEndpoint:            pdsEndpoint,
+		Repository:             repositoryName,
+		ServiceToken:           serviceToken, // Cached service token from puller's PDS
+		InvalidateServiceToken: invalidateServiceToken,
+		ATProtoClient:          atprotoClient,
+		AuthMethod:             authMethod,        // Auth method from JWT token
+		PullerDID:              pullerDID,         // Authenticated user making the request
+		PullerPDSEndpoint:      pullerPDSEndpoint, // Puller's PDS for service token refresh
+		HasPushScope:           hasPushScope,      // Whether JWT has push scope (for pull stats filtering)
+		Anonymous:              pullerDID == "",   // No puller identity: hold decides via captain.Public
+		AutoRemoveUntagged:     prefs.AutoRemoveUntagged,
+		Database:               nr.database,
+		Authorizer:             nr.authorizer,
+		Refresher:              nr.refresher,
+		ReadmeFetcher:          nr.readmeFetcher,
+		WebhookDispatcher:      nr.webhookDispatcher,
+		ManifestRefChecker:     nr.manifestRefChecker,
 	}

 	return storage.NewRoutingRepository(repo, registryCtx), nil
@@ -0,0 +1,151 @@
+package middleware
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"testing"
+	"time"
+
+	"atcr.io/pkg/auth"
+)
+
+// serviceTokenExpiring builds an unsigned JWT whose exp claim is the given
+// time. The validation cache only reads exp; the hold is what verifies
+// signatures.
+func serviceTokenExpiring(at time.Time) string {
+	payload := fmt.Sprintf(`{"exp":%d}`, at.Unix())
+	return "header." + base64.RawURLEncoding.EncodeToString([]byte(payload)) + ".signature"
+}
+
+// entryValidUntil reads back the cached entry's deadline.
+func entryValidUntil(t *testing.T, vc *validationCache, cacheKey string) time.Time {
+	t.Helper()
+
+	vc.mu.RLock()
+	entry, ok := vc.entries[cacheKey]
+	vc.mu.RUnlock()
+	if !ok {
+		t.Fatalf("no validation cache entry for %q", cacheKey)
+	}
+
+	entry.mu.Lock()
+	defer entry.mu.Unlock()
+	return entry.validUntil
+}
+
+// A token with only 20s of life must not be pinned for the flat 45s TTL: that
+// is the production bug, where the appview kept presenting a dead service token
+// and the hold answered 403 "token has expired" on blobs that exist.
+func TestValidationCache_BoundsEntryByTokenExpiry(t *testing.T) {
+	vc := newValidationCache()
+	cacheKey := "did:plc:puller:did:web:hold.example.com"
+
+	realExp := time.Now().Add(20 * time.Second)
+	token := serviceTokenExpiring(realExp)
+
+	got, err := vc.getOrFetch(context.Background(), cacheKey, func() (string, error) {
+		return token, nil
+	})
+	if err != nil {
+		t.Fatalf("getOrFetch() error = %v", err)
+	}
+	if got != token {
+		t.Fatalf("getOrFetch() returned %q, want the fetched token", got)
+	}
+
+	validUntil := entryValidUntil(t, vc, cacheKey)
+
+	want := realExp.Add(-auth.ServiceTokenSafetyMargin)
+	if validUntil.After(want) {
+		t.Errorf("entry valid until %v, want no later than %v (exp minus %v)",
+			validUntil, want, auth.ServiceTokenSafetyMargin)
+	}
+
+	if flat := time.Now().Add(validationCacheTTL); !validUntil.Before(flat) {
+		t.Errorf("entry valid until %v, which is the flat %v TTL: the token's own exp was ignored",
+			validUntil, validationCacheTTL)
+	}
+}
+
+// A long-lived token is still capped by the flat TTL, so the cache keeps
+// rechecking the underlying session rather than holding one token for minutes.
+func TestValidationCache_LongLivedTokenKeepsFlatTTL(t *testing.T) {
+	vc := newValidationCache()
+	cacheKey := "did:plc:puller:did:web:hold.example.com"
+
+	token := serviceTokenExpiring(time.Now().Add(1 * time.Hour))
+
+	if _, err := vc.getOrFetch(context.Background(), cacheKey, func() (string, error) {
+		return token, nil
+	}); err != nil {
+		t.Fatalf("getOrFetch() error = %v", err)
+	}
+
+	validUntil := entryValidUntil(t, vc, cacheKey)
+
+	want := time.Now().Add(validationCacheTTL)
+	if diff := validUntil.Sub(want); diff < -2*time.Second || diff > 2*time.Second {
+		t.Errorf("entry valid until %v, want ~%v (the flat TTL)", validUntil, want)
+	}
+}
+
+// A token whose exp cannot be read falls back to the behaviour that shipped
+// before: the flat TTL. The appview can't do better for a shape it doesn't
+// understand, and refusing to cache would stampede the PDS.
+func TestValidationCache_UnparsableTokenKeepsFlatTTL(t *testing.T) {
+	vc := newValidationCache()
+	cacheKey := "did:plc:puller:did:web:hold.example.com"
+
+	if _, err := vc.getOrFetch(context.Background(), cacheKey, func() (string, error) {
+		return "not-a-jwt", nil
+	}); err != nil {
+		t.Fatalf("getOrFetch() error = %v", err)
+	}
+
+	validUntil := entryValidUntil(t, vc, cacheKey)
+
+	want := time.Now().Add(validationCacheTTL)
+	if diff := validUntil.Sub(want); diff < -2*time.Second || diff > 2*time.Second {
+		t.Errorf("entry valid until %v, want ~%v (the flat TTL)", validUntil, want)
+	}
+}
+
+// invalidate must force the next call to re-mint, which is what makes the 401
+// we send to Docker worth retrying.
+func TestValidationCache_InvalidateForcesRefetch(t *testing.T) {
+	vc := newValidationCache()
+	cacheKey := "did:plc:puller:did:web:hold.example.com"
+
+	fetches := 0
+	fetch := func() (string, error) {
+		fetches++
+		return serviceTokenExpiring(time.Now().Add(5 * time.Minute)), nil
+	}
+
+	if _, err := vc.getOrFetch(context.Background(), cacheKey, fetch); err != nil {
+		t.Fatalf("getOrFetch() error = %v", err)
+	}
+	if _, err := vc.getOrFetch(context.Background(), cacheKey, fetch); err != nil {
+		t.Fatalf("getOrFetch() error = %v", err)
+	}
+	if fetches != 1 {
+		t.Fatalf("second call fetched again (%d fetches), expected a cache hit", fetches)
+	}
+
+	vc.invalidate(cacheKey)
+
+	if _, err := vc.getOrFetch(context.Background(), cacheKey, fetch); err != nil {
+		t.Fatalf("getOrFetch() after invalidate error = %v", err)
+	}
+	if fetches != 2 {
+		t.Errorf("got %d fetches after invalidate, want 2", fetches)
+	}
+}
+
+// invalidate on a key that was never cached is a no-op, not a panic: the blob
+// store calls it on whatever request hit the rejection.
+func TestValidationCache_InvalidateUnknownKey(t *testing.T) {
+	vc := newValidationCache()
+	vc.invalidate("did:plc:nobody:did:web:hold.example.com")
+}