diff --git a/go.mod b/go.mod index d749b661..197c6ce0 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/conductorone/baton-hubspot go 1.25.2 require ( - github.com/conductorone/baton-sdk v0.26.0 + github.com/conductorone/baton-sdk v0.28.0 github.com/ennyjfrick/ruleguard-logfatal v0.0.2 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 github.com/quasilyte/go-ruleguard/dsl v0.3.23 @@ -49,9 +49,9 @@ require ( github.com/cockroachdb/redact v1.1.5 // indirect github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b // indirect github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect - github.com/conductorone/dpop v0.2.6 // indirect - github.com/conductorone/dpop/integrations/dpop_grpc v0.2.4 // indirect - github.com/conductorone/dpop/integrations/dpop_oauth2 v0.2.5 // indirect + github.com/conductorone/dpop v0.3.0 // indirect + github.com/conductorone/dpop/integrations/dpop_grpc v0.3.0 // indirect + github.com/conductorone/dpop/integrations/dpop_oauth2 v0.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/deckarep/golang-set/v2 v2.9.0 // indirect github.com/doug-martin/goqu/v9 v9.19.0 // indirect diff --git a/go.sum b/go.sum index 9be8b467..4ba6cc26 100644 --- a/go.sum +++ b/go.sum @@ -84,14 +84,14 @@ github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b h1:VXvSNzmr8hMj8 github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b/go.mod h1:yBRu/cnL4ks9bgy4vAASdjIW+/xMlFwuHKqtmh3GZQg= github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 h1:zuQyyAKVxetITBuuhv3BI9cMrmStnpT18zmgmTxunpo= github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ= -github.com/conductorone/baton-sdk v0.26.0 h1:aNKg81BhPAVGyYe+W4czZJnL9hzJEUkDinbt0klOeo4= -github.com/conductorone/baton-sdk v0.26.0/go.mod h1:SKm95z4KkQ23Tufo2ys88lVzbwKb0AQEbKee5GE0Lig= -github.com/conductorone/dpop v0.2.6 h1:fakwai/Xm2b/fcDUwJN41WtcSI/2UhQOyRIVvnnrrNA= -github.com/conductorone/dpop v0.2.6/go.mod h1:gyo8TtzB9SCFCsjsICH4IaLZ7y64CcrDXMOPBwfq/3s= -github.com/conductorone/dpop/integrations/dpop_grpc v0.2.4 h1:lYxYi9/WTSL9sE96CO0QF2BY3kehs8dTTApI134TGCA= -github.com/conductorone/dpop/integrations/dpop_grpc v0.2.4/go.mod h1:LYNoUc1lkvozk9HBio+xI2w8YyfYy0v2cAJtIgrkj8o= -github.com/conductorone/dpop/integrations/dpop_oauth2 v0.2.5 h1:x/ZtD0YLNwlmoSv9SE4OBPJB9Hj2cpwyE5BAfia7aY8= -github.com/conductorone/dpop/integrations/dpop_oauth2 v0.2.5/go.mod h1:2eI0qv+XaEhoCw0GKFF1yH4X8Mp4KLVEVnQKRFEy4zs= +github.com/conductorone/baton-sdk v0.28.0 h1:XDOkPYeC8f3sc3UupzfqYRShjD1uvY3ux0BT5gMmSio= +github.com/conductorone/baton-sdk v0.28.0/go.mod h1:i4DXDGaiyHyg164r4VbLu734z+MF0TyYjscOaRYd92M= +github.com/conductorone/dpop v0.3.0 h1:j5fZk0VqepGKYo+/NDikCOMsZcgs4HO4i0k56wRel5g= +github.com/conductorone/dpop v0.3.0/go.mod h1:gyo8TtzB9SCFCsjsICH4IaLZ7y64CcrDXMOPBwfq/3s= +github.com/conductorone/dpop/integrations/dpop_grpc v0.3.0 h1:R2uxHBtStgUn7cxbAnT3mAj6/e1akfP2Pj/hEPFB0D8= +github.com/conductorone/dpop/integrations/dpop_grpc v0.3.0/go.mod h1:f30gFNZHGkbPlufIDqzg5cr7llQQS0r5VQ+bJCnrDik= +github.com/conductorone/dpop/integrations/dpop_oauth2 v0.3.0 h1:g9OX0PW9DQyrGy/Tp2lesky/V1VH55cTY8z0Iq0ksOo= +github.com/conductorone/dpop/integrations/dpop_oauth2 v0.3.0/go.mod h1:RZqiSQdi4NXnoZtB3qG3XYN/6L4dHZy6jctEr5K5oPk= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/pkg/hubspot/client.go b/pkg/hubspot/client.go index c3e27f01..537d4d17 100644 --- a/pkg/hubspot/client.go +++ b/pkg/hubspot/client.go @@ -28,12 +28,6 @@ func (c *Client) usersURL() string { return c.baseURL.JoinPath("settings/users/2026-03").String() } -// listUsersURL is the paginated list endpoint. 2026-09-beta returns -// paging.next.after with limit=50; 2026-03 omits paging and silently truncates. -func (c *Client) listUsersURL() string { - return c.baseURL.JoinPath("settings/users/2026-09-beta").String() -} - func (c *Client) userURL(userID string) string { return c.baseURL.JoinPath("settings/users/2026-03", userID).String() } @@ -59,8 +53,20 @@ func (c *Client) accountLastLoginURL() string { } type UsersResponse struct { - Results []User `json:"results"` - Paging PaginationData `json:"paging"` + Results []User `json:"results"` + Paging *PaginationData `json:"paging"` +} + +// HasPaginationData makes UsersResponse a uhttp.PaginatedResponse. +// WithPaginationData unmarshals the whole body into whatever it is given, so the +// receiver has to mirror the top level of the response; an inner field would +// decode against the wrong level and always report nothing. +// +// Paging is a pointer because encoding/json leaves a value field zero whether +// the key was absent or empty, and those mean opposite things here: absent is +// the endpoint truncating silently, empty is a legitimate last page. +func (u *UsersResponse) HasPaginationData() bool { + return u.Paging != nil } type AccountLoginResponse struct { @@ -144,16 +150,17 @@ func (c *Client) GetUsers(ctx context.Context, getUsersVars GetUsersVars) ([]Use annos, err := c.get( ctx, - c.listUsersURL(), + c.usersURL(), &userResponse, queryParams, + uhttp.WithPaginationData(&userResponse), ) if err != nil { return nil, "", nil, err } - if (userResponse.Paging != PaginationData{}) { + if userResponse.Paging != nil { return userResponse.Results, userResponse.Paging.Next.After, annos, nil } @@ -330,8 +337,14 @@ func (c *Client) GetUserLastLogin(ctx context.Context, userId string) (*time.Tim return nil, annos, nil } -func (c *Client) get(ctx context.Context, url string, resourceResponse interface{}, queryParams url.Values) (annotations.Annotations, error) { - return c.doRequest(ctx, url, http.MethodGet, nil, resourceResponse, queryParams) +func (c *Client) get( + ctx context.Context, + url string, + resourceResponse interface{}, + queryParams url.Values, + doOptions ...uhttp.DoOption, +) (annotations.Annotations, error) { + return c.doRequest(ctx, url, http.MethodGet, nil, resourceResponse, queryParams, doOptions...) } func (c *Client) put(ctx context.Context, url string, data interface{}, resourceResponse interface{}) (annotations.Annotations, error) { @@ -353,6 +366,7 @@ func (c *Client) doRequest( data interface{}, resourceResponse interface{}, queryParams url.Values, + doOptions ...uhttp.DoOption, ) (annotations.Annotations, error) { parsedURL, err := url.Parse(urlAddress) if err != nil { @@ -376,7 +390,6 @@ func (c *Client) doRequest( return nil, err } - var doOptions []uhttp.DoOption if resourceResponse != nil { doOptions = append(doOptions, uhttp.WithJSONResponse(resourceResponse)) } diff --git a/pkg/hubspot/client_test.go b/pkg/hubspot/client_test.go new file mode 100644 index 00000000..2b8a33ae --- /dev/null +++ b/pkg/hubspot/client_test.go @@ -0,0 +1,75 @@ +package hubspot + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/conductorone/baton-sdk/pkg/uhttp" +) + +// newTestClient spins up a server that always replies with body and returns a +// client pointed at it. Each case gets its own server so uhttp's response cache +// never serves one case's body to another. +func newTestClient(t *testing.T, body string) *Client { + t.Helper() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(server.Close) + + client, err := NewClient("token", server.Client(), server.URL+"/") + if err != nil { + t.Fatalf("NewClient: %v", err) + } + return client +} + +func TestGetUsersReturnsNextPage(t *testing.T) { + client := newTestClient(t, `{"results":[{"id":"1","email":"a@example.com"}],"paging":{"next":{"after":"50"}}}`) + + users, nextPage, _, err := client.GetUsers(context.Background(), GetUsersVars{Limit: 50}) + if err != nil { + t.Fatalf("GetUsers: %v", err) + } + if len(users) != 1 || users[0].Id != "1" { + t.Errorf("got users %+v, want a single user with id 1", users) + } + if nextPage != "50" { + t.Errorf("got next page %q, want %q", nextPage, "50") + } +} + +// The last page still carries a paging object, just without a cursor. +func TestGetUsersLastPage(t *testing.T) { + client := newTestClient(t, `{"results":[{"id":"1","email":"a@example.com"}],"paging":{}}`) + + users, nextPage, _, err := client.GetUsers(context.Background(), GetUsersVars{Limit: 50}) + if err != nil { + t.Fatalf("GetUsers: %v", err) + } + if len(users) != 1 { + t.Errorf("got %d users, want 1", len(users)) + } + if nextPage != "" { + t.Errorf("got next page %q, want empty", nextPage) + } +} + +// A 200 with results but no paging object means the API truncated the list +// without telling us. That must be an error, not a short sync. +func TestGetUsersMissingPagingErrors(t *testing.T) { + client := newTestClient(t, `{"results":[{"id":"1","email":"a@example.com"}]}`) + + _, _, _, err := client.GetUsers(context.Background(), GetUsersVars{Limit: 50}) + if err == nil { + t.Fatal("GetUsers: expected an error when the response omits paging") + } + if !errors.Is(err, uhttp.ErrMissingPaginationData) { + t.Errorf("got error %v, want it to wrap ErrMissingPaginationData", err) + } +} diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/dotc1z/source_cache.go b/vendor/github.com/conductorone/baton-sdk/pkg/dotc1z/source_cache.go index fe1eb55d..90cca8ae 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/dotc1z/source_cache.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/dotc1z/source_cache.go @@ -38,6 +38,11 @@ type sourceCacheStoreTestSeams struct { // implemented ONLY by the Pebble engine; the syncer type-asserts for it and // treats a store without it as "source cache unsupported" (no-op lookup, // no replay). It is deliberately NOT part of c1zstore.Store. +// +// Advanced: this interface exists for the SDK's replay orchestration, not +// for direct use. The correctness obligations (preflight, scope poisoning, +// compat validation) live in the callers; see pkg/sourcecache for the +// connector-facing contract. type SourceCacheStore interface { // LookupSourceCacheEntry returns this store's manifest entry for // (kind, scopeKey). Backs the connector-facing lookup when this diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/sdk/version.go b/vendor/github.com/conductorone/baton-sdk/pkg/sdk/version.go index 00d0d815..4a12abc4 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/sdk/version.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/sdk/version.go @@ -1,3 +1,3 @@ package sdk -const Version = "v0.25.1" +const Version = "v0.27.0" diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/sourcecache/sourcecache.go b/vendor/github.com/conductorone/baton-sdk/pkg/sourcecache/sourcecache.go index d3b3c5fd..191d4c77 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/sourcecache/sourcecache.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/sourcecache/sourcecache.go @@ -1,6 +1,12 @@ // Package sourcecache defines the connector-facing surface of source-cache // replay (see proto/c1/connector/v2/annotation_source_cache.proto). // +// ADVANCED FUNCTIONALITY. Source-cache replay is an advanced, opt-in +// capability with strict correctness obligations on the connector (scope +// partitioning, validator lifetime — see the invariants below). Most +// connectors should not use this package; adopt it only in coordination +// with the SDK maintainers. +// // NOT YET WIRED. This package describes the intended contract, and the // storage and eligibility machinery beneath it is in place, but the syncer // does not install a Lookup or consume these annotations yet: SyncOpAttrs diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/sync/syncer.go b/vendor/github.com/conductorone/baton-sdk/pkg/sync/syncer.go index cadd2fef..c3eb71e4 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/sync/syncer.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/sync/syncer.go @@ -4028,6 +4028,9 @@ func WithExternalResourceC1ZPath(path string) SyncOpt { // WithPreviousSyncC1ZPath registers a separate c1z holding the previous sync // for replay features. // +// Advanced: source-cache replay is advanced, opt-in functionality (see +// pkg/sourcecache); most callers should not set this option. +// // This is required for the single-sync v3 (Pebble) engine: a Pebble c1z // holds exactly one sync by contract, so there is no in-file "previous // sync" to replay from (StartNewSync replaces the prior sync). NewSyncer @@ -4053,7 +4056,8 @@ func WithPreviousSyncC1ZPath(path string) SyncOpt { // WithOptionalPreviousSyncC1ZPath is WithPreviousSyncC1ZPath with // best-effort semantics: if the file is missing, corrupt, or written by // an incompatible SDK, NewSyncer logs and proceeds WITHOUT replay -// instead of failing. Intended for cache-style replay sources the +// instead of failing. Advanced, opt-in functionality like its strict +// twin — see pkg/sourcecache. Intended for cache-style replay sources the // caller maintains automatically (the service-mode previous-sync spare) // — a bad cache file must never fail a sync. Callers that name a // specific file deliberately should use WithPreviousSyncC1ZPath, which diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/pagination.go b/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/pagination.go index 397eb4b9..fdb54e28 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/pagination.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/pagination.go @@ -1,8 +1,14 @@ package uhttp import ( + "errors" + "fmt" + "reflect" "strings" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "github.com/conductorone/baton-sdk/pkg/pagination" ) @@ -97,3 +103,44 @@ func WithNextLinkPagination(bag *pagination.Bag, config *NextLinkConfig) DoOptio return nil } } + +// PaginatedResponse is implemented by response types that can report whether the API returned the pagination data the caller needs to fetch the next page. +type PaginatedResponse interface { + HasPaginationData() bool +} + +// ErrMissingPaginationData is the sentinel returned when a successful response decoded fine but carried no pagination data; match it with errors.Is. +var ErrMissingPaginationData = errors.New("uhttp: response is missing pagination data") + +// WithPaginationData decodes the body into response and fails the request if its pagination data is absent, so an API that silently drops its cursor errors instead of ending the sync after one page. +// response must be a non-nil pointer, since the body is decoded into it. +func WithPaginationData(response PaginatedResponse) DoOption { + return func(resp *WrapperResponse) error { + if response == nil { + return status.Error(codes.InvalidArgument, "WithPaginationData: response is nil") + } + + rv := reflect.ValueOf(response) + if rv.Kind() != reflect.Pointer || rv.IsNil() { + return status.Errorf(codes.InvalidArgument, "WithPaginationData: response must be a non-nil pointer, got %T", response) + } + + if !isSuccessStatusCode(resp.StatusCode) { + return nil + } + + if err := WithResponse(response)(resp); err != nil { + return err + } + + if !response.HasPaginationData() { + return WrapErrors( + codes.FailedPrecondition, + fmt.Sprintf("%T reported no pagination data. status code: %d", response, resp.StatusCode), + ErrMissingPaginationData, + ) + } + + return nil + } +} diff --git a/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/wrapper.go b/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/wrapper.go index 22806159..a12ff18d 100644 --- a/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/wrapper.go +++ b/vendor/github.com/conductorone/baton-sdk/pkg/uhttp/wrapper.go @@ -259,7 +259,7 @@ func WithAlwaysXMLResponse(response any) DoOption { if resp.StatusCode == http.StatusNoContent { return nil } - if resp.StatusCode >= 200 && resp.StatusCode < 300 && len(resp.Body) == 0 { + if isSuccessStatusCode(resp.StatusCode) && len(resp.Body) == 0 { return nil } return unmarshalXMLToMap(genericResponse, resp) @@ -317,6 +317,13 @@ type ErrorResponse interface { Message() string } +// isSuccessStatusCode reports whether code is in the 2xx success class. +// http.StatusOK (200) is the inclusive lower bound and +// http.StatusMultipleChoices (300) the exclusive upper bound. +func isSuccessStatusCode(code int) bool { + return code >= http.StatusOK && code < http.StatusMultipleChoices +} + // GrpcCodeFromHTTPStatus maps an HTTP status code to the appropriate gRPC status code. func GrpcCodeFromHTTPStatus(httpStatus int) codes.Code { switch httpStatus { @@ -349,7 +356,7 @@ func GrpcCodeFromHTTPStatus(httpStatus int) codes.Code { func WithErrorResponse(resource ErrorResponse) DoOption { return func(resp *WrapperResponse) error { - if resp.StatusCode < 300 { + if resp.StatusCode < http.StatusMultipleChoices { return nil } @@ -425,7 +432,7 @@ func WithGenericResponse(response *map[string]any) DoOption { return nil } - if resp.StatusCode >= 200 && resp.StatusCode < 300 && len(resp.Body) == 0 { + if isSuccessStatusCode(resp.StatusCode) && len(resp.Body) == 0 { return nil } @@ -587,7 +594,7 @@ func (c *BaseHttpClient) Do(req *http.Request, options ...DoOption) (*http.Respo } } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { + if !isSuccessStatusCode(resp.StatusCode) { grpcCode := GrpcCodeFromHTTPStatus(resp.StatusCode) return resp, WrapErrorsWithRateLimitInfo(grpcCode, resp, optErrs...) } diff --git a/vendor/github.com/conductorone/dpop/integrations/dpop_grpc/client_credential.go b/vendor/github.com/conductorone/dpop/integrations/dpop_grpc/client_credential.go index 4a8b4ce9..b0e229b9 100644 --- a/vendor/github.com/conductorone/dpop/integrations/dpop_grpc/client_credential.go +++ b/vendor/github.com/conductorone/dpop/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/vendor/github.com/conductorone/dpop/integrations/dpop_oauth2/retry.go b/vendor/github.com/conductorone/dpop/integrations/dpop_oauth2/retry.go new file mode 100644 index 00000000..888be662 --- /dev/null +++ b/vendor/github.com/conductorone/dpop/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/vendor/github.com/conductorone/dpop/integrations/dpop_oauth2/token_client_assertion.go b/vendor/github.com/conductorone/dpop/integrations/dpop_oauth2/token_client_assertion.go index edb8588c..2ebaaace 100644 --- a/vendor/github.com/conductorone/dpop/integrations/dpop_oauth2/token_client_assertion.go +++ b/vendor/github.com/conductorone/dpop/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,15 +195,66 @@ 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 } -func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2.Token, error) { +// tryToken performs a single token request. retryNonce, when non-empty, is the +// nonce returned by a prior use_dpop_nonce challenge and is attached to this +// 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. +// +// 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, @@ -190,7 +262,7 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. }, 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 @@ -198,6 +270,11 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. 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}, @@ -220,13 +297,13 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. 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 @@ -234,34 +311,37 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. proofOpts := make([]dpop.ProofOption, 0, len(c.proofOptions)+2) proofOpts = append(proofOpts, c.proofOptions...) - // Add nonce if available from store - if c.nonceStore != nil { - nonce := c.nonceStore.GetNonce() - if nonce != "" { - proofOpts = append(proofOpts, dpop.WithStaticNonce(nonce)) - } + // Attach a nonce when available. Prefer the nonce from a use_dpop_nonce + // challenge on this same Token() call (retryNonce); otherwise fall back to + // a cached nonce from the configured store for cross-call reuse. + nonce := retryNonce + if nonce == "" && c.nonceStore != nil { + nonce = c.nonceStore.GetNonce() + } + if nonce != "" { + proofOpts = append(proofOpts, dpop.WithStaticNonce(nonce)) } 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) @@ -271,7 +351,23 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. 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() @@ -282,58 +378,67 @@ func (c *tokenSource) tryToken(ctx context.Context, firstAttempt bool) (*oauth2. 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 if we have a store + // 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) } - // Try again with the new nonce - return c.tryToken(ctx, false) + // Retry with the challenged nonce. Passing it explicitly means the + // retry is nonce-aware even with no NonceStore configured. + 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() { token.Expiry = time.Now() if token.ExpiresIn > 0 { - token.Expiry = time.Now().Add(time.Duration(token.ExpiresIn-10) * time.Second) // 10 seconds before the token expires + expiresIn := token.ExpiresIn - 10 // 10 seconds before the token expires + if expiresIn < 0 { + expiresIn = 0 + } + token.Expiry = time.Now().Add(time.Duration(expiresIn) * time.Second) } } // 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 } diff --git a/vendor/modules.txt b/vendor/modules.txt index 440e03fc..fe54c9a2 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -272,7 +272,7 @@ github.com/cockroachdb/swiss # github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 ## explicit; go 1.19 github.com/cockroachdb/tokenbucket -# github.com/conductorone/baton-sdk v0.26.0 +# github.com/conductorone/baton-sdk v0.28.0 ## explicit; go 1.25.2 github.com/conductorone/baton-sdk/internal/connector github.com/conductorone/baton-sdk/pb/c1/c1z/v1 @@ -345,13 +345,13 @@ github.com/conductorone/baton-sdk/pkg/uotel github.com/conductorone/baton-sdk/pkg/uotel/uotelzap github.com/conductorone/baton-sdk/pkg/ustrings github.com/conductorone/baton-sdk/pkg/utls -# github.com/conductorone/dpop v0.2.6 +# github.com/conductorone/dpop v0.3.0 ## explicit; go 1.23.4 github.com/conductorone/dpop/pkg/dpop -# github.com/conductorone/dpop/integrations/dpop_grpc v0.2.4 +# github.com/conductorone/dpop/integrations/dpop_grpc v0.3.0 ## explicit; go 1.23.4 github.com/conductorone/dpop/integrations/dpop_grpc -# github.com/conductorone/dpop/integrations/dpop_oauth2 v0.2.5 +# github.com/conductorone/dpop/integrations/dpop_oauth2 v0.3.0 ## explicit; go 1.23.4 github.com/conductorone/dpop/integrations/dpop_oauth2 # github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc