package github import ( "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" "go.uber.org/zap" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" ) const ( maxGitHubWebhookPayloadSize = 10 * 1024 * 1024 webhookDeliveryStaleAfter = 5 * time.Minute ) var ErrInvalidWebhookSignature = errors.New("invalid GitHub webhook signature") type githubWebhookStore interface { ClaimWebhookDelivery(context.Context, *WebhookDelivery, time.Time) (WebhookDeliveryClaim, error) CompleteAppRegistrationWebhookDelivery( context.Context, string, string, WebhookDeliveryStatus, string, time.Time, ) error ListWorkspaceConnectionsByAppInstallation(context.Context, string, int64) ([]*WorkspaceConnection, error) TransitionWorkspaceInstallationConnection( context.Context, *WorkspaceConnection, *WorkspaceConnection, ) (bool, error) ListUserConnectionsByAppGitHubUser(context.Context, string, int64) ([]*UserConnection, error) } type personalConnectionRevoker interface { RevokePersonalConnection(context.Context, string, string) error } type personalAuthorizationReconciler interface { ReconcileAuthorizationRevocation(context.Context, *UserConnection) (bool, error) } type GitHubWebhookReconciliation struct { Installations appInstallationVerifier Personal personalAuthorizationReconciler } type InstallationRepository struct { ID int64 `json:"id"` FullName string `json:"full_name"` } type InstallationRepositoriesChange struct { WorkspaceID string AppRegistrationID string InstallationID int64 ConnectionSource ConnectionSource CredentialGeneration int64 Action string Added []InstallationRepository Removed []InstallationRepository } type installationRepositoryUpdater interface { ApplyInstallationRepositories(context.Context, InstallationRepositoriesChange) (bool, error) } type GitHubWebhookRequest struct { DeliveryID string Event string Signature string Payload []byte } func (s *Service) HandleAppRegistrationWebhook( ctx context.Context, registrationID string, request GitHubWebhookRequest, ) (GitHubWebhookResult, error) { registrationID = strings.TrimSpace(registrationID) runtime := s.currentAppRegistrationRuntime(registrationID) if registrationID == "" || runtime == nil || runtime.webhookAuth == nil || runtime.webhookAuth.registrationID != registrationID { return GitHubWebhookResult{}, ErrGitHubNotConfigured } service := runtime.webhookAuth signed := service.Authenticates(request) result, err := service.Handle(ctx, request) if !signed { return result, err } status := DeploymentAppWebhookVerified reason := "" if err != nil { status = DeploymentAppWebhookFailing reason = "signed webhook processing failed" } observedAt := service.now().UTC() if s.store != nil { if _, healthErr := s.store.updateAppRegistrationWebhookHealth( ctx, registrationID, runtime.generation, status, observedAt, reason, ); healthErr != nil && s.logger != nil { s.logger.Warn("App registration webhook health persistence failed", zap.Error(healthErr)) } } return result, err } type GitHubWebhookResult struct { Duplicate bool Status WebhookDeliveryStatus Affected int } type installationWebhookEvent struct { Action string `json:"action"` Installation struct { ID int64 `json:"id"` Account struct { Login string `json:"login"` Type string `json:"type"` } `json:"account"` } `json:"installation"` } type installationTransition struct { status ConnectionStatus lastError string login string accountType string verified bool } type GitHubWebhookService struct { registrationID string secret []byte store githubWebhookStore repos installationRepositoryUpdater personal personalConnectionRevoker installations appInstallationVerifier personalReconciler personalAuthorizationReconciler publisher bus.EventBus now func() time.Time } func NewGitHubWebhookService( secret string, store githubWebhookStore, repositories installationRepositoryUpdater, personal personalConnectionRevoker, publisher bus.EventBus, reconciliation ...GitHubWebhookReconciliation, ) *GitHubWebhookService { service := &GitHubWebhookService{ secret: []byte(secret), store: store, repos: repositories, personal: personal, publisher: publisher, now: time.Now, } if len(reconciliation) > 0 { service.installations = reconciliation[0].Installations service.personalReconciler = reconciliation[0].Personal } return service } func NewAppRegistrationWebhookService( registrationID, secret string, store githubWebhookStore, repositories installationRepositoryUpdater, personal personalConnectionRevoker, publisher bus.EventBus, reconciliation ...GitHubWebhookReconciliation, ) *GitHubWebhookService { service := NewGitHubWebhookService(secret, store, repositories, personal, publisher, reconciliation...) service.registrationID = strings.TrimSpace(registrationID) return service } func (s *GitHubWebhookService) Authenticates(request GitHubWebhookRequest) bool { return s != nil && len(s.secret) > 0 && request.DeliveryID != "" && request.Event != "" && len(request.Payload) <= maxGitHubWebhookPayloadSize && validGitHubWebhookSignature(s.secret, request.Payload, request.Signature) } func (s *GitHubWebhookService) Handle( ctx context.Context, request GitHubWebhookRequest, ) (GitHubWebhookResult, error) { if s == nil || s.store == nil || len(s.secret) == 0 { return GitHubWebhookResult{}, errors.New("GitHub webhook service is not configured") } if request.DeliveryID == "" || request.Event == "" || len(request.Payload) > maxGitHubWebhookPayloadSize { return GitHubWebhookResult{}, errors.New("invalid GitHub webhook request") } if !validGitHubWebhookSignature(s.secret, request.Payload, request.Signature) { return GitHubWebhookResult{}, ErrInvalidWebhookSignature } now := s.now().UTC() claim, err := s.store.ClaimWebhookDelivery(ctx, &WebhookDelivery{ AppRegistrationID: s.registrationID, DeliveryID: request.DeliveryID, Event: request.Event, Status: WebhookDeliveryStatusReceived, ReceivedAt: now, }, now.Add(-webhookDeliveryStaleAfter)) if err != nil { return GitHubWebhookResult{}, fmt.Errorf("record GitHub webhook delivery: %w", err) } if !claim.Acquired { return GitHubWebhookResult{Duplicate: true, Status: claim.Status}, nil } affected, result, processErr := s.process(ctx, request.Event, request.Payload) status := WebhookDeliveryStatusProcessed if affected == 0 { status = WebhookDeliveryStatusIgnored } if processErr != nil { status = WebhookDeliveryStatusFailed result = processErr.Error() } completeErr := s.completeDelivery(ctx, request.DeliveryID, status, result, s.now().UTC()) if processErr != nil || completeErr != nil { return GitHubWebhookResult{Status: status, Affected: affected}, errors.Join(processErr, completeErr) } return GitHubWebhookResult{Status: status, Affected: affected}, nil } func (s *GitHubWebhookService) completeDelivery( ctx context.Context, deliveryID string, status WebhookDeliveryStatus, result string, processedAt time.Time, ) error { return s.store.CompleteAppRegistrationWebhookDelivery( ctx, s.registrationID, deliveryID, status, result, processedAt, ) } func (s *GitHubWebhookService) process(ctx context.Context, event string, payload []byte) (int, string, error) { switch event { case "installation": return s.processInstallation(ctx, payload) case "installation_repositories": return s.processInstallationRepositories(ctx, payload) case "github_app_authorization": return s.processAuthorization(ctx, payload) case "push": return s.processPush(ctx, payload) case "check_run": return s.processCheckRun(ctx, payload) default: return 0, "event ignored", nil } } func (s *GitHubWebhookService) processInstallation( ctx context.Context, payload []byte, ) (int, string, error) { var event installationWebhookEvent if err := json.Unmarshal(payload, &event); err != nil { return 0, "", fmt.Errorf("decode installation webhook: %w", err) } transition, supported, err := s.installationTransition(ctx, event) if err != nil { return 0, "", err } if !supported || event.Installation.ID <= 0 { return 0, "installation action ignored", nil } connections, err := s.listWorkspaceConnections(ctx, event.Installation.ID) if err != nil { return 0, "", fmt.Errorf("load installation bindings: %w", err) } affected := 0 for _, connection := range connections { if !shouldApplyInstallationTransition(s.registrationID, connection, event, transition) { continue } updated, err := s.applyInstallationTransition(ctx, connection, transition) if err != nil { return affected, "", fmt.Errorf("update installation binding: %w", err) } if !updated { continue } affected++ } return affected, fmt.Sprintf("installation %s: %d binding(s)", event.Action, affected), nil } func (s *GitHubWebhookService) listWorkspaceConnections( ctx context.Context, installationID int64, ) ([]*WorkspaceConnection, error) { return s.store.ListWorkspaceConnectionsByAppInstallation(ctx, s.registrationID, installationID) } func (s *GitHubWebhookService) installationTransition( ctx context.Context, event installationWebhookEvent, ) (installationTransition, bool, error) { status, lastError, supported := installationWebhookStatus(event.Action) transition := installationTransition{ status: status, lastError: lastError, login: event.Installation.Account.Login, accountType: event.Installation.Account.Type, } if !supported || event.Installation.ID <= 0 || s.installations == nil { return transition, supported, nil } status, lastError, login, accountType, err := s.reconcileInstallation(ctx, event.Installation.ID) if err != nil { return installationTransition{}, false, err } return installationTransition{ status: status, lastError: lastError, login: login, accountType: accountType, verified: true, }, true, nil } func shouldApplyInstallationTransition( registrationID string, connection *WorkspaceConnection, event installationWebhookEvent, transition installationTransition, ) bool { if !matchesInstallation(connection, registrationID, event.Installation.ID) { return false } if !transition.verified { return canApplyInstallationTransition(connection.Status, event.Action) } return connection.Status != transition.status || (transition.login != "" && connection.InstallationAccountLogin != transition.login) || (transition.accountType != "" && connection.InstallationAccountType != transition.accountType) } func (s *GitHubWebhookService) applyInstallationTransition( ctx context.Context, connection *WorkspaceConnection, transition installationTransition, ) (bool, error) { expected := *connection next := expected next.Status = transition.status next.LastError = transition.lastError next.CredentialGeneration++ if transition.login != "" { next.InstallationAccountLogin = transition.login } if transition.accountType != "" { next.InstallationAccountType = transition.accountType } updated, err := s.store.TransitionWorkspaceInstallationConnection(ctx, &expected, &next) if updated { *connection = next } return updated, err } func (s *GitHubWebhookService) processInstallationRepositories( ctx context.Context, payload []byte, ) (int, string, error) { var event struct { Action string `json:"action"` Installation struct { ID int64 `json:"id"` } `json:"installation"` Added []InstallationRepository `json:"repositories_added"` Removed []InstallationRepository `json:"repositories_removed"` } if err := json.Unmarshal(payload, &event); err != nil { return 0, "", fmt.Errorf("decode installation repositories webhook: %w", err) } if event.Installation.ID <= 0 || (event.Action != "added" && event.Action != "removed") { return 0, "installation repositories action ignored", nil } connections, err := s.listWorkspaceConnections(ctx, event.Installation.ID) if err != nil { return 0, "", fmt.Errorf("load installation bindings: %w", err) } affected := 0 for _, connection := range connections { if !matchesInstallation(connection, s.registrationID, event.Installation.ID) { continue } if s.repos == nil { return affected, "", errors.New("installation repository updater is not configured") } updated, err := s.repos.ApplyInstallationRepositories(ctx, InstallationRepositoriesChange{ WorkspaceID: connection.WorkspaceID, AppRegistrationID: s.registrationID, InstallationID: event.Installation.ID, ConnectionSource: connection.Source, CredentialGeneration: connection.CredentialGeneration, Action: event.Action, Added: event.Added, Removed: event.Removed, }) if err != nil { return affected, "", fmt.Errorf("update installation repository access: %w", err) } if updated { affected++ } } return affected, fmt.Sprintf("installation repositories %s: %d binding(s)", event.Action, affected), nil } func (s *GitHubWebhookService) processAuthorization( ctx context.Context, payload []byte, ) (int, string, error) { var event struct { Action string `json:"action"` Sender struct { ID int64 `json:"id"` } `json:"sender"` } if err := json.Unmarshal(payload, &event); err != nil { return 0, "", fmt.Errorf("decode GitHub App authorization webhook: %w", err) } if event.Action != "revoked" || event.Sender.ID <= 0 { return 0, "authorization action ignored", nil } connections, err := s.store.ListUserConnectionsByAppGitHubUser(ctx, s.registrationID, event.Sender.ID) if err != nil { return 0, "", fmt.Errorf("load personal GitHub bindings: %w", err) } affected := 0 for _, connection := range connections { if connection == nil || connection.AppRegistrationID != s.registrationID || connection.GitHubUserID != event.Sender.ID { continue } if s.personalReconciler != nil { revoked, reconcileErr := s.personalReconciler.ReconcileAuthorizationRevocation(ctx, connection) if reconcileErr != nil { return affected, "", fmt.Errorf("reconcile personal GitHub connection: %w", reconcileErr) } if revoked { affected++ } continue } if s.personal == nil { return affected, "", errors.New("personal GitHub revoker is not configured") } if err := s.personal.RevokePersonalConnection(ctx, connection.WorkspaceID, connection.UserID); err != nil { return affected, "", fmt.Errorf("revoke personal GitHub connection: %w", err) } affected++ } return affected, fmt.Sprintf("authorization revoked: %d binding(s)", affected), nil } type pushWebhookEvent struct { Ref string `json:"ref"` After string `json:"after"` Deleted bool `json:"deleted"` Repository struct { Name string `json:"name"` Owner struct { Login string `json:"login"` } `json:"owner"` } `json:"repository"` Installation struct { ID int64 `json:"id"` } `json:"installation"` Pusher struct { Name string `json:"name"` } `json:"pusher"` HeadCommit struct { Message string `json:"message"` } `json:"head_commit"` } const gitBranchRefPrefix = "refs/heads/" func isZeroSHA(sha string) bool { if sha == "" { return true } for _, c := range sha { if c != '0' { return false } } return true } // processPush handles the "push" webhook, publishing events.GitHubPushReceived // with the resolved workspace IDs when the push targets a branch ref and // isn't a branch delete. func (s *GitHubWebhookService) processPush(ctx context.Context, payload []byte) (int, string, error) { var event pushWebhookEvent if err := json.Unmarshal(payload, &event); err != nil { return 0, "", fmt.Errorf("decode push webhook: %w", err) } if event.Deleted || isZeroSHA(event.After) || !strings.HasPrefix(event.Ref, gitBranchRefPrefix) { return 0, "push ignored", nil } workspaceIDs, err := s.resolveWorkspaceIDs(ctx, event.Installation.ID) if err != nil { return 0, "", err } if len(workspaceIDs) == 0 { return 0, "push ignored", nil } s.publishEvent(ctx, events.GitHubPushReceived, &GitHubPushEventPayload{ WorkspaceIDs: workspaceIDs, Owner: event.Repository.Owner.Login, Name: event.Repository.Name, Branch: strings.TrimPrefix(event.Ref, gitBranchRefPrefix), SHA: event.After, PusherLogin: event.Pusher.Name, HeadCommitMsg: event.HeadCommit.Message, }) return len(workspaceIDs), fmt.Sprintf("push published: %d workspace(s)", len(workspaceIDs)), nil } type checkRunWebhookEvent struct { Action string `json:"action"` CheckRun struct { ID int64 `json:"id"` Name string `json:"name"` HeadSHA string `json:"head_sha"` Conclusion string `json:"conclusion"` HTMLURL string `json:"html_url"` CheckSuite struct { HeadBranch string `json:"head_branch"` } `json:"check_suite"` } `json:"check_run"` Repository struct { Name string `json:"name"` Owner struct { Login string `json:"login"` } `json:"owner"` } `json:"repository"` Installation struct { ID int64 `json:"id"` } `json:"installation"` } // processCheckRun handles the "check_run" webhook, publishing // events.GitHubCheckRunCompleted with the resolved workspace IDs for // completed check runs only. func (s *GitHubWebhookService) processCheckRun(ctx context.Context, payload []byte) (int, string, error) { var event checkRunWebhookEvent if err := json.Unmarshal(payload, &event); err != nil { return 0, "", fmt.Errorf("decode check_run webhook: %w", err) } if event.Action != "completed" { return 0, "check_run ignored", nil } workspaceIDs, err := s.resolveWorkspaceIDs(ctx, event.Installation.ID) if err != nil { return 0, "", err } if len(workspaceIDs) == 0 { return 0, "check_run ignored", nil } s.publishEvent(ctx, events.GitHubCheckRunCompleted, &GitHubCheckRunEventPayload{ WorkspaceIDs: workspaceIDs, Owner: event.Repository.Owner.Login, Name: event.Repository.Name, Branch: event.CheckRun.CheckSuite.HeadBranch, SHA: event.CheckRun.HeadSHA, CheckName: event.CheckRun.Name, Conclusion: event.CheckRun.Conclusion, CheckRunID: event.CheckRun.ID, HTMLURL: event.CheckRun.HTMLURL, }) return len(workspaceIDs), fmt.Sprintf("check_run published: %d workspace(s)", len(workspaceIDs)), nil } // resolveWorkspaceIDs resolves an installation ID to the distinct set of // workspace IDs connected to it under this registration. func (s *GitHubWebhookService) resolveWorkspaceIDs(ctx context.Context, installationID int64) ([]string, error) { if installationID <= 0 { return nil, nil } connections, err := s.listWorkspaceConnections(ctx, installationID) if err != nil { return nil, fmt.Errorf("load installation bindings: %w", err) } seen := make(map[string]bool, len(connections)) workspaceIDs := make([]string, 0, len(connections)) for _, connection := range connections { // Only active installations may drive automation events: a suspended // or revoked App binding must not still fire push/CI automations. if connection == nil || connection.WorkspaceID == "" || connection.Status != ConnectionStatusActive || seen[connection.WorkspaceID] { continue } seen[connection.WorkspaceID] = true workspaceIDs = append(workspaceIDs, connection.WorkspaceID) } return workspaceIDs, nil } // publishEvent is nil-safe and never fails webhook delivery: publish errors // are swallowed so a bus outage doesn't turn a real webhook into a failed // delivery. func (s *GitHubWebhookService) publishEvent(ctx context.Context, eventType string, data interface{}) { if s.publisher == nil { return } _ = s.publisher.Publish(ctx, eventType, bus.NewEvent(eventType, "github_webhook_service", data)) } func (s *GitHubWebhookService) reconcileInstallation( ctx context.Context, installationID int64, ) (ConnectionStatus, string, string, string, error) { installation, err := s.installations.GetInstallation(ctx, installationID) if err != nil { var apiErr *GitHubAPIError if errors.As(err, &apiErr) && apiErr.StatusCode == 404 { return ConnectionStatusRevoked, "GitHub App installation deleted", "", "", nil } return "", "", "", "", fmt.Errorf("reconcile GitHub App installation: %w", err) } if installation.ID != installationID { return "", "", "", "", errors.New("reconciled GitHub App installation ID does not match webhook") } if installation.SuspendedAt != nil { return ConnectionStatusSuspended, "GitHub App installation suspended", installation.AccountLogin, installation.AccountType, nil } return ConnectionStatusActive, "", installation.AccountLogin, installation.AccountType, nil } func validGitHubWebhookSignature(secret, payload []byte, provided string) bool { if !strings.HasPrefix(provided, "sha256=") { return false } providedMAC, err := hex.DecodeString(strings.TrimPrefix(provided, "sha256=")) if err != nil { return false } mac := hmac.New(sha256.New, secret) _, _ = mac.Write(payload) return hmac.Equal(mac.Sum(nil), providedMAC) } func installationWebhookStatus(action string) (ConnectionStatus, string, bool) { switch action { case "suspend": return ConnectionStatusSuspended, "GitHub App installation suspended", true case "unsuspend": return ConnectionStatusActive, "", true case "deleted": return ConnectionStatusRevoked, "GitHub App installation deleted", true default: return "", "", false } } func canApplyInstallationTransition(current ConnectionStatus, action string) bool { switch action { case "suspend": return current != ConnectionStatusSuspended && current != ConnectionStatusRevoked case "unsuspend": return current == ConnectionStatusSuspended || current == ConnectionStatusInvalid case "deleted": return current != ConnectionStatusRevoked default: return false } } func matchesInstallation(connection *WorkspaceConnection, registrationID string, installationID int64) bool { return connection != nil && connection.Source == ConnectionSourceGitHubAppInstallation && connection.AppRegistrationID == registrationID && connection.InstallationID != nil && *connection.InstallationID == installationID } func (s *Store) updateAppRegistrationWebhookHealth( ctx context.Context, registrationID string, generation int64, status DeploymentAppWebhookStatus, lastWebhookAt time.Time, lastError string, ) (bool, error) { result, err := s.db.ExecContext(ctx, s.db.Rebind(` UPDATE github_app_registrations SET webhook_status = ?, last_webhook_at = ?, last_error = ?, updated_at = ? WHERE id = ? AND credential_generation = ?`), status, lastWebhookAt, nullString(lastError), time.Now().UTC(), registrationID, generation, ) if err != nil { return false, err } count, err := result.RowsAffected() return count == 1, err }