diff --git a/pkg/accessreview/drivers/posthog.go b/pkg/accessreview/drivers/posthog.go index e952d230e..4b2620ef3 100644 --- a/pkg/accessreview/drivers/posthog.go +++ b/pkg/accessreview/drivers/posthog.go @@ -23,7 +23,9 @@ package drivers import ( "context" "encoding/json" + "errors" "fmt" + "io" "net/http" "net/url" "strconv" @@ -40,10 +42,14 @@ type PostHogDriver struct { var _ Driver = (*PostHogDriver)(nil) +// PostHogOrganizationPath is the @current organization endpoint. It is the +// single source of truth shared with the connection probe so the two never +// duplicate the path. +const PostHogOrganizationPath = "/api/organizations/@current/" + const ( - posthogMembersPath = "/api/organizations/@current/members/" - posthogOrganizationPath = "/api/organizations/@current/" - posthogMembersPageSize = 100 + posthogMembersPath = "/api/organizations/@current/members/" + posthogMembersPageSize = 100 // PostHog Cloud regional data hosts. OAuth connections carry no region // (empty baseURL): the region-agnostic oauth.posthog.com gateway used @@ -59,6 +65,13 @@ const ( posthogMembershipLevelOwner = 15 ) +// ErrPostHogCredentialRejected reports that every PostHog Cloud region refused +// the token with 401/403 — a definitively dead or revoked credential, as +// opposed to a transient failure (5xx/network) on the token's own region. The +// connection probe uses it to tell a rejected credential apart from an +// inconclusive result, which must not flap the source to disconnected. +var ErrPostHogCredentialRejected = errors.New("posthog rejected the credential on every region") + type ( posthogMembersResponse struct { Next string `json:"next"` @@ -100,7 +113,7 @@ func (d *PostHogDriver) resolveBaseURL(ctx context.Context) error { return nil } - host, err := resolvePostHogRegion(ctx, d.httpClient) + host, err := ResolvePostHogRegion(ctx, d.httpClient) if err != nil { return err } @@ -221,20 +234,32 @@ func PostHogRegionBaseURL(region string) (string, bool) { } } -// resolvePostHogRegion probes the PostHog Cloud region hosts with the given +// ResolvePostHogRegion probes the PostHog Cloud region hosts with the given // token-bearing client and returns the first that answers 2xx on the @current // organization endpoint. OAuth connections authenticate via the region-agnostic // oauth.posthog.com gateway, which does not serve /api, so the actual data // region (us/eu) must be discovered against the regional hosts directly. -func resolvePostHogRegion(ctx context.Context, client *http.Client) (string, error) { +// +// A token is valid on exactly one region; the other rejects it with 401/403. +// The result distinguishes the two failure classes the connection probe needs: +// ErrPostHogCredentialRejected when every region refused the token (dead/revoked +// credential), and a generic error when a region was merely unreachable or +// errored transiently, which must not be read as a rejection. +func ResolvePostHogRegion(ctx context.Context, client *http.Client) (string, error) { + allRejected := true + for _, host := range []string{posthogUSBaseURL, posthogEUBaseURL} { - endpoint, err := url.JoinPath(host, posthogOrganizationPath) + endpoint, err := url.JoinPath(host, PostHogOrganizationPath) if err != nil { + allRejected = false + continue } req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { + allRejected = false + continue } @@ -248,15 +273,28 @@ func resolvePostHogRegion(ctx context.Context, client *http.Client) (string, err return "", fmt.Errorf("cannot resolve posthog region: %w", ctx.Err()) } + allRejected = false + continue } status := resp.StatusCode + _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() if status >= http.StatusOK && status < http.StatusMultipleChoices { return host, nil } + + // Only 401/403 is a credential rejection; a 5xx/429 is transient and + // leaves the verdict inconclusive rather than rejected. + if status != http.StatusUnauthorized && status != http.StatusForbidden { + allRejected = false + } + } + + if allRejected { + return "", fmt.Errorf("cannot resolve posthog region: %w", ErrPostHogCredentialRejected) } return "", fmt.Errorf("cannot resolve posthog region: no region accepted the connection") @@ -375,7 +413,7 @@ func NewPostHogNameResolver(httpClient *http.Client, baseURL string) NameResolve func (r *posthogNameResolver) ResolveInstanceName(ctx context.Context) (string, error) { baseURL := r.baseURL if baseURL == "" { - host, err := resolvePostHogRegion(ctx, r.httpClient) + host, err := ResolvePostHogRegion(ctx, r.httpClient) if err != nil { // Terminal: cannot determine the region (e.g. revoked token). // Keep the generic source name rather than making the @@ -386,7 +424,7 @@ func (r *posthogNameResolver) ResolveInstanceName(ctx context.Context) (string, baseURL = host } - endpoint, err := url.JoinPath(baseURL, posthogOrganizationPath) + endpoint, err := url.JoinPath(baseURL, PostHogOrganizationPath) if err != nil { return "", fmt.Errorf("cannot build posthog organization URL: %w", err) } diff --git a/pkg/connector/provider/probe.go b/pkg/connector/provider/probe.go index 729574385..c04773c22 100644 --- a/pkg/connector/provider/probe.go +++ b/pkg/connector/provider/probe.go @@ -24,6 +24,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -31,6 +32,7 @@ import ( "slices" "strings" + "go.probo.inc/probo/pkg/accessreview/drivers" "go.probo.inc/probo/pkg/connector" "go.probo.inc/probo/pkg/coredata" ) @@ -43,9 +45,6 @@ const ( linearGraphQLEndpoint = "https://api.linear.app/graphql" mondayGraphQLEndpoint = "https://api.monday.com/v2" railwayGraphQLEndpoint = "https://backboard.railway.com/graphql/v2" - posthogOrganizationPath = "/api/organizations/@current/" - posthogUSBaseURL = "https://us.posthog.com" - posthogEUBaseURL = "https://eu.posthog.com" crispAPIBaseURL = "https://api.crisp.chat/v1" crispTierHeader = "X-Crisp-Tier" crispTierValue = "plugin" @@ -406,7 +405,7 @@ func buildPostHogProbeURL(conn *coredata.Connector) (string, error) { return "", nil } - return url.JoinPath(baseURL, posthogOrganizationPath) + return url.JoinPath(baseURL, drivers.PostHogOrganizationPath) } func probeLinear( @@ -597,44 +596,22 @@ func probePostHog( return err } + // Explicit host (API-key region or self-hosted): probe it directly. if probeURL != "" { return probeGET(ctx, httpClient, probeURL) } - for _, host := range []string{posthogUSBaseURL, posthogEUBaseURL} { - endpoint, err := url.JoinPath(host, posthogOrganizationPath) - if err != nil { - continue + // Cloud OAuth (empty BaseURL): reuse the driver's region resolver so the + // probe and the campaign never drift. Only a credential every region + // rejected is disconnected; a transient failure on the token's own region + // stays connected rather than flapping the badge. + if _, err := drivers.ResolvePostHogRegion(ctx, httpClient); err != nil { + if errors.Is(err, drivers.ErrPostHogCredentialRejected) { + return fmt.Errorf("cannot probe posthog: %w", err) } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) - if err != nil { - continue - } - - req.Header.Set("Accept", "application/json") - - resp, err := httpClient.Do(req) - if err != nil { - if ctx.Err() != nil { - return fmt.Errorf("cannot probe posthog region: %w", ctx.Err()) - } - - continue - } - - status := resp.StatusCode - _, _ = io.Copy(io.Discard, resp.Body) - _ = resp.Body.Close() - - if status == http.StatusUnauthorized || status == http.StatusForbidden { - return fmt.Errorf("credential rejected: status %d", status) - } - - if status >= http.StatusOK && status < http.StatusMultipleChoices { - return nil - } + return nil } - return fmt.Errorf("credential rejected: no posthog region accepted the connection") + return nil } diff --git a/pkg/connector/provider/probe_test.go b/pkg/connector/provider/probe_test.go index 13cf5be49..0b9ca68be 100644 --- a/pkg/connector/provider/probe_test.go +++ b/pkg/connector/provider/probe_test.go @@ -30,6 +30,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.probo.inc/probo/pkg/accessreview/drivers" "go.probo.inc/probo/pkg/coredata" ) @@ -120,6 +121,91 @@ func TestBuildPostHogProbeURL(t *testing.T) { assert.Equal(t, "https://us.posthog.com/api/organizations/@current/", probeURL) } +func TestProbePostHog(t *testing.T) { + t.Parallel() + + // A cloud OAuth connection carries no region (empty BaseURL): the token is + // valid on exactly one PostHog region and the other rejects it with + // 401/403. The probe must try every region and only report the credential + // rejected when none accept it — mirroring the access-review driver — so an + // EU token hitting us.posthog.com (probed first) does not falsely mark the + // source disconnected while its access reviews keep working. A transient + // 5xx on the token's own region is inconclusive, not a rejection. + cases := []struct { + name string + baseURL string + hostStatus map[string]int + wantErr bool + wantRejected bool + }{ + { + name: "explicit region accepts", + baseURL: "https://us.posthog.com", + hostStatus: map[string]int{"us.posthog.com": http.StatusOK}, + wantErr: false, + }, + { + name: "explicit region rejects", + baseURL: "https://us.posthog.com", + hostStatus: map[string]int{"us.posthog.com": http.StatusUnauthorized}, + wantErr: true, + }, + { + name: "oauth EU token: US refuses, EU accepts", + baseURL: "", + hostStatus: map[string]int{"us.posthog.com": http.StatusUnauthorized, "eu.posthog.com": http.StatusOK}, + wantErr: false, + }, + { + name: "oauth transient: US refuses, EU errors", + baseURL: "", + hostStatus: map[string]int{"us.posthog.com": http.StatusUnauthorized, "eu.posthog.com": http.StatusInternalServerError}, + wantErr: false, + }, + { + name: "oauth dead token: every region refuses", + baseURL: "", + hostStatus: map[string]int{"us.posthog.com": http.StatusForbidden, "eu.posthog.com": http.StatusForbidden}, + wantErr: true, + wantRejected: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + client := &http.Client{Transport: probeRoundTripFunc(func(r *http.Request) (*http.Response, error) { + status, ok := tc.hostStatus[r.URL.Host] + if !ok { + status = http.StatusNotFound + } + + return &http.Response{StatusCode: status, Body: http.NoBody, Header: make(http.Header)}, nil + })} + + conn := &coredata.Connector{Provider: coredata.ConnectorProviderPostHog} + require.NoError(t, conn.SetSettings(&coredata.PostHogConnectorSettings{BaseURL: tc.baseURL})) + + err := probePostHog(context.Background(), client, conn) + + if !tc.wantErr { + require.NoError(t, err) + + return + } + + require.Error(t, err) + + // A credential every region refused must surface the sentinel so the + // probe distinguishes it from an inconclusive/transient failure. + if tc.wantRejected { + require.ErrorIs(t, err, drivers.ErrPostHogCredentialRejected) + } + }) + } +} + func TestBuildScalewayProbeURL(t *testing.T) { t.Parallel()