From 3d4b215b8ff88a2eaaee00c1f03e29a5e17bff34 Mon Sep 17 00:00:00 2001 From: Bryan Frimin Date: Sun, 1 Feb 2026 21:22:31 +0100 Subject: [PATCH] Add bridge backend for sync Signed-off-by: Bryan Frimin --- pkg/connector/oauth2.go | 119 ++++++++++-- pkg/connector/registry.go | 23 +++ pkg/coredata/connector.go | 54 ++++++ pkg/coredata/migrations/20260129T120000Z.sql | 7 + pkg/coredata/migrations/20260201T120000Z.sql | 8 + pkg/coredata/scim_bridge.go | 137 ++++++++++++-- pkg/coredata/scim_bridge_state.go | 12 +- .../syncer.go => iam/scim/bridge/bridge.go} | 76 +++----- .../scim => iam/scim/bridge/client}/client.go | 45 +++-- .../provider/googleworkspace/provider.go | 40 ++-- .../scim/bridge}/provider/provider.go | 4 +- pkg/iam/scim/bridge_runner.go | 166 +++++++++++++++++ pkg/iam/scim/bridge_runner_backoff.go | 48 +++++ pkg/iam/scim/bridge_runner_state.go | 173 ++++++++++++++++++ pkg/iam/scim/bridge_runner_sync.go | 173 ++++++++++++++++++ pkg/iam/scim/service.go | 38 +++- pkg/iam/service.go | 26 ++- pkg/probod/probod.go | 11 +- .../user.go => probod/scim_bridge_config.go} | 23 +-- 19 files changed, 1038 insertions(+), 145 deletions(-) create mode 100644 pkg/coredata/migrations/20260129T120000Z.sql create mode 100644 pkg/coredata/migrations/20260201T120000Z.sql rename pkg/{scimbridge/syncer.go => iam/scim/bridge/bridge.go} (58%) rename pkg/{scimbridge/scim => iam/scim/bridge/client}/client.go (87%) rename pkg/{scimbridge => iam/scim/bridge}/provider/googleworkspace/provider.go (73%) rename pkg/{scimbridge => iam/scim/bridge}/provider/provider.go (89%) create mode 100644 pkg/iam/scim/bridge_runner.go create mode 100644 pkg/iam/scim/bridge_runner_backoff.go create mode 100644 pkg/iam/scim/bridge_runner_state.go create mode 100644 pkg/iam/scim/bridge_runner_sync.go rename pkg/{scimbridge/scim/user.go => probod/scim_bridge_config.go} (61%) diff --git a/pkg/connector/oauth2.go b/pkg/connector/oauth2.go index 2104d9c39..725a814cc 100644 --- a/pkg/connector/oauth2.go +++ b/pkg/connector/oauth2.go @@ -15,7 +15,6 @@ package connector import ( - "bytes" "context" "encoding/json" "fmt" @@ -25,8 +24,10 @@ import ( "strings" "time" + "go.gearno.de/kit/httpclient" "go.probo.inc/probo/pkg/gid" "go.probo.inc/probo/pkg/statelesstoken" + "golang.org/x/oauth2" ) // NOTE: I use client_secret as a salt for the state token, it's an antipattern to @@ -60,6 +61,13 @@ type ( TokenType string `json:"token_type"` Scope string `json:"scope,omitempty"` } + + // OAuth2RefreshConfig contains the OAuth2 credentials needed for token refresh. + OAuth2RefreshConfig struct { + ClientID string + ClientSecret string + TokenURL string + } ) var ( @@ -212,14 +220,30 @@ func (c *OAuth2Connector) CompleteWithState(ctx context.Context, r *http.Request return nil, nil, fmt.Errorf("cannot read token response body: %w", err) } - var oauth2Conn OAuth2Connection - var buf bytes.Buffer - buf.Write(body) - err = json.NewDecoder(&buf).Decode(&oauth2Conn) - if err != nil { + // Parse the raw token response (OAuth2 uses expires_in, not expires_at) + var rawToken struct { + AccessToken string `json:"access_token"` + RefreshToken string `json:"refresh_token"` + ExpiresIn int64 `json:"expires_in"` + TokenType string `json:"token_type"` + Scope string `json:"scope"` + } + if err := json.Unmarshal(body, &rawToken); err != nil { return nil, nil, fmt.Errorf("cannot decode token response: %w", err) } + oauth2Conn := OAuth2Connection{ + AccessToken: rawToken.AccessToken, + RefreshToken: rawToken.RefreshToken, + TokenType: rawToken.TokenType, + Scope: rawToken.Scope, + } + + // Convert expires_in (seconds) to expires_at (absolute time) + if rawToken.ExpiresIn > 0 { + oauth2Conn.ExpiresAt = time.Now().Add(time.Duration(rawToken.ExpiresIn) * time.Second) + } + if provider == SlackProvider { conn, _, err := ParseSlackTokenResponse(body, oauth2Conn, organizationID) return conn, &payload.Data, err @@ -232,17 +256,88 @@ func (c *OAuth2Connection) Type() ProtocolType { return ProtocolOAuth2 } -func (c OAuth2Connection) Client(ctx context.Context) (*http.Client, error) { +func (c *OAuth2Connection) Client(ctx context.Context) (*http.Client, error) { + return c.ClientWithOptions(ctx) +} + +// ClientWithOptions returns an HTTP client with the given options. +// Use this to add logging and tracing to the HTTP client. +func (c *OAuth2Connection) ClientWithOptions(ctx context.Context, opts ...httpclient.Option) (*http.Client, error) { + transport := &oauth2Transport{ + token: c.AccessToken, + tokenType: c.TokenType, + underlying: httpclient.DefaultPooledTransport(opts...), + } client := &http.Client{ - Transport: &oauth2Transport{ - token: c.AccessToken, - tokenType: c.TokenType, - underlying: http.DefaultTransport, - }, + Transport: transport, } return client, nil } +// RefreshableClient returns an HTTP client that automatically refreshes the token when expired. +// It also updates the connection's token fields if a refresh occurs. +func (c *OAuth2Connection) RefreshableClient(ctx context.Context, cfg OAuth2RefreshConfig, opts ...httpclient.Option) (*http.Client, error) { + if c.RefreshToken == "" { + return c.ClientWithOptions(ctx, opts...) + } + + config := &oauth2.Config{ + ClientID: cfg.ClientID, + ClientSecret: cfg.ClientSecret, + Endpoint: oauth2.Endpoint{ + TokenURL: cfg.TokenURL, + }, + } + + // Determine the token expiry + // If ExpiresAt is zero or in the past, set expiry to force a refresh + expiry := c.ExpiresAt + if expiry.IsZero() || expiry.Before(time.Now()) { + // Set expiry to the past to force oauth2 library to refresh + expiry = time.Now().Add(-time.Hour) + } + + token := &oauth2.Token{ + AccessToken: c.AccessToken, + RefreshToken: c.RefreshToken, + Expiry: expiry, + TokenType: c.TokenType, + } + + // Create an HTTP client with telemetry for the oauth2 library to use + // This ensures token refresh requests are also logged + baseClient := &http.Client{ + Transport: httpclient.DefaultPooledTransport(opts...), + } + ctx = context.WithValue(ctx, oauth2.HTTPClient, baseClient) + + // Create a token source that will automatically refresh when expired + tokenSource := config.TokenSource(ctx, token) + + // Get the current (possibly refreshed) token + newToken, err := tokenSource.Token() + if err != nil { + return nil, fmt.Errorf("cannot refresh token: %w", err) + } + + // Update the connection with the potentially refreshed token + c.AccessToken = newToken.AccessToken + c.ExpiresAt = newToken.Expiry + c.TokenType = newToken.TokenType + if newToken.RefreshToken != "" { + c.RefreshToken = newToken.RefreshToken + } + + // Return a client with telemetry that uses the refreshed token + return &http.Client{ + Transport: &oauth2Transport{ + token: newToken.AccessToken, + tokenType: newToken.TokenType, + underlying: httpclient.DefaultPooledTransport(opts...), + }, + }, nil +} + func (c OAuth2Connection) MarshalJSON() ([]byte, error) { type Alias OAuth2Connection return json.Marshal(&struct { diff --git a/pkg/connector/registry.go b/pkg/connector/registry.go index 5b2a37958..53a902f7e 100644 --- a/pkg/connector/registry.go +++ b/pkg/connector/registry.go @@ -73,3 +73,26 @@ func (cr *ConnectorRegistry) Complete(ctx context.Context, provider string, r *h return connector.Complete(ctx, r) } + +// GetOAuth2RefreshConfig returns the OAuth2 refresh configuration for a provider. +// Returns nil if the provider is not found or is not an OAuth2 connector. +func (cr *ConnectorRegistry) GetOAuth2RefreshConfig(provider string) *OAuth2RefreshConfig { + cr.RLock() + defer cr.RUnlock() + + connector, ok := cr.connectors[provider] + if !ok { + return nil + } + + oauth2Connector, ok := connector.(*OAuth2Connector) + if !ok { + return nil + } + + return &OAuth2RefreshConfig{ + ClientID: oauth2Connector.ClientID, + ClientSecret: oauth2Connector.ClientSecret, + TokenURL: oauth2Connector.TokenURL, + } +} diff --git a/pkg/coredata/connector.go b/pkg/coredata/connector.go index dc4ac461d..b02d7c396 100644 --- a/pkg/coredata/connector.go +++ b/pkg/coredata/connector.go @@ -507,6 +507,60 @@ ORDER BY return nil } +func (c *Connector) Update( + ctx context.Context, + conn pg.Conn, + scope Scoper, + encryptionKey cipher.EncryptionKey, +) error { + q := ` +UPDATE connectors +SET + settings = @settings, + encrypted_connection = @encrypted_connection, + updated_at = @updated_at +WHERE + %s + AND id = @id +` + + q = fmt.Sprintf(q, scope.SQLFragment()) + + if c.Connection == nil { + return fmt.Errorf("connection is nil") + } + + c.extractSlackSettings() + + connection, err := json.Marshal(c.Connection) + if err != nil { + return fmt.Errorf("cannot marshal connection: %w", err) + } + + encryptedConnection, err := cipher.Encrypt(connection, encryptionKey) + if err != nil { + return fmt.Errorf("cannot encrypt connection: %w", err) + } + + args := pgx.StrictNamedArgs{ + "id": c.ID, + "settings": c.Settings, + "encrypted_connection": encryptedConnection, + "updated_at": c.UpdatedAt, + } + maps.Copy(args, scope.SQLArguments()) + + _, err = conn.Exec(ctx, q, args) + if err != nil { + return fmt.Errorf("cannot update connector: %w", err) + } + + c.EncryptedConnection = encryptedConnection + c.populateSlackSettings() + + return nil +} + func (c *Connectors) decryptConnections(encryptionKey cipher.EncryptionKey) error { for _, cnnctr := range *c { if len(cnnctr.EncryptedConnection) == 0 { diff --git a/pkg/coredata/migrations/20260129T120000Z.sql b/pkg/coredata/migrations/20260129T120000Z.sql new file mode 100644 index 000000000..1d5cd4d5d --- /dev/null +++ b/pkg/coredata/migrations/20260129T120000Z.sql @@ -0,0 +1,7 @@ +-- Add sync tracking fields to iam_scim_bridges table +ALTER TABLE iam_scim_bridges ADD COLUMN last_synced_at TIMESTAMP WITH TIME ZONE; +ALTER TABLE iam_scim_bridges ADD COLUMN next_sync_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(); +ALTER TABLE iam_scim_bridges ADD COLUMN sync_error TEXT; + +-- Create index for efficient polling of bridges due for sync +CREATE INDEX idx_iam_scim_bridges_next_sync ON iam_scim_bridges (next_sync_at) WHERE state = 'ACTIVE'; diff --git a/pkg/coredata/migrations/20260201T120000Z.sql b/pkg/coredata/migrations/20260201T120000Z.sql new file mode 100644 index 000000000..69b89a8a9 --- /dev/null +++ b/pkg/coredata/migrations/20260201T120000Z.sql @@ -0,0 +1,8 @@ +-- Add failure tracking columns to iam_scim_bridges table +ALTER TABLE iam_scim_bridges ADD COLUMN consecutive_failures INTEGER NOT NULL DEFAULT 0; +ALTER TABLE iam_scim_bridges ADD COLUMN total_sync_count INTEGER NOT NULL DEFAULT 0; +ALTER TABLE iam_scim_bridges ADD COLUMN total_failure_count INTEGER NOT NULL DEFAULT 0; + +-- Drop old index and create new one that covers all processable states +DROP INDEX IF EXISTS idx_iam_scim_bridges_next_sync; +CREATE INDEX idx_iam_scim_bridges_next_sync ON iam_scim_bridges (next_sync_at) WHERE state IN ('ACTIVE', 'FAILED', 'SYNCING'); diff --git a/pkg/coredata/scim_bridge.go b/pkg/coredata/scim_bridge.go index 6db56d3e5..b7518c348 100644 --- a/pkg/coredata/scim_bridge.go +++ b/pkg/coredata/scim_bridge.go @@ -29,19 +29,27 @@ import ( type ( SCIMBridge struct { - ID gid.GID `db:"id"` - OrganizationID gid.GID `db:"organization_id"` - ScimConfigurationID gid.GID `db:"scim_configuration_id"` - ConnectorID *gid.GID `db:"connector_id"` - Type SCIMBridgeType `db:"type"` - State SCIMBridgeState `db:"state"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` + ID gid.GID `db:"id"` + OrganizationID gid.GID `db:"organization_id"` + ScimConfigurationID gid.GID `db:"scim_configuration_id"` + ConnectorID *gid.GID `db:"connector_id"` + Type SCIMBridgeType `db:"type"` + State SCIMBridgeState `db:"state"` + LastSyncedAt *time.Time `db:"last_synced_at"` + NextSyncAt *time.Time `db:"next_sync_at"` + SyncError *string `db:"sync_error"` + ConsecutiveFailures int `db:"consecutive_failures"` + TotalSyncCount int `db:"total_sync_count"` + TotalFailureCount int `db:"total_failure_count"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` } SCIMBridges []*SCIMBridge ) +var ErrNoSCIMBridgeAvailable = errors.New("no SCIM bridge available for sync") + func (s *SCIMBridge) CursorKey(orderBy SCIMBridgeOrderField) page.CursorKey { switch orderBy { case SCIMBridgeOrderFieldCreatedAt: @@ -79,6 +87,12 @@ SELECT connector_id, type, state, + last_synced_at, + next_sync_at, + sync_error, + consecutive_failures, + total_sync_count, + total_failure_count, created_at, updated_at FROM @@ -127,6 +141,12 @@ SELECT connector_id, type, state, + last_synced_at, + next_sync_at, + sync_error, + consecutive_failures, + total_sync_count, + total_failure_count, created_at, updated_at FROM @@ -175,6 +195,12 @@ SELECT connector_id, type, state, + last_synced_at, + next_sync_at, + sync_error, + consecutive_failures, + total_sync_count, + total_failure_count, created_at, updated_at FROM @@ -223,6 +249,12 @@ INSERT INTO iam_scim_bridges ( connector_id, type, state, + last_synced_at, + next_sync_at, + sync_error, + consecutive_failures, + total_sync_count, + total_failure_count, created_at, updated_at ) VALUES ( @@ -233,6 +265,12 @@ INSERT INTO iam_scim_bridges ( @connector_id, @type, @state, + @last_synced_at, + @next_sync_at, + @sync_error, + @consecutive_failures, + @total_sync_count, + @total_failure_count, @created_at, @updated_at ) @@ -246,6 +284,12 @@ INSERT INTO iam_scim_bridges ( "connector_id": s.ConnectorID, "type": s.Type, "state": s.State, + "last_synced_at": s.LastSyncedAt, + "next_sync_at": s.NextSyncAt, + "sync_error": s.SyncError, + "consecutive_failures": s.ConsecutiveFailures, + "total_sync_count": s.TotalSyncCount, + "total_failure_count": s.TotalFailureCount, "created_at": s.CreatedAt, "updated_at": s.UpdatedAt, } @@ -268,6 +312,12 @@ UPDATE iam_scim_bridges SET connector_id = @connector_id, state = @state, + last_synced_at = @last_synced_at, + next_sync_at = @next_sync_at, + sync_error = @sync_error, + consecutive_failures = @consecutive_failures, + total_sync_count = @total_sync_count, + total_failure_count = @total_failure_count, updated_at = @updated_at WHERE %s @@ -277,10 +327,16 @@ WHERE q = fmt.Sprintf(q, scope.SQLFragment()) args := pgx.StrictNamedArgs{ - "id": s.ID, - "connector_id": s.ConnectorID, - "state": s.State, - "updated_at": s.UpdatedAt, + "id": s.ID, + "connector_id": s.ConnectorID, + "state": s.State, + "last_synced_at": s.LastSyncedAt, + "next_sync_at": s.NextSyncAt, + "sync_error": s.SyncError, + "consecutive_failures": s.ConsecutiveFailures, + "total_sync_count": s.TotalSyncCount, + "total_failure_count": s.TotalFailureCount, + "updated_at": s.UpdatedAt, } maps.Copy(args, scope.SQLArguments()) @@ -293,6 +349,63 @@ WHERE return nil } +func (s *SCIMBridge) LoadNextForSyncSkipLocked( + ctx context.Context, + conn pg.Conn, + staleSyncThreshold time.Duration, +) error { + staleCutoff := time.Now().Add(-staleSyncThreshold) + + q := ` +SELECT + id, + organization_id, + scim_configuration_id, + connector_id, + type, + state, + last_synced_at, + next_sync_at, + sync_error, + consecutive_failures, + total_sync_count, + total_failure_count, + created_at, + updated_at +FROM + iam_scim_bridges +WHERE + (state IN (@state_active, @state_failed) AND (next_sync_at IS NULL OR next_sync_at <= NOW())) + OR (state = @state_syncing AND updated_at < @stale_cutoff) +ORDER BY + next_sync_at ASC NULLS FIRST +LIMIT 1 +FOR UPDATE SKIP LOCKED +` + args := pgx.StrictNamedArgs{ + "state_active": SCIMBridgeStateActive, + "state_failed": SCIMBridgeStateFailed, + "state_syncing": SCIMBridgeStateSyncing, + "stale_cutoff": staleCutoff, + } + + rows, err := conn.Query(ctx, q, args) + if err != nil { + return fmt.Errorf("cannot query iam_scim_bridges: %w", err) + } + + bridge, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[SCIMBridge]) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrNoSCIMBridgeAvailable + } + return fmt.Errorf("cannot collect scim_bridge: %w", err) + } + + *s = bridge + return nil +} + func (s *SCIMBridge) Delete( ctx context.Context, conn pg.Conn, diff --git a/pkg/coredata/scim_bridge_state.go b/pkg/coredata/scim_bridge_state.go index c61aa2e1f..e0e32ce0b 100644 --- a/pkg/coredata/scim_bridge_state.go +++ b/pkg/coredata/scim_bridge_state.go @@ -22,9 +22,11 @@ import ( type SCIMBridgeState string const ( - SCIMBridgeStatePending SCIMBridgeState = "PENDING" - SCIMBridgeStateActive SCIMBridgeState = "ACTIVE" - SCIMBridgeStateFailed SCIMBridgeState = "FAILED" + SCIMBridgeStatePending SCIMBridgeState = "PENDING" + SCIMBridgeStateActive SCIMBridgeState = "ACTIVE" + SCIMBridgeStateSyncing SCIMBridgeState = "SYNCING" + SCIMBridgeStateFailed SCIMBridgeState = "FAILED" + SCIMBridgeStateDisabled SCIMBridgeState = "DISABLED" ) func (s SCIMBridgeState) String() string { @@ -47,8 +49,12 @@ func (s *SCIMBridgeState) Scan(value any) error { *s = SCIMBridgeStatePending case "ACTIVE": *s = SCIMBridgeStateActive + case "SYNCING": + *s = SCIMBridgeStateSyncing case "FAILED": *s = SCIMBridgeStateFailed + case "DISABLED": + *s = SCIMBridgeStateDisabled default: return fmt.Errorf("invalid SCIMBridgeState value: %q", str) } diff --git a/pkg/scimbridge/syncer.go b/pkg/iam/scim/bridge/bridge.go similarity index 58% rename from pkg/scimbridge/syncer.go rename to pkg/iam/scim/bridge/bridge.go index 922b967ce..aca95c2df 100644 --- a/pkg/scimbridge/syncer.go +++ b/pkg/iam/scim/bridge/bridge.go @@ -14,46 +14,43 @@ // Package scimbridge provides a bridge for synchronizing users from identity // providers to SCIM-compliant systems. -package scimbridge +package bridge import ( "context" + "errors" "fmt" "strings" - "go.gearno.de/kit/log" - - "go.probo.inc/probo/pkg/scimbridge/provider" - "go.probo.inc/probo/pkg/scimbridge/scim" + scimclient "go.probo.inc/probo/pkg/iam/scim/bridge/client" + "go.probo.inc/probo/pkg/iam/scim/bridge/provider" ) type ( - Syncer struct { + Bridge struct { provider provider.Provider - scimClient *scim.Client + scimClient *scimclient.Client forceUpdate bool dryRun bool - logger *log.Logger } - Option func(*Syncer) + Option func(*Bridge) ) func WithDryRun(dryRun bool) Option { - return func(s *Syncer) { + return func(s *Bridge) { s.dryRun = dryRun } } func WithForceUpdate(forceUpdate bool) Option { - return func(s *Syncer) { + return func(s *Bridge) { s.forceUpdate = forceUpdate } } -func NewSyncer(logger *log.Logger, provider provider.Provider, scimClient *scim.Client, opts ...Option) *Syncer { - s := &Syncer{ - logger: logger, +func NewBridge(provider provider.Provider, scimClient *scimclient.Client, opts ...Option) *Bridge { + s := &Bridge{ provider: provider, scimClient: scimClient, } @@ -65,31 +62,18 @@ func NewSyncer(logger *log.Logger, provider provider.Provider, scimClient *scim. return s } -func (s *Syncer) Run(ctx context.Context) (created, updated, deactivated, skipped, errors int, err error) { - s.logger.InfoCtx(ctx, "starting SCIM bridge sync", - log.String("provider", s.provider.Name()), - log.Bool("dry_run", s.dryRun), - log.Bool("force_update", s.forceUpdate), - ) - +func (s *Bridge) Run(ctx context.Context) (created, updated, deactivated, skipped int, err error) { providerUsers, err := s.provider.ListUsers(ctx) if err != nil { - return 0, 0, 0, 0, 0, fmt.Errorf("cannot list provider users: %w", err) + return 0, 0, 0, 0, fmt.Errorf("cannot list provider users: %w", err) } - s.logger.InfoCtx(ctx, "fetched users from provider", - log.String("provider", s.provider.Name()), - log.Int("count", len(providerUsers)), - ) scimUsers, err := s.scimClient.ListUsers(ctx) if err != nil { - return 0, 0, 0, 0, 0, fmt.Errorf("cannot list scim users: %w", err) + return 0, 0, 0, 0, fmt.Errorf("cannot list scim users: %w", err) } - s.logger.InfoCtx(ctx, "fetched existing SCIM users", - log.Int("count", len(scimUsers)), - ) - scimUsersByEmail := make(map[string]*scim.User) + scimUsersByEmail := make(map[string]*scimclient.User) for i := range scimUsers { email := strings.ToLower(scimUsers[i].UserName) scimUsersByEmail[email] = &scimUsers[i] @@ -97,6 +81,8 @@ func (s *Syncer) Run(ctx context.Context) (created, updated, deactivated, skippe providerEmails := make(map[string]bool) + var errs []error + for _, pu := range providerUsers { email := strings.ToLower(pu.UserName) providerEmails[email] = true @@ -105,11 +91,7 @@ func (s *Syncer) Run(ctx context.Context) (created, updated, deactivated, skippe if !exists { if !s.dryRun { if err := s.scimClient.CreateUser(ctx, &pu); err != nil { - s.logger.ErrorCtx(ctx, "cannot create user", - log.String("email", pu.UserName), - log.Error(err), - ) - errors++ + errs = append(errs, fmt.Errorf("cannot create user %q: %w", pu.UserName, err)) continue } } @@ -127,11 +109,7 @@ func (s *Syncer) Run(ctx context.Context) (created, updated, deactivated, skippe if needsUpdate { if !s.dryRun { if err := s.scimClient.UpdateUser(ctx, existingSCIM.ID, &pu); err != nil { - s.logger.ErrorCtx(ctx, "cannot update user", - log.String("email", pu.UserName), - log.Error(err), - ) - errors++ + errs = append(errs, fmt.Errorf("cannot update user %q: %w", pu.UserName, err)) continue } } @@ -153,24 +131,12 @@ func (s *Syncer) Run(ctx context.Context) (created, updated, deactivated, skippe if !s.dryRun { if err := s.scimClient.DeactivateUser(ctx, scimUser.ID); err != nil { - s.logger.ErrorCtx(ctx, "cannot deactivate user", - log.String("email", email), - log.Error(err), - ) - errors++ + errs = append(errs, fmt.Errorf("cannot deactivate user %q: %w", email, err)) continue } } deactivated++ } - s.logger.InfoCtx(ctx, "sync completed", - log.Int("created", created), - log.Int("updated", updated), - log.Int("deactivated", deactivated), - log.Int("skipped", skipped), - log.Int("errors", errors), - ) - - return created, updated, deactivated, skipped, errors, nil + return created, updated, deactivated, skipped, errors.Join(errs...) } diff --git a/pkg/scimbridge/scim/client.go b/pkg/iam/scim/bridge/client/client.go similarity index 87% rename from pkg/scimbridge/scim/client.go rename to pkg/iam/scim/bridge/client/client.go index a320fb27c..66da7e7d8 100644 --- a/pkg/scimbridge/scim/client.go +++ b/pkg/iam/scim/bridge/client/client.go @@ -12,7 +12,7 @@ // OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR // PERFORMANCE OF THIS SOFTWARE. -package scim +package scimclient import ( "bytes" @@ -23,26 +23,45 @@ import ( "net/http" "net/url" "strings" - - "go.gearno.de/kit/httpclient" ) -type Client struct { - endpoint string - token string - httpClient *http.Client -} +type ( + Client struct { + endpoint string + token string + httpClient *http.Client + } -func NewClient(endpoint, token string) *Client { + User struct { + ID string `json:"id,omitempty"` + UserName string `json:"userName"` + DisplayName string `json:"displayName"` + GivenName string `json:"-"` + FamilyName string `json:"-"` + Active bool `json:"active"` + } + + Users []User + + ListResponse struct { + Schemas []string `json:"schemas"` + TotalResults int `json:"totalResults"` + StartIndex int `json:"startIndex"` + ItemsPerPage int `json:"itemsPerPage"` + Resources Users `json:"Resources"` + } +) + +func NewClient(httpClient *http.Client, endpoint, token string) *Client { return &Client{ endpoint: strings.TrimSuffix(endpoint, "/"), token: token, - httpClient: httpclient.DefaultPooledClient(), + httpClient: httpClient, } } -func (c *Client) ListUsers(ctx context.Context) ([]User, error) { - var allUsers []User +func (c *Client) ListUsers(ctx context.Context) (Users, error) { + var allUsers Users startIndex := 1 count := 100 @@ -64,7 +83,7 @@ func (c *Client) ListUsers(ctx context.Context) ([]User, error) { return allUsers, nil } -func (c *Client) listUsersPage(ctx context.Context, startIndex, count int) ([]User, int, error) { +func (c *Client) listUsersPage(ctx context.Context, startIndex, count int) (Users, int, error) { reqURL := fmt.Sprintf("%s/Users?startIndex=%d&count=%d", c.endpoint, startIndex, count) req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil) diff --git a/pkg/scimbridge/provider/googleworkspace/provider.go b/pkg/iam/scim/bridge/provider/googleworkspace/provider.go similarity index 73% rename from pkg/scimbridge/provider/googleworkspace/provider.go rename to pkg/iam/scim/bridge/provider/googleworkspace/provider.go index c0bb59346..07cb3e7a0 100644 --- a/pkg/scimbridge/provider/googleworkspace/provider.go +++ b/pkg/iam/scim/bridge/provider/googleworkspace/provider.go @@ -13,31 +13,30 @@ // PERFORMANCE OF THIS SOFTWARE. // Package googleworkspace provides a Google Workspace identity provider -// for SCIM synchronization. +// for SCIM synchronization using OAuth2. package googleworkspace import ( "context" "fmt" + "net/http" - "golang.org/x/oauth2/google" admin "google.golang.org/api/admin/directory/v1" "google.golang.org/api/option" - "go.probo.inc/probo/pkg/scimbridge/scim" + scimclient "go.probo.inc/probo/pkg/iam/scim/bridge/client" + "go.probo.inc/probo/pkg/iam/scim/bridge/provider" ) -type ( - Provider struct { - serviceAccountKey []byte - adminEmail string - } -) +var _ provider.Provider = (*Provider)(nil) -func New(serviceAccountKey []byte, adminEmail string) *Provider { +type Provider struct { + httpClient *http.Client +} + +func New(httpClient *http.Client) *Provider { return &Provider{ - serviceAccountKey: serviceAccountKey, - adminEmail: adminEmail, + httpClient: httpClient, } } @@ -45,24 +44,17 @@ func (p *Provider) Name() string { return "google-workspace" } -func (p *Provider) ListUsers(ctx context.Context) ([]scim.User, error) { - config, err := google.JWTConfigFromJSON(p.serviceAccountKey, admin.AdminDirectoryUserReadonlyScope) - if err != nil { - return nil, fmt.Errorf("cannot create JWT config: %w", err) - } - - config.Subject = p.adminEmail - - adminService, err := admin.NewService(ctx, option.WithHTTPClient(config.Client(ctx))) +func (p *Provider) ListUsers(ctx context.Context) (scimclient.Users, error) { + adminService, err := admin.NewService(ctx, option.WithHTTPClient(p.httpClient)) if err != nil { return nil, fmt.Errorf("cannot create admin service: %w", err) } - var allUsers []scim.User + var allUsers scimclient.Users pageToken := "" for { - call := adminService.Users.List().Customer("my_customer").MaxResults(500) + call := adminService.Users.List().Customer("my_customer").MaxResults(500).Context(ctx) if pageToken != "" { call = call.PageToken(pageToken) } @@ -75,7 +67,7 @@ func (p *Provider) ListUsers(ctx context.Context) ([]scim.User, error) { for _, u := range resp.Users { allUsers = append( allUsers, - scim.User{ + scimclient.User{ UserName: u.PrimaryEmail, DisplayName: u.Name.FullName, GivenName: u.Name.GivenName, diff --git a/pkg/scimbridge/provider/provider.go b/pkg/iam/scim/bridge/provider/provider.go similarity index 89% rename from pkg/scimbridge/provider/provider.go rename to pkg/iam/scim/bridge/provider/provider.go index 586f1c0e6..8a7eb53d2 100644 --- a/pkg/scimbridge/provider/provider.go +++ b/pkg/iam/scim/bridge/provider/provider.go @@ -19,10 +19,10 @@ package provider import ( "context" - "go.probo.inc/probo/pkg/scimbridge/scim" + scimclient "go.probo.inc/probo/pkg/iam/scim/bridge/client" ) type Provider interface { Name() string - ListUsers(ctx context.Context) ([]scim.User, error) + ListUsers(ctx context.Context) (scimclient.Users, error) } diff --git a/pkg/iam/scim/bridge_runner.go b/pkg/iam/scim/bridge_runner.go new file mode 100644 index 000000000..860ed3445 --- /dev/null +++ b/pkg/iam/scim/bridge_runner.go @@ -0,0 +1,166 @@ +// Copyright (c) 2025 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package scim + +import ( + "context" + "errors" + "time" + + "github.com/prometheus/client_golang/prometheus" + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + "go.probo.inc/probo/pkg/baseurl" + "go.probo.inc/probo/pkg/connector" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/crypto/cipher" +) + +type ( + // BridgeRunnerConfig holds the configuration for the SCIM bridge runner. + BridgeRunnerConfig struct { + // Interval is the time between sync attempts for each bridge. + Interval time.Duration + // PollInterval is the time between polling for bridges to sync. + PollInterval time.Duration + // SyncTimeout is the maximum time allowed for a single sync operation. + SyncTimeout time.Duration + // BaseURL is the base URL of the API server (used to construct SCIM endpoint). + BaseURL *baseurl.BaseURL + // MaxBackoff is the maximum backoff duration between retries for failed bridges. + MaxBackoff time.Duration + // MaxConsecutiveFailures is the maximum number of consecutive failures + // before a bridge is automatically disabled. + MaxConsecutiveFailures int + // StaleSyncThreshold is the time after which a SYNCING bridge is considered + // stale and can be recovered by another runner (handles crashed runners). + StaleSyncThreshold time.Duration + } + + // BridgeRunner is the SCIM bridge background runner. + BridgeRunner struct { + pg *pg.Client + logger *log.Logger + tp trace.TracerProvider + tracer trace.Tracer + registerer prometheus.Registerer + encryptionKey cipher.EncryptionKey + connectorRegistry *connector.ConnectorRegistry + cfg BridgeRunnerConfig + } +) + +// NewBridgeRunner creates a new SCIM bridge runner. +func NewBridgeRunner( + pgClient *pg.Client, + logger *log.Logger, + tp trace.TracerProvider, + registerer prometheus.Registerer, + encryptionKey cipher.EncryptionKey, + connectorRegistry *connector.ConnectorRegistry, + cfg BridgeRunnerConfig, +) *BridgeRunner { + if cfg.Interval == 0 { + cfg.Interval = 15 * time.Minute + } + if cfg.PollInterval == 0 { + cfg.PollInterval = 30 * time.Second + } + if cfg.SyncTimeout == 0 { + cfg.SyncTimeout = 5 * time.Minute + } + if cfg.MaxBackoff == 0 { + cfg.MaxBackoff = DefaultMaxBackoff + } + if cfg.MaxConsecutiveFailures == 0 { + cfg.MaxConsecutiveFailures = DefaultMaxConsecutiveFailures + } + if cfg.StaleSyncThreshold == 0 { + cfg.StaleSyncThreshold = DefaultStaleSyncThreshold + } + + return &BridgeRunner{ + pg: pgClient, + logger: logger, + tp: tp, + tracer: tp.Tracer("scim-bridge-runner"), + registerer: registerer, + encryptionKey: encryptionKey, + connectorRegistry: connectorRegistry, + cfg: cfg, + } +} + +// Run starts the runner loop that processes SCIM bridges. +func (r *BridgeRunner) Run(ctx context.Context) error { + r.logger.InfoCtx(ctx, "starting SCIM bridge runner", + log.Duration("poll_interval", r.cfg.PollInterval), + log.Duration("sync_interval", r.cfg.Interval), + log.Duration("sync_timeout", r.cfg.SyncTimeout), + log.Duration("max_backoff", r.cfg.MaxBackoff), + log.Int("max_consecutive_failures", r.cfg.MaxConsecutiveFailures), + log.Duration("stale_sync_threshold", r.cfg.StaleSyncThreshold), + ) + + ticker := time.NewTicker(r.cfg.PollInterval) + defer ticker.Stop() + + for { + if err := r.processBridge(ctx); err != nil { + if !errors.Is(err, coredata.ErrNoSCIMBridgeAvailable) { + r.logger.ErrorCtx(ctx, "cannot process SCIM bridge", log.Error(err)) + } + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func (r *BridgeRunner) processBridge(ctx context.Context) error { + bridge, scope, err := r.acquireNextBridge(ctx) + if err != nil { + return err + } + + ctx, span := r.tracer.Start(ctx, "scim-bridge-runner.processBridge") + defer span.End() + + logger := r.logger.Named("bridge-sync").With( + log.String("bridge_id", bridge.ID.String()), + log.String("scim_configuration_id", bridge.ScimConfigurationID.String()), + log.String("bridge_type", string(bridge.Type)), + log.Int("consecutive_failures", bridge.ConsecutiveFailures), + ) + + logger.InfoCtx(ctx, "starting sync") + + syncCtx, cancel := context.WithTimeout(ctx, r.cfg.SyncTimeout) + defer cancel() + + stats, duration, connector, err := r.executeSync(syncCtx, bridge, scope, logger) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "sync failed") + return r.transitionToFailed(ctx, bridge, scope, err, duration, logger) + } + + return r.transitionToSuccess(ctx, bridge, scope, stats, duration, connector, logger) +} diff --git a/pkg/iam/scim/bridge_runner_backoff.go b/pkg/iam/scim/bridge_runner_backoff.go new file mode 100644 index 000000000..6cf6abfb0 --- /dev/null +++ b/pkg/iam/scim/bridge_runner_backoff.go @@ -0,0 +1,48 @@ +// Copyright (c) 2025 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package scim + +import "time" + +const ( + // DefaultMaxConsecutiveFailures is the maximum number of consecutive failures + // before a bridge is disabled. + DefaultMaxConsecutiveFailures = 10 + + // DefaultMaxBackoff is the maximum backoff duration between retries. + DefaultMaxBackoff = 24 * time.Hour + + // DefaultStaleSyncThreshold is the time after which a SYNCING bridge is + // considered stale and can be recovered by another runner. + DefaultStaleSyncThreshold = 10 * time.Minute +) + +func (r *BridgeRunner) calculateBackoff(consecutiveFailures int) time.Duration { + if consecutiveFailures <= 0 { + return r.cfg.Interval + } + + backoff := r.cfg.Interval * time.Duration(1< r.cfg.MaxBackoff { + return r.cfg.MaxBackoff + } + + return backoff +} + +func (r *BridgeRunner) shouldDisable(consecutiveFailures int) bool { + return consecutiveFailures >= r.cfg.MaxConsecutiveFailures +} diff --git a/pkg/iam/scim/bridge_runner_state.go b/pkg/iam/scim/bridge_runner_state.go new file mode 100644 index 000000000..94a5eebe1 --- /dev/null +++ b/pkg/iam/scim/bridge_runner_state.go @@ -0,0 +1,173 @@ +// Copyright (c) 2025 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package scim + +import ( + "context" + "time" + + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/coredata" +) + +// SyncStats holds the statistics from a sync operation. +type SyncStats struct { + Created int + Updated int + Deactivated int + Skipped int +} + +func (r *BridgeRunner) acquireNextBridge(ctx context.Context) (*coredata.SCIMBridge, coredata.Scoper, error) { + var bridge *coredata.SCIMBridge + var scope coredata.Scoper + + err := r.pg.WithTx( + ctx, + func(tx pg.Conn) error { + bridge = &coredata.SCIMBridge{} + if err := bridge.LoadNextForSyncSkipLocked(ctx, tx, r.cfg.StaleSyncThreshold); err != nil { + return err + } + + scope = coredata.NewScope(bridge.ID.TenantID()) + + now := time.Now() + bridge.State = coredata.SCIMBridgeStateSyncing + bridge.UpdatedAt = now + + return bridge.Update(ctx, tx, scope) + }, + ) + + if err != nil { + return nil, nil, err + } + + return bridge, scope, nil +} + +func (r *BridgeRunner) transitionToSuccess( + ctx context.Context, + bridge *coredata.SCIMBridge, + scope coredata.Scoper, + stats SyncStats, + duration time.Duration, + connector *coredata.Connector, + logger *log.Logger, +) error { + return r.pg.WithConn( + ctx, + func(conn pg.Conn) error { + now := time.Now() + nextSync := now.Add(r.cfg.Interval) + + bridge.State = coredata.SCIMBridgeStateActive + bridge.LastSyncedAt = &now + bridge.NextSyncAt = &nextSync + bridge.SyncError = nil + bridge.ConsecutiveFailures = 0 + bridge.TotalSyncCount++ + bridge.UpdatedAt = now + + if err := bridge.Update(ctx, conn, scope); err != nil { + logger.ErrorCtx(ctx, "cannot update bridge after successful sync", + log.Error(err), + ) + return err + } + + if connector != nil { + connector.UpdatedAt = now + if err := connector.Update(ctx, conn, scope, r.encryptionKey); err != nil { + logger.WarnCtx(ctx, "cannot persist refreshed OAuth2 token", + log.String("connector_id", connector.ID.String()), + log.Error(err), + ) + } + } + + logger.InfoCtx(ctx, "sync completed successfully", + log.Duration("sync_duration", duration), + log.Int("users_created", stats.Created), + log.Int("users_updated", stats.Updated), + log.Int("users_deactivated", stats.Deactivated), + log.Int("users_skipped", stats.Skipped), + ) + + return nil + }, + ) +} + +func (r *BridgeRunner) transitionToFailed( + ctx context.Context, + bridge *coredata.SCIMBridge, + scope coredata.Scoper, + syncErr error, + duration time.Duration, + logger *log.Logger, +) error { + return r.pg.WithConn( + ctx, + func(conn pg.Conn) error { + now := time.Now() + + bridge.ConsecutiveFailures++ + bridge.TotalFailureCount++ + bridge.TotalSyncCount++ + bridge.LastSyncedAt = &now + bridge.UpdatedAt = now + + errStr := syncErr.Error() + bridge.SyncError = &errStr + + if r.shouldDisable(bridge.ConsecutiveFailures) { + bridge.State = coredata.SCIMBridgeStateDisabled + bridge.NextSyncAt = nil + + logger.ErrorCtx(ctx, "bridge disabled due to max consecutive failures", + log.Duration("sync_duration", duration), + log.Int("consecutive_failures", bridge.ConsecutiveFailures), + log.Int("max_consecutive_failures", r.cfg.MaxConsecutiveFailures), + log.Error(syncErr), + ) + } else { + bridge.State = coredata.SCIMBridgeStateFailed + backoff := r.calculateBackoff(bridge.ConsecutiveFailures) + nextSync := now.Add(backoff) + bridge.NextSyncAt = &nextSync + + logger.ErrorCtx(ctx, "sync failed, will retry with backoff", + log.Duration("sync_duration", duration), + log.Int("consecutive_failures", bridge.ConsecutiveFailures), + log.Duration("next_retry_in", backoff), + log.Error(syncErr), + ) + } + + if err := bridge.Update(ctx, conn, scope); err != nil { + logger.ErrorCtx(ctx, "cannot update bridge after failed sync", + log.String("new_state", string(bridge.State)), + log.Error(err), + ) + return err + } + + return nil + }, + ) +} diff --git a/pkg/iam/scim/bridge_runner_sync.go b/pkg/iam/scim/bridge_runner_sync.go new file mode 100644 index 000000000..38faf48d0 --- /dev/null +++ b/pkg/iam/scim/bridge_runner_sync.go @@ -0,0 +1,173 @@ +// Copyright (c) 2025 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package scim + +import ( + "context" + "fmt" + "time" + + "go.gearno.de/kit/httpclient" + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/connector" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/iam/scim/bridge" + scimclient "go.probo.inc/probo/pkg/iam/scim/bridge/client" + "go.probo.inc/probo/pkg/iam/scim/bridge/provider" + "go.probo.inc/probo/pkg/iam/scim/bridge/provider/googleworkspace" +) + +func (r *BridgeRunner) executeSync( + ctx context.Context, + bridge *coredata.SCIMBridge, + scope coredata.Scoper, + logger *log.Logger, +) (stats SyncStats, duration time.Duration, connector *coredata.Connector, err error) { + start := time.Now() + + err = r.pg.WithConn( + ctx, + func(conn pg.Conn) error { + var syncErr error + stats, connector, syncErr = r.doSync(ctx, conn, bridge, scope, logger) + return syncErr + }, + ) + + duration = time.Since(start) + return stats, duration, connector, err +} + +func (r *BridgeRunner) doSync( + ctx context.Context, + conn pg.Conn, + scimBridge *coredata.SCIMBridge, + scope coredata.Scoper, + logger *log.Logger, +) (SyncStats, *coredata.Connector, error) { + if scimBridge.ConnectorID == nil { + return SyncStats{}, nil, fmt.Errorf("bridge has no connector configured") + } + + dbConnector := &coredata.Connector{} + if err := dbConnector.LoadByID(ctx, conn, scope, *scimBridge.ConnectorID, r.encryptionKey); err != nil { + return SyncStats{}, nil, fmt.Errorf("cannot load connector: %w", err) + } + + idp, err := r.createProvider(ctx, logger, scimBridge.Type, dbConnector) + if err != nil { + return SyncStats{}, nil, fmt.Errorf("cannot create provider: %w", err) + } + + var scimConfig coredata.SCIMConfiguration + if err := scimConfig.LoadByID(ctx, conn, scope, scimBridge.ScimConfigurationID); err != nil { + return SyncStats{}, nil, fmt.Errorf("cannot load SCIM configuration: %w", err) + } + + token, err := GenerateToken() + if err != nil { + return SyncStats{}, nil, fmt.Errorf("cannot generate SCIM token: %w", err) + } + + scimConfig.HashedToken = HashToken(token) + scimConfig.UpdatedAt = time.Now() + if err := scimConfig.Update(ctx, conn, scope); err != nil { + return SyncStats{}, nil, fmt.Errorf("cannot update SCIM configuration token: %w", err) + } + + scimClient := r.createSCIMClient(logger, token) + syncer := bridge.NewBridge(idp, scimClient) + created, updated, deactivated, skipped, err := syncer.Run(ctx) + if err != nil { + return SyncStats{}, nil, fmt.Errorf("sync failed: %w", err) + } + + stats := SyncStats{ + Created: created, + Updated: updated, + Deactivated: deactivated, + Skipped: skipped, + } + + return stats, dbConnector, nil +} + +func (r *BridgeRunner) createSCIMClient(logger *log.Logger, token string) *scimclient.Client { + scimEndpoint := r.cfg.BaseURL.WithPath("/api/connect/v1/scim/2.0").MustString() + httpClient := httpclient.DefaultPooledClient( + httpclient.WithLogger(logger), + httpclient.WithTracerProvider(r.tp), + httpclient.WithRegisterer(r.registerer), + ) + + return scimclient.NewClient(httpClient, scimEndpoint, token) +} + +func (r *BridgeRunner) createProvider( + ctx context.Context, + logger *log.Logger, + bridgeType coredata.SCIMBridgeType, + dbConnector *coredata.Connector, +) (provider.Provider, error) { + switch bridgeType { + case coredata.SCIMBridgeTypeGoogleWorkspace: + return r.createGoogleWorkspaceProvider(ctx, logger, dbConnector) + default: + return nil, fmt.Errorf("unsupported bridge type: %s", bridgeType) + } +} + +func (r *BridgeRunner) createGoogleWorkspaceProvider( + ctx context.Context, + logger *log.Logger, + dbConnector *coredata.Connector, +) (provider.Provider, error) { + if dbConnector.Connection == nil { + return nil, fmt.Errorf("connector has no connection configured") + } + + oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection) + if !ok { + return nil, fmt.Errorf("connector is not an OAuth2 connection") + } + + httpClientOpts := []httpclient.Option{ + httpclient.WithLogger(logger), + httpclient.WithTracerProvider(r.tp), + httpclient.WithRegisterer(r.registerer), + } + + providerName := dbConnector.Provider.String() + refreshCfg := r.connectorRegistry.GetOAuth2RefreshConfig(providerName) + if refreshCfg == nil { + logger.WarnCtx(ctx, "no OAuth2 refresh config found, using static token", + log.String("connector_id", dbConnector.ID.String()), + log.String("connector_provider", providerName), + ) + httpClient, err := oauth2Conn.ClientWithOptions(ctx, httpClientOpts...) + if err != nil { + return nil, fmt.Errorf("cannot create HTTP client: %w", err) + } + return googleworkspace.New(httpClient), nil + } + + httpClient, err := oauth2Conn.RefreshableClient(ctx, *refreshCfg, httpClientOpts...) + if err != nil { + return nil, fmt.Errorf("cannot create refreshable HTTP client: %w", err) + } + + return googleworkspace.New(httpClient), nil +} diff --git a/pkg/iam/scim/service.go b/pkg/iam/scim/service.go index 26b312f3e..72a0a6e21 100644 --- a/pkg/iam/scim/service.go +++ b/pkg/iam/scim/service.go @@ -27,10 +27,14 @@ import ( "github.com/elimity-com/scim" scimerrors "github.com/elimity-com/scim/errors" "github.com/elimity-com/scim/optional" + "github.com/prometheus/client_golang/prometheus" scimfilter "github.com/scim2/filter-parser/v2" "go.gearno.de/kit/log" "go.gearno.de/kit/pg" + "go.opentelemetry.io/otel/trace" + "go.probo.inc/probo/pkg/connector" "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/crypto/cipher" "go.probo.inc/probo/pkg/gid" "go.probo.inc/probo/pkg/mail" "go.probo.inc/probo/pkg/page" @@ -38,21 +42,47 @@ import ( type ( Service struct { - pg *pg.Client - logger *log.Logger + pg *pg.Client + logger *log.Logger + bridgeRunner *BridgeRunner + } + + ServiceConfig struct { + TracerProvider trace.TracerProvider + Registerer prometheus.Registerer + EncryptionKey cipher.EncryptionKey + ConnectorRegistry *connector.ConnectorRegistry + BridgeRunner BridgeRunnerConfig } ) func NewService( pg *pg.Client, logger *log.Logger, + cfg ServiceConfig, ) *Service { + bridgeRunner := NewBridgeRunner( + pg, + logger.Named("bridge-runner"), + cfg.TracerProvider, + cfg.Registerer, + cfg.EncryptionKey, + cfg.ConnectorRegistry, + cfg.BridgeRunner, + ) + return &Service{ - pg: pg, - logger: logger, + pg: pg, + logger: logger, + bridgeRunner: bridgeRunner, } } +// Run starts the SCIM service background processes. +func (s *Service) Run(ctx context.Context) error { + return s.bridgeRunner.Run(ctx) +} + func HashToken(token string) []byte { hash := sha256.Sum256([]byte(token)) return hash[:] diff --git a/pkg/iam/service.go b/pkg/iam/service.go index ba4b9b3be..7644ba768 100644 --- a/pkg/iam/service.go +++ b/pkg/iam/service.go @@ -7,9 +7,12 @@ import ( "fmt" "time" + "github.com/prometheus/client_golang/prometheus" "go.gearno.de/kit/log" "go.gearno.de/kit/pg" "go.opentelemetry.io/otel/trace" + "go.probo.inc/probo/pkg/baseurl" + "go.probo.inc/probo/pkg/connector" "go.probo.inc/probo/pkg/coredata" "go.probo.inc/probo/pkg/crypto/cipher" "go.probo.inc/probo/pkg/crypto/passwdhash" @@ -58,14 +61,18 @@ type ( SessionDuration time.Duration Bucket string TokenSecret string - BaseURL string + BaseURL *baseurl.BaseURL EncryptionKey cipher.EncryptionKey Certificate *x509.Certificate PrivateKey *rsa.PrivateKey Logger *log.Logger TracerProvider trace.TracerProvider + Registerer prometheus.Registerer + ConnectorRegistry *connector.ConnectorRegistry DomainVerificationInterval time.Duration DomainVerificationResolverAddr string + SCIMBridgeSyncInterval time.Duration + SCIMBridgePollInterval time.Duration } ) @@ -84,7 +91,7 @@ func NewService( return nil, fmt.Errorf("token secret is required") } - if cfg.BaseURL == "" { + if cfg.BaseURL == nil { return nil, fmt.Errorf("base URL is required") } @@ -96,7 +103,7 @@ func NewService( pg: pgClient, fm: fm, hp: hp, - baseURL: cfg.BaseURL, + baseURL: cfg.BaseURL.String(), tokenSecret: cfg.TokenSecret, disableSignup: cfg.DisableSignup, invitationTokenValidity: cfg.InvitationTokenValidity, @@ -124,7 +131,17 @@ func NewService( } svc.SAMLService = samlService - svc.SCIMService = scim.NewService(svc.pg, cfg.Logger.Named("scim")) + svc.SCIMService = scim.NewService(svc.pg, cfg.Logger.Named("scim"), scim.ServiceConfig{ + TracerProvider: cfg.TracerProvider, + Registerer: cfg.Registerer, + EncryptionKey: cfg.EncryptionKey, + ConnectorRegistry: cfg.ConnectorRegistry, + BridgeRunner: scim.BridgeRunnerConfig{ + Interval: cfg.SCIMBridgeSyncInterval, + PollInterval: cfg.SCIMBridgePollInterval, + BaseURL: cfg.BaseURL, + }, + }) svc.samlDomainVerifier = NewSAMLDomainVerifier( pgClient, @@ -142,6 +159,7 @@ func (s *Service) Run(ctx context.Context) error { g.Go(func() error { return s.SAMLService.Run(ctx) }) g.Go(func() error { return s.samlDomainVerifier.Run(ctx) }) + g.Go(func() error { return s.SCIMService.Run(ctx) }) return g.Wait() } diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index ea743b118..5295a164e 100644 --- a/pkg/probod/probod.go +++ b/pkg/probod/probod.go @@ -80,6 +80,7 @@ type ( OpenAI openaiConfig `json:"openai"` ChromeDPAddr string `json:"chrome-dp-addr"` CustomDomains customDomainsConfig `json:"custom-domains"` + SCIMBridge scimBridgeConfig `json:"scim-bridge"` } trustCenterConfig struct { @@ -163,6 +164,10 @@ func New() *Implm { KeyType: "EC256", }, }, + SCIMBridge: scimBridgeConfig{ + SyncInterval: 60, // 15 minutes + PollInterval: 30, // 30 seconds + }, }, } } @@ -300,14 +305,18 @@ func (impl *Implm) Run( SessionDuration: time.Duration(impl.cfg.Auth.Cookie.Duration) * time.Hour, Bucket: impl.cfg.AWS.Bucket, TokenSecret: impl.cfg.Auth.Cookie.Secret, - BaseURL: impl.cfg.BaseURL.String(), + BaseURL: impl.cfg.BaseURL, EncryptionKey: impl.cfg.EncryptionKey, Certificate: samlCert, PrivateKey: samlKey, Logger: l.Named("iam"), TracerProvider: tp, + Registerer: r, + ConnectorRegistry: defaultConnectorRegistry, DomainVerificationInterval: impl.cfg.Auth.SAML.DomainVerificationInterval(), DomainVerificationResolverAddr: impl.cfg.Auth.SAML.DomainVerificationResolverAddr, + SCIMBridgeSyncInterval: time.Duration(impl.cfg.SCIMBridge.SyncInterval) * time.Second, + SCIMBridgePollInterval: time.Duration(impl.cfg.SCIMBridge.PollInterval) * time.Second, }, ) if err != nil { diff --git a/pkg/scimbridge/scim/user.go b/pkg/probod/scim_bridge_config.go similarity index 61% rename from pkg/scimbridge/scim/user.go rename to pkg/probod/scim_bridge_config.go index aac2b7e8d..5415a90de 100644 --- a/pkg/scimbridge/scim/user.go +++ b/pkg/probod/scim_bridge_config.go @@ -12,21 +12,14 @@ // OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR // PERFORMANCE OF THIS SOFTWARE. -package scim +package probod -type User struct { - ID string `json:"id,omitempty"` - UserName string `json:"userName"` - DisplayName string `json:"displayName"` - GivenName string `json:"-"` - FamilyName string `json:"-"` - Active bool `json:"active"` -} +type scimBridgeConfig struct { + // SyncInterval is the time between sync attempts for each bridge (in seconds). + // Default: 900 (15 minutes) + SyncInterval int `json:"sync-interval"` -type ListResponse struct { - Schemas []string `json:"schemas"` - TotalResults int `json:"totalResults"` - StartIndex int `json:"startIndex"` - ItemsPerPage int `json:"itemsPerPage"` - Resources []User `json:"Resources"` + // PollInterval is the time between polling for bridges to sync (in seconds). + // Default: 30 + PollInterval int `json:"poll-interval"` }