rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix PostHog source disconnected on EU cloud OAuth

Browse Source 
The connection-status probe and the access-review driver each had
their own copy of the "try us.posthog.com, then eu.posthog.com"
region-discovery loop, and they drifted. The driver skips a region
that rejects the token (wrong region) and uses the one that answers
2xx; the probe instead returned "credential rejected" on the first
region's 401/403, before ever trying the second.

PostHog Cloud US and EU are separate deployments, so an EU OAuth
token is a 401 on us.posthog.com, which is probed first. The probe
bailed there and marked the source disconnected, while access-review
campaigns -- which use the driver -- kept working.

Delete the probe's copy and delegate to the driver's now-exported
ResolvePostHogRegion, the single resolver the campaign also uses. It
flags a credential every region rejected (ErrPostHogCredentialRejected)
apart from a transient failure on the token's own region, so the probe
marks a source disconnected only for a genuinely dead token and does
not flap on a passing 5xx.

Signed-off-by: Aurélien Sibiril <81782+aureliensibiril@users.noreply.github.com>
This commit is contained in:
Aurélien Sibiril
2026-07-22 22:32:51 +02:00
parent 386962dff6
commit 3805afc806
3 changed files with 146 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
pkg/accessreview/drivers/posthog.go
View File

@@ -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)
}

49
pkg/connector/provider/probe.go
View File

@@ -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
}

86
pkg/connector/provider/probe_test.go
View File

@@ -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()