diff --git a/Makefile b/Makefile index b937a68..508e906 100644 --- a/Makefile +++ b/Makefile @@ -88,6 +88,12 @@ tag: echo "❌ ERROR: No tag supplied. Usage: make tag TAG="; \ exit 1; \ fi + @if ! git show HEAD:integrations/dpop_grpc/go.mod | awk -v tag="$(TAG)" '$$1 == "github.com/conductorone/dpop/integrations/dpop_oauth2" && $$2 == tag { found = 1 } END { exit !found }'; then \ + echo "❌ ERROR: integrations/dpop_grpc/go.mod at HEAD must require dpop_oauth2 $(TAG) before tagging."; \ + echo " The local replace directive is ignored by consumers; a published dpop_grpc"; \ + echo " pinned to an older dpop_oauth2 will not compile. Bump the require, commit, re-tag."; \ + exit 1; \ + fi @echo "🔖 Tagging all Go modules with $(TAG)..." @git tag "$(TAG)"; @echo "$(TAG)"; diff --git a/integrations/dpop_grpc/client_credential.go b/integrations/dpop_grpc/client_credential.go index 4a8b4ce..b0e229b 100644 --- a/integrations/dpop_grpc/client_credential.go +++ b/integrations/dpop_grpc/client_credential.go @@ -5,9 +5,12 @@ import ( "errors" "net/url" + "github.com/conductorone/dpop/integrations/dpop_oauth2" "github.com/conductorone/dpop/pkg/dpop" "golang.org/x/oauth2" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/status" ) // DPoPCredentials implements the credentials.PerRPCCredentials interface @@ -54,7 +57,7 @@ func (d *DPoPCredentials) GetRequestMetadata(ctx context.Context, uri ...string) // Get the OAuth2 token token, err := d.tokenSource.Token() if err != nil { - return nil, err + return nil, tokenStatusError(err) } // Add access token to proof options @@ -78,3 +81,19 @@ func (d *DPoPCredentials) GetRequestMetadata(ctx context.Context, uri ...string) func (d *DPoPCredentials) RequireTransportSecurity() bool { return d.requireTLS } + +// tokenStatusError maps a token source failure onto a gRPC status so the +// transient/definitive classification survives the per-RPC credentials +// boundary — grpc-go flattens any non-status credentials error to +// codes.Unauthenticated. Transient failures (5xx responses, transport errors, +// timeouts; see dpop_oauth2.IsTransient) become codes.Unavailable so callers' +// retry policies treat them as retryable. Definitive failures (e.g. +// invalid_client, a disabled credential) become codes.Unauthenticated and +// fail fast. +func tokenStatusError(err error) error { + code := codes.Unauthenticated + if dpop_oauth2.IsTransient(err) { + code = codes.Unavailable + } + return status.Errorf(code, "dpop_grpc: failed to fetch token: %v", err) +} diff --git a/integrations/dpop_grpc/client_credential_test.go b/integrations/dpop_grpc/client_credential_test.go new file mode 100644 index 0000000..dd48e6c --- /dev/null +++ b/integrations/dpop_grpc/client_credential_test.go @@ -0,0 +1,152 @@ +package dpop_grpc + +import ( + "context" + "crypto/ed25519" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + pb "github.com/conductorone/dpop/integrations/dpop_grpc/testdata" + "github.com/conductorone/dpop/integrations/dpop_oauth2" + "github.com/conductorone/dpop/pkg/dpop" + "github.com/go-jose/go-jose/v4" + "github.com/stretchr/testify/require" + "golang.org/x/oauth2" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +// newCredentialTestClient wires a DPoPCredentials backed by the given token +// source to a bufconn-hosted test service and returns a client for it. +func newCredentialTestClient(t *testing.T, tokenSource oauth2.TokenSource) pb.TestServiceClient { + t.Helper() + registerBufnetResolver() + lis := bufconn.Listen(bufSize) + + s := grpc.NewServer() + pb.RegisterTestServiceServer(s, &testServer{}) + go func() { + _ = s.Serve(lis) + }() + t.Cleanup(s.Stop) + + _, priv, err := ed25519.GenerateKey(nil) + require.NoError(t, err) + jwk := &jose.JSONWebKey{ + Key: priv, + KeyID: "test-key", + Algorithm: string(jose.EdDSA), + Use: "sig", + } + proofer, err := dpop.NewProofer(jwk) + require.NoError(t, err) + + creds, err := NewDPoPCredentials(proofer, tokenSource, "test-endpoint", nil) + require.NoError(t, err) + creds.requireTLS = false + + conn, err := grpc.NewClient( + "bufnet://test-endpoint", + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return lis.Dial() }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithPerRPCCredentials(creds), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return pb.NewTestServiceClient(conn) +} + +// TestDPoPCredentials_TokenErrorClassification asserts that token source +// failures surface as gRPC status codes that preserve the transient vs +// definitive distinction: transient failures map to Unavailable (retryable by +// callers' retry policies) and definitive failures map to Unauthenticated +// (fail fast). +func TestDPoPCredentials_TokenErrorClassification(t *testing.T) { + tests := []struct { + name string + tokenErr error + wantCode codes.Code + }{ + { + name: "transient token failure maps to Unavailable", + tokenErr: fmt.Errorf("%w: unexpected status code: 503 Service Unavailable", dpop_oauth2.ErrTokenRequestTransient), + wantCode: codes.Unavailable, + }, + { + name: "definitive OAuth rejection maps to Unauthenticated", + tokenErr: fmt.Errorf("%w: invalid_client - client authentication failed", dpop_oauth2.ErrTokenRequestFailed), + wantCode: codes.Unauthenticated, + }, + { + name: "unclassified error maps to Unauthenticated", + tokenErr: errors.New("boom"), + wantCode: codes.Unauthenticated, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + client := newCredentialTestClient(t, &mockTokenSource{tokenErr: tc.tokenErr}) + + _, err := client.TestUnary(context.Background(), &pb.TestRequest{Message: "test"}) + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + require.Equal(t, tc.wantCode, st.Code()) + require.Contains(t, st.Message(), tc.tokenErr.Error()) + }) + } +} + +// TestDPoPCredentials_EndToEndTransient503 exercises the full path: a real +// dpop_oauth2 token source hitting a token endpoint that persistently returns +// 503 must surface as codes.Unavailable on the gRPC call. +func TestDPoPCredentials_EndToEndTransient503(t *testing.T) { + tokenSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"error":"unavailable"}`, http.StatusServiceUnavailable) + })) + defer tokenSrv.Close() + + privJWK := &jose.JSONWebKey{ + KeyID: "test-key", + Algorithm: string(jose.EdDSA), + Use: "sig", + } + _, priv, err := ed25519.GenerateKey(nil) + require.NoError(t, err) + privJWK.Key = priv + + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(tokenSrv.URL + "/token") + require.NoError(t, err) + + ts, err := dpop_oauth2.NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + dpop_oauth2.WithHTTPClient(tokenSrv.Client()), + dpop_oauth2.WithRetryConfig(dpop_oauth2.RetryConfig{MaxAttempts: 1}), + ) + require.NoError(t, err) + + client := newCredentialTestClient(t, ts) + + _, err = client.TestUnary(context.Background(), &pb.TestRequest{Message: "test"}) + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + require.Equal(t, codes.Unavailable, st.Code(), "a 503 from the token endpoint must surface as Unavailable") + require.Contains(t, st.Message(), "503") +} diff --git a/integrations/dpop_grpc/go.mod b/integrations/dpop_grpc/go.mod index 1150052..3f2b13a 100644 --- a/integrations/dpop_grpc/go.mod +++ b/integrations/dpop_grpc/go.mod @@ -4,6 +4,12 @@ go 1.23.4 require ( github.com/conductorone/dpop v0.0.2 + // RELEASE NOTE: this module uses dpop_oauth2 APIs newer than v0.2.5 + // (IsTransient, ErrTokenRequestTransient). The replace directive below + // covers local builds only — consumers ignore it. When tagging a release + // (make tag tags every module together), bump this require to that same + // new tag or published dpop_grpc will not compile. + github.com/conductorone/dpop/integrations/dpop_oauth2 v0.2.5 github.com/go-jose/go-jose/v4 v4.0.4 github.com/stretchr/testify v1.9.0 golang.org/x/oauth2 v0.26.0 @@ -26,3 +32,5 @@ require ( ) replace github.com/conductorone/dpop => ../.. + +replace github.com/conductorone/dpop/integrations/dpop_oauth2 => ../dpop_oauth2 diff --git a/integrations/dpop_oauth2/go.mod b/integrations/dpop_oauth2/go.mod index b00d741..939acde 100644 --- a/integrations/dpop_oauth2/go.mod +++ b/integrations/dpop_oauth2/go.mod @@ -5,6 +5,7 @@ go 1.23.4 require ( github.com/conductorone/dpop v0.0.2 github.com/go-jose/go-jose/v4 v4.0.4 + github.com/google/uuid v1.6.0 github.com/stretchr/testify v1.9.0 golang.org/x/oauth2 v0.26.0 ) @@ -13,7 +14,6 @@ replace github.com/conductorone/dpop => ../.. require ( github.com/davecgh/go-spew v1.1.1 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/jellydator/ttlcache/v3 v3.3.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect golang.org/x/crypto v0.34.0 // indirect diff --git a/integrations/dpop_oauth2/retry.go b/integrations/dpop_oauth2/retry.go new file mode 100644 index 0000000..888be66 --- /dev/null +++ b/integrations/dpop_oauth2/retry.go @@ -0,0 +1,115 @@ +package dpop_oauth2 + +import ( + "context" + "errors" + "math/rand/v2" + "net/http" + "time" +) + +// Defaults chosen so a full retry cycle (attempts plus backoff) fits well +// within the 30 second budget Token() imposes on each call. +const ( + defaultRetryMaxAttempts = 3 + defaultRetryInitialDelay = 500 * time.Millisecond + defaultRetryMaxDelay = 2 * time.Second +) + +// RetryConfig controls how Token() retries transient token request failures: +// 5xx or 429 responses, transport-level errors, and timeouts. Every attempt +// re-runs the full token request with a freshly signed DPoP proof and client +// assertion; a proof's jti may be single-use, so an identical request is never +// replayed. Definitive OAuth protocol errors (e.g. invalid_client) are never +// retried. +type RetryConfig struct { + // MaxAttempts is the total number of attempts, including the first. + // Values below 1 are treated as 1 (retries disabled). + MaxAttempts int + // InitialDelay is the backoff before the first retry. It doubles on each + // subsequent retry, capped at MaxDelay, with jitter applied. + InitialDelay time.Duration + // MaxDelay caps the backoff between attempts. + MaxDelay time.Duration +} + +// DefaultRetryConfig returns the retry behavior used when no WithRetryConfig +// option is supplied. +func DefaultRetryConfig() RetryConfig { + return RetryConfig{ + MaxAttempts: defaultRetryMaxAttempts, + InitialDelay: defaultRetryInitialDelay, + MaxDelay: defaultRetryMaxDelay, + } +} + +func (c RetryConfig) normalized() RetryConfig { + if c.MaxAttempts < 1 { + c.MaxAttempts = 1 + } + if c.InitialDelay <= 0 { + c.InitialDelay = defaultRetryInitialDelay + } + if c.MaxDelay < c.InitialDelay { + c.MaxDelay = c.InitialDelay + } + return c +} + +// retryDelay computes the backoff preceding retry number `retry` (1-based): +// exponential doubling capped at MaxDelay, with equal jitter (half the delay +// is fixed, the other half randomized) so concurrent clients hitting the same +// outage don't retry in lockstep. +func (c RetryConfig) retryDelay(retry int) time.Duration { + delay := c.InitialDelay + for i := 1; i < retry; i++ { + delay *= 2 + if delay >= c.MaxDelay { + delay = c.MaxDelay + break + } + } + half := delay / 2 + return half + rand.N(half+1) +} + +// sleepBeforeRetry blocks for the backoff delay preceding the given retry. +// It returns false if ctx expires first. +func sleepBeforeRetry(ctx context.Context, cfg RetryConfig, retry int) bool { + timer := time.NewTimer(cfg.retryDelay(retry)) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +// IsTransient reports whether err is a token request failure that was +// classified as transient: the failure mode gives no indication the +// credential itself is bad, so retrying (with a fresh proof and assertion) +// may succeed. +func IsTransient(err error) bool { + return errors.Is(err, ErrTokenRequestTransient) +} + +// markTransient tags err as a transient token request failure. The result +// matches ErrTokenRequestTransient in addition to everything err already +// matched, and its message is unchanged. +func markTransient(err error) error { + return &transientError{error: err} +} + +type transientError struct{ error } + +func (e *transientError) Unwrap() []error { + return []error{e.error, ErrTokenRequestTransient} +} + +// isRetryableStatus reports whether an HTTP response status is worth +// retrying: any 5xx (upstream failure) or 429 (throttling). 4xx OAuth +// protocol rejections are definitive and must not be retried. +func isRetryableStatus(code int) bool { + return code >= 500 || code == http.StatusTooManyRequests +} diff --git a/integrations/dpop_oauth2/retry_test.go b/integrations/dpop_oauth2/retry_test.go new file mode 100644 index 0000000..8b35147 --- /dev/null +++ b/integrations/dpop_oauth2/retry_test.go @@ -0,0 +1,551 @@ +package dpop_oauth2 + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + "time" + + "github.com/conductorone/dpop/pkg/dpop" + "github.com/go-jose/go-jose/v4" + "github.com/stretchr/testify/require" +) + +// fastRetry keeps retry tests quick while preserving the default attempt count. +func fastRetry() RetryConfig { + return RetryConfig{ + MaxAttempts: 3, + InitialDelay: time.Millisecond, + MaxDelay: 4 * time.Millisecond, + } +} + +// scriptedTokenServer answers each token request with the next status in the +// script (the last entry repeats if calls continue past the end). A 200 +// returns a valid token response; a 400 returns an invalid_client OAuth +// protocol error; anything else returns a bare error status. Every request's +// DPoP proof jti is recorded so tests can assert each attempt signed a fresh +// proof. +type scriptedTokenServer struct { + t *testing.T + server *httptest.Server + mu sync.Mutex + script []int + calls int + proofJTIs []string + assertions []string +} + +func newScriptedTokenServer(t *testing.T, script []int) *scriptedTokenServer { + s := &scriptedTokenServer{t: t, script: script} + s.server = httptest.NewServer(http.HandlerFunc(s.handle)) + t.Cleanup(s.server.Close) + return s +} + +func (s *scriptedTokenServer) recordProof(r *http.Request) { + proof := r.Header.Get(dpop.HeaderName) + if proof == "" { + s.proofJTIs = append(s.proofJTIs, "") + return + } + token, err := jose.ParseSigned(proof, []jose.SignatureAlgorithm{jose.EdDSA}) + require.NoError(s.t, err) + var claims struct { + JTI string `json:"jti"` + } + require.NoError(s.t, json.Unmarshal(token.UnsafePayloadWithoutVerification(), &claims)) + s.proofJTIs = append(s.proofJTIs, claims.JTI) +} + +func (s *scriptedTokenServer) handle(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + defer s.mu.Unlock() + + s.recordProof(r) + require.NoError(s.t, r.ParseForm()) + s.assertions = append(s.assertions, r.PostFormValue("client_assertion")) + + idx := s.calls + if idx >= len(s.script) { + idx = len(s.script) - 1 + } + s.calls++ + + status := s.script[idx] + w.Header().Set("Content-Type", "application/json") + switch { + case status == http.StatusOK: + json.NewEncoder(w).Encode(map[string]interface{}{ + "access_token": "test_access_token", + "token_type": "DPoP", + "expires_in": 3600, + }) + case status == http.StatusBadRequest: + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]string{ + "error": "invalid_client", + "error_description": "client authentication failed", + }) + default: + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]string{"error": "unavailable"}) + } +} + +func (s *scriptedTokenServer) callCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls +} + +func (s *scriptedTokenServer) seenProofJTIs() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.proofJTIs...) +} + +func (s *scriptedTokenServer) seenAssertions() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.assertions...) +} + +func newScriptedTokenSource(t *testing.T, srv *scriptedTokenServer, opts ...TokenSourceOption) *tokenSource { + t.Helper() + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(srv.server.URL + "/token") + require.NoError(t, err) + + opts = append([]TokenSourceOption{ + WithHTTPClient(srv.server.Client()), + WithRetryConfig(fastRetry()), + }, opts...) + + ts, err := NewTokenSource(proofer, tokenURL, "test-client", privJWK, opts...) + require.NoError(t, err) + return ts +} + +// TestTokenSource_RetriesTransient5xx asserts that 5xx responses are retried +// until success and that every attempt carries a freshly signed DPoP proof +// and client assertion (distinct jtis) — an identical request is never +// replayed, even when retries land within the same second. +func TestTokenSource_RetriesTransient5xx(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusServiceUnavailable, http.StatusInternalServerError, http.StatusOK}) + ts := newScriptedTokenSource(t, srv) + + token, err := ts.Token() + require.NoError(t, err, "transient 5xx responses should be retried to success") + require.Equal(t, "test_access_token", token.AccessToken) + require.Equal(t, 3, srv.callCount(), "expected two failed attempts plus one success") + + jtis := srv.seenProofJTIs() + require.Len(t, jtis, 3) + seenJTIs := make(map[string]bool, len(jtis)) + for _, jti := range jtis { + require.NotEmpty(t, jti, "every attempt must carry a DPoP proof") + require.False(t, seenJTIs[jti], "each attempt must sign a fresh proof (jti %q reused)", jti) + seenJTIs[jti] = true + } + + assertions := srv.seenAssertions() + require.Len(t, assertions, 3) + seenAssertions := make(map[string]bool, len(assertions)) + for _, assertion := range assertions { + require.NotEmpty(t, assertion, "every attempt must carry a client assertion") + require.False(t, seenAssertions[assertion], "each attempt must sign a fresh client assertion") + seenAssertions[assertion] = true + } +} + +// TestTokenSource_RetryOn429 asserts throttling responses are retried. +func TestTokenSource_RetryOn429(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusTooManyRequests, http.StatusOK}) + ts := newScriptedTokenSource(t, srv) + + token, err := ts.Token() + require.NoError(t, err) + require.Equal(t, "test_access_token", token.AccessToken) + require.Equal(t, 2, srv.callCount()) +} + +// TestTokenSource_RetriesExhausted asserts a persistent 5xx fails after +// MaxAttempts and that the returned error is classified transient while still +// matching ErrTokenRequestFailed. +func TestTokenSource_RetriesExhausted(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusServiceUnavailable}) + ts := newScriptedTokenSource(t, srv) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.Equal(t, 3, srv.callCount(), "expected exactly MaxAttempts attempts") + require.True(t, IsTransient(err), "persistent 5xx must classify as transient") + require.ErrorIs(t, err, ErrTokenRequestTransient) + require.ErrorIs(t, err, ErrTokenRequestFailed, "transient errors must still match ErrTokenRequestFailed") + require.Contains(t, err.Error(), "503") +} + +// TestTokenSource_NoRetryOnOAuthProtocolError asserts a definitive OAuth +// protocol rejection is returned immediately, without retries, and is not +// classified transient. +func TestTokenSource_NoRetryOnOAuthProtocolError(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusBadRequest}) + ts := newScriptedTokenSource(t, srv) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.Equal(t, 1, srv.callCount(), "OAuth protocol errors must never be retried") + require.False(t, IsTransient(err), "invalid_client is definitive, not transient") + require.ErrorIs(t, err, ErrTokenRequestFailed) + require.Contains(t, err.Error(), "invalid_client") +} + +// TestTokenSource_RetriesDisabled asserts MaxAttempts=1 restores the old +// single-shot behavior. +func TestTokenSource_RetriesDisabled(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusServiceUnavailable}) + ts := newScriptedTokenSource(t, srv, WithRetryConfig(RetryConfig{MaxAttempts: 1})) + + _, err := ts.Token() + require.Error(t, err) + require.Equal(t, 1, srv.callCount()) + require.True(t, IsTransient(err), "classification applies even when retries are disabled") +} + +// TestTokenSource_TransportErrorIsTransient asserts an error before any HTTP +// response (connection refused) is classified transient. +func TestTokenSource_TransportErrorIsTransient(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusOK}) + ts := newScriptedTokenSource(t, srv) + srv.server.Close() + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.True(t, IsTransient(err), "transport errors must classify as transient") + require.ErrorIs(t, err, ErrTokenRequestFailed) +} + +// TestTokenSource_TimeoutIsTransient asserts a client-side timeout on the +// token POST is classified transient. +func TestTokenSource_TimeoutIsTransient(t *testing.T) { + blocked := make(chan struct{}) + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-blocked: + case <-r.Context().Done(): + } + })) + defer slow.Close() + // LIFO: unblock the handlers before slow.Close() waits on them. + defer close(blocked) + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(slow.URL + "/token") + require.NoError(t, err) + + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithHTTPClient(&http.Client{Timeout: 50 * time.Millisecond}), + WithRetryConfig(RetryConfig{MaxAttempts: 2, InitialDelay: time.Millisecond, MaxDelay: time.Millisecond}), + ) + require.NoError(t, err) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.True(t, IsTransient(err), "timeouts must classify as transient") +} + +// TestTokenSource_NonceChallengeThenTransientRetry exercises the interplay of +// the two retry mechanisms: a use_dpop_nonce challenge is satisfied within an +// attempt, a subsequent 503 triggers the transient retry loop, and the retry +// succeeds using the nonce cached from the earlier challenge. +func TestTokenSource_NonceChallengeThenTransientRetry(t *testing.T) { + const serverNonce = "interplay-nonce" + + var mu sync.Mutex + calls := 0 + handler := func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + calls++ + + proof := r.Header.Get(dpop.HeaderName) + token, err := jose.ParseSigned(proof, []jose.SignatureAlgorithm{jose.EdDSA}) + require.NoError(t, err) + var claims struct { + Nonce string `json:"nonce"` + } + require.NoError(t, json.Unmarshal(token.UnsafePayloadWithoutVerification(), &claims)) + + w.Header().Set("Content-Type", "application/json") + switch { + case claims.Nonce != serverNonce: + w.Header().Set(dpop.NonceHeaderName, serverNonce) + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(map[string]string{"error": "use_dpop_nonce"}) + case calls == 2: + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(map[string]string{"error": "unavailable"}) + default: + json.NewEncoder(w).Encode(map[string]interface{}{ + "access_token": "test_access_token", + "token_type": "DPoP", + "expires_in": 3600, + }) + } + } + + srv := httptest.NewServer(http.HandlerFunc(handler)) + defer srv.Close() + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(srv.URL + "/token") + require.NoError(t, err) + + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithHTTPClient(srv.Client()), + WithNonceStore(NewNonceStore()), + WithRetryConfig(fastRetry()), + ) + require.NoError(t, err) + + token, err := ts.Token() + require.NoError(t, err, "challenge + transient failure should still converge on success") + require.Equal(t, "test_access_token", token.AccessToken) + // Call 1: challenged. Call 2 (nonce retry): 503. Call 3 (transient retry, + // cached nonce sent up front): success. + require.Equal(t, 3, calls) +} + +// TestTokenSource_NonceCarriedAcrossRetries asserts that a bare consumer (no +// NonceStore) does not get re-challenged on every transient retry: the nonce +// learned from the first use_dpop_nonce challenge is carried into subsequent +// outer attempts. +func TestTokenSource_NonceCarriedAcrossRetries(t *testing.T) { + const serverNonce = "carried-nonce" + + var mu sync.Mutex + calls := 0 + var seenNonces []string + handler := func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + calls++ + + proof := r.Header.Get(dpop.HeaderName) + token, err := jose.ParseSigned(proof, []jose.SignatureAlgorithm{jose.EdDSA}) + require.NoError(t, err) + var claims struct { + Nonce string `json:"nonce"` + } + require.NoError(t, json.Unmarshal(token.UnsafePayloadWithoutVerification(), &claims)) + seenNonces = append(seenNonces, claims.Nonce) + + w.Header().Set("Content-Type", "application/json") + switch { + case claims.Nonce != serverNonce: + w.Header().Set(dpop.NonceHeaderName, serverNonce) + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(map[string]string{"error": "use_dpop_nonce"}) + case calls == 2: + w.WriteHeader(http.StatusServiceUnavailable) + json.NewEncoder(w).Encode(map[string]string{"error": "unavailable"}) + default: + json.NewEncoder(w).Encode(map[string]interface{}{ + "access_token": "test_access_token", + "token_type": "DPoP", + "expires_in": 3600, + }) + } + } + + srv := httptest.NewServer(http.HandlerFunc(handler)) + defer srv.Close() + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(srv.URL + "/token") + require.NoError(t, err) + + // Deliberately no NonceStore. + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithHTTPClient(srv.Client()), + WithRetryConfig(fastRetry()), + ) + require.NoError(t, err) + + token, err := ts.Token() + require.NoError(t, err) + require.Equal(t, "test_access_token", token.AccessToken) + // Call 1: challenged. Call 2 (inner nonce retry): 503. Call 3 (outer + // transient retry): carries the learned nonce up front, so the server + // does not challenge again. + require.Equal(t, 3, calls, "the outer retry must not trigger a second challenge round trip") + require.Equal(t, []string{"", serverNonce, serverNonce}, seenNonces) +} + +// TestTokenSource_CanceledContextIsNotTransient asserts that a caller +// abandoning the call (context cancellation) is not classified as a retryable +// transport failure. +func TestTokenSource_CanceledContextIsNotTransient(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusOK}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(srv.server.URL + "/token") + require.NoError(t, err) + + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithBaseContext(ctx), + WithHTTPClient(srv.server.Client()), + WithRetryConfig(fastRetry()), + ) + require.NoError(t, err) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.False(t, IsTransient(err), "cancellation is not a transport failure and must not classify as transient") + require.ErrorIs(t, err, ErrTokenRequestFailed) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, 0, srv.callCount(), "no request should reach the server on a canceled context") +} + +// TestTokenSource_CancelDuringBackoffIsNotTransient asserts that a caller +// cancel landing mid-backoff strips the transient classification instead of +// surfacing the previous attempt's transient error. +func TestTokenSource_CancelDuringBackoffIsNotTransient(t *testing.T) { + srv := newScriptedTokenServer(t, []int{http.StatusServiceUnavailable}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(srv.server.URL + "/token") + require.NoError(t, err) + + // A very long backoff guarantees the cancel below lands during + // sleepBeforeRetry, not during an HTTP attempt. + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithBaseContext(ctx), + WithHTTPClient(srv.server.Client()), + WithRetryConfig(RetryConfig{MaxAttempts: 3, InitialDelay: time.Minute, MaxDelay: time.Minute}), + ) + require.NoError(t, err) + + time.AfterFunc(100*time.Millisecond, cancel) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.False(t, IsTransient(err), "a cancel during backoff must not surface as transient") + require.ErrorIs(t, err, ErrTokenRequestFailed) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, 1, srv.callCount(), "cancel during backoff must stop further attempts") +} + +// TestTokenSource_CancelCauseIsNotTransient asserts that cancellation via +// context.WithCancelCause on an in-flight request is not classified +// transient, even though http.Client.Do surfaces the cause instead of +// context.Canceled. +func TestTokenSource_CancelCauseIsNotTransient(t *testing.T) { + done := make(chan struct{}) + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-done: + case <-r.Context().Done(): + } + })) + defer slow.Close() + // LIFO: unblock the handler before slow.Close() waits on it. + defer close(done) + + ctx, cancel := context.WithCancelCause(context.Background()) + defer cancel(nil) + + privJWK := newTestProoferKey(t) + proofer, err := dpop.NewProofer(privJWK) + require.NoError(t, err) + + tokenURL, err := url.Parse(slow.URL + "/token") + require.NoError(t, err) + + ts, err := NewTokenSource( + proofer, + tokenURL, + "test-client", + privJWK, + WithBaseContext(ctx), + WithHTTPClient(slow.Client()), + WithRetryConfig(fastRetry()), + ) + require.NoError(t, err) + + cause := errors.New("caller abandoned the sync") + time.AfterFunc(100*time.Millisecond, func() { cancel(cause) }) + + token, err := ts.Token() + require.Error(t, err) + require.Nil(t, token) + require.False(t, IsTransient(err), "a cancel cause must not classify as transient") + require.ErrorIs(t, err, ErrTokenRequestFailed) + require.ErrorIs(t, err, cause, "the cancel cause should be preserved in the chain") +} + +// TestIsTransient_Wrapping asserts classification survives additional +// wrapping by callers. +func TestIsTransient_Wrapping(t *testing.T) { + base := markTransient(errors.New("boom")) + wrapped := errors.Join(errors.New("outer"), base) + require.True(t, IsTransient(wrapped)) + require.False(t, IsTransient(errors.New("boom"))) + require.False(t, IsTransient(nil)) +} diff --git a/integrations/dpop_oauth2/token_client_assertion.go b/integrations/dpop_oauth2/token_client_assertion.go index 5703c4b..2ebaaac 100644 --- a/integrations/dpop_oauth2/token_client_assertion.go +++ b/integrations/dpop_oauth2/token_client_assertion.go @@ -12,6 +12,7 @@ import ( "github.com/go-jose/go-jose/v4" "github.com/go-jose/go-jose/v4/jwt" + "github.com/google/uuid" "golang.org/x/oauth2" "github.com/conductorone/dpop/pkg/dpop" @@ -30,6 +31,14 @@ var ( // ErrTokenRequestFailed indicates the token request failed ErrTokenRequestFailed = errors.New("dpop_oauth2: token request failed") + // ErrTokenRequestTransient classifies a token request failure as likely + // transient: a 5xx or 429 response, a transport-level error, or a + // timeout. Errors matching this sentinel always also match + // ErrTokenRequestFailed; definitive OAuth protocol rejections (e.g. + // invalid_client) match only ErrTokenRequestFailed. Use IsTransient to + // test for it. + ErrTokenRequestTransient = errors.New("dpop_oauth2: transient token request failure") + // ErrProofCreationFailed indicates failure to create or sign DPoP proof ErrProofCreationFailed = errors.New("dpop_oauth2: failed to create or sign DPoP proof") ) @@ -88,6 +97,7 @@ type tokenSourceOptions struct { proofOptions []dpop.ProofOption nonceStore *NonceStore requestOptions []TokenRequestOption + retry RetryConfig } // WithBaseContext sets a custom base context for the token source @@ -125,6 +135,15 @@ func WithRequestOption(opt TokenRequestOption) TokenSourceOption { } } +// WithRetryConfig overrides how transient token request failures are retried. +// See RetryConfig for field semantics; set MaxAttempts to 1 to disable +// retries entirely. +func WithRetryConfig(cfg RetryConfig) TokenSourceOption { + return func(opts *tokenSourceOptions) { + opts.retry = cfg + } +} + func NewTokenSource(proofer *dpop.Proofer, tokenURL *url.URL, clientID string, clientSecret *jose.JSONWebKey, opts ...TokenSourceOption) (*tokenSource, error) { if proofer == nil { return nil, fmt.Errorf("%w: dpop-proofer", ErrMissingRequiredField) @@ -145,6 +164,7 @@ func NewTokenSource(proofer *dpop.Proofer, tokenURL *url.URL, clientID string, c options := &tokenSourceOptions{ baseCtx: context.Background(), httpClient: http.DefaultClient, + retry: DefaultRetryConfig(), } for _, opt := range opts { @@ -161,6 +181,7 @@ func NewTokenSource(proofer *dpop.Proofer, tokenURL *url.URL, clientID string, c requestOptions: options.requestOptions, proofOptions: options.proofOptions, nonceStore: options.nonceStore, + retry: options.retry.normalized(), }, nil } @@ -174,12 +195,54 @@ type tokenSource struct { requestOptions []TokenRequestOption proofOptions []dpop.ProofOption nonceStore *NonceStore + retry RetryConfig } func (c *tokenSource) Token() (*oauth2.Token, error) { ctx, done := context.WithTimeout(c.baseCtx, time.Second*30) defer done() - return c.tryToken(ctx, true, "") + + // Transient failures (5xx/429, transport errors, timeouts) are retried + // with capped exponential backoff + jitter. The retry re-enters tryToken, + // so every attempt signs a fresh DPoP proof and client assertion — both + // carry unique jtis, so an identical request is never replayed. + // Definitive failures (OAuth protocol rejections) return immediately. + // + // A nonce learned from a use_dpop_nonce challenge is carried across + // attempts so a bare consumer (no NonceStore) isn't re-challenged on + // every retry. + var lastErr error + retryNonce := "" + for attempt := 0; attempt < c.retry.MaxAttempts; attempt++ { + if attempt > 0 { + if !sleepBeforeRetry(ctx, c.retry, attempt) { + // The context died mid-backoff. A deadline expiry (the 30s + // Token() budget) is a timeout: surface the last transient + // failure so callers can still classify it. A caller cancel + // is not a timeout — strip the transient classification so + // nothing retries abandoned work. + if errors.Is(ctx.Err(), context.Canceled) { + // context.Cause preserves a WithCancelCause cause in the + // chain; for a plain cancel it is context.Canceled. + return nil, fmt.Errorf("%w: %w during retry backoff (last error: %v)", ErrTokenRequestFailed, context.Cause(ctx), lastErr) + } + break + } + } + + token, nonce, err := c.tryToken(ctx, true, retryNonce) + if err == nil { + return token, nil + } + if nonce != "" { + retryNonce = nonce + } + lastErr = err + if !IsTransient(err) { + return nil, err + } + } + return nil, lastErr } // tryToken performs a single token request. retryNonce, when non-empty, is the @@ -187,7 +250,11 @@ func (c *tokenSource) Token() (*oauth2.Token, error) { // attempt's proof regardless of whether a NonceStore is configured. This is // what makes a bare consumer (no NonceStore) nonce-aware: the challenge/retry // is self-contained within a single Token() call. -func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonce string) (*oauth2.Token, error) { +// +// The second return value is the nonce in effect for this attempt (the +// carried retryNonce, a cached store nonce, or a newly challenged one), so +// the transient retry loop in Token() can carry it into the next attempt. +func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonce string) (*oauth2.Token, string, error) { jsigner, err := jose.NewSigner( jose.SigningKey{ Algorithm: jose.EdDSA, @@ -195,7 +262,7 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc }, nil) if err != nil { - return nil, fmt.Errorf("%w: failed to create signer: %v", ErrProofCreationFailed, err) + return nil, retryNonce, fmt.Errorf("%w: failed to create signer: %v", ErrProofCreationFailed, err) } // Our token host may include a port, but the audience never expects a port @@ -203,6 +270,11 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc now := time.Now() claims := &jwt.Claims{ + // A unique jti makes every signed assertion distinct. Without it, + // second-precision timestamps plus deterministic Ed25519 signatures + // would make fast retries re-send a byte-identical assertion, which a + // server enforcing RFC 7523 single-use may reject. + ID: uuid.New().String(), Issuer: c.clientID, Subject: c.clientID, Audience: jwt.Audience{aud}, @@ -225,13 +297,13 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc for _, opt := range c.requestOptions { err = opt(tr) if err != nil { - return nil, fmt.Errorf("%w: failed to modify request: %v", ErrTokenRequestFailed, err) + return nil, retryNonce, fmt.Errorf("%w: failed to modify request: %v", ErrTokenRequestFailed, err) } } marshalledClaims, err := tr.Marshaler(claims) if err != nil { - return nil, fmt.Errorf("%w: failed to marshal claims: %v", ErrTokenRequestFailed, err) + return nil, retryNonce, fmt.Errorf("%w: failed to marshal claims: %v", ErrTokenRequestFailed, err) } method := http.MethodPost @@ -252,24 +324,24 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc dpopProof, err := c.proofer.CreateProof(ctx, method, c.tokenURL.String(), proofOpts...) if err != nil { - return nil, fmt.Errorf("%w: failed to create proof: %v", ErrProofCreationFailed, err) + return nil, nonce, fmt.Errorf("%w: failed to create proof: %v", ErrProofCreationFailed, err) } rv, err := jsigner.Sign(marshalledClaims) if err != nil { - return nil, fmt.Errorf("%w: failed to sign proof: %v", ErrProofCreationFailed, err) + return nil, nonce, fmt.Errorf("%w: failed to sign proof: %v", ErrProofCreationFailed, err) } s, err := rv.CompactSerialize() if err != nil { - return nil, fmt.Errorf("%w: failed to serialize proof: %v", ErrProofCreationFailed, err) + return nil, nonce, fmt.Errorf("%w: failed to serialize proof: %v", ErrProofCreationFailed, err) } tr.Body["client_assertion"] = []string{s} req, err := http.NewRequestWithContext(ctx, method, c.tokenURL.String(), strings.NewReader(tr.Body.Encode())) if err != nil { - return nil, fmt.Errorf("%w: failed to create request: %v", ErrTokenRequestFailed, err) + return nil, nonce, fmt.Errorf("%w: failed to create request: %v", ErrTokenRequestFailed, err) } req.Header.Set(dpop.HeaderName, dpopProof) @@ -279,7 +351,23 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc resp, err := c.httpClient.Do(req) if err != nil { - return nil, fmt.Errorf("%w: failed to execute request: %v", ErrTokenRequestFailed, err) + // The transport error stays in the chain (%w) so callers can inspect + // the underlying cause (context.Canceled, net errors, ...). + reqErr := fmt.Errorf("%w: failed to execute request: %w", ErrTokenRequestFailed, err) + // A canceled context means the caller abandoned the call — that is + // not a transport failure, so don't classify it as retryable. Check + // the context as well as the returned error: when the context was + // canceled via context.WithCancelCause, Do returns the cause, which + // need not match context.Canceled. + if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) { + return nil, nonce, reqErr + } + // Everything else that fails before an HTTP response (connection + // resets, proxy errors, timeouts — including a deadline expiry, which + // is exactly the timed-out token POST class) never reached the + // authorization server's OAuth logic: it carries no verdict about the + // credential, so it is safe to classify as retryable. + return nil, nonce, markTransient(reqErr) } defer resp.Body.Close() @@ -290,45 +378,49 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc ErrorDescription string `json:"error_description"` } if err := json.NewDecoder(resp.Body).Decode(&errorResp); err != nil { - return nil, fmt.Errorf("%w: failed to decode error response: %v", ErrTokenRequestFailed, err) + return nil, nonce, fmt.Errorf("%w: failed to decode error response: %v", ErrTokenRequestFailed, err) } if errorResp.Error == "use_dpop_nonce" { // Get the new nonce from header - nonce := resp.Header.Get(dpop.NonceHeaderName) - if nonce == "" { - return nil, ErrNonceMissing + challengeNonce := resp.Header.Get(dpop.NonceHeaderName) + if challengeNonce == "" { + return nil, nonce, ErrNonceMissing } // Store the nonce for cross-call reuse if we have a store if c.nonceStore != nil { - c.nonceStore.SetNonce(nonce) + c.nonceStore.SetNonce(challengeNonce) } // Only retry once on first attempt if !firstAttempt { - return nil, fmt.Errorf("%w: token request failed after retry: %s - %s", ErrTokenRequestFailed, errorResp.Error, errorResp.ErrorDescription) + return nil, challengeNonce, fmt.Errorf("%w: token request failed after retry: %s - %s", ErrTokenRequestFailed, errorResp.Error, errorResp.ErrorDescription) } // Retry with the challenged nonce. Passing it explicitly means the // retry is nonce-aware even with no NonceStore configured. - return c.tryToken(ctx, false, nonce) + return c.tryToken(ctx, false, challengeNonce) } - return nil, fmt.Errorf("%w: %s - %s", ErrTokenRequestFailed, errorResp.Error, errorResp.ErrorDescription) + return nil, nonce, fmt.Errorf("%w: %s - %s", ErrTokenRequestFailed, errorResp.Error, errorResp.ErrorDescription) + } + + if isRetryableStatus(resp.StatusCode) { + return nil, nonce, markTransient(fmt.Errorf("%w: unexpected status code: %s", ErrTokenRequestFailed, resp.Status)) } if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("%w: unexpected status code: %s", ErrTokenRequestFailed, resp.Status) + return nil, nonce, fmt.Errorf("%w: unexpected status code: %s", ErrTokenRequestFailed, resp.Status) } token := &oauth2.Token{} err = json.NewDecoder(resp.Body).Decode(token) if err != nil { - return nil, fmt.Errorf("%w: failed to decode token response: %v", ErrInvalidToken, err) + return nil, nonce, fmt.Errorf("%w: failed to decode token response: %v", ErrInvalidToken, err) } if token.AccessToken == "" { - return nil, fmt.Errorf("%w: empty access token", ErrInvalidToken) + return nil, nonce, fmt.Errorf("%w: empty access token", ErrInvalidToken) } if token.Expiry.IsZero() { @@ -345,8 +437,8 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool, retryNonc // Accept both DPoP and Bearer tokens // If we sent a DPoP proof but got a Bearer token, that means the AS doesn't support DPoP if !strings.EqualFold(token.TokenType, "DPoP") && !strings.EqualFold(token.TokenType, "Bearer") { - return nil, fmt.Errorf("%w: invalid token type: %s", ErrInvalidToken, token.TokenType) + return nil, nonce, fmt.Errorf("%w: invalid token type: %s", ErrInvalidToken, token.TokenType) } - return token, nil + return token, nonce, nil }