fix(auth): let concurrent agent processes refresh one MCP credential

docs/authentication.md

@@ -247,7 +247,7 @@ it only with a provider you administer.
 | Browser session | Interactive access | Idle and absolute TTL; revoked on role, email, or status change |
 | Ingest token | OTLP ingest | Settings page |
 | Metrics token | `/metrics` scrape | `FANOUT_METRICS_TOKEN` |
-| MCP OAuth tokens | MCP clients | Refresh rotation with replay-family revocation |
+| MCP OAuth tokens | MCP clients | One-hour access tokens; refresh rotation with replay-family revocation, admitting the same client's just-rotated token for one minute so processes sharing a credential can refresh together |
 
 Local email codes allow three attempts per code and expire after five minutes.
 Attempts are reserved atomically, so concurrent guesses cannot exceed the limit.

internal/api/oauth_test.go

@@ -700,10 +700,11 @@ func TestMCPOAuthRefreshGrantOverHTTP(t *testing.T) {
 		t.Fatalf("MCP with rotated token = %d %s", mcp.Code, mcp.Body.String())
 	}
 
-	// Replaying the consumed refresh token is invalid_grant at the HTTP layer.
-	rec = serve(t, e, http.MethodPost, "/oauth/token", rotate.Encode(), formHeaders)
-	if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "invalid_grant") {
-		t.Fatalf("refresh replay = %d %s, want 400 invalid_grant", rec.Code, rec.Body.String())
+	// A concurrent sibling replaying the consumed refresh token within the
+	// grace window gets its own pair; later reuse is covered by the store tests.
+	sibling := decodeTokens(t, serve(t, e, http.MethodPost, "/oauth/token", rotate.Encode(), formHeaders))
+	if sibling["refresh_token"] == rotated["refresh_token"] || sibling["access_token"] == rotated["access_token"] {
+		t.Fatalf("sibling refresh reused credentials: %v", sibling)
 	}
 }
 

internal/auth/oauth_store.go

@@ -16,9 +16,15 @@ import (
 )
 
 const (
-	OAuthAccessTTL  = 15 * time.Minute
+	// Access tokens are checked against the database on every request, so
+	// revocation is immediate whatever their lifetime.
+	OAuthAccessTTL  = time.Hour
 	OAuthRefreshTTL = 30 * 24 * time.Hour
 	OAuthCodeTTL    = 5 * time.Minute
+	// OAuthRefreshReuseGrace admits a refresh token its own client rotated
+	// moments ago, as when agent processes sharing one stored credential
+	// refresh together; later reuse still revokes the family.
+	OAuthRefreshReuseGrace = time.Minute
 )
 
 var (
@@ -273,8 +279,18 @@ func (s *OAuthStore) RotateRefreshToken(ctx context.Context, clientID, raw, reso
 	}
 	// Reuse detection runs BEFORE the expiry check: replaying a refresh token
 	// that was rotated and has since expired is still evidence of theft and
-	// must revoke the family, not fall through to a plain invalid grant.
-	if record.RevokedAt.Valid {
+	// must revoke the family, not fall through to a plain invalid grant. Only
+	// a rotation within the grace window, in a family that is still live, is
+	// a concurrent sibling refresh; it gets its own pair in the family.
+	sibling := false
+	// A rotation dated after now (a clock stepped back) is never in the window,
+	// and an expired rotated token is reuse, not a sibling.
+	if elapsed := now - record.RevokedAt.Int64; record.RevokedAt.Valid && elapsed >= 0 && elapsed <= int64(OAuthRefreshReuseGrace/time.Second) && record.ExpiresAt.Unix() > now {
+		if sibling, err = liveRefreshFamily(ctx, conn, record.FamilyID, now); err != nil {
+			return OAuthTokenPair{}, err
+		}
+	}
+	if record.RevokedAt.Valid && !sibling {
 		// The family revocation is committed BEFORE returning the error on
 		// purpose: the call fails, but the revocation must persist — rolling
 		// it back with the failed transaction would undo the defense.
@@ -326,8 +342,11 @@ func (s *OAuthStore) RotateRefreshToken(ctx context.Context, clientID, raw, reso
 	if !ok {
 		return OAuthTokenPair{}, ErrInvalidOAuthScope
 	}
-	if _, err := conn.ExecContext(ctx, `UPDATE oauth_tokens SET revoked_at = ? WHERE token_hash = ?`, now, oauthHash(raw)); err != nil {
-		return OAuthTokenPair{}, fmt.Errorf("oauth: rotate refresh token: %w", err)
+	// A sibling's token keeps its rotation time, so the grace window cannot extend.
+	if !sibling {
+		if _, err := conn.ExecContext(ctx, `UPDATE oauth_tokens SET revoked_at = ? WHERE token_hash = ?`, now, oauthHash(raw)); err != nil {
+			return OAuthTokenPair{}, fmt.Errorf("oauth: rotate refresh token: %w", err)
+		}
 	}
 	pair, err := s.insertTokenPair(ctx, conn, record.FamilyID, record.ClientID, record.UserID, rotationScope, record.Resource)
 	if err != nil {
@@ -444,6 +463,18 @@ func (s *OAuthStore) insertTokenPair(ctx context.Context, db oauthDB, familyID,
 	return OAuthTokenPair{AccessToken: access, RefreshToken: refresh, ExpiresIn: int64(OAuthAccessTTL.Seconds()), Scope: scope}, nil
 }
 
+func liveRefreshFamily(ctx context.Context, db oauthDB, familyID string, now int64) (bool, error) {
+	var live bool
+	if err := db.QueryRowContext(ctx, `
+		SELECT EXISTS (SELECT 1 FROM oauth_tokens
+		WHERE family_id = ? AND kind = ? AND revoked_at IS NULL AND expires_at > ?)`,
+		familyID, string(TokenKindRefresh), now,
+	).Scan(&live); err != nil {
+		return false, fmt.Errorf("oauth: read refresh family: %w", err)
+	}
+	return live, nil
+}
+
 func readOAuthToken(ctx context.Context, db oauthDB, raw string) (OAuthTokenRecord, error) {
 	var record OAuthTokenRecord
 	var expires int64

internal/auth/oauth_store_test.go

@@ -82,6 +82,8 @@ func TestOAuthStoreRefreshRotationAndReuseRevokesFamily(t *testing.T) {
 		t.Fatalf("Create user: %v", err)
 	}
 	store := NewOAuthStore(sqlite.DB)
+	base := time.Now().UTC()
+	store.now = func() time.Time { return base }
 	client, err := store.RegisterClient(t.Context(), "Claude Code", "", []string{"http://localhost:8765/callback"})
 	if err != nil {
 		t.Fatalf("RegisterClient: %v", err)
@@ -98,6 +100,7 @@ func TestOAuthStoreRefreshRotationAndReuseRevokesFamily(t *testing.T) {
 	if second.RefreshToken == first.RefreshToken || second.AccessToken == first.AccessToken {
 		t.Fatal("rotation must issue fresh credentials")
 	}
+	store.now = func() time.Time { return base.Add(OAuthRefreshReuseGrace + time.Second) }
 	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
 		t.Fatalf("reused refresh = %v, want reuse detection", err)
 	}
@@ -106,6 +109,167 @@ func TestOAuthStoreRefreshRotationAndReuseRevokesFamily(t *testing.T) {
 	}
 }
 
+// Agent processes sharing one stored credential refresh together; the
+// sibling that presents the just-rotated token gets its own pair, once per
+// grace window, and only from the same client while the family is live.
+func TestOAuthStoreRefreshGraceAdmitsConcurrentSibling(t *testing.T) {
+	sqlite := newTestSQLite(t)
+	users := NewUserStore(sqlite.DB)
+	user, err := users.CreateWithAudit("sibling@example.com", "", "admin", AuditEvent{EventType: "user.created", Outcome: "success"})
+	if err != nil {
+		t.Fatalf("Create user: %v", err)
+	}
+	store := NewOAuthStore(sqlite.DB)
+	base := time.Now().UTC()
+	store.now = func() time.Time { return base }
+	client, err := store.RegisterClient(t.Context(), "Codex", "", []string{"http://127.0.0.1:14550/callback"})
+	if err != nil {
+		t.Fatalf("RegisterClient: %v", err)
+	}
+	other, err := store.RegisterClient(t.Context(), "Other", "", []string{"http://127.0.0.1:14551/callback"})
+	if err != nil {
+		t.Fatalf("RegisterClient: %v", err)
+	}
+	resource := "https://fanout.example.com/mcp"
+	first, err := store.IssueTokenPair(t.Context(), client.ClientID, user.ID, MCPScopeTelemetryRead, resource)
+	if err != nil {
+		t.Fatalf("IssueTokenPair: %v", err)
+	}
+	second, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, "")
+	if err != nil {
+		t.Fatalf("RotateRefreshToken: %v", err)
+	}
+	if _, err := store.RotateRefreshToken(t.Context(), other.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrInvalidOAuthGrant) {
+		t.Fatalf("another client's sibling refresh = %v, want invalid grant", err)
+	}
+	store.now = func() time.Time { return base.Add(OAuthRefreshReuseGrace) }
+	sibling, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, "")
+	if err != nil {
+		t.Fatalf("sibling refresh within grace = %v", err)
+	}
+	if sibling.RefreshToken == second.RefreshToken || sibling.AccessToken == second.AccessToken {
+		t.Fatal("a sibling must receive its own credentials")
+	}
+	for name, pair := range map[string]OAuthTokenPair{"rotated": second, "sibling": sibling} {
+		if _, err := store.VerifyAccessToken(t.Context(), pair.AccessToken, resource); err != nil {
+			t.Fatalf("%s access token = %v", name, err)
+		}
+	}
+	// The sibling refresh must not move the rotation time and extend the window.
+	store.now = func() time.Time { return base.Add(OAuthRefreshReuseGrace + time.Second) }
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
+		t.Fatalf("refresh after grace = %v, want reuse detection", err)
+	}
+	for name, pair := range map[string]OAuthTokenPair{"rotated": second, "sibling": sibling} {
+		if _, err := store.VerifyAccessToken(t.Context(), pair.AccessToken, resource); !errors.Is(err, ErrInvalidOAuthToken) {
+			t.Fatalf("%s access token after reuse = %v, want invalid token", name, err)
+		}
+	}
+}
+
+// A clock stepped back after a rotation must not stretch the window: the
+// rotation is then dated in the future, which counts as reuse.
+func TestOAuthStoreRefreshGraceRejectsFutureRotation(t *testing.T) {
+	sqlite := newTestSQLite(t)
+	users := NewUserStore(sqlite.DB)
+	user, err := users.CreateWithAudit("clock@example.com", "", "admin", AuditEvent{EventType: "user.created", Outcome: "success"})
+	if err != nil {
+		t.Fatalf("Create user: %v", err)
+	}
+	store := NewOAuthStore(sqlite.DB)
+	base := time.Now().UTC()
+	store.now = func() time.Time { return base }
+	client, err := store.RegisterClient(t.Context(), "Codex", "", []string{"http://127.0.0.1:14550/callback"})
+	if err != nil {
+		t.Fatalf("RegisterClient: %v", err)
+	}
+	resource := "https://fanout.example.com/mcp"
+	first, err := store.IssueTokenPair(t.Context(), client.ClientID, user.ID, MCPScopeTelemetryRead, resource)
+	if err != nil {
+		t.Fatalf("IssueTokenPair: %v", err)
+	}
+	second, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, "")
+	if err != nil {
+		t.Fatalf("RotateRefreshToken: %v", err)
+	}
+	store.now = func() time.Time { return base.Add(-time.Hour) }
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
+		t.Fatalf("refresh after a clock step back = %v, want reuse detection", err)
+	}
+	store.now = func() time.Time { return base }
+	if _, err := store.VerifyAccessToken(t.Context(), second.AccessToken, resource); !errors.Is(err, ErrInvalidOAuthToken) {
+		t.Fatalf("access token after rejected replay = %v, want family revoked", err)
+	}
+}
+
+// A rotated token replayed just after it expires, though within the grace
+// window, is reuse: the family is revoked rather than left live.
+func TestOAuthStoreRefreshGraceRevokesExpiredReplay(t *testing.T) {
+	sqlite := newTestSQLite(t)
+	users := NewUserStore(sqlite.DB)
+	user, err := users.CreateWithAudit("expiring@example.com", "", "admin", AuditEvent{EventType: "user.created", Outcome: "success"})
+	if err != nil {
+		t.Fatalf("Create user: %v", err)
+	}
+	store := NewOAuthStore(sqlite.DB)
+	base := time.Now().UTC()
+	store.now = func() time.Time { return base }
+	client, err := store.RegisterClient(t.Context(), "Codex", "", []string{"http://127.0.0.1:14550/callback"})
+	if err != nil {
+		t.Fatalf("RegisterClient: %v", err)
+	}
+	resource := "https://fanout.example.com/mcp"
+	first, err := store.IssueTokenPair(t.Context(), client.ClientID, user.ID, MCPScopeTelemetryRead, resource)
+	if err != nil {
+		t.Fatalf("IssueTokenPair: %v", err)
+	}
+	rotatedAt := base.Add(OAuthRefreshTTL - 30*time.Second)
+	store.now = func() time.Time { return rotatedAt }
+	second, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, "")
+	if err != nil {
+		t.Fatalf("RotateRefreshToken: %v", err)
+	}
+	store.now = func() time.Time { return base.Add(OAuthRefreshTTL + time.Second) }
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
+		t.Fatalf("expired replay within grace = %v, want reuse detection", err)
+	}
+	if _, err := store.VerifyAccessToken(t.Context(), second.AccessToken, resource); !errors.Is(err, ErrInvalidOAuthToken) {
+		t.Fatalf("descendant access token = %v, want family revoked", err)
+	}
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, second.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
+		t.Fatalf("descendant refresh token = %v, want family revoked", err)
+	}
+}
+
+// A revoked family has no live sibling, so the grace window never revives it.
+func TestOAuthStoreRefreshGraceDoesNotReviveRevokedFamily(t *testing.T) {
+	sqlite := newTestSQLite(t)
+	users := NewUserStore(sqlite.DB)
+	user, err := users.CreateWithAudit("revoked-sibling@example.com", "", "admin", AuditEvent{EventType: "user.created", Outcome: "success"})
+	if err != nil {
+		t.Fatalf("Create user: %v", err)
+	}
+	store := NewOAuthStore(sqlite.DB)
+	client, err := store.RegisterClient(t.Context(), "Codex", "", []string{"http://127.0.0.1:14550/callback"})
+	if err != nil {
+		t.Fatalf("RegisterClient: %v", err)
+	}
+	resource := "https://fanout.example.com/mcp"
+	first, err := store.IssueTokenPair(t.Context(), client.ClientID, user.ID, MCPScopeTelemetryRead, resource)
+	if err != nil {
+		t.Fatalf("IssueTokenPair: %v", err)
+	}
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); err != nil {
+		t.Fatalf("RotateRefreshToken: %v", err)
+	}
+	if err := users.RevokeAllSessionsWithAudit(user.ID, AuditEvent{EventType: "session.revoked", Outcome: "success"}); err != nil {
+		t.Fatalf("RevokeAllSessions: %v", err)
+	}
+	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
+		t.Fatalf("sibling refresh of a revoked family = %v, want reuse detection", err)
+	}
+}
+
 func TestRevokeAllSessionsAlsoRevokesOAuthCredentials(t *testing.T) {
 	sqlite := newTestSQLite(t)
 	users := NewUserStore(sqlite.DB)
@@ -314,7 +478,8 @@ func TestOAuthStoreCleanupExpired(t *testing.T) {
 	if deleted != 0 {
 		t.Fatalf("early cleanup deleted %d rows, want 0", deleted)
 	}
-	// The revoked row survived, so reuse detection still fires.
+	// The revoked row survived, so reuse detection still fires after the grace window.
+	store.now = func() time.Time { return base.Add(OAuthRefreshReuseGrace + time.Second) }
 	if _, err := store.RotateRefreshToken(t.Context(), client.ClientID, first.RefreshToken, resource, ""); !errors.Is(err, ErrOAuthRefreshReuse) {
 		t.Fatalf("reuse after cleanup = %v, want reuse detection", err)
 	}