remove mutex usage around failFast.

* updating sendgrid detector to use tri-state verification

main.go

@@ -53,6 +53,7 @@ var (
 	printAvgDetectorTime = cli.Flag("print-avg-detector-time", "Print the average time spent on each detector.").Bool()
 	noUpdate             = cli.Flag("no-update", "Don't check for updates.").Bool()
 	fail                 = cli.Flag("fail", "Exit with code 183 if results are found.").Bool()
+	failFast             = cli.Flag("fail-fast", "Exit with code 1 as soon as a single result is found.").Bool()
 	verifiers            = cli.Flag("verifier", "Set custom verification endpoints.").StringMap()
 	archiveMaxSize       = cli.Flag("archive-max-size", "Maximum size of archive to scan. (Byte units eg. 512B, 2KB, 4MB)").Bytes()
 	archiveMaxDepth      = cli.Flag("archive-max-depth", "Maximum depth of archive to scan.").Int()
@@ -105,11 +106,11 @@ var (
 	filesystemScanIncludePaths = filesystemScan.Flag("include-paths", "Path to file with newline separated regexes for files to include in scan.").Short('i').String()
 	filesystemScanExcludePaths = filesystemScan.Flag("exclude-paths", "Path to file with newline separated regexes for files to exclude in scan.").Short('x').String()
 
-	s3Scan    = cli.Command("s3", "Find credentials in S3 buckets.")
-	s3ScanKey = s3Scan.Flag("key", "S3 key used to authenticate. Can be provided with environment variable AWS_ACCESS_KEY_ID.").Envar("AWS_ACCESS_KEY_ID").String()
-	s3ScanRoleArns     = s3Scan.Flag("role-arn", "Specify the ARN of an IAM role to assume for scanning. You can repeat this flag.").Strings()
-	s3ScanSecret       = s3Scan.Flag("secret", "S3 secret used to authenticate. Can be provided with environment variable AWS_SECRET_ACCESS_KEY.").Envar("AWS_SECRET_ACCESS_KEY").String()
-	s3ScanSessionToken = s3Scan.Flag("session-token", "S3 session token used to authenticate temporary credentials. Can be provided with environment variable AWS_SESSION_TOKEN.").Envar("AWS_SESSION_TOKEN").String()
+	s3Scan              = cli.Command("s3", "Find credentials in S3 buckets.")
+	s3ScanKey           = s3Scan.Flag("key", "S3 key used to authenticate. Can be provided with environment variable AWS_ACCESS_KEY_ID.").Envar("AWS_ACCESS_KEY_ID").String()
+	s3ScanRoleArns      = s3Scan.Flag("role-arn", "Specify the ARN of an IAM role to assume for scanning. You can repeat this flag.").Strings()
+	s3ScanSecret        = s3Scan.Flag("secret", "S3 secret used to authenticate. Can be provided with environment variable AWS_SECRET_ACCESS_KEY.").Envar("AWS_SECRET_ACCESS_KEY").String()
+	s3ScanSessionToken  = s3Scan.Flag("session-token", "S3 session token used to authenticate temporary credentials. Can be provided with environment variable AWS_SESSION_TOKEN.").Envar("AWS_SESSION_TOKEN").String()
 	s3ScanCloudEnv      = s3Scan.Flag("cloud-environment", "Use IAM credentials in cloud environment.").Bool()
 	s3ScanBuckets       = s3Scan.Flag("bucket", "Name of S3 bucket to scan. You can repeat this flag.").Strings()
 	s3ScanMaxObjectSize = s3Scan.Flag("max-object-size", "Maximum size of objects to scan. Objects larger than this will be skipped. (Byte units eg. 512B, 2KB, 4MB)").Default("250MB").Bytes()
@@ -366,6 +367,7 @@ func run(state overseer.State) {
 		engine.WithOnlyVerified(*onlyVerified),
 		engine.WithPrintAvgDetectorTime(*printAvgDetectorTime),
 		engine.WithPrinter(printer),
+		engine.WithFailFast(*failFast),
 	)
 	if err != nil {
 		logFatal(err, "error initializing engine")
@@ -469,7 +471,7 @@ func run(state overseer.State) {
 			Secret:        *s3ScanSecret,
 			SessionToken:  *s3ScanSessionToken,
 			Buckets:       *s3ScanBuckets,
-			Roles:	       *s3ScanRoleArns,
+			Roles:         *s3ScanRoleArns,
 			CloudCred:     *s3ScanCloudEnv,
 			MaxObjectSize: int64(*s3ScanMaxObjectSize),
 		}
@@ -533,6 +535,9 @@ func run(state overseer.State) {
 	if err = e.Finish(ctx); err != nil {
 		logFatal(err, "engine failed to finish execution")
 	}
+	if e.IsTerminated() {
+		os.Exit(1)
+	}
 
 	metrics := e.GetMetrics()
 	// Print results.

pkg/detectors/jiratoken/jiratoken.go

@@ -13,16 +13,18 @@ import (
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/detectorspb"
 )
 
-type Scanner struct{}
+type Scanner struct {
+	client *http.Client
+}
 
 // Ensure the Scanner satisfies the interface at compile time.
 var _ detectors.Detector = (*Scanner)(nil)
 var _ detectors.Versioner = (*Scanner)(nil)
 
-func (Scanner) Version() int            { return 1 }
+func (Scanner) Version() int { return 1 }
 
 var (
-	client = common.SaneHttpClient()
+	defaultClient = common.SaneHttpClient()
 
 	// Make sure that your group is surrounded in boundary characters such as below to reduce false positives.
 	tokenPat  = regexp.MustCompile(detectors.PrefixRegex([]string{"jira"}) + `\b([a-zA-Z-0-9]{24})\b`)
@@ -73,6 +75,11 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 				}
 
 				if verify {
+					client := s.client
+					if client == nil {
+						client = defaultClient
+					}
+
 					data := fmt.Sprintf("%s:%s", resEmail, resToken)
 					sEnc := b64.StdEncoding.EncodeToString([]byte(data))
 					req, err := http.NewRequestWithContext(ctx, "GET", "https://"+resDomain+"/rest/api/3/dashboard", nil)
@@ -88,9 +95,15 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 						// If the request is successful and the login reason is not failed authentication, then the token is valid.
 						// This is because Jira returns a 200 status code even if the token is invalid.
 						// Jira returns a default dashboard page.
-						if (res.StatusCode >= 200 && res.StatusCode < 300) && res.Header.Get(loginReasonHeaderKey) != failedAuth {
-							s1.Verified = true
+						if res.StatusCode >= 200 && res.StatusCode < 300 {
+							if res.Header.Get(loginReasonHeaderKey) != failedAuth {
+								s1.Verified = true
+							}
+						} else {
+							s1.VerificationError = fmt.Errorf("unexpected HTTP response status %d", res.StatusCode)
 						}
+					} else {
+						s1.VerificationError = err
 					}
 				}
 

pkg/detectors/jiratoken/jiratoken_test.go

@@ -9,7 +9,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/kylelemons/godebug/pretty"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
@@ -23,8 +24,8 @@ func TestJiraToken_FromChunk(t *testing.T) {
 	if err != nil {
 		t.Fatalf("could not get test secrets from GCP: %s", err)
 	}
-	token := testSecrets.MustGetField("JIRA_TOKEN")
-	inactiveToken := testSecrets.MustGetField("JIRA_INACTIVE")
+	secret := testSecrets.MustGetField("JIRA_TOKEN")
+	inactiveSecret := testSecrets.MustGetField("JIRA_INACTIVE")
 	email := testSecrets.MustGetField("JIRA_EMAIL")
 	domain := testSecrets.MustGetField("JIRA_DOMAIN")
 
@@ -34,18 +35,19 @@ func TestJiraToken_FromChunk(t *testing.T) {
 		verify bool
 	}
 	tests := []struct {
-		name    string
-		s       Scanner
-		args    args
-		want    []detectors.Result
-		wantErr bool
+		name                string
+		s                   Scanner
+		args                args
+		want                []detectors.Result
+		wantErr             bool
+		wantVerificationErr bool
 	}{
 		{
 			name: "found, verified",
 			s:    Scanner{},
 			args: args{
 				ctx:    context.Background(),
-				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", token, email, domain)),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
 				verify: true,
 			},
 			want: []detectors.Result{
@@ -54,14 +56,15 @@ func TestJiraToken_FromChunk(t *testing.T) {
 					Verified:     true,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "found, unverified",
 			s:    Scanner{},
 			args: args{
 				ctx:    context.Background(),
-				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s but not jira %s valid", inactiveToken, email, domain)), // the secret would satisfy the regex but not pass validation
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s but not jira %s valid", inactiveSecret, email, domain)), // the secret would satisfy the regex but not pass validation
 				verify: true,
 			},
 			want: []detectors.Result{
@@ -70,7 +73,8 @@ func TestJiraToken_FromChunk(t *testing.T) {
 					Verified:     false,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "not found",
@@ -80,14 +84,48 @@ func TestJiraToken_FromChunk(t *testing.T) {
 				data:   []byte("You cannot find the secret within"),
 				verify: true,
 			},
-			want:    nil,
-			wantErr: false,
+			want:                nil,
+			wantErr:             false,
+			wantVerificationErr: false,
+		},
+		{
+			name: "found, would be verified if not for timeout",
+			s:    Scanner{client: common.SaneHttpClientTimeOut(1 * time.Microsecond)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_JiraToken,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but unexpected api surface",
+			s:    Scanner{client: common.ConstantResponseHttpClient(404, "")},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_JiraToken,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s := Scanner{}
-			got, err := s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
+			got, err := tt.s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("JiraToken.FromData() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -96,10 +134,12 @@ func TestJiraToken_FromChunk(t *testing.T) {
 				if len(got[i].Raw) == 0 {
 					t.Fatalf("no raw secret present: \n %+v", got[i])
 				}
-				got[i].Raw = nil
-				got[i].RawV2 = nil
+				if (got[i].VerificationError != nil) != tt.wantVerificationErr {
+					t.Fatalf("wantVerificationError = %v, verification error = %v", tt.wantVerificationErr, got[i].VerificationError)
+				}
 			}
-			if diff := pretty.Compare(got, tt.want); diff != "" {
+			ignoreOpts := cmpopts.IgnoreFields(detectors.Result{}, "Raw", "RawV2", "VerificationError")
+			if diff := cmp.Diff(got, tt.want, ignoreOpts); diff != "" {
 				t.Errorf("JiraToken.FromData() %s diff: (-got +want)\n%s", tt.name, diff)
 			}
 		})

pkg/detectors/jiratoken_v2/jiratoken_v2.go

@@ -13,16 +13,18 @@ import (
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/detectorspb"
 )
 
-type Scanner struct{}
+type Scanner struct {
+	client *http.Client
+}
 
 // Ensure the Scanner satisfies the interface at compile time.
 var _ detectors.Detector = (*Scanner)(nil)
 var _ detectors.Versioner = (*Scanner)(nil)
 
-func (Scanner) Version() int            { return 2 }
+func (Scanner) Version() int { return 2 }
 
 var (
-	client = common.SaneHttpClient()
+	defaultClient = common.SaneHttpClient()
 
 	// https://support.atlassian.com/atlassian-account/docs/manage-api-tokens-for-your-atlassian-account/
 	// Tokens created after Jan 18 2023 use a variable length
@@ -74,6 +76,11 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 				}
 
 				if verify {
+					client := s.client
+					if client == nil {
+						client = defaultClient
+					}
+
 					data := fmt.Sprintf("%s:%s", resEmail, resToken)
 					sEnc := b64.StdEncoding.EncodeToString([]byte(data))
 					req, err := http.NewRequestWithContext(ctx, "GET", "https://"+resDomain+"/rest/api/3/dashboard", nil)
@@ -89,11 +96,15 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 						// If the request is successful and the login reason is not failed authentication, then the token is valid.
 						// This is because Jira returns a 200 status code even if the token is invalid.
 						// Jira returns a default dashboard page.
-						if (res.StatusCode >= 200 && res.StatusCode < 300) && res.Header.Get(loginReasonHeaderKey) != failedAuth {
-							s1.Verified = true
+						if res.StatusCode >= 200 && res.StatusCode < 300 {
+							if res.Header.Get(loginReasonHeaderKey) != failedAuth {
+								s1.Verified = true
+							}
 						} else {
-							s1.VerificationError = err
+							s1.VerificationError = fmt.Errorf("unexpected HTTP response status %d", res.StatusCode)
 						}
+					} else {
+						s1.VerificationError = err
 					}
 				}
 

pkg/detectors/jiratoken_v2/jiratoken_v2_test.go

@@ -9,7 +9,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/kylelemons/godebug/pretty"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
@@ -23,8 +24,8 @@ func TestJiraToken_FromChunk(t *testing.T) {
 	if err != nil {
 		t.Fatalf("could not get test secrets from GCP: %s", err)
 	}
-	token := testSecrets.MustGetField("JIRA_V2_TOKEN")
-	inactiveToken := testSecrets.MustGetField("JIRA_V2_INACTIVE")
+	secret := testSecrets.MustGetField("JIRA_V2_TOKEN")
+	inactiveSecret := testSecrets.MustGetField("JIRA_V2_INACTIVE")
 	email := testSecrets.MustGetField("JIRA_V2_EMAIL")
 	domain := testSecrets.MustGetField("JIRA_V2_DOMAIN")
 
@@ -34,18 +35,19 @@ func TestJiraToken_FromChunk(t *testing.T) {
 		verify bool
 	}
 	tests := []struct {
-		name    string
-		s       Scanner
-		args    args
-		want    []detectors.Result
-		wantErr bool
+		name                string
+		s                   Scanner
+		args                args
+		want                []detectors.Result
+		wantErr             bool
+		wantVerificationErr bool
 	}{
 		{
 			name: "found, verified",
 			s:    Scanner{},
 			args: args{
 				ctx:    context.Background(),
-				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", token, email, domain)),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
 				verify: true,
 			},
 			want: []detectors.Result{
@@ -54,14 +56,15 @@ func TestJiraToken_FromChunk(t *testing.T) {
 					Verified:     true,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "found, unverified",
 			s:    Scanner{},
 			args: args{
 				ctx:    context.Background(),
-				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s but not jira %s valid", inactiveToken, email, domain)), // the secret would satisfy the regex but not pass validation
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s but not jira %s valid", inactiveSecret, email, domain)), // the secret would satisfy the regex but not pass validation
 				verify: true,
 			},
 			want: []detectors.Result{
@@ -70,7 +73,8 @@ func TestJiraToken_FromChunk(t *testing.T) {
 					Verified:     false,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "not found",
@@ -80,14 +84,48 @@ func TestJiraToken_FromChunk(t *testing.T) {
 				data:   []byte("You cannot find the secret within"),
 				verify: true,
 			},
-			want:    nil,
-			wantErr: false,
+			want:                nil,
+			wantErr:             false,
+			wantVerificationErr: false,
+		},
+		{
+			name: "found, would be verified if not for timeout",
+			s:    Scanner{client: common.SaneHttpClientTimeOut(1 * time.Microsecond)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_JiraToken,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but unexpected api surface",
+			s:    Scanner{client: common.ConstantResponseHttpClient(404, "")},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a jira secret %s within jira %s with jira %s", secret, email, domain)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_JiraToken,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s := Scanner{}
-			got, err := s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
+			got, err := tt.s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("JiraToken.FromData() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -96,10 +134,12 @@ func TestJiraToken_FromChunk(t *testing.T) {
 				if len(got[i].Raw) == 0 {
 					t.Fatalf("no raw secret present: \n %+v", got[i])
 				}
-				got[i].Raw = nil
-				got[i].RawV2 = nil
+				if (got[i].VerificationError != nil) != tt.wantVerificationErr {
+					t.Fatalf("wantVerificationError = %v, verification error = %v", tt.wantVerificationErr, got[i].VerificationError)
+				}
 			}
-			if diff := pretty.Compare(got, tt.want); diff != "" {
+			ignoreOpts := cmpopts.IgnoreFields(detectors.Result{}, "Raw", "RawV2", "VerificationError")
+			if diff := cmp.Diff(got, tt.want, ignoreOpts); diff != "" {
 				t.Errorf("JiraToken.FromData() %s diff: (-got +want)\n%s", tt.name, diff)
 			}
 		})

pkg/detectors/pagerdutyapikey/pagerdutyapikey.go

@@ -21,13 +21,13 @@ var (
 	client = common.SaneHttpClient()
 
 	// Make sure that your group is surrounded in boundary characters such as below to reduce false positives.
-	keyPat = regexp.MustCompile(detectors.PrefixRegex([]string{"pagerduty"}) + `\b([a-z]{1}\+[a-zA-Z]{9}\-[a-z]{2}\-[a-z0-9]{5})\b`)
+	keyPat = regexp.MustCompile(detectors.PrefixRegex([]string{"pagerduty", "pd"}) + `\b([a-zA-Z0-9_+-]{20})\b`)
 )
 
 // Keywords are used for efficiently pre-filtering chunks.
 // Use identifiers in the secret preferably, or the provider name.
 func (s Scanner) Keywords() []string {
-	return []string{"pagerduty"}
+	return []string{"pagerduty", "pd"}
 }
 
 // FromData will find and optionally verify PagerDutyApiKey secrets in a given set of bytes.

pkg/detectors/sendbird/sendbird.go

@@ -2,6 +2,7 @@ package sendbird
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"regexp"
@@ -12,19 +13,25 @@ import (
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/detectorspb"
 )
 
-type Scanner struct{}
+type Scanner struct {
+	client *http.Client
+}
 
 // Ensure the Scanner satisfies the interface at compile time.
 var _ detectors.Detector = (*Scanner)(nil)
 
 var (
-	client = common.SaneHttpClient()
+	defaultClient = common.SaneHttpClient()
 
 	// Make sure that your group is surrounded in boundary characters such as below to reduce false positives.
 	keyPat   = regexp.MustCompile(detectors.PrefixRegex([]string{"sendbird"}) + `\b([0-9a-f]{40})\b`)
 	appIdPat = regexp.MustCompile(detectors.PrefixRegex([]string{"sendbird"}) + `\b([0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12})\b`)
 )
 
+type userResp struct {
+	Code int `json:"code"`
+}
+
 // Keywords are used for efficiently pre-filtering chunks.
 // Use identifiers in the secret preferably, or the provider name.
 func (s Scanner) Keywords() []string {
@@ -61,20 +68,40 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 					continue
 				}
 				req.Header.Add("Api-Token", resMatch)
+
+				client := s.client
+				if client == nil {
+					client = defaultClient
+				}
+
 				res, err := client.Do(req)
 				if err == nil {
 					defer res.Body.Close()
 					if res.StatusCode >= 200 && res.StatusCode < 300 {
 						s1.Verified = true
-					} else {
-						// This function will check false positives for common test words, but also it will make sure the key appears 'random' enough to be a real key.
-						if detectors.IsKnownFalsePositive(resMatch, detectors.DefaultFalsePositives, true) {
-							continue
+					} else if res.StatusCode == 400 { // Sendbird returns 400 for all errors
+						var userResp userResp
+						err := json.NewDecoder(res.Body).Decode(&userResp)
+						if err != nil {
+							s1.VerificationError = fmt.Errorf("error decoding json response body: %w", err)
+						} else if userResp.Code != 400401 {
+							// https://sendbird.com/docs/chat/platform-api/v3/error-codes
+							// Sendbird always includes its own error codes with 400 responses
+							// 400401 (InvalidApiToken) is the only one that indicates a bad token
+							s1.VerificationError = fmt.Errorf("unexpected response code: %d", userResp.Code)
 						}
+					} else {
+						s1.VerificationError = fmt.Errorf("unexpected HTTP response status %d", res.StatusCode)
 					}
+				} else {
+					s1.VerificationError = err
 				}
 			}
 
+			// This function will check false positives for common test words, but also it will make sure the key appears 'random' enough to be a real key.
+			if !s1.Verified && detectors.IsKnownFalsePositive(resMatch, detectors.DefaultFalsePositives, true) {
+				continue
+			}
 			results = append(results, s1)
 		}
 	}

pkg/detectors/sendbird/sendbird_test.go

@@ -9,8 +9,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/kylelemons/godebug/pretty"
-
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/detectorspb"
@@ -19,7 +19,7 @@ import (
 func TestSendbird_FromChunk(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()
-	testSecrets, err := common.GetSecret(ctx, "trufflehog-testing", "detectors3")
+	testSecrets, err := common.GetSecret(ctx, "trufflehog-testing", "detectors5")
 	if err != nil {
 		t.Fatalf("could not get test secrets from GCP: %s", err)
 	}
@@ -33,11 +33,12 @@ func TestSendbird_FromChunk(t *testing.T) {
 		verify bool
 	}
 	tests := []struct {
-		name    string
-		s       Scanner
-		args    args
-		want    []detectors.Result
-		wantErr bool
+		name                string
+		s                   Scanner
+		args                args
+		want                []detectors.Result
+		wantErr             bool
+		wantVerificationErr bool
 	}{
 		{
 			name: "found, verified",
@@ -53,7 +54,8 @@ func TestSendbird_FromChunk(t *testing.T) {
 					Verified:     true,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "found, unverified",
@@ -69,7 +71,8 @@ func TestSendbird_FromChunk(t *testing.T) {
 					Verified:     false,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "not found",
@@ -79,14 +82,82 @@ func TestSendbird_FromChunk(t *testing.T) {
 				data:   []byte("You cannot find the secret within"),
 				verify: true,
 			},
-			want:    nil,
-			wantErr: false,
+			want:                nil,
+			wantErr:             false,
+			wantVerificationErr: false,
+		},
+		{
+			name: "found, would be verified if not for timeout",
+			s:    Scanner{client: common.SaneHttpClientTimeOut(1 * time.Microsecond)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendbird application ID %s with sendbird secret %s within", appId, secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_Sendbird,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but unexpected api surface",
+			s:    Scanner{client: common.ConstantResponseHttpClient(404, "")},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendbird application ID %s with sendbird secret %s within", appId, secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_Sendbird,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but unexpected error code",
+			s:    Scanner{client: common.ConstantResponseHttpClient(400, `{"code":400402}`)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendbird application ID %s with sendbird secret %s within", appId, secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_Sendbird,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but failed to decode json",
+			s:    Scanner{client: common.ConstantResponseHttpClient(400, `{`)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendbird application ID %s with sendbird secret %s within", appId, secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_Sendbird,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s := Scanner{}
-			got, err := s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
+			got, err := tt.s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Sendbird.FromData() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -95,9 +166,12 @@ func TestSendbird_FromChunk(t *testing.T) {
 				if len(got[i].Raw) == 0 {
 					t.Fatalf("no raw secret present: \n %+v", got[i])
 				}
-				got[i].Raw = nil
+				if (got[i].VerificationError != nil) != tt.wantVerificationErr {
+					t.Fatalf("wantVerificationError = %v, verification error = %v", tt.wantVerificationErr, got[i].VerificationError)
+				}
 			}
-			if diff := pretty.Compare(got, tt.want); diff != "" {
+			ignoreOpts := cmpopts.IgnoreFields(detectors.Result{}, "Raw", "VerificationError")
+			if diff := cmp.Diff(got, tt.want, ignoreOpts); diff != "" {
 				t.Errorf("Sendbird.FromData() %s diff: (-got +want)\n%s", tt.name, diff)
 			}
 		})

pkg/detectors/sendgrid/sendgrid.go

@@ -13,13 +13,15 @@ import (
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 )
 
-type Scanner struct{}
+type Scanner struct {
+	client *http.Client
+}
 
 // Ensure the Scanner satisfies the interface at compile time.
 var _ detectors.Detector = (*Scanner)(nil)
 
 var (
-	client = common.SaneHttpClient()
+	defaultClient = common.SaneHttpClient()
 
 	keyPat = regexp.MustCompile(detectors.PrefixRegex([]string{"sendgrid"}) + `(SG\.[\w\-_]{20,24}\.[\w\-_]{39,50})\b`)
 )
@@ -42,7 +44,7 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 		}
 		res := strings.TrimSpace(match[1])
 
-		s := detectors.Result{
+		s1 := detectors.Result{
 			DetectorType: detectorspb.DetectorType_SendGrid,
 			Raw:          []byte(res),
 		}
@@ -52,6 +54,11 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 			// 403 will be issued if the scope is wrong but the key is correct
 			baseURL := "https://api.sendgrid.com/v3/templates"
 
+			client := s.client
+			if client == nil {
+				client = defaultClient
+			}
+
 			// test `read_user` scope
 			req, err := http.NewRequestWithContext(ctx, "GET", baseURL, nil)
 			if err != nil {
@@ -67,16 +74,18 @@ func (s Scanner) FromData(ctx context.Context, verify bool, data []byte) (result
 				// 403 means good key but not the right scope
 				// 401 is bad key
 				if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusForbidden {
-					s.Verified = true
+					s1.Verified = true
+				} else if res.StatusCode != http.StatusUnauthorized {
+					s1.VerificationError = fmt.Errorf("unexpected HTTP response status %d", res.StatusCode)
 				}
 			}
 		}
 
-		if !s.Verified && detectors.IsKnownFalsePositive(string(s.Raw), detectors.DefaultFalsePositives, true) {
+		if !s1.Verified && detectors.IsKnownFalsePositive(string(s1.Raw), detectors.DefaultFalsePositives, true) {
 			continue
 		}
 
-		results = append(results, s)
+		results = append(results, s1)
 	}
 
 	return

pkg/detectors/sendgrid/sendgrid_test.go

@@ -9,7 +9,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/kylelemons/godebug/pretty"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/detectorspb"
@@ -18,7 +19,7 @@ import (
 func TestSendgrid_FromChunk(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()
-	testSecrets, err := common.GetSecret(ctx, "trufflehog-testing", "detectors2")
+	testSecrets, err := common.GetSecret(ctx, "trufflehog-testing", "detectors5")
 	if err != nil {
 		t.Fatalf("could not get test secrets from GCP: %s", err)
 	}
@@ -30,11 +31,12 @@ func TestSendgrid_FromChunk(t *testing.T) {
 		verify bool
 	}
 	tests := []struct {
-		name    string
-		s       Scanner
-		args    args
-		want    []detectors.Result
-		wantErr bool
+		name                string
+		s                   Scanner
+		args                args
+		want                []detectors.Result
+		wantErr             bool
+		wantVerificationErr bool
 	}{
 		{
 			name: "found, verified",
@@ -50,7 +52,8 @@ func TestSendgrid_FromChunk(t *testing.T) {
 					Verified:     true,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "found, unverified",
@@ -66,7 +69,8 @@ func TestSendgrid_FromChunk(t *testing.T) {
 					Verified:     false,
 				},
 			},
-			wantErr: false,
+			wantErr:             false,
+			wantVerificationErr: false,
 		},
 		{
 			name: "not found",
@@ -76,22 +80,54 @@ func TestSendgrid_FromChunk(t *testing.T) {
 				data:   []byte("You cannot find the secret within"),
 				verify: true,
 			},
-			want:    nil,
-			wantErr: false,
+			want:                nil,
+			wantErr:             false,
+			wantVerificationErr: false,
+		},
+		{
+			name: "found, would be verified if not for timeout",
+			s:    Scanner{client: common.SaneHttpClientTimeOut(1 * time.Microsecond)},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendgrid secret %s within", secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_SendGrid,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
+		},
+		{
+			name: "found, verified but unexpected api surface",
+			s:    Scanner{client: common.ConstantResponseHttpClient(404, "")},
+			args: args{
+				ctx:    context.Background(),
+				data:   []byte(fmt.Sprintf("You can find a sendgrid secret %s within", secret)),
+				verify: true,
+			},
+			want: []detectors.Result{
+				{
+					DetectorType: detectorspb.DetectorType_SendGrid,
+					Verified:     false,
+				},
+			},
+			wantErr:             false,
+			wantVerificationErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s := Scanner{}
-			got, err := s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
+			got, err := tt.s.FromData(tt.args.ctx, tt.args.verify, tt.args.data)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Sendgrid.FromData() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
-			for i := range got {
-				got[i].Raw = nil
-			}
-			if diff := pretty.Compare(got, tt.want); diff != "" {
+			ignoreOpts := cmpopts.IgnoreFields(detectors.Result{}, "Raw", "VerificationError")
+			if diff := cmp.Diff(got, tt.want, ignoreOpts); diff != "" {
 				t.Errorf("Sendgrid.FromData() %s diff: (-got +want)\n%s", tt.name, diff)
 			}
 		})

pkg/engine/engine.go

@@ -51,6 +51,53 @@ type Printer interface {
 	Print(ctx context.Context, r *detectors.ResultWithMetadata) error
 }
 
+// terminationSignal provides a mechanism to safely signal a termination
+// event across different goroutines. It uses a combination of a condition variable
+// and a mutex to enable safe signaling and waiting, ensuring that the signaling
+// is consistent and that there are no race conditions when checking or updating
+// the status of the termination.
+type terminationSignal struct {
+	// isTerminated indicates the signal state (true if the engine is terminated).
+	isTerminated bool
+	// isFailFast indicates if the engine is in fail-fast mode.
+	isFailFast bool
+	// mutex is used to protect access to isFailFast and to synchronize
+	// the condition variable.
+	mutex sync.Mutex
+	// failFastCondition is a condition variable that's signaled when
+	// a fail-fast event occurs. It's used in Finish to wait for or
+	// react to this signal.
+	failFastCondition *sync.Cond
+}
+
+// newTerminationSignal initializes a new terminationSignal.
+// The failFast argument determines the initial state.
+func newTerminationSignal(failFast bool) *terminationSignal {
+	ts := &terminationSignal{isFailFast: failFast}
+	// Initialize the condition variable with the mutex.
+	ts.failFastCondition = sync.NewCond(&ts.mutex)
+	return ts
+}
+
+// signalFinish signals that the engine has finished its work.
+// It uses the condition variable to notify any waiting goroutines
+// (e.g., the Finish method).
+func (t *terminationSignal) signalFinish() {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	t.isTerminated = true
+
+	t.failFastCondition.Signal() // Signal the condition to unblock Finish.
+}
+
+func (t *terminationSignal) wait() {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	for !t.isTerminated {
+		t.failFastCondition.Wait()
+	}
+}
+
 type Engine struct {
 	// CLI flags.
 	concurrency uint8
@@ -74,6 +121,11 @@ type Engine struct {
 	workersWg            sync.WaitGroup
 	wgDetectorWorkers    sync.WaitGroup
 	WgNotifier           sync.WaitGroup
+	// termSignal is used to signal the engine to terminate.
+	// It is used to implement the fail-fast feature,
+	// where the engine will terminate scanning as soon as
+	// the first potential secret is detected.
+	termSignal *terminationSignal
 
 	// Runtime information.
 	metrics runtimeMetrics
@@ -169,6 +221,19 @@ func WithPrinter(printer Printer) EngineOption {
 	}
 }
 
+// WithFailFast returns an EngineOption that configures the engine's termination behavior.
+// If 'failFast' is enabled, the engine will terminate scanning upon detecting the first potential secret.
+// This mode is valuable in contexts like CI/CD pipelines or preliminary scans where
+// immediate feedback is more important than a comprehensive review. In the non-failFast mode,
+// the engine continues its operation without waiting for a result.
+// The boolean parameter 'failFast' dictates the termination behavior.
+func WithFailFast(failFast bool) EngineOption {
+	return func(e *Engine) {
+		// Configure the termination signal with the desired mode (failFast or not).
+		e.termSignal = newTerminationSignal(failFast)
+	}
+}
+
 func filterDetectors(filterFunc func(detectors.Detector) bool, input []detectors.Detector) []detectors.Detector {
 	var output []detectors.Detector
 	for _, detector := range input {
@@ -276,6 +341,7 @@ func Start(ctx context.Context, options ...EngineOption) (*Engine, error) {
 		dedupeCache:          cache,
 		printer:              new(output.PlainPrinter), // default printer
 		metrics:              runtimeMetrics{Metrics: Metrics{scanStartTime: time.Now()}},
+		termSignal:           &terminationSignal{},
 	}
 
 	for _, option := range options {
@@ -367,22 +433,15 @@ func Start(ctx context.Context, options ...EngineOption) (*Engine, error) {
 		}()
 	}
 
-	// We want 1/4th of the notifier workers as the number of scanner workers.
-	const notifierWorkerRatio = 4
-	maxNotifierWorkers := 1
-	if numWorkers := e.concurrency / notifierWorkerRatio; numWorkers > 0 {
-		maxNotifierWorkers = int(numWorkers)
-	}
-	ctx.Logger().V(2).Info("starting notifier workers", "count", maxNotifierWorkers)
-	for worker := 0; worker < maxNotifierWorkers; worker++ {
-		e.WgNotifier.Add(1)
-		go func() {
-			ctx := context.WithValue(ctx, "notifier_worker_id", common.RandomID(5))
-			defer common.Recover(ctx)
-			defer e.WgNotifier.Done()
-			e.notifyResults(ctx)
-		}()
-	}
+	// We want only one notifier worker.
+	ctx.Logger().V(2).Info("starting notifier worker")
+	e.WgNotifier.Add(1)
+	go func() {
+		ctx := context.WithValue(ctx, "notifier_worker_id", common.RandomID(5))
+		defer common.Recover(ctx)
+		defer e.WgNotifier.Done()
+		e.notifyResults(ctx)
+	}()
 
 	return e, nil
 }
@@ -392,6 +451,13 @@ func Start(ctx context.Context, options ...EngineOption) (*Engine, error) {
 // more sources may be scanned by the engine.
 func (e *Engine) Finish(ctx context.Context) error {
 	defer common.RecoverWithExit(ctx)
+
+	// Handle the fail-fast termination signal.
+	if e.termSignal.isFailFast {
+		e.termSignal.wait()
+		return fmt.Errorf("terminated early due to fail-fast flag after printing the first result")
+	}
+
 	// Wait for the sources to finish putting chunks onto the chunks channel.
 	err := e.sourceManager.Wait()
 
@@ -407,6 +473,15 @@ func (e *Engine) Finish(ctx context.Context) error {
 	return err
 }
 
+// IsTerminated checks if the engine received a termination signal.
+// This can be used, for example, to determine if the engine
+// terminated early due to the fail-fast feature being activated.
+func (e *Engine) IsTerminated() bool {
+	e.termSignal.mutex.Lock()
+	defer e.termSignal.mutex.Unlock()
+	return e.termSignal.isTerminated
+}
+
 func (e *Engine) ChunksChan() <-chan *sources.Chunk {
 	return e.sourceManager.Chunks()
 }
@@ -559,27 +634,51 @@ func (e *Engine) processResult(data detectableChunk, res detectors.Result) {
 }
 
 func (e *Engine) notifyResults(ctx context.Context) {
+	if e.termSignal.isFailFast {
+		e.processResultsWithFailFast(ctx)
+	} else {
+		e.processRegularResults(ctx)
+	}
+}
+
+func (e *Engine) processResultsWithFailFast(ctx context.Context) {
 	for r := range e.ResultsChan() {
-		if e.onlyVerified && !r.Verified {
-			continue
-		}
-		atomic.AddUint32(&e.numFoundResults, 1)
+		e.processSingleResult(ctx, &r)
 
-		key := fmt.Sprintf("%s%s%s%+v", r.DetectorType.String(), r.Raw, r.RawV2, r.SourceMetadata)
-		if _, ok := e.dedupeCache.Get(key); ok {
-			continue
-		}
-		e.dedupeCache.Add(key, struct{}{})
+		// If fail-fast is enabled, signal finish after processing the first result.
+		e.termSignal.signalFinish()
+		return
+	}
+}
 
-		if r.Verified {
-			atomic.AddUint64(&e.metrics.VerifiedSecretsFound, 1)
-		} else {
-			atomic.AddUint64(&e.metrics.UnverifiedSecretsFound, 1)
-		}
+func (e *Engine) processRegularResults(ctx context.Context) {
+	for r := range e.ResultsChan() {
+		e.processSingleResult(ctx, &r)
+	}
+}
 
-		if err := e.printer.Print(ctx, &r); err != nil {
-			ctx.Logger().Error(err, "error printing result")
-		}
+func (e *Engine) processSingleResult(ctx context.Context, r *detectors.ResultWithMetadata) {
+	if e.onlyVerified && !r.Verified {
+		return
+	}
+
+	atomic.AddUint32(&e.numFoundResults, 1)
+
+	key := fmt.Sprintf("%s%s%s%+v", r.DetectorType.String(), r.Raw, r.RawV2, r.SourceMetadata)
+	if _, ok := e.dedupeCache.Get(key); ok {
+		return
+	}
+
+	e.dedupeCache.Add(key, struct{}{})
+
+	if r.Verified {
+		atomic.AddUint64(&e.metrics.VerifiedSecretsFound, 1)
+	} else {
+		atomic.AddUint64(&e.metrics.UnverifiedSecretsFound, 1)
+	}
+
+	if err := e.printer.Print(ctx, r); err != nil {
+		ctx.Logger().Error(err, "error printing result")
 	}
 }
 

pkg/handlers/archive.go

@@ -17,12 +17,15 @@ import (
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	logContext "github.com/trufflesecurity/trufflehog/v3/pkg/context"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/sources"
 )
 
 type ctxKey int
 
 const (
 	depthKey ctxKey = iota
+
+	errMaxArchiveDepthReached = "max archive depth reached"
 )
 
 var (
@@ -81,51 +84,66 @@ func (a *Archive) FromFile(originalCtx context.Context, data io.Reader) chan []b
 	return archiveChan
 }
 
+type decompressorInfo struct {
+	depth       int
+	reader      io.Reader
+	archiveChan chan []byte
+	archiver    archiver.Decompressor
+}
+
 // openArchive takes a reader and extracts the contents up to the maximum depth.
 func (a *Archive) openArchive(ctx context.Context, depth int, reader io.Reader, archiveChan chan []byte) error {
 	if depth >= maxDepth {
-		return fmt.Errorf("max archive depth reached")
+		return fmt.Errorf(errMaxArchiveDepthReached)
 	}
-	format, reader, err := archiver.Identify("", reader)
-	if err != nil {
-		if errors.Is(err, archiver.ErrNoMatch) && depth > 0 {
-			chunkSize := 10 * 1024
-			for {
-				chunk := make([]byte, chunkSize)
-				n, _ := reader.Read(chunk)
-				archiveChan <- chunk
-				if n < chunkSize {
-					break
-				}
-			}
-			return nil
-		}
-		return err
+
+	format, arReader, err := archiver.Identify("", reader)
+	if errors.Is(err, archiver.ErrNoMatch) && depth > 0 {
+		return a.handleNonArchiveContent(ctx, arReader, archiveChan)
 	}
+
 	switch archive := format.(type) {
 	case archiver.Decompressor:
-		compReader, err := archive.OpenReader(reader)
-		if err != nil {
-			return err
-		}
-		fileBytes, err := a.ReadToMax(ctx, compReader)
-		if err != nil {
-			return err
-		}
-		newReader := bytes.NewReader(fileBytes)
-		return a.openArchive(ctx, depth+1, newReader, archiveChan)
+		info := decompressorInfo{depth: depth, reader: arReader, archiveChan: archiveChan, archiver: archive}
+		return a.handleDecompressor(ctx, info)
 	case archiver.Extractor:
-		err := archive.Extract(context.WithValue(ctx, depthKey, depth+1), reader, nil, a.extractorHandler(archiveChan))
-		if err != nil {
+		return archive.Extract(context.WithValue(ctx, depthKey, depth+1), reader, nil, a.extractorHandler(archiveChan))
+	default:
+		return fmt.Errorf("unknown archive type: %s", format.Name())
+	}
+}
+
+func (a *Archive) handleNonArchiveContent(ctx context.Context, reader io.Reader, archiveChan chan []byte) error {
+	aCtx := logContext.AddLogger(ctx)
+	chunkReader := sources.NewChunkReader()
+	chunkResChan := chunkReader(aCtx, reader)
+	for data := range chunkResChan {
+		if err := data.Error(); err != nil {
+			aCtx.Logger().Error(err, "error reading chunk")
+			continue
+		}
+		if err := common.CancellableWrite(ctx, archiveChan, data.Bytes()); err != nil {
 			return err
 		}
-		return nil
 	}
-	return fmt.Errorf("Unknown archive type: %s", format.Name())
+	return nil
+}
+
+func (a *Archive) handleDecompressor(ctx context.Context, info decompressorInfo) error {
+	compReader, err := info.archiver.OpenReader(info.reader)
+	if err != nil {
+		return err
+	}
+	fileBytes, err := a.ReadToMax(ctx, compReader)
+	if err != nil {
+		return err
+	}
+	newReader := bytes.NewReader(fileBytes)
+	return a.openArchive(ctx, info.depth+1, newReader, info.archiveChan)
 }
 
 // IsFiletype returns true if the provided reader is an archive.
-func (a *Archive) IsFiletype(ctx context.Context, reader io.Reader) (io.Reader, bool) {
+func (a *Archive) IsFiletype(_ context.Context, reader io.Reader) (io.Reader, bool) {
 	format, readerB, err := archiver.Identify("", reader)
 	if err != nil {
 		return readerB, false
@@ -135,8 +153,9 @@ func (a *Archive) IsFiletype(ctx context.Context, reader io.Reader) (io.Reader,
 		return readerB, true
 	case archiver.Decompressor:
 		return readerB, true
+	default:
+		return readerB, false
 	}
-	return readerB, false
 }
 
 // extractorHandler is applied to each file in an archiver.Extractor file.
@@ -180,7 +199,7 @@ func (a *Archive) ReadToMax(ctx context.Context, reader io.Reader) (data []byte,
 			if e, ok := r.(error); ok {
 				err = e
 			} else {
-				err = fmt.Errorf("Panic occurred: %v", r)
+				err = fmt.Errorf("panic occurred: %v", r)
 			}
 			logger.Error(err, "Panic occurred when reading archive")
 		}
@@ -283,7 +302,7 @@ func (a *Archive) HandleSpecialized(ctx logContext.Context, reader io.Reader) (i
 // The caller is responsible for closing the returned reader.
 func (a *Archive) extractDebContent(ctx logContext.Context, file io.Reader) (io.ReadCloser, error) {
 	if a.currentDepth >= maxDepth {
-		return nil, fmt.Errorf("max archive depth reached")
+		return nil, fmt.Errorf(errMaxArchiveDepthReached)
 	}
 
 	tmpEnv, err := a.createTempEnv(ctx, file)

pkg/output/github_actions.go

@@ -4,7 +4,6 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
-	"sync"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
@@ -14,7 +13,7 @@ import (
 var dedupeCache = make(map[string]struct{})
 
 // GitHubActionsPrinter is a printer that prints results in GitHub Actions format.
-type GitHubActionsPrinter struct{ mu sync.Mutex }
+type GitHubActionsPrinter struct{}
 
 func (p *GitHubActionsPrinter) Print(_ context.Context, r *detectors.ResultWithMetadata) error {
 	out := gitHubActionsOutputFormat{
@@ -57,8 +56,6 @@ func (p *GitHubActionsPrinter) Print(_ context.Context, r *detectors.ResultWithM
 	}
 	dedupeCache[key] = struct{}{}
 
-	p.mu.Lock()
-	defer p.mu.Unlock()
 	message := fmt.Sprintf("Found %s %s result 🐷🔑\n", verifiedStatus, out.DetectorType)
 	if r.Result.DecoderType != detectorspb.DecoderType_PLAIN {
 		message = fmt.Sprintf("Found %s %s result with %s encoding 🐷🔑\n", verifiedStatus, out.DetectorType, out.DecoderType)

pkg/output/json.go

@@ -3,7 +3,6 @@ package output
 import (
 	"encoding/json"
 	"fmt"
-	"sync"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/detectors"
@@ -13,7 +12,7 @@ import (
 )
 
 // JSONPrinter is a printer that prints results in JSON format.
-type JSONPrinter struct{ mu sync.Mutex }
+type JSONPrinter struct{}
 
 func (p *JSONPrinter) Print(_ context.Context, r *detectors.ResultWithMetadata) error {
 	v := &struct {
@@ -62,8 +61,6 @@ func (p *JSONPrinter) Print(_ context.Context, r *detectors.ResultWithMetadata)
 		return fmt.Errorf("could not marshal result: %w", err)
 	}
 
-	p.mu.Lock()
 	fmt.Println(string(out))
-	p.mu.Unlock()
 	return nil
 }

pkg/output/legacy_json.go

@@ -7,7 +7,6 @@ import (
 	"net/url"
 	"os"
 	"strings"
-	"sync"
 
 	gogit "github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/plumbing"
@@ -21,7 +20,7 @@ import (
 )
 
 // LegacyJSONPrinter is a printer that prints results in legacy JSON format for backwards compatibility.
-type LegacyJSONPrinter struct{ mu sync.Mutex }
+type LegacyJSONPrinter struct{}
 
 func (p *LegacyJSONPrinter) Print(ctx context.Context, r *detectors.ResultWithMetadata) error {
 	var repo string
@@ -54,9 +53,7 @@ func (p *LegacyJSONPrinter) Print(ctx context.Context, r *detectors.ResultWithMe
 		return fmt.Errorf("could not marshal result: %w", err)
 	}
 
-	p.mu.Lock()
 	fmt.Println(string(out))
-	p.mu.Unlock()
 	return nil
 }
 

pkg/output/plain.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"sort"
 	"strings"
-	"sync"
 
 	"github.com/fatih/color"
 	"golang.org/x/text/cases"
@@ -23,7 +22,7 @@ var (
 )
 
 // PlainPrinter is a printer that prints results in plain text format.
-type PlainPrinter struct{ mu sync.Mutex }
+type PlainPrinter struct{}
 
 func (p *PlainPrinter) Print(_ context.Context, r *detectors.ResultWithMetadata) error {
 	out := outputFormat{
@@ -40,8 +39,6 @@ func (p *PlainPrinter) Print(_ context.Context, r *detectors.ResultWithMetadata)
 	}
 
 	printer := greenPrinter
-	p.mu.Lock()
-	defer p.mu.Unlock()
 
 	if out.Verified {
 		yellowPrinter.Print("Found verified result 🐷🔑\n")

pkg/sources/circleci/circleci.go

@@ -75,7 +75,7 @@ func (s *Source) Init(_ context.Context, name string, jobId, sourceId int64, ver
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	projects, err := s.projects(ctx)
 	if err != nil {
 		return fmt.Errorf("error getting projects: %w", err)

pkg/sources/docker/docker.go

@@ -2,7 +2,6 @@ package docker
 
 import (
 	"archive/tar"
-	"bytes"
 	"compress/gzip"
 	"errors"
 	"fmt"
@@ -14,9 +13,11 @@ import (
 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/google/go-containerregistry/pkg/v1/tarball"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/source_metadatapb"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/sourcespb"
@@ -24,17 +25,16 @@ import (
 )
 
 type Source struct {
-	name     string
-	sourceId int64
-	jobId    int64
-	verify   bool
-	conn     sourcespb.Docker
+	name        string
+	sourceId    int64
+	jobId       int64
+	verify      bool
+	concurrency int
+	conn        sourcespb.Docker
 	sources.Progress
 	sources.CommonSourceUnitUnmarshaller
 }
 
-var FilesizeLimitBytes int64 = 50 * 1024 * 1024 // 50MB
-
 // Ensure the Source satisfies the interfaces at compile time.
 var _ sources.Source = (*Source)(nil)
 var _ sources.SourceUnitUnmarshaller = (*Source)(nil)
@@ -59,6 +59,7 @@ func (s *Source) Init(_ context.Context, name string, jobId, sourceId int64, ver
 	s.sourceId = sourceId
 	s.jobId = jobId
 	s.verify = verify
+	s.concurrency = concurrency
 
 	// Reset metrics for this source at initialization time.
 	dockerImagesScanned.WithLabelValues(s.name).Set(0)
@@ -71,164 +72,207 @@ func (s *Source) Init(_ context.Context, name string, jobId, sourceId int64, ver
 	return nil
 }
 
+type imageInfo struct {
+	image v1.Image
+	base  string
+	tag   string
+}
+
+type layerInfo struct {
+	digest v1.Hash
+	base   string
+	tag    string
+}
+
 // Chunks emits data over a channel that is decoded and scanned for secrets.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
-	remoteOpts, err := s.remoteOpts()
-	if err != nil {
-		return err
-	}
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
+	ctx = context.WithValues(ctx, "source_type", s.Type(), "source_name", s.name)
 
-	logger := ctx.Logger().WithValues("source_type", s.Type(), "source_name", s.name)
+	workers := new(errgroup.Group)
+	workers.SetLimit(s.concurrency)
 
+	scanErrs := sources.NewScanErrors()
 	for _, image := range s.conn.GetImages() {
-		var img v1.Image
-		var err error
-		var base, tag string
-		var hasDigest bool
-		var imageName name.Reference
+		image := image
+		workers.Go(func() error {
+			if common.IsDone(ctx) {
+				return nil
+			}
 
-		if strings.HasPrefix(image, "file://") {
-			image = strings.TrimPrefix(image, "file://")
-			base = image
-			img, err = tarball.ImageFromPath(image, nil)
+			imgInfo, err := s.processImage(ctx, image)
 			if err != nil {
-				return err
-			}
-		} else {
-			base, tag, hasDigest = baseAndTagFromImage(image)
-
-			if hasDigest {
-				imageName, err = name.NewDigest(image)
-				if err != nil {
-					return err
-				}
-			} else {
-				imageName, err = name.NewTag(image)
-				if err != nil {
-					return err
-				}
+				scanErrs.Add(err)
+				return nil
 			}
 
-			img, err = remote.Image(imageName, remoteOpts...)
+			ctx = context.WithValues(ctx, "image", imgInfo.base, "tag", imgInfo.tag)
+			ctx.Logger().V(2).Info("scanning image")
+
+			layers, err := imgInfo.image.Layers()
 			if err != nil {
-				return err
-			}
-		}
-
-		logger = logger.WithValues("image", base, "tag", tag)
-		logger.V(2).Info("scanning image")
-
-		layers, err := img.Layers()
-		if err != nil {
-			return err
-		}
-
-		for _, layer := range layers {
-
-			digest, err := layer.Digest()
-			if err != nil {
-				return err
+				scanErrs.Add(err)
+				return nil
 			}
 
-			logger = logger.WithValues("layer", digest.String())
-			logger.V(2).Info("scanning layer")
-
-			rc, err := layer.Compressed()
-			if err != nil {
-				return err
+			for _, layer := range layers {
+				if err := s.processLayer(ctx, layer, imgInfo, chunksChan); err != nil {
+					scanErrs.Add(err)
+					return nil
+				}
+				dockerLayersScanned.WithLabelValues(s.name).Inc()
 			}
 
-			defer rc.Close()
+			dockerImagesScanned.WithLabelValues(s.name).Inc()
 
-			gzipReader, err := gzip.NewReader(rc)
-			if err != nil {
-				return err
-			}
-
-			defer gzipReader.Close()
-
-			tarReader := tar.NewReader(gzipReader)
-
-			for {
-				header, err := tarReader.Next()
-				if errors.Is(err, io.EOF) {
-					break // End of archive
-				}
-				if err != nil {
-					return err
-				}
-
-				// Skip files larger than FilesizeLimitBytes
-				if header.Size > FilesizeLimitBytes {
-					continue
-				}
-
-				file := bytes.NewBuffer(make([]byte, 0, header.Size))
-
-				_, err = io.Copy(file, tarReader)
-				if err != nil {
-					return err
-				}
-
-				chunk := &sources.Chunk{
-					SourceType: s.Type(),
-					SourceName: s.name,
-					SourceID:   s.SourceID(),
-					JobID:      s.JobID(),
-					Data:       file.Bytes(),
-					SourceMetadata: &source_metadatapb.MetaData{
-						Data: &source_metadatapb.MetaData_Docker{
-							Docker: &source_metadatapb.Docker{
-								File:  "/" + header.Name, // Prepend /, because the file is relative to the root of the image
-								Image: base,
-								Tag:   tag,
-								Layer: digest.String(),
-							},
-						},
-					},
-					Verify: s.verify,
-				}
-
-				select {
-				case chunksChan <- chunk:
-				case <-ctx.Done():
-					return ctx.Err()
-				}
-			}
-
-			dockerLayersScanned.WithLabelValues(s.name).Inc()
-		}
-
-		dockerImagesScanned.WithLabelValues(s.name).Inc()
+			return nil
+		})
+	}
+	_ = workers.Wait()
+	if scanErrs.Count() > 0 {
+		ctx.Logger().V(2).Info("scan errors", "errors", scanErrs.String())
 	}
 
 	return nil
 }
 
-func baseAndTagFromImage(image string) (base, tag string, hasDigest bool) {
-	regRepoDelimiter := "/"
-	tagDelim := ":"
-	digestDelim := "@"
+// processImage processes an individual image and prepares it for further processing.
+func (s *Source) processImage(ctx context.Context, image string) (imageInfo, error) {
+	var (
+		imgInfo   imageInfo
+		hasDigest bool
+		imageName name.Reference
+	)
 
-	// Split on digest first, if present.
-	parts := strings.SplitN(image, digestDelim, 2)
-	if len(parts) > 1 {
-		base = parts[0]
-		tag = parts[1]
-		hasDigest = true
-		return
+	remoteOpts, err := s.remoteOpts()
+	if err != nil {
+		return imgInfo, err
 	}
-	parts = strings.Split(image, tagDelim)
-	// Verify that we aren't confusing a tag for a hostname w/ port for the purposes of weak validation.
-	if len(parts) > 1 && !strings.Contains(parts[len(parts)-1], regRepoDelimiter) {
-		base = strings.Join(parts[:len(parts)-1], tagDelim)
-		tag = parts[len(parts)-1]
+
+	const filePrefix = "file://"
+	if strings.HasPrefix(image, filePrefix) {
+		image = strings.TrimPrefix(image, filePrefix)
+		imgInfo.base = image
+		imgInfo.image, err = tarball.ImageFromPath(image, nil)
+		if err != nil {
+			return imgInfo, err
+		}
 	} else {
-		base = image
-		tag = "latest"
+		imgInfo.base, imgInfo.tag, hasDigest = baseAndTagFromImage(image)
+
+		if hasDigest {
+			imageName, err = name.NewDigest(image)
+		} else {
+			imageName, err = name.NewTag(image)
+		}
+		if err != nil {
+			return imgInfo, err
+		}
+
+		imgInfo.image, err = remote.Image(imageName, remoteOpts...)
+		if err != nil {
+			return imgInfo, err
+		}
 	}
 
-	return
+	ctx.Logger().WithValues("image", imgInfo.base, "tag", imgInfo.tag).V(2).Info("scanning image")
+
+	return imgInfo, nil
+}
+
+// processLayer processes an individual layer of an image.
+func (s *Source) processLayer(ctx context.Context, layer v1.Layer, imgInfo imageInfo, chunksChan chan *sources.Chunk) error {
+	layerInfo := layerInfo{
+		base: imgInfo.base,
+		tag:  imgInfo.tag,
+	}
+
+	var err error
+	layerInfo.digest, err = layer.Digest()
+	if err != nil {
+		return err
+	}
+
+	ctx.Logger().WithValues("layer", layerInfo.digest.String()).V(2).Info("scanning layer")
+
+	rc, err := layer.Compressed()
+	if err != nil {
+		return err
+	}
+	defer rc.Close()
+
+	gzipReader, err := gzip.NewReader(rc)
+	if err != nil {
+		return err
+	}
+	defer gzipReader.Close()
+
+	tarReader := tar.NewReader(gzipReader)
+	for {
+		header, err := tarReader.Next()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return err
+		}
+
+		info := chunkProcessingInfo{size: header.Size, name: header.Name, reader: tarReader, layer: layerInfo}
+		if err := s.processChunk(ctx, info, chunksChan); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+type chunkProcessingInfo struct {
+	size   int64
+	name   string
+	reader io.Reader
+	layer  layerInfo
+}
+
+// processChunk processes an individual chunk of a layer.
+func (s *Source) processChunk(ctx context.Context, info chunkProcessingInfo, chunksChan chan *sources.Chunk) error {
+	const filesizeLimitBytes int64 = 50 * 1024 * 1024 // 50MB
+	if info.size > filesizeLimitBytes {
+		ctx.Logger().V(4).Info("skipping large file", "file", info.name, "size", info.size)
+		return nil
+	}
+
+	chunkReader := sources.NewChunkReader()
+	chunkResChan := chunkReader(ctx, info.reader)
+
+	for data := range chunkResChan {
+		if err := data.Error(); err != nil {
+			ctx.Logger().Error(err, "error reading chunk.")
+			continue
+		}
+
+		chunk := &sources.Chunk{
+			SourceType: s.Type(),
+			SourceName: s.name,
+			SourceID:   s.SourceID(),
+			SourceMetadata: &source_metadatapb.MetaData{
+				Data: &source_metadatapb.MetaData_Docker{
+					Docker: &source_metadatapb.Docker{
+						File:  "/" + info.name,
+						Image: info.layer.base,
+						Tag:   info.layer.tag,
+						Layer: info.layer.digest.String(),
+					},
+				},
+			},
+			Verify: s.verify,
+		}
+		chunk.Data = data.Bytes()
+
+		if err := common.CancellableWrite(ctx, chunksChan, chunk); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (s *Source) remoteOpts() ([]remote.Option, error) {
@@ -256,3 +300,40 @@ func (s *Source) remoteOpts() ([]remote.Option, error) {
 		return nil, fmt.Errorf("unknown credential type: %T", s.conn.Credential)
 	}
 }
+
+func baseAndTagFromImage(image string) (base, tag string, hasDigest bool) {
+	if base, tag, hasDigest = extractDigest(image); hasDigest {
+		return base, tag, true
+	}
+
+	base, tag = extractTagOrUseDefault(image)
+	return base, tag, false
+}
+
+// extractDigest tries to split the image string on the digest delimiter.
+// If successful, it means the image has a digest.
+func extractDigest(image string) (base, tag string, hasDigest bool) {
+	const digestDelim = "@"
+
+	if parts := strings.SplitN(image, digestDelim, 2); len(parts) > 1 {
+		return parts[0], parts[1], true
+	}
+	return "", "", false
+}
+
+// extractTagOrUseDefault extracts the tag from the image string.
+// If no tag is found, it defaults to "latest".
+func extractTagOrUseDefault(image string) (base, tag string) {
+	const (
+		tagDelim     = ":"
+		regRepoDelim = "/"
+	)
+
+	parts := strings.Split(image, tagDelim)
+
+	// Check if the last part is not a hostname with a port (for weak validation)
+	if len(parts) > 1 && !strings.Contains(parts[len(parts)-1], regRepoDelim) {
+		return strings.Join(parts[:len(parts)-1], tagDelim), parts[len(parts)-1]
+	}
+	return image, "latest"
+}

pkg/sources/filesystem/filesystem.go

@@ -77,7 +77,7 @@ func (s *Source) WithFilter(filter *common.Filter) {
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	for i, path := range s.paths {
 		logger := ctx.Logger().WithValues("path", path)
 		if common.IsDone(ctx) {

pkg/sources/gcs/gcs.go

@@ -248,7 +248,7 @@ func (s *Source) enumerate(ctx context.Context) error {
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	persistableCache := s.setupCache(ctx)
 
 	objectCh, err := s.gcsManager.ListObjects(ctx)

pkg/sources/git/git.go

@@ -153,7 +153,7 @@ func (s *Source) Init(aCtx context.Context, name string, jobId, sourceId int64,
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	if err := s.scanRepos(ctx, chunksChan); err != nil {
 		return err
 	}

pkg/sources/github/github.go

@@ -413,7 +413,7 @@ func (s *Source) visibilityOf(ctx context.Context, repoURL string) (visibility s
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	apiEndpoint := s.conn.Endpoint
 	if len(apiEndpoint) == 0 || endsWithGithub.MatchString(apiEndpoint) {
 		apiEndpoint = "https://api.github.com"

pkg/sources/gitlab/gitlab.go

@@ -139,7 +139,7 @@ func (s *Source) Init(_ context.Context, name string, jobId, sourceId int64, ver
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	// Start client.
 	apiClient, err := s.newClient()
 	if err != nil {

pkg/sources/s3/s3.go

@@ -31,6 +31,7 @@ import (
 )
 
 const (
+	defaultAWSRegion     = "us-east-1"
 	defaultMaxObjectSize = 250 * 1024 * 1024 // 250 MiB
 	maxObjectSizeLimit   = 250 * 1024 * 1024 // 250 MiB
 )
@@ -53,6 +54,7 @@ type Source struct {
 // Ensure the Source satisfies the interfaces at compile time
 var _ sources.Source = (*Source)(nil)
 var _ sources.SourceUnitUnmarshaller = (*Source)(nil)
+var _ sources.Validator = (*Source)(nil)
 
 // Type returns the type of source
 func (s *Source) Type() sourcespb.SourceType {
@@ -93,6 +95,23 @@ func (s *Source) Init(aCtx context.Context, name string, jobId, sourceId int64,
 	return nil
 }
 
+func (s *Source) Validate(ctx context.Context) []error {
+	var errors []error
+	visitor := func(c context.Context, defaultRegionClient *s3.S3, roleArn string, buckets []string) {
+		roleErrs := s.validateBucketAccess(c, defaultRegionClient, roleArn, buckets)
+		if len(roleErrs) > 0 {
+			errors = append(errors, roleErrs...)
+		}
+	}
+
+	err := s.visitRoles(ctx, visitor)
+	if err != nil {
+		errors = append(errors, err)
+	}
+
+	return errors
+}
+
 // setMaxObjectSize sets the maximum size of objects that will be scanned. If
 // not set, set to a negative number, or set larger than the
 // maxObjectSizeLimit, the defaultMaxObjectSizeLimit will be used.
@@ -153,7 +172,7 @@ func (s *Source) getBucketsToScan(client *s3.S3) ([]string, error) {
 
 	res, err := client.ListBuckets(&s3.ListBucketsInput{})
 	if err != nil {
-		return nil, fmt.Errorf("could not list s3 buckets: %w", err)
+		return nil, err
 	}
 
 	var bucketsToScan []string
@@ -163,32 +182,28 @@ func (s *Source) getBucketsToScan(client *s3.S3) ([]string, error) {
 	return bucketsToScan, nil
 }
 
-func (s *Source) scanBuckets(ctx context.Context, client *s3.S3, role string, bucketsToScan []string, chunksChan chan *sources.Chunk) error {
-	const defaultAWSRegion = "us-east-1"
+func (s *Source) scanBuckets(ctx context.Context, client *s3.S3, role string, bucketsToScan []string, chunksChan chan *sources.Chunk) {
 	objectCount := uint64(0)
 
+	logger := s.log
+	if role != "" {
+		logger = logger.WithValues("roleArn", role)
+	}
+
 	for i, bucket := range bucketsToScan {
+		logger := logger.WithValues("bucket", bucket)
+
 		if common.IsDone(ctx) {
-			return nil
+			return
 		}
 
 		s.SetProgressComplete(i, len(bucketsToScan), fmt.Sprintf("Bucket: %s", bucket), "")
-		s.log.Info("Scanning bucket", "bucket", bucket)
-		region, err := s3manager.GetBucketRegionWithClient(ctx, client, bucket)
-		if err != nil {
-			s.log.Error(err, "could not get s3 region for bucket", "bucket: ", bucket)
-			continue
-		}
+		logger.Info("Scanning bucket")
 
-		var regionalClient *s3.S3
-		if region != defaultAWSRegion {
-			regionalClient, err = s.newClient(region, role)
-			if err != nil {
-				s.log.Error(err, "could not make regional s3 client")
-				continue
-			}
-		} else {
-			regionalClient = client
+		regionalClient, err := s.getRegionalClientForBucket(ctx, client, role, bucket)
+		if err != nil {
+			logger.Error(err, "could not get regional client for bucket")
+			continue
 		}
 
 		errorCount := sync.Map{}
@@ -201,40 +216,46 @@ func (s *Source) scanBuckets(ctx context.Context, client *s3.S3, role string, bu
 			})
 
 		if err != nil {
-			s.log.Error(err, "could not list objects in s3 bucket", "bucket: ", bucket)
-			continue
+			if role == "" {
+				logger.Error(err, "could not list objects in bucket")
+			} else {
+				// Our documentation blesses specifying a role to assume without specifying buckets to scan, which will
+				// often cause this to happen a lot (because in that case the scanner tries to scan every bucket in the
+				// account, but the role probably doesn't have access to all of them). This makes it expected behavior
+				// and therefore not an error.
+				logger.V(3).Info("could not list objects in bucket",
+					"err", err)
+			}
 		}
 	}
 	s.SetProgressComplete(len(bucketsToScan), len(bucketsToScan), fmt.Sprintf("Completed scanning source %s. %d objects scanned.", s.name, objectCount), "")
-	return nil
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
-	const defaultAWSRegion = "us-east-1"
-
-	roles := s.conn.Roles
-	if len(roles) == 0 {
-		roles = []string{""}
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
+	visitor := func(c context.Context, defaultRegionClient *s3.S3, roleArn string, buckets []string) {
+		s.scanBuckets(c, defaultRegionClient, roleArn, buckets, chunksChan)
 	}
 
-	for _, role := range roles {
-		client, err := s.newClient(defaultAWSRegion, role)
-		if err != nil {
-			return errors.WrapPrefix(err, "could not create s3 client", 0)
-		}
+	return s.visitRoles(ctx, visitor)
+}
 
-		bucketsToScan, err := s.getBucketsToScan(client)
-		if err != nil {
-			return err
-		}
-
-		if err := s.scanBuckets(ctx, client, role, bucketsToScan, chunksChan); err != nil {
-			return err
-		}
+func (s *Source) getRegionalClientForBucket(ctx context.Context, defaultRegionClient *s3.S3, role, bucket string) (*s3.S3, error) {
+	region, err := s3manager.GetBucketRegionWithClient(ctx, defaultRegionClient, bucket)
+	if err != nil {
+		return nil, errors.WrapPrefix(err, "could not get s3 region for bucket", 0)
 	}
 
-	return nil
+	if region == defaultAWSRegion {
+		return defaultRegionClient, nil
+	}
+
+	regionalClient, err := s.newClient(region, role)
+	if err != nil {
+		return nil, errors.WrapPrefix(err, "could not create regional s3 client", 0)
+	}
+
+	return regionalClient, nil
 }
 
 // pageChunker emits chunks onto the given channel from a page
@@ -396,6 +417,65 @@ func (s *Source) pageChunker(ctx context.Context, client *s3.S3, chunksChan chan
 	_ = s.jobPool.Wait()
 }
 
+func (s *Source) validateBucketAccess(ctx context.Context, client *s3.S3, roleArn string, buckets []string) []error {
+	shouldHaveAccessToAllBuckets := roleArn == ""
+	wasAbleToListAnyBucket := false
+	var errors []error
+
+	for _, bucket := range buckets {
+		if common.IsDone(ctx) {
+			return append(errors, ctx.Err())
+		}
+
+		regionalClient, err := s.getRegionalClientForBucket(ctx, client, roleArn, bucket)
+		if err != nil {
+			errors = append(errors, fmt.Errorf("could not get regional client for bucket %q: %w", bucket, err))
+			continue
+		}
+
+		_, err = regionalClient.ListObjectsV2(&s3.ListObjectsV2Input{Bucket: &bucket})
+
+		if err == nil {
+			wasAbleToListAnyBucket = true
+		} else if shouldHaveAccessToAllBuckets {
+			errors = append(errors, fmt.Errorf("could not list objects in bucket %q: %w", bucket, err))
+		}
+	}
+
+	if !wasAbleToListAnyBucket {
+		if roleArn == "" {
+			errors = append(errors, fmt.Errorf("could not list objects in any bucket"))
+		} else {
+			errors = append(errors, fmt.Errorf("role %q could not list objects in any bucket", roleArn))
+		}
+	}
+
+	return errors
+}
+
+func (s *Source) visitRoles(ctx context.Context, f func(c context.Context, defaultRegionClient *s3.S3, roleArn string, buckets []string)) error {
+	roles := s.conn.Roles
+	if len(roles) == 0 {
+		roles = []string{""}
+	}
+
+	for _, role := range roles {
+		client, err := s.newClient(defaultAWSRegion, role)
+		if err != nil {
+			return errors.WrapPrefix(err, "could not create s3 client", 0)
+		}
+
+		bucketsToScan, err := s.getBucketsToScan(client)
+		if err != nil {
+			return fmt.Errorf("role %q could not list any s3 buckets for scanning: %w", role, err)
+		}
+
+		f(ctx, client, role, bucketsToScan)
+	}
+
+	return nil
+}
+
 // S3 links currently have the general format of:
 // https://[bucket].s3[.region unless us-east-1].amazonaws.com/[key]
 func makeS3Link(bucket, region, key string) string {

pkg/sources/s3/s3_integration_test.go

@@ -4,10 +4,14 @@
 package s3
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/credentialspb"
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
@@ -45,3 +49,127 @@ func TestSource_ChunksCount(t *testing.T) {
 	}
 	assert.Greater(t, got, wantChunkCount)
 }
+
+func TestSource_Validate(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+	defer cancel()
+
+	secret, err := common.GetTestSecret(ctx)
+	if err != nil {
+		t.Fatal(fmt.Errorf("failed to access secret: %v", err))
+	}
+
+	s3key := secret.MustGetField("AWS_S3_KEY")
+	s3secret := secret.MustGetField("AWS_S3_SECRET")
+
+	tests := []struct {
+		name         string
+		roles        []string
+		buckets      []string
+		wantErrCount int
+	}{
+		{
+			name: "buckets without roles, can access all buckets",
+			buckets: []string{
+				"truffletestbucket-s3-tests",
+			},
+			wantErrCount: 0,
+		},
+		{
+			name: "buckets without roles, one error per inaccessible bucket",
+			buckets: []string{
+				"truffletestbucket-s3-tests",
+				"truffletestbucket-s3-role-assumption",
+				"truffletestbucket-no-access",
+			},
+			wantErrCount: 2,
+		},
+		{
+			name: "roles without buckets, all can access at least one account bucket",
+			roles: []string{
+				"arn:aws:iam::619888638459:role/s3-test-assume-role",
+			},
+			wantErrCount: 0,
+		},
+		{
+			name: "roles without buckets, one error per role that cannot access any account buckets",
+			roles: []string{
+				"arn:aws:iam::619888638459:role/s3-test-assume-role",
+				"arn:aws:iam::619888638459:role/test-no-access",
+			},
+			wantErrCount: 1,
+		},
+		{
+			name: "role and buckets, can access at least one bucket",
+			roles: []string{
+				"arn:aws:iam::619888638459:role/s3-test-assume-role",
+			},
+			buckets: []string{
+				"truffletestbucket-s3-role-assumption",
+				"truffletestbucket-no-access",
+			},
+			wantErrCount: 0,
+		},
+		{
+			name: "roles and buckets, one error per role that cannot access at least one bucket",
+			roles: []string{
+				"arn:aws:iam::619888638459:role/s3-test-assume-role",
+				"arn:aws:iam::619888638459:role/test-no-access",
+			},
+			buckets: []string{
+				"truffletestbucket-s3-role-assumption",
+				"truffletestbucket-no-access",
+			},
+			wantErrCount: 1,
+		},
+		{
+			name: "role and buckets, a bucket doesn't even exist",
+			roles: []string{
+				"arn:aws:iam::619888638459:role/s3-test-assume-role",
+			},
+			buckets: []string{
+				"truffletestbucket-s3-role-assumption",
+				"not-a-real-bucket-asljdhmglasjgvklhsdaljfh", // need a bucket name that nobody is likely to ever create
+			},
+			wantErrCount: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+			var cancelOnce sync.Once
+			defer cancelOnce.Do(cancel)
+
+			// These are used by the tests that assume roles
+			t.Setenv("AWS_ACCESS_KEY_ID", s3key)
+			t.Setenv("AWS_SECRET_ACCESS_KEY", s3secret)
+
+			s := &Source{}
+
+			conn, err := anypb.New(&sourcespb.S3{
+				// These are used by the tests that don't assume roles
+				Credential: &sourcespb.S3_AccessKey{
+					AccessKey: &credentialspb.KeySecret{
+						Key:    s3key,
+						Secret: s3secret,
+					},
+				},
+				Buckets: tt.buckets,
+				Roles:   tt.roles,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			err = s.Init(ctx, tt.name, 0, 0, false, conn, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			errs := s.Validate(ctx)
+
+			assert.Equal(t, tt.wantErrCount, len(errs))
+		})
+	}
+}

pkg/sources/source_manager_test.go

@@ -6,10 +6,11 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/types/known/anypb"
+
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/sourcespb"
-	"google.golang.org/protobuf/types/known/anypb"
 )
 
 // DummySource implements Source and is used for testing a SourceManager.
@@ -31,7 +32,7 @@ func (d *DummySource) GetProgress() *Progress { return nil }
 
 // Interface to easily test different chunking methods.
 type chunker interface {
-	Chunks(context.Context, chan *Chunk) error
+	Chunks(context.Context, chan *Chunk, ...ChunkingTarget) error
 	ChunkUnit(ctx context.Context, unit SourceUnit, reporter ChunkReporter) error
 	Enumerate(ctx context.Context, reporter UnitReporter) error
 }
@@ -42,7 +43,7 @@ type counterChunker struct {
 	count        int
 }
 
-func (c *counterChunker) Chunks(ctx context.Context, ch chan *Chunk) error {
+func (c *counterChunker) Chunks(ctx context.Context, ch chan *Chunk, _ ...ChunkingTarget) error {
 	for i := 0; i < c.count; i++ {
 		select {
 		case ch <- &Chunk{Data: []byte{c.chunkCounter}}:
@@ -75,9 +76,9 @@ func (c *counterChunker) ChunkUnit(ctx context.Context, unit SourceUnit, reporte
 // Chunk method that always returns an error.
 type errorChunker struct{ error }
 
-func (c errorChunker) Chunks(context.Context, chan *Chunk) error                  { return c }
-func (c errorChunker) Enumerate(context.Context, UnitReporter) error              { return c }
-func (c errorChunker) ChunkUnit(context.Context, SourceUnit, ChunkReporter) error { return c }
+func (c errorChunker) Chunks(context.Context, chan *Chunk, ...ChunkingTarget) error { return c }
+func (c errorChunker) Enumerate(context.Context, UnitReporter) error                { return c }
+func (c errorChunker) ChunkUnit(context.Context, SourceUnit, ChunkReporter) error   { return c }
 
 // enrollDummy is a helper function to enroll a DummySource with a SourceManager.
 func enrollDummy(mgr *SourceManager, chunkMethod chunker) (handle, error) {
@@ -176,7 +177,7 @@ type unitChunk struct {
 
 type unitChunker struct{ steps []unitChunk }
 
-func (c *unitChunker) Chunks(ctx context.Context, ch chan *Chunk) error {
+func (c *unitChunker) Chunks(ctx context.Context, ch chan *Chunk, _ ...ChunkingTarget) error {
 	for _, step := range c.steps {
 		if step.err != "" {
 			continue
@@ -294,7 +295,9 @@ type callbackChunker struct {
 	cb func(context.Context, chan *Chunk) error
 }
 
-func (c callbackChunker) Chunks(ctx context.Context, ch chan *Chunk) error           { return c.cb(ctx, ch) }
+func (c callbackChunker) Chunks(ctx context.Context, ch chan *Chunk, _ ...ChunkingTarget) error {
+	return c.cb(ctx, ch)
+}
 func (c callbackChunker) Enumerate(context.Context, UnitReporter) error              { return nil }
 func (c callbackChunker) ChunkUnit(context.Context, SourceUnit, ChunkReporter) error { return nil }
 

pkg/sources/sources.go

@@ -30,6 +30,16 @@ type Chunk struct {
 	Verify bool
 }
 
+// ChunkingTarget specifies criteria for a targeted chunking process.
+// Instead of collecting data indiscriminately, this struct allows the caller
+// to specify particular subsets of data they're interested in. This becomes
+// especially useful when one needs to verify or recheck specific data points
+// without processing the entire dataset.
+type ChunkingTarget struct {
+	// QueryCriteria represents specific parameters or conditions to target the chunking process.
+	QueryCriteria source_metadatapb.MetaData
+}
+
 // Source defines the interface required to implement a source chunker.
 type Source interface {
 	// Type returns the source type, used for matching against configuration and jobs.
@@ -40,8 +50,12 @@ type Source interface {
 	JobID() int64
 	// Init initializes the source.
 	Init(aCtx context.Context, name string, jobId, sourceId int64, verify bool, connection *anypb.Any, concurrency int) error
-	// Chunks emits data over a channel that is decoded and scanned for secrets.
-	Chunks(ctx context.Context, chunksChan chan *Chunk) error
+	// Chunks emits data over a channel which is then decoded and scanned for secrets.
+	// By default, data is obtained indiscriminately. However, by providing one or more
+	// ChunkingTarget parameters, the caller can direct the function to retrieve
+	// specific chunks of data. This targeted approach allows for efficient and
+	// intentional data processing, beneficial when verifying or rechecking specific data points.
+	Chunks(ctx context.Context, chunksChan chan *Chunk, targets ...ChunkingTarget) error
 	// GetProgress is the completion progress (percentage) for Scanned Source.
 	GetProgress() *Progress
 }

pkg/sources/syslog/syslog.go

@@ -182,7 +182,7 @@ func (s *Source) verifyConnectionConfig() error {
 }
 
 // Chunks emits chunks of bytes over a channel.
-func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
+func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ ...sources.ChunkingTarget) error {
 	switch {
 	case s.conn.TlsCert != nilString || s.conn.TlsKey != nilString:
 		cert, err := tls.X509KeyPair([]byte(s.conn.TlsCert), []byte(s.conn.TlsKey))