package webhooks import ( "context" "fmt" neturl "net/url" "slices" "strings" "time" "github.com/google/uuid" "github.com/lib/pq" "go.opentelemetry.io/otel/attribute" otelcodes "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" svcerrors "github.com/sarathsp06/sparrow/internal/errors" "github.com/sarathsp06/sparrow/internal/storage" "github.com/sarathsp06/sparrow/internal/tenant" "github.com/sarathsp06/sparrow/internal/webhooks/client" "github.com/sarathsp06/sparrow/internal/webhooks/store" ) // setWebhookActive is the shared implementation of PauseWebhook and ResumeWebhook. // It loads the webhook, checks whether a state transition is needed, and persists the change. func (s *WebhookService) setWebhookActive(ctx context.Context, webhookID string, consumer string, active bool) error { if webhookID == "" { return svcerrors.Error(svcerrors.InvalidArgument, "webhook ID is required") } if consumer == "" { return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } tenantID := tenant.FromContext(ctx) id, err := parseUUID(webhookID, "webhook ID") if err != nil { return err } webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer) if err != nil { s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err) return fmt.Errorf("failed to retrieve webhook: %w", err) } if webhook.Active == active { if active { return svcerrors.Error(svcerrors.FailedPrecondition, "webhook is already active") } return svcerrors.Error(svcerrors.FailedPrecondition, "webhook is already paused") } webhook.Active = active webhook.UpdatedAt = time.Now() if err := s.webhookRepo.UpdateWebhook(ctx, tenantID, webhook); err != nil { action := "pause" if active { action = "resume" } s.logger.ErrorContext(ctx, "Failed to "+action+" webhook", "error", err) return fmt.Errorf("failed to %s webhook: %w", action, err) } action := "paused" if active { action = "resumed" } s.logger.InfoContext(ctx, "Webhook "+action+" successfully", "webhook_id", webhookID) return nil } // CreateWebhook creates a webhook registration with HTTP configuration support. // It returns the registration as stored and the plaintext signing secret, // which callers show once: the stored registration only has it encrypted. func (s *WebhookService) CreateWebhook(ctx context.Context, req WebhookRegistrationRequest) (*store.WebhookRegistration, string, error) { redactedURL := client.RedactURL(req.URL) span := trace.SpanFromContext(ctx) span.SetAttributes( attribute.String("consumer", req.Consumer), attribute.StringSlice("events", req.Events), attribute.String("url", redactedURL), ) s.logger.InfoContext(ctx, "Processing enhanced webhook creation request", "consumer", req.Consumer, "events", req.Events, "url", redactedURL, ) tenantID := tenant.FromContext(ctx) if req.Consumer == "" { return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } // Validate webhook URL against SSRF if err := ValidateWebhookURL(req.URL, s.networkPolicy); err != nil { return nil, "", err } if err := validateHeaders("headers", stringHeaders(req.Headers)); err != nil { return nil, "", err } secretHeaders, err := canonicalHeaders("secret_headers", req.SecretHeaders) if err != nil { return nil, "", err } if err := validateHeaders("secret_headers", secretHeaders); err != nil { return nil, "", err } req.SecretHeaders = secretHeaders if err := req.TemplateSettings.validate(); err != nil { return nil, "", err } if err := validateTransformTemplate(req.TransformTemplate); err != nil { return nil, "", err } if req.RequiresTransform && len(req.Events) > 0 && strings.TrimSpace(req.TransformTemplate) == "" { return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "requires_transform is set, so transform_template is required for the subscriptions created from events (or register without events and add each subscription with its own template)") } // Convert request to internal webhook registration webhookReg, err := req.ToWebhookRegistration() if err != nil { return nil, "", svcerrors.Wrapf(err, svcerrors.InvalidArgument, "invalid webhook configuration: %v", err) } // Generate ID if not provided if webhookReg.ID == "" { webhookReg.ID = uuid.New().String() } // Validate event names for _, event := range req.Events { if event == "" { return nil, "", svcerrors.Error(svcerrors.InvalidArgument, "empty event name not allowed") } if err := checkEventForConsumer(req.Consumer, event); err != nil { return nil, "", err } } webhookID, err := parseUUID(webhookReg.ID, "webhook ID") if err != nil { return nil, "", err } // Convert internal webhook to store model for database operation storeWebhook := &store.WebhookRegistration{ ID: webhookID, Consumer: webhookReg.Consumer, URL: webhookReg.URL, Timeout: webhookReg.HTTPConfig.RequestTimeoutSeconds, Active: webhookReg.Active, Description: webhookReg.Description, Health: store.WebhookHealth(webhookReg.Health), MaxRetries: DerefIntOr(webhookReg.HTTPConfig.MaxRetries, DefaultMaxRetries), RetryBackoffSeconds: webhookReg.HTTPConfig.RetryBackoffSeconds, CaptureResponseBody: DerefBoolOr(webhookReg.HTTPConfig.CaptureResponseBody, false), FollowRedirects: DerefBoolOr(webhookReg.HTTPConfig.FollowRedirects, true), VerifySSL: DerefBoolOr(webhookReg.HTTPConfig.VerifySSL, true), RequestTimeoutSeconds: webhookReg.HTTPConfig.RequestTimeoutSeconds, UserAgent: webhookReg.HTTPConfig.UserAgent, ContentType: webhookReg.HTTPConfig.ContentType, RateLimitRPS: webhookReg.HTTPConfig.RateLimitRPS, RequiresTransform: req.RequiresTransform, CreatedAt: time.Now(), UpdatedAt: time.Now(), } // Set signature type, defaulting to HMAC sigType := store.SignatureType(webhookReg.SignatureType) if sigType != store.SignatureTypeEd25519 { sigType = store.SignatureTypeHMAC } storeWebhook.SignatureType = sigType // Convert headers to string map for store model headersMap := make(map[string]string) for k, v := range webhookReg.Headers { if str, ok := v.(string); ok { headersMap[k] = str } } storeWebhook.Headers = headersMap // Encrypt webhook secret if provided, or generate one if missing if webhookReg.HTTPConfig.WebhookSecret != "" { encSecret, err := s.EncryptWebhookSecret(webhookReg.HTTPConfig.WebhookSecret) if err != nil { return nil, "", fmt.Errorf("failed to encrypt webhook secret: %w", err) } storeWebhook.WebhookSecret = encSecret } else { // Auto-generate a secret if missing to ensure all webhooks are signed secret, err := generateWebhookSecret() if err != nil { return nil, "", fmt.Errorf("failed to generate webhook secret: %w", err) } encSecret, err := s.EncryptWebhookSecret(secret) if err != nil { return nil, "", fmt.Errorf("failed to encrypt generated webhook secret: %w", err) } storeWebhook.WebhookSecret = encSecret webhookReg.HTTPConfig.WebhookSecret = secret // returned in plaintext once } // Generate Ed25519 keypair only when signature_type is "ed25519". // The private key is envelope-encrypted and stored; the public key is derived at runtime. if sigType == store.SignatureTypeEd25519 { encPrivKey, err := s.generateEncryptedEd25519Key() if err != nil { return nil, "", err } storeWebhook.Ed25519PrivateKey = encPrivKey } // Encrypt secret headers if provided if len(req.SecretHeaders) > 0 { encrypted, err := s.EncryptSecretHeaders(req.SecretHeaders) if err != nil { return nil, "", fmt.Errorf("failed to encrypt secret headers: %w", err) } storeWebhook.SecretHeaders = encrypted } // Convert expected status codes expectedCodes := make(pq.Int64Array, len(webhookReg.HTTPConfig.ExpectedStatusCodes)) for i, code := range webhookReg.HTTPConfig.ExpectedStatusCodes { expectedCodes[i] = int64(code) } storeWebhook.ExpectedStatusCodes = expectedCodes // Build subscriptions slice for atomic creation var subscriptions []*store.EventSubscription for _, event := range req.Events { subscriptions = append(subscriptions, &store.EventSubscription{ EventName: event, Consumer: req.Consumer, TransformEnabled: strings.TrimSpace(req.TransformTemplate) != "", TransformTemplate: req.TransformTemplate, OnTransformError: req.TemplateSettings.OnTransformError, TemplateMissingKey: req.TemplateSettings.TemplateMissingKey, }) } // Atomically register the webhook and all subscriptions in a single transaction firstVersion := store.SubscriptionTemplateVersion{Source: req.TemplateMeta.Source, Notes: req.TemplateMeta.Notes, SavedBy: req.TemplateMeta.SavedBy} if err := s.webhookRepo.RegisterWebhookWithSubscriptions(ctx, tenantID, storeWebhook, subscriptions, firstVersion); err != nil { s.logger.ErrorContext(ctx, "Failed to register webhook", "consumer", req.Consumer, "events", req.Events, "url", client.RedactURL(req.URL), "error", err, ) span.RecordError(err) span.SetStatus(otelcodes.Error, "failed to register webhook") return nil, "", fmt.Errorf("failed to register webhook: %w", err) } // Initialize rate limit state if rate limiting is configured if storeWebhook.RateLimitRPS != nil { if err := s.webhookRepo.UpsertRateLimitState(ctx, tenantID, webhookID); err != nil { s.logger.WarnContext(ctx, "Failed to initialize rate limit state", "webhook_id", webhookReg.ID, "error", err, ) // Non-fatal: the rate limit row will be created on first delivery slot acquisition } } // Update metrics if s.metrics != nil { s.metrics.WebhookRegistrations.Add(ctx, 1) } s.logger.InfoContext(ctx, "Enhanced webhook registered successfully", "webhook_id", webhookReg.ID, "consumer", req.Consumer, "events", req.Events, "url", client.RedactURL(req.URL), "http_config_provided", req.HTTPConfig != nil, ) span.SetStatus(otelcodes.Ok, "webhook created successfully") return storeWebhook, webhookReg.HTTPConfig.WebhookSecret, nil } // UnregisterWebhook removes a webhook registration func (s *WebhookService) UnregisterWebhook(ctx context.Context, webhookID string, consumer string) error { s.logger.InfoContext(ctx, "Processing webhook un registration request", "webhook_id", webhookID, "consumer", consumer, ) if webhookID == "" { return svcerrors.Error(svcerrors.InvalidArgument, "webhook_id is required") } if consumer == "" { return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } tenantID := tenant.FromContext(ctx) id, err := parseUUID(webhookID, "webhook ID") if err != nil { return err } // Check if webhook exists in consumer _, err = s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer) if err != nil { return fmt.Errorf("failed to retrieve webhook: %w", err) } if err := s.webhookRepo.UnregisterWebhook(ctx, tenantID, id); err != nil { s.logger.ErrorContext(ctx, "Failed to unregister webhook", "webhook_id", webhookID, "error", err, ) return fmt.Errorf("failed to unregister webhook: %w", err) } s.logger.InfoContext(ctx, "Webhook unregistered successfully", "webhook_id", webhookID, ) return nil } // ListWebhooks lists all registered webhooks with optional consumer and other filters. // When consumer is empty, returns webhooks across all consumers. // When health is non-empty, only webhooks with that health status are returned. func (s *WebhookService) ListWebhooks(ctx context.Context, consumer string, webhookID string, event string, activeOnly bool, health string, limit, offset int32) ([]*store.WebhookRegistration, int32, error) { s.logger.InfoContext(ctx, "Processing list webhooks request", "consumer", consumer, "webhook_id", webhookID, "event", event, "active_only", activeOnly, "limit", limit, "offset", offset, ) tenantID := tenant.FromContext(ctx) l, o := normalizePagination(int(limit), int(offset)) limit, offset = clampInt32(l), clampInt32(o) if webhookID != "" { id, err := parseUUID(webhookID, "webhook ID") if err != nil { return nil, 0, err } reg, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer) if err != nil { if storage.IsNotFound(err) { return []*store.WebhookRegistration{}, 0, nil } return nil, 0, fmt.Errorf("failed to retrieve webhook: %w", err) } if activeOnly && !reg.Active { return []*store.WebhookRegistration{}, 0, nil } if health != "" && string(reg.Health) != health { return []*store.WebhookRegistration{}, 0, nil } return []*store.WebhookRegistration{reg}, 1, nil } registrations, totalCount, err := s.webhookRepo.ListWebhooksPaginated(ctx, tenantID, consumer, event, activeOnly, store.WebhookHealth(health), int(limit), int(offset)) if err != nil { s.logger.ErrorContext(ctx, "Failed to list webhooks", "consumer", consumer, "error", err, ) return nil, 0, fmt.Errorf("failed to list webhooks: %w", err) } s.logger.InfoContext(ctx, "Listed webhooks successfully", "consumer", consumer, "count", len(registrations), "total", totalCount, ) return registrations, clampInt32(totalCount), nil } // PauseWebhook temporarily disables webhook deliveries func (s *WebhookService) PauseWebhook(ctx context.Context, webhookID string, consumer string, reason string) error { s.logger.InfoContext(ctx, "Pausing webhook", "webhook_id", webhookID, "consumer", consumer, "reason", reason) return s.setWebhookActive(ctx, webhookID, consumer, false) } // WebhookResumeResult describes a resumed webhook. type WebhookResumeResult struct { // PausedDeliveries counts the webhook's deliveries held as paused. They // are not sent automatically; the caller decides what to retry. PausedDeliveries int } // ResumeWebhook re-enables webhook deliveries, whether it was paused by an // operator or auto-disabled. Deliveries held while it was paused stay paused // until retried, like those of a resumed subscription. func (s *WebhookService) ResumeWebhook(ctx context.Context, webhookID string, consumer string) (*WebhookResumeResult, error) { s.logger.InfoContext(ctx, "Resuming webhook", "webhook_id", webhookID, "consumer", consumer) if err := s.setWebhookActive(ctx, webhookID, consumer, true); err != nil { return nil, err } id, err := parseUUID(webhookID, "webhook ID") if err != nil { return nil, err } status := string(store.StatusPaused) count, err := s.webhookRepo.CountDeliveries(ctx, tenant.FromContext(ctx), store.DeliveryFilter{ Consumer: consumer, WebhookID: &id, Status: &status, }) if err != nil { return nil, fmt.Errorf("failed to count paused deliveries: %w", err) } return &WebhookResumeResult{PausedDeliveries: count}, nil } // UpdateWebhookConfig updates webhook configuration. // When updateMask is non-empty, only the listed field paths are applied. // When updateMask is empty, falls back to legacy behavior (all non-zero fields applied). // // Supported mask paths: // // "url", "active", "description", "events", "headers", "secret_headers", // "signature_type", "http_config", "http_config.webhook_secret", // "http_config.rate_limit_rps", "requires_transform" // // requires_transform is only changed when named in the mask. Turning it on // fails while any subscription lacks a transform, and a webhook that requires // one cannot have its events replaced in bulk (that recreates subscriptions // without templates). func (s *WebhookService) UpdateWebhookConfig(ctx context.Context, webhookID string, consumer string, events []string, url string, headers map[string]string, active bool, description string, httpConfig *WebhookHTTPConfig, secretHeaders map[string]string, signatureType string, requiresTransform bool, updateMask []string) error { s.logger.InfoContext(ctx, "Processing update webhook config request", "webhook_id", webhookID, "consumer", consumer, "update_mask", updateMask) if webhookID == "" { return svcerrors.Error(svcerrors.InvalidArgument, "webhook ID is required") } if consumer == "" { return svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } tenantID := tenant.FromContext(ctx) webhookUUID, err := parseUUID(webhookID, "webhook ID") if err != nil { return err } webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenantID, webhookUUID, consumer) if err != nil { s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err) return fmt.Errorf("failed to retrieve webhook: %w", err) } oldURL := webhook.URL oldSecretHeaders := webhook.SecretHeaders // Build a set for O(1) lookup. When empty, all non-zero fields are applied (legacy). mask := make(map[string]bool, len(updateMask)) for _, p := range updateMask { mask[p] = true } useMask := len(mask) > 0 shouldUpdate := func(field string) bool { if !useMask { return true // legacy: apply everything } return mask[field] } // Build new subscription list (pure logic, no DB calls) var newSubs []*store.EventSubscription replaceEvents := shouldUpdate("events") && len(events) > 0 if replaceEvents { for _, event := range events { if err := checkEventForConsumer(consumer, event); err != nil { return err } newSubs = append(newSubs, &store.EventSubscription{ EventName: event, }) } } // Apply config mutations to the in-memory webhook (pure logic, no DB calls) if shouldUpdate("url") && url != "" { normalizedURL := strings.TrimSpace(url) if normalizedURL == "" { return svcerrors.Error(svcerrors.InvalidArgument, "URL is required") } if err := ValidateWebhookURL(normalizedURL, s.networkPolicy); err != nil { return err } webhook.URL = normalizedURL } if shouldUpdate("headers") && headers != nil { if err := validateHeaders("headers", headers); err != nil { return err } webhook.Headers = headers } if shouldUpdate("active") { webhook.Active = active } if shouldUpdate("description") && description != "" { webhook.Description = description } // Apply HTTP config updates if provided if shouldUpdate("http_config") && httpConfig != nil { if httpConfig.MaxRetries != nil { webhook.MaxRetries = *httpConfig.MaxRetries } if httpConfig.RetryBackoffSeconds > 0 { webhook.RetryBackoffSeconds = httpConfig.RetryBackoffSeconds } if httpConfig.RequestTimeoutSeconds > 0 { webhook.RequestTimeoutSeconds = httpConfig.RequestTimeoutSeconds } if len(httpConfig.ExpectedStatusCodes) > 0 { int64Codes := make([]int64, len(httpConfig.ExpectedStatusCodes)) for i, c := range httpConfig.ExpectedStatusCodes { int64Codes[i] = int64(c) } webhook.ExpectedStatusCodes = int64Codes } // Only update the webhook secret if explicitly requested via mask. // Without mask (legacy mode), non-empty secret is applied. // With mask, "http_config.webhook_secret" must be in the mask. updateSecret := false if useMask { updateSecret = mask["http_config.webhook_secret"] } else { updateSecret = httpConfig.WebhookSecret != "" } if updateSecret && httpConfig.WebhookSecret != "" { encSecret, err := s.EncryptWebhookSecret(httpConfig.WebhookSecret) if err != nil { return fmt.Errorf("failed to encrypt webhook secret: %w", err) } webhook.WebhookSecret = encSecret } if httpConfig.UserAgent != "" { webhook.UserAgent = httpConfig.UserAgent } if httpConfig.ContentType != "" { webhook.ContentType = httpConfig.ContentType } if httpConfig.CaptureResponseBody != nil { webhook.CaptureResponseBody = *httpConfig.CaptureResponseBody } if httpConfig.FollowRedirects != nil { webhook.FollowRedirects = *httpConfig.FollowRedirects } if httpConfig.VerifySSL != nil { webhook.VerifySSL = *httpConfig.VerifySSL } // RateLimitRPS: pointer field — nil means "not provided" (keep existing), // non-nil overrides. With mask, "http_config.rate_limit_rps" must be in the mask. updateRateLimit := false if useMask { updateRateLimit = mask["http_config.rate_limit_rps"] } else { updateRateLimit = httpConfig.RateLimitRPS != nil } if updateRateLimit { webhook.RateLimitRPS = httpConfig.RateLimitRPS } // Enforce the same bounds the create path enforces: validate the // merged result, not just the delta (e.g. request_timeout_seconds // must stay within 1-300 after a PATCH). codes := make(IntArray, len(webhook.ExpectedStatusCodes)) for i, code := range webhook.ExpectedStatusCodes { codes[i] = int(code) } merged := WebhookHTTPConfig{ MaxRetries: &webhook.MaxRetries, RetryBackoffSeconds: webhook.RetryBackoffSeconds, RequestTimeoutSeconds: webhook.RequestTimeoutSeconds, ExpectedStatusCodes: codes, ContentType: webhook.ContentType, RateLimitRPS: webhook.RateLimitRPS, } if err := merged.ValidateConfig(); err != nil { return err } } // Merge secret header updates into the existing encrypted map. // Empty-string values remove a header; omitted keys stay untouched. if shouldUpdate("secret_headers") { if secretHeaders, err = canonicalHeaders("secret_headers", secretHeaders); err != nil { return err } if err := validateHeaders("secret_headers", secretHeaders); err != nil { return err } mergedSecretHeaders, err := s.DecryptSecretHeaders(webhook.SecretHeaders) if err != nil { return fmt.Errorf("failed to decrypt existing secret headers: %w", err) } mergedSecretHeaders = canonicalNames(mergedSecretHeaders) if len(secretHeaders) == 0 { webhook.SecretHeaders = nil } else { if mergedSecretHeaders == nil { mergedSecretHeaders = map[string]string{} } for key, value := range secretHeaders { if value == "" { delete(mergedSecretHeaders, key) continue } mergedSecretHeaders[key] = value } encrypted, err := s.EncryptSecretHeaders(mergedSecretHeaders) if err != nil { return fmt.Errorf("failed to encrypt secret headers: %w", err) } webhook.SecretHeaders = encrypted } } if err := s.checkSecretHeadersFollowURL(oldURL, webhook.URL, oldSecretHeaders, secretHeaders, shouldUpdate("secret_headers")); err != nil { return err } // Update signature_type if in mask (or legacy non-empty) if shouldUpdate("signature_type") && signatureType != "" { newSigType := store.SignatureType(signatureType) if newSigType != store.SignatureTypeHMAC && newSigType != store.SignatureTypeEd25519 { return svcerrors.Errorf(svcerrors.InvalidArgument, "invalid signature_type: %q (must be \"hmac\" or \"ed25519\")", signatureType) } oldSigType := webhook.SignatureType webhook.SignatureType = newSigType // Generate Ed25519 keypair when switching to ed25519 if newSigType == store.SignatureTypeEd25519 && oldSigType != store.SignatureTypeEd25519 { encPrivKey, err := s.generateEncryptedEd25519Key() if err != nil { return err } webhook.Ed25519PrivateKey = encPrivKey } // Clear Ed25519 key when switching away from ed25519 if newSigType == store.SignatureTypeHMAC && oldSigType == store.SignatureTypeEd25519 { webhook.Ed25519PrivateKey = nil } } // Turning requires_transform on is checked inside the transaction, under // an exclusive lock that subscription writes (shared lock) wait on. turningOnTransform := mask["requires_transform"] && requiresTransform && !webhook.RequiresTransform if mask["requires_transform"] { webhook.RequiresTransform = requiresTransform } if replaceEvents && webhook.RequiresTransform { return svcerrors.Error(svcerrors.FailedPrecondition, "this webhook requires a payload transform, so its events cannot be replaced in bulk (that recreates subscriptions without templates); add or delete subscriptions one by one instead") } // Persist subscription replacement + webhook update atomically. err = s.webhookRepo.RunInTransaction(func(txRepo store.RepositoryInterface) error { if turningOnTransform { if _, err := txRepo.LockWebhook(ctx, tenantID, webhookUUID, consumer, true); err != nil { return fmt.Errorf("failed to lock webhook: %w", err) } subs, err := txRepo.ListSubscriptions(ctx, tenantID, webhookUUID) if err != nil { return fmt.Errorf("failed to list subscriptions: %w", err) } var missing []string for _, sub := range subs { if !sub.HasTransform() { missing = append(missing, sub.EventName) } } if len(missing) > 0 { return svcerrors.Errorf(svcerrors.FailedPrecondition, "cannot require a payload transform: the subscriptions for %s have none; add a transform_template to each first", strings.Join(missing, ", ")) } } if replaceEvents { if err := txRepo.ReplaceWebhookSubscriptions(ctx, tenantID, webhookUUID, consumer, newSubs); err != nil { return fmt.Errorf("failed to update webhook subscriptions: %w", err) } } return txRepo.UpdateWebhook(ctx, tenantID, webhook) }) if err != nil { s.logger.ErrorContext(ctx, "Failed to update webhook config", "webhook_id", webhookID, "error", err) return fmt.Errorf("failed to update webhook configuration: %w", err) } // Manage rate limit state: upsert when rate limit is set, delete when cleared if webhook.RateLimitRPS != nil { if err := s.webhookRepo.UpsertRateLimitState(ctx, tenantID, webhookUUID); err != nil { s.logger.WarnContext(ctx, "Failed to upsert rate limit state", "webhook_id", webhookID, "error", err, ) } } else { // Rate limit removed (or was never set) — clean up any existing state if err := s.webhookRepo.DeleteRateLimitState(ctx, tenantID, webhookUUID); err != nil { s.logger.WarnContext(ctx, "Failed to delete rate limit state", "webhook_id", webhookID, "error", err, ) } } s.logger.InfoContext(ctx, "Webhook configuration updated successfully", "webhook_id", webhookID) return nil } // RotateWebhookSecret replaces a webhook's HMAC signing secret with a freshly // generated one and returns it in plaintext. This is the only time the new // secret is readable; every later read masks it. Deliveries signed after the // rotation use the new secret, so receivers must be updated with it. func (s *WebhookService) RotateWebhookSecret(ctx context.Context, webhookID string, consumer string) (string, error) { if consumer == "" { return "", svcerrors.Error(svcerrors.InvalidArgument, "consumer is required") } webhookUUID, err := parseUUID(webhookID, "webhook ID") if err != nil { return "", err } webhook, err := s.webhookRepo.GetWebhookByID(ctx, tenant.FromContext(ctx), webhookUUID, consumer) if err != nil { return "", fmt.Errorf("failed to retrieve webhook: %w", err) } secret, err := generateWebhookSecret() if err != nil { return "", fmt.Errorf("failed to generate webhook secret: %w", err) } encSecret, err := s.EncryptWebhookSecret(secret) if err != nil { return "", fmt.Errorf("failed to encrypt webhook secret: %w", err) } webhook.WebhookSecret = encSecret if err := s.webhookRepo.UpdateWebhook(ctx, tenant.FromContext(ctx), webhook); err != nil { return "", fmt.Errorf("failed to store rotated webhook secret: %w", err) } s.logger.InfoContext(ctx, "Rotated webhook signing secret", "webhook_id", webhookID, "consumer", consumer) return secret, nil } // checkSecretHeadersFollowURL refuses to point a webhook at another origin // while it keeps secret headers this request did not set. Secret header // values are never returned by the API, so without this anyone who may edit // the URL (a portal consumer, say) could read them by moving the webhook to // a server they control. Each existing header must be given a new value, or // "" to remove it; an empty secret_headers object removes them all. func (s *WebhookService) checkSecretHeadersFollowURL(oldURL, newURL string, oldEncrypted []byte, set map[string]string, updated bool) error { if len(oldEncrypted) == 0 || sameOrigin(oldURL, newURL) { return nil } if updated && len(set) == 0 { return nil // all removed } old, err := s.DecryptSecretHeaders(oldEncrypted) if err != nil { return fmt.Errorf("failed to decrypt existing secret headers: %w", err) } var missing []string for name := range canonicalNames(old) { if _, ok := set[name]; !ok || !updated { missing = append(missing, name) } } if len(missing) == 0 { return nil } slices.Sort(missing) return svcerrors.Errorf(svcerrors.InvalidArgument, "changing the URL to another origin requires secret_headers to be set again: give %s a new value, or \"\" to remove it", strings.Join(missing, ", ")) } // sameOrigin is client.SameOrigin for URL strings; one that does not parse // is another origin. func sameOrigin(a, b string) bool { ua, errA := neturl.Parse(a) ub, errB := neturl.Parse(b) if errA != nil || errB != nil { return false } return client.SameOrigin(ua, ub) }