From 3ee9c2f1af6365e0db6dc036826196bae21d34e7 Mon Sep 17 00:00:00 2001 From: 0x1 <13666360+0x1@users.noreply.github.com> Date: Tue, 4 Feb 2025 13:44:37 -0500 Subject: [PATCH] [SCAN-165] Use Err Reporting (#3862) * nit * update rate limit handler to use reporter * update process repos to use rate limit handler with unit reporter * update getReposByOrgOrUser to report err * update dedupreporter to report err * add errReporter interface to handle both types of reporters * convert to error reporter types * update handleRateLimit signature * use reporters for all rate limit handlers in github * get repo url before err check * use iterator * remove err log * nit * pluralize * remove err log * update tests * make linter happy --- pkg/sources/github/connector_token.go | 4 +- pkg/sources/github/github.go | 100 ++++++++++++++++++-------- pkg/sources/github/github_test.go | 10 +-- pkg/sources/github/repo.go | 6 +- pkg/sources/source_manager.go | 3 +- 5 files changed, 83 insertions(+), 40 deletions(-) diff --git a/pkg/sources/github/connector_token.go b/pkg/sources/github/connector_token.go index 88ef8f8f1..469f50b98 100644 --- a/pkg/sources/github/connector_token.go +++ b/pkg/sources/github/connector_token.go @@ -17,14 +17,14 @@ type tokenConnector struct { apiClient *github.Client token string isGitHubEnterprise bool - handleRateLimit func(context.Context, error) bool + handleRateLimit func(context.Context, error, ...errorReporter) bool user string userMu sync.Mutex } var _ connector = (*tokenConnector)(nil) -func newTokenConnector(apiEndpoint string, token string, handleRateLimit func(context.Context, error) bool) (*tokenConnector, error) { +func newTokenConnector(apiEndpoint string, token string, handleRateLimit func(context.Context, error, ...errorReporter) bool) (*tokenConnector, error) { const httpTimeoutSeconds = 60 httpClient := common.RetryableHTTPClientTimeout(int64(httpTimeoutSeconds)) tokenSource := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token}) diff --git a/pkg/sources/github/github.go b/pkg/sources/github/github.go index 4dce437fb..b0b77f3f6 100644 --- a/pkg/sources/github/github.go +++ b/pkg/sources/github/github.go @@ -381,7 +381,7 @@ func (s *Source) Enumerate(ctx context.Context, reporter sources.UnitReporter) e // See: https://github.com/trufflesecurity/trufflehog/pull/2379#discussion_r1487454788 for _, name := range s.filteredRepoCache.Keys() { url, _ := s.filteredRepoCache.Get(name) - url, err := s.ensureRepoInfoCache(ctx, url) + url, err := s.ensureRepoInfoCache(ctx, url, &unitErrorReporter{reporter}) if err != nil { if err := dedupeReporter.UnitErr(ctx, err); err != nil { return err @@ -417,9 +417,10 @@ func (s *Source) Enumerate(ctx context.Context, reporter sources.UnitReporter) e for _, repo := range s.filteredRepoCache.Values() { ctx := context.WithValue(ctx, "repo", repo) - repo, err := s.ensureRepoInfoCache(ctx, repo) + repo, err := s.ensureRepoInfoCache(ctx, repo, &unitErrorReporter{reporter}) if err != nil { ctx.Logger().Error(err, "error caching repo info") + _ = dedupeReporter.UnitErr(ctx, fmt.Errorf("error caching repo info: %w", err)) } s.repos = append(s.repos, repo) } @@ -434,7 +435,7 @@ func (s *Source) Enumerate(ctx context.Context, reporter sources.UnitReporter) e // provided repository URL. If not, it fetches and stores the metadata for the // repository. In some cases, the gist URL needs to be normalized, which is // returned by this function. -func (s *Source) ensureRepoInfoCache(ctx context.Context, repo string) (string, error) { +func (s *Source) ensureRepoInfoCache(ctx context.Context, repo string, reporter errorReporter) (string, error) { if _, ok := s.repoInfoCache.get(repo); ok { return repo, nil } @@ -453,12 +454,15 @@ func (s *Source) ensureRepoInfoCache(ctx context.Context, repo string) (string, // Normalize the URL to the Gist's pull URL. // See https://github.com/trufflesecurity/trufflehog/pull/2625#issuecomment-2025507937 repo = gist.GetGitPullURL() - if s.handleRateLimit(ctx, err) { + + if s.handleRateLimit(ctx, err, reporter) { continue } + if err != nil { return repo, fmt.Errorf("failed to fetch gist") } + s.cacheGistInfo(gist) break } @@ -466,7 +470,7 @@ func (s *Source) ensureRepoInfoCache(ctx context.Context, repo string) (string, // Cache repository info. for { ghRepo, _, err := s.connector.APIClient().Repositories.Get(ctx, urlParts[1], urlParts[2]) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimit(ctx, err, reporter) { continue } if err != nil { @@ -491,7 +495,7 @@ func (s *Source) enumerateBasicAuth(ctx context.Context, reporter sources.UnitRe // TODO: This modifies s.memberCache but it doesn't look like // we do anything with it. if userType == organization && s.conn.ScanUsers { - if err := s.addMembersByOrg(ctx, org); err != nil { + if err := s.addMembersByOrg(ctx, org, reporter); err != nil { orgCtx.Logger().Error(err, "Unable to add members by org") } } @@ -526,7 +530,7 @@ func (s *Source) enumerateWithToken(ctx context.Context, isGithubEnterprise bool var err error for { ghUser, _, err = s.connector.APIClient().Users.Get(ctx, "") - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { @@ -546,11 +550,11 @@ func (s *Source) enumerateWithToken(ctx context.Context, isGithubEnterprise bool } if isGithubEnterprise { - s.addAllVisibleOrgs(ctx) + s.addAllVisibleOrgs(ctx, reporter) } else { // Scan for orgs is default with a token. // GitHub App enumerates the repos that were assigned to it in GitHub App settings. - s.addOrgsByUser(ctx, ghUser.GetLogin()) + s.addOrgsByUser(ctx, ghUser.GetLogin(), reporter) } } @@ -564,7 +568,7 @@ func (s *Source) enumerateWithToken(ctx context.Context, isGithubEnterprise bool } if userType == organization && s.conn.ScanUsers { - if err := s.addMembersByOrg(ctx, org); err != nil { + if err := s.addMembersByOrg(ctx, org, reporter); err != nil { orgCtx.Logger().Error(err, "Unable to add members for org") } } @@ -588,7 +592,7 @@ func (s *Source) enumerateWithApp(ctx context.Context, installationClient *githu // Check if we need to find user repos. if s.conn.ScanUsers { - err := s.addMembersByApp(ctx, installationClient) + err := s.addMembersByApp(ctx, installationClient, reporter) if err != nil { return err } @@ -739,13 +743,37 @@ var ( rateLimitResumeTime time.Time ) -// handleRateLimit returns true if a rate limit was handled +// errorReporter is an interface that captures just the error reporting functionality +type errorReporter interface { + Err(ctx context.Context, err error) error +} + +// wrapper to adapt UnitReporter to errorReporter +type unitErrorReporter struct { + reporter sources.UnitReporter +} + +func (u unitErrorReporter) Err(ctx context.Context, err error) error { + return u.reporter.UnitErr(ctx, err) +} + +// wrapper to adapt ChunkReporter to errorReporter +type chunkErrorReporter struct { + reporter sources.ChunkReporter +} + +func (c chunkErrorReporter) Err(ctx context.Context, err error) error { + return c.reporter.ChunkErr(ctx, err) +} + +// handleRateLimit handles GitHub API rate limiting with an optional error reporter. +// Returns true if a rate limit was handled. // // Unauthenticated users have a rate limit of 60 requests per hour. // Authenticated users have a rate limit of 5,000 requests per hour, // however, certain actions are subject to a stricter "secondary" limit. // https://docs.github.com/en/rest/overview/rate-limits-for-the-rest-api -func (s *Source) handleRateLimit(ctx context.Context, errIn error) bool { +func (s *Source) handleRateLimit(ctx context.Context, errIn error, reporters ...errorReporter) bool { if errIn == nil { return false } @@ -757,7 +785,6 @@ func (s *Source) handleRateLimit(ctx context.Context, errIn error) bool { var retryAfter time.Duration if resumeTime.IsZero() || time.Now().After(resumeTime) { rateLimitMu.Lock() - var ( now = time.Now() @@ -785,6 +812,10 @@ func (s *Source) handleRateLimit(ctx context.Context, errIn error) bool { retryAfter = retryAfter + jitter rateLimitResumeTime = now.Add(retryAfter) ctx.Logger().Info(fmt.Sprintf("exceeded %s rate limit", limitType), "retry_after", retryAfter.String(), "resume_time", rateLimitResumeTime.Format(time.RFC3339)) + // Only report the error if a reporter was provided + for _, reporter := range reporters { + _ = reporter.Err(ctx, fmt.Errorf("exceeded %s rate limit", limitType)) + } } else { retryAfter = (5 * time.Minute) + jitter rateLimitResumeTime = now.Add(retryAfter) @@ -803,6 +834,16 @@ func (s *Source) handleRateLimit(ctx context.Context, errIn error) bool { return true } +// handleRateLimitWithUnitReporter is a wrapper around handleRateLimit that includes unit reporting +func (s *Source) handleRateLimitWithUnitReporter(ctx context.Context, reporter sources.UnitReporter, errIn error) bool { + return s.handleRateLimit(ctx, errIn, &unitErrorReporter{reporter: reporter}) +} + +// handleRateLimitWithChunkReporter is a wrapper around handleRateLimit that includes chunk reporting +func (s *Source) handleRateLimitWithChunkReporter(ctx context.Context, reporter sources.ChunkReporter, errIn error) bool { + return s.handleRateLimit(ctx, errIn, &chunkErrorReporter{reporter: reporter}) +} + func (s *Source) addReposForMembers(ctx context.Context, reporter sources.UnitReporter) { ctx.Logger().Info("Fetching repos from members", "members", len(s.memberCache)) for member := range s.memberCache { @@ -823,7 +864,7 @@ func (s *Source) addUserGistsToCache(ctx context.Context, user string, reporter for { gists, res, err := s.connector.APIClient().Gists.List(ctx, user, gistOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { @@ -847,7 +888,7 @@ func (s *Source) addUserGistsToCache(ctx context.Context, user string, reporter return nil } -func (s *Source) addMembersByApp(ctx context.Context, installationClient *github.Client) error { +func (s *Source) addMembersByApp(ctx context.Context, installationClient *github.Client, reporter sources.UnitReporter) error { opts := &github.ListOptions{ PerPage: membersAppPagination, } @@ -862,7 +903,7 @@ func (s *Source) addMembersByApp(ctx context.Context, installationClient *github if org.Account.GetType() != "Organization" { continue } - if err := s.addMembersByOrg(ctx, *org.Account.Login); err != nil { + if err := s.addMembersByOrg(ctx, *org.Account.Login, reporter); err != nil { return err } } @@ -870,7 +911,7 @@ func (s *Source) addMembersByApp(ctx context.Context, installationClient *github return nil } -func (s *Source) addAllVisibleOrgs(ctx context.Context) { +func (s *Source) addAllVisibleOrgs(ctx context.Context, reporter sources.UnitReporter) { ctx.Logger().V(2).Info("enumerating all visible organizations on GHE") // Enumeration on this endpoint does not use pages it uses a since ID. // The endpoint will return organizations with an ID greater than the given since ID. @@ -883,7 +924,7 @@ func (s *Source) addAllVisibleOrgs(ctx context.Context) { } for { orgs, _, err := s.connector.APIClient().Organizations.ListAll(ctx, orgOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { @@ -915,14 +956,14 @@ func (s *Source) addAllVisibleOrgs(ctx context.Context) { } } -func (s *Source) addOrgsByUser(ctx context.Context, user string) { +func (s *Source) addOrgsByUser(ctx context.Context, user string, reporter sources.UnitReporter) { orgOpts := &github.ListOptions{ PerPage: defaultPagination, } logger := ctx.Logger().WithValues("user", user) for { orgs, resp, err := s.connector.APIClient().Organizations.List(ctx, "", orgOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { @@ -944,7 +985,7 @@ func (s *Source) addOrgsByUser(ctx context.Context, user string) { } } -func (s *Source) addMembersByOrg(ctx context.Context, org string) error { +func (s *Source) addMembersByOrg(ctx context.Context, org string, reporter sources.UnitReporter) error { opts := &github.ListMembersOptions{ PublicOnly: false, ListOptions: github.ListOptions{ @@ -955,7 +996,7 @@ func (s *Source) addMembersByOrg(ctx context.Context, org string) error { logger := ctx.Logger().WithValues("org", org) for { members, res, err := s.connector.APIClient().Organizations.ListMembers(ctx, org, opts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { @@ -1087,7 +1128,7 @@ func (s *Source) processGistComments(ctx context.Context, gistURL string, urlPar } for { comments, _, err := s.connector.APIClient().Gists.ListComments(ctx, gistID, options) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithChunkReporter(ctx, reporter, err) { continue } if err != nil { @@ -1187,7 +1228,6 @@ func (s *Source) processRepoComments(ctx context.Context, repoInfo repoInfo, rep } return nil - } func (s *Source) processIssues(ctx context.Context, repoInfo repoInfo, reporter sources.ChunkReporter) error { @@ -1203,7 +1243,7 @@ func (s *Source) processIssues(ctx context.Context, repoInfo repoInfo, reporter for { issues, _, err := s.connector.APIClient().Issues.ListByRepo(ctx, repoInfo.owner, repoInfo.name, bodyTextsOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithChunkReporter(ctx, reporter, err) { continue } @@ -1272,7 +1312,7 @@ func (s *Source) processIssueComments(ctx context.Context, repoInfo repoInfo, re for { issueComments, _, err := s.connector.APIClient().Issues.ListComments(ctx, repoInfo.owner, repoInfo.name, allComments, issueOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithChunkReporter(ctx, reporter, err) { continue } if err != nil { @@ -1340,7 +1380,7 @@ func (s *Source) processPRs(ctx context.Context, repoInfo repoInfo, reporter sou for { prs, _, err := s.connector.APIClient().PullRequests.List(ctx, repoInfo.owner, repoInfo.name, prOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithChunkReporter(ctx, reporter, err) { continue } if err != nil { @@ -1372,7 +1412,7 @@ func (s *Source) processPRComments(ctx context.Context, repoInfo repoInfo, repor for { prComments, _, err := s.connector.APIClient().PullRequests.ListComments(ctx, repoInfo.owner, repoInfo.name, allComments, prOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithChunkReporter(ctx, reporter, err) { continue } if err != nil { @@ -1528,7 +1568,7 @@ func (s *Source) ChunkUnit(ctx context.Context, unit sources.SourceUnit, reporte ctx = context.WithValue(ctx, "repo", repoURL) // ChunkUnit is not guaranteed to be called from Enumerate, so we must // check and fetch the repoInfoCache for this repo. - repoURL, err := s.ensureRepoInfoCache(ctx, repoURL) + repoURL, err := s.ensureRepoInfoCache(ctx, repoURL, &chunkErrorReporter{reporter: reporter}) if err != nil { return err } diff --git a/pkg/sources/github/github_test.go b/pkg/sources/github/github_test.go index 3516017db..01401f7d0 100644 --- a/pkg/sources/github/github_test.go +++ b/pkg/sources/github/github_test.go @@ -197,7 +197,7 @@ func TestAddMembersByOrg(t *testing.T) { }) s := initTestSource(&sourcespb.GitHub{Credential: &sourcespb.GitHub_Unauthenticated{}}) - err := s.addMembersByOrg(context.Background(), "org1") + err := s.addMembersByOrg(context.Background(), "org1", noopReporter()) assert.Nil(t, err) assert.Equal(t, 2, len(s.memberCache)) _, ok := s.memberCache["testman1"] @@ -221,7 +221,7 @@ func TestAddMembersByOrg_AuthFailure(t *testing.T) { }}) s := initTestSource(&sourcespb.GitHub{Credential: &sourcespb.GitHub_Unauthenticated{}}) - err := s.addMembersByOrg(context.Background(), "org1") + err := s.addMembersByOrg(context.Background(), "org1", noopReporter()) assert.True(t, strings.HasPrefix(err.Error(), "could not list organization")) assert.False(t, gock.HasUnmatchedRequest()) assert.True(t, gock.IsDone()) @@ -236,7 +236,7 @@ func TestAddMembersByOrg_NoMembers(t *testing.T) { JSON([]map[string]string{}) s := initTestSource(&sourcespb.GitHub{Credential: &sourcespb.GitHub_Unauthenticated{}}) - err := s.addMembersByOrg(context.Background(), "org1") + err := s.addMembersByOrg(context.Background(), "org1", noopReporter()) assert.Equal(t, fmt.Sprintf("organization (%q) had 0 members: account may not have access to list organization members", "org1"), err.Error()) assert.False(t, gock.HasUnmatchedRequest()) @@ -276,7 +276,7 @@ func TestAddMembersByApp(t *testing.T) { AppId: "4141", }, }}) - err := s.addMembersByApp(context.Background(), s.connector.(*appConnector).InstallationClient()) + err := s.addMembersByApp(context.Background(), s.connector.(*appConnector).InstallationClient(), noopReporter()) assert.Nil(t, err) assert.Equal(t, 3, len(s.memberCache)) _, ok := s.memberCache["ssm1"] @@ -327,7 +327,7 @@ func TestAddOrgsByUser(t *testing.T) { }) s := initTestSource(&sourcespb.GitHub{Credential: &sourcespb.GitHub_Unauthenticated{}}) - s.addOrgsByUser(context.Background(), "super-secret-user") + s.addOrgsByUser(context.Background(), "super-secret-user", noopReporter()) assert.Equal(t, 1, s.orgsCache.Count()) ok := s.orgsCache.Exists("sso2") assert.True(t, ok) diff --git a/pkg/sources/github/repo.go b/pkg/sources/github/repo.go index c5e813578..660dfd955 100644 --- a/pkg/sources/github/repo.go +++ b/pkg/sources/github/repo.go @@ -154,6 +154,9 @@ func (s *Source) getReposByOrgOrUser(ctx context.Context, name string, reporter if err == nil { return organization, nil } else if !isGitHub404Error(err) { + if err := reporter.UnitErr(ctx, fmt.Errorf("error getting repos by org: %w", err)); err != nil { + return unknown, err + } return unknown, err } @@ -181,6 +184,7 @@ func isGitHub404Error(err error) bool { return ghErr.Response.StatusCode == http.StatusNotFound } +// processRepos is the main function for getting repositories from a source. func (s *Source) processRepos(ctx context.Context, target string, reporter sources.UnitReporter, listRepos repoLister, listOpts repoListOptions) error { logger := ctx.Logger().WithValues("target", target) opts := listOpts.getListOptions() @@ -192,7 +196,7 @@ func (s *Source) processRepos(ctx context.Context, target string, reporter sourc for { someRepos, res, err := listRepos(ctx, target, listOpts) - if s.handleRateLimit(ctx, err) { + if s.handleRateLimitWithUnitReporter(ctx, reporter, err) { continue } if err != nil { diff --git a/pkg/sources/source_manager.go b/pkg/sources/source_manager.go index 43784141f..2e03be1d8 100644 --- a/pkg/sources/source_manager.go +++ b/pkg/sources/source_manager.go @@ -406,8 +406,7 @@ func (s *SourceManager) enumerate(ctx context.Context, source Source, report *Jo // Check if source units are supported and configured. canUseSourceUnits := s.useSourceUnitsFunc != nil if enumChunker, ok := source.(SourceUnitEnumerator); ok && canUseSourceUnits && s.useSourceUnitsFunc() { - ctx.Logger().Info("running source", - "with_units", true) + ctx.Logger().Info("running source", "with_units", true) return s.enumerateWithUnits(ctx, enumChunker, report, reporter) } return fmt.Errorf("Enumeration not supported or configured for source: %s", source.Type().String())