From dfd604b10642dcdb3438571f26e55603469488c7 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Wed, 2 Sep 2026 21:12:13 -0500 Subject: [PATCH] hold/scanner: stop one undispatchable job freezing all scanning Vulnerability scanning produced nothing across the whole deployment for nine days, from 2026-08-25 01:20:48 until a scanner restart on 2026-09-03. The scanner was connected and idle, the hold's discovery pass kept reporting unscannedFound=15 every four hours, and no scan_jobs row was created in that entire window. hasActiveJobs counted pending, assigned and processing rows globally with no age bound, and waitForCapacity spins while it is true. dispatchLoop calls it before popping any candidate, so a single pending row that never reached a terminal state reported "busy" forever: discovery kept pushing candidates into unscannedQueue and nothing ever popped them. That is why the symptom was an empty queue rather than a growing one. Nothing papered over it because push-triggered enqueue only fires for owner or a tier with scan_on_push, which in production means pro alone. All 210 manifests pushed to this hold in that window came from free, supporter, or accounts with no crew row, so the frozen proactive loop was the only source of jobs. Nor could it recover on its own. Only Enqueue and drainPendingJobs dispatch a pending row, and drainPendingJobs runs only when a scanner newly connects; reDispatchTimedOut considered assigned rows only. The hold had been up since Aug 14 and the scanner since Aug 21 on the same websocket, so the drain path had not run since the row appeared. So bound the capacity gate to pending rows younger than pendingStaleAfter, give reDispatchTimedOut a pending reclaim, and check RowsAffected on the assign UPDATE now that two dispatchers can race for a row. waitForCapacity warns and names the blocking jobs after ten minutes without capacity, because the failure mode above was completely silent. Two adjacent fixes for the same outage. The scanner never called InitLogger, so log_level and log_shipper were dead config and an idle scanner was mute, which is what made nine days invisible. And skipReason now also skips a job whose layers contain nothing tar-shaped: the job that wedged this queue was an in-toto attestation whose config mediaType is an ordinary image config, so the existing config-type check missed it and buildOCILayout would have handed Syft an empty image. The regression tests were verified against the old logic first: three of them fail on it and pass on the fix. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01DPWkeCKcbtGoyXyyeMhSps --- pkg/hold/pds/scan_broadcaster.go | 139 +++++++++++++--- pkg/hold/pds/scan_broadcaster_stall_test.go | 167 ++++++++++++++++++++ scanner/cmd/scanner/main.go | 15 ++ scanner/internal/scan/grype.go | 4 +- scanner/internal/scan/worker.go | 36 ++++- scanner/internal/scan/worker_skip_test.go | 83 ++++++++++ scanner/types.go | 20 +-- 7 files changed, 430 insertions(+), 34 deletions(-) create mode 100644 pkg/hold/pds/scan_broadcaster_stall_test.go create mode 100644 scanner/internal/scan/worker_skip_test.go diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index 52e94bb..25d7012 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -21,6 +21,25 @@ import ( "github.com/gorilla/websocket" ) +const ( + // pendingReclaimAfter is how long a job may sit in 'pending' before the + // re-dispatch loop offers it to a scanner again. Enqueue and + // drainPendingJobs are both one-shot, so without a periodic re-offer a row + // whose only dispatch attempt failed waits for the next scanner connect. + pendingReclaimAfter = 1 * time.Minute + + // pendingStaleAfter is how long a pending job counts as active for dispatch + // throttling. Assigned and processing jobs need no such bound — the ack and + // processing timeouts always resolve them — but 'pending' is otherwise + // unbounded, and one row stuck there froze proactive scanning for nine days. + pendingStaleAfter = 15 * time.Minute + + // capacityStallWarnAfter is how long the dispatch loop waits with no + // capacity before it says so. A silent stall is what made this class of + // failure invisible until users noticed missing scans. + capacityStallWarnAfter = 10 * time.Minute +) + // ScanBroadcaster manages scanner WebSocket connections and dispatches scan jobs // using a competing-consumer pattern. Jobs are persisted in SQLite and dispatched // round-robin to connected scanners. @@ -396,7 +415,7 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) { sb.nextIdx++ // Mark as assigned in database - _, err := sb.db.Exec(` + res, err := sb.db.Exec(` UPDATE scan_jobs SET status = 'assigned', assigned_to = ?, assigned_at = ? WHERE seq = ? AND status = 'pending' `, sub.id, time.Now(), job.Seq) @@ -404,6 +423,13 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) { slog.Error("Failed to assign scan job", "seq", job.Seq, "error", err) return } + // Two paths hand out pending rows now (drainPendingJobs on connect, and the + // re-dispatch loop), so a row that is no longer pending was claimed by the + // other one. Sending it anyway would scan it twice. + if n, err := res.RowsAffected(); err == nil && n == 0 { + slog.Debug("Scan job no longer pending, skipping dispatch", "seq", job.Seq) + return + } // Send to subscriber select { @@ -790,9 +816,16 @@ func (sb *ScanBroadcaster) reDispatchLoop() { } // reDispatchTimedOut finds jobs that were assigned but not acked/completed within timeout, +// re-offers jobs that have been sitting in 'pending' with nobody to hand them to, // and also marks stuck processing jobs as failed. // Collects timed-out rows first, closes cursor, then resets and re-dispatches // to avoid holding a SELECT cursor open during UPDATEs (prevents SQLite BUSY). +// +// Pending rows matter as much as assigned ones: Enqueue and drainPendingJobs are +// both one-shot dispatchers, so before this loop covered 'pending' a row whose +// single dispatch attempt failed (or that a disconnect unassigned while no +// scanner reconnected afterwards) stayed pending forever — and hasActiveJobs() +// counted it, which stopped proactive scanning deployment-wide. func (sb *ScanBroadcaster) reDispatchTimedOut() { timeout := time.Now().Add(-sb.ackTimeout) @@ -809,25 +842,31 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { } rows, err := sb.db.Query(` - SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json + SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json, status FROM scan_jobs - WHERE status = 'assigned' AND assigned_at < ? + WHERE (status = 'assigned' AND assigned_at < ?) + OR (status = 'pending' AND datetime(created_at) < datetime('now', ?)) ORDER BY seq ASC - `, timeout) + `, timeout, sqliteAgoModifier(pendingReclaimAfter)) if err != nil { slog.Error("Failed to query timed-out scan jobs", "error", err) return } - var jobs []*ScanJobEvent + type reclaimable struct { + job *ScanJobEvent + status string + } + + var jobs []reclaimable for rows.Next() { job := &ScanJobEvent{Type: "job"} - var configJSON, layersJSON string + var configJSON, layersJSON, status string err := rows.Scan( &job.Seq, &job.ManifestDigest, &job.Repository, &job.Tag, &job.UserDID, &job.UserHandle, &job.HoldDID, &job.HoldEndpoint, - &job.Tier, &configJSON, &layersJSON, + &job.Tier, &configJSON, &layersJSON, &status, ) if err != nil { continue @@ -835,27 +874,41 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { job.Config = json.RawMessage(configJSON) job.Layers = json.RawMessage(layersJSON) - jobs = append(jobs, job) + jobs = append(jobs, reclaimable{job: job, status: status}) } rows.Close() - for _, job := range jobs { - _, err = sb.db.Exec(` - UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL - WHERE seq = ? - `, job.Seq) - if err != nil { - continue + for _, r := range jobs { + job := r.job + + // Already-pending rows need no reset — and skipping the UPDATE keeps a + // row another dispatcher claimed between the SELECT and here assigned, + // so dispatchJob's status guard can drop it instead of double-sending. + if r.status != "pending" { + _, err = sb.db.Exec(` + UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL + WHERE seq = ? + `, job.Seq) + if err != nil { + continue + } } - slog.Info("Re-dispatching timed-out scan job", + slog.Info("Re-dispatching scan job", "seq", job.Seq, - "repository", job.Repository) + "repository", job.Repository, + "previousStatus", r.status) sb.dispatchJob(job) } } +// sqliteAgoModifier renders a duration as a SQLite datetime() modifier that +// walks backwards from 'now', e.g. 15m becomes "-900 seconds". +func sqliteAgoModifier(d time.Duration) string { + return fmt.Sprintf("-%d seconds", int64(d.Seconds())) +} + // Close stops background goroutines and closes the scan broadcaster's database connection func (sb *ScanBroadcaster) Close() error { if sb.stopCh != nil { @@ -1273,10 +1326,20 @@ func (sb *ScanBroadcaster) dispatchLoop() { // waitForCapacity blocks until there are no active proactive scan jobs. // Returns false if stopCh is closed. func (sb *ScanBroadcaster) waitForCapacity() bool { + blockedSince := time.Now() + var lastWarn time.Time + for { if !sb.hasActiveJobs() { return true } + + if blocked := time.Since(blockedSince); blocked >= capacityStallWarnAfter && + time.Since(lastWarn) >= capacityStallWarnAfter { + sb.logStalledCapacity(blocked) + lastWarn = time.Now() + } + select { case <-sb.stopCh: return false @@ -1488,13 +1551,23 @@ func (sb *ScanBroadcaster) hasConnectedScanners() bool { return len(sb.subscribers) > 0 } -// hasActiveJobs returns true if there are any pending, assigned, or processing scan jobs. +// hasActiveJobs returns true if there are any assigned or processing scan jobs, +// or any recently-created pending ones. +// +// Pending rows older than pendingStaleAfter are deliberately not counted. A job +// that has been pending that long is one no scanner can be given (dispatch +// failed, or no scanner is connected), and counting it blocks the proactive +// dispatch loop for as long as the row exists — which is how a single +// undispatchable job stopped scanning deployment-wide. The re-dispatch loop +// keeps re-offering such rows, so ignoring them here costs nothing when a +// scanner is available. func (sb *ScanBroadcaster) hasActiveJobs() bool { var count int err := sb.db.QueryRow(` SELECT COUNT(*) FROM scan_jobs - WHERE status IN ('pending', 'assigned', 'processing') - `).Scan(&count) + WHERE status IN ('assigned', 'processing') + OR (status = 'pending' AND datetime(created_at) > datetime('now', ?)) + `, sqliteAgoModifier(pendingStaleAfter)).Scan(&count) if err != nil { slog.Error("Failed to check active scan jobs", "error", err) return true // Assume busy on error @@ -1502,6 +1575,32 @@ func (sb *ScanBroadcaster) hasActiveJobs() bool { return count > 0 } +// logStalledCapacity reports what is holding the dispatch loop back, so a stall +// shows up in the hold's logs instead of only as an absence of scan results. +func (sb *ScanBroadcaster) logStalledCapacity(blockedFor time.Duration) { + var ( + count int + oldestSeq sql.NullInt64 + statuses sql.NullString + ) + err := sb.db.QueryRow(` + SELECT COUNT(*), MIN(seq), GROUP_CONCAT(DISTINCT status) + FROM scan_jobs + WHERE status IN ('pending', 'assigned', 'processing') + `).Scan(&count, &oldestSeq, &statuses) + if err != nil { + slog.Warn("Proactive scan dispatch stalled; could not inspect active jobs", + "blockedFor", blockedFor.Truncate(time.Minute), "error", err) + return + } + + slog.Warn("Proactive scan dispatch stalled waiting on active jobs", + "blockedFor", blockedFor.Truncate(time.Minute), + "activeJobs", count, + "oldestSeq", oldestSeq.Int64, + "statuses", statuses.String) +} + func generateSubscriberID() string { b := make([]byte, 8) _, _ = rand.Read(b) diff --git a/pkg/hold/pds/scan_broadcaster_stall_test.go b/pkg/hold/pds/scan_broadcaster_stall_test.go new file mode 100644 index 0000000..d590b41 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_stall_test.go @@ -0,0 +1,167 @@ +package pds + +import ( + "testing" + "time" +) + +// backdateJob ages a job's created_at so the staleness windows in +// hasActiveJobs and reDispatchTimedOut can be exercised without sleeping. +func backdateJob(t *testing.T, sb *ScanBroadcaster, seq int64, minutes int) { + t.Helper() + + _, err := sb.db.Exec( + `UPDATE scan_jobs SET created_at = datetime('now', ?) WHERE seq = ?`, + sqliteAgoModifier(time.Duration(minutes)*time.Minute), seq) + if err != nil { + t.Fatalf("backdate job %d: %v", seq, err) + } +} + +func jobStatus(t *testing.T, sb *ScanBroadcaster, seq int64) string { + t.Helper() + + var status string + if err := sb.db.QueryRow(`SELECT status FROM scan_jobs WHERE seq = ?`, seq).Scan(&status); err != nil { + t.Fatalf("query status for %d: %v", seq, err) + } + return status +} + +// TestScanHasActiveJobs_IgnoresLongPendingJob is the regression test for the +// nine-day deployment-wide scanning outage. A single job sat in 'pending' with +// nothing left to dispatch it, hasActiveJobs() counted it forever, and +// waitForCapacity() therefore never let the proactive dispatch loop enqueue +// another job — so discovery kept finding unscanned images and creating none. +func TestScanHasActiveJobs_IgnoresLongPendingJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + + if !sb.hasActiveJobs() { + t.Fatal("a freshly enqueued pending job must count as active") + } + + backdateJob(t, sb, 1, 60) + + if sb.hasActiveJobs() { + t.Error("a job pending for an hour must not block dispatch capacity") + } +} + +// TestScanHasActiveJobs_CountsAssignedAndProcessing guards the other half: +// assigned and processing jobs have their own reclaim timeouts, so they must +// still hold capacity no matter how old the row is. +func TestScanHasActiveJobs_CountsAssignedAndProcessing(t *testing.T) { + for _, status := range []string{"assigned", "processing"} { + t.Run(status, func(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 60) + + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = ? WHERE seq = 1`, status); err != nil { + t.Fatalf("set status: %v", err) + } + + if !sb.hasActiveJobs() { + t.Errorf("%s job must hold dispatch capacity", status) + } + }) + } +} + +// TestScanHasActiveJobs_IgnoresTerminalJobs keeps completed and failed rows out +// of the capacity check — the table holds tens of thousands of them. +func TestScanHasActiveJobs_IgnoresTerminalJobs(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 2) + + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'completed' WHERE seq = 1`); err != nil { + t.Fatalf("complete: %v", err) + } + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'failed' WHERE seq = 2`); err != nil { + t.Fatalf("fail: %v", err) + } + + if sb.hasActiveJobs() { + t.Error("terminal jobs must not hold dispatch capacity") + } +} + +// TestScanReDispatch_ReoffersStalePendingJob covers the missing lease reclaim. +// Enqueue and drainPendingJobs are both one-shot, so a pending row used to wait +// for the next scanner connect — nine days, in production. The re-dispatch loop +// now hands it to a connected scanner on its own. +func TestScanReDispatch_ReoffersStalePendingJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 30) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "assigned" { + t.Errorf("stale pending job not re-dispatched, status=%q", got) + } + if len(sub.send) != 1 { + t.Errorf("scanner received %d jobs, want 1", len(sub.send)) + } +} + +// TestScanReDispatch_LeavesFreshPendingJobAlone stops the loop from racing the +// dispatch Enqueue just did, which would scan the same manifest twice. +func TestScanReDispatch_LeavesFreshPendingJobAlone(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "pending" { + t.Errorf("just-enqueued job was re-dispatched, status=%q", got) + } + if len(sub.send) != 0 { + t.Errorf("scanner received %d jobs, want 0", len(sub.send)) + } +} + +// TestScanReDispatch_PendingSurvivesWithNoScanner verifies the reclaim is a +// no-op rather than a corruption when no scanner is connected: the row stays +// pending and gets offered again on the next tick. +func TestScanReDispatch_PendingSurvivesWithNoScanner(t *testing.T) { + sb := newTestScanBroadcaster(t) + seedPendingJobs(t, sb, 1) + backdateJob(t, sb, 1, 30) + + sb.reDispatchTimedOut() + + if got := jobStatus(t, sb, 1); got != "pending" { + t.Errorf("job status changed with no scanner connected, status=%q", got) + } +} + +// TestScanDispatchJob_SkipsClaimedJob covers the double-send window opened by +// having two dispatchers for pending rows: whichever loses the UPDATE race must +// not also push the job onto a scanner's queue. +func TestScanDispatchJob_SkipsClaimedJob(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + // Another dispatcher got there first. + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status='assigned', assigned_to='other' WHERE seq = 1`); err != nil { + t.Fatalf("claim: %v", err) + } + + sb.dispatchJob(&ScanJobEvent{Type: "job", Seq: 1, Repository: "repo"}) + + if len(sub.send) != 0 { + t.Errorf("already-claimed job was dispatched again (%d sends)", len(sub.send)) + } + var assignedTo string + if err := sb.db.QueryRow(`SELECT assigned_to FROM scan_jobs WHERE seq = 1`).Scan(&assignedTo); err != nil { + t.Fatalf("query assigned_to: %v", err) + } + if assignedTo != "other" { + t.Errorf("assignment stolen from the first dispatcher, assigned_to=%q", assignedTo) + } +} diff --git a/scanner/cmd/scanner/main.go b/scanner/cmd/scanner/main.go index 0491392..b2d8c01 100644 --- a/scanner/cmd/scanner/main.go +++ b/scanner/cmd/scanner/main.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" + "atcr.io/pkg/logging" "atcr.io/scanner/internal/client" "atcr.io/scanner/internal/config" "atcr.io/scanner/internal/queue" @@ -50,6 +51,20 @@ Environment variables always override file values (SCANNER_ prefix).`, return fmt.Errorf("failed to load config: %w", err) } + // Without this the scanner runs on slog's default handler: log_level and + // log_shipper are silently ignored, nothing is shipped anywhere, and an + // idle scanner is indistinguishable from a wedged one. + logging.InitLoggerWithShipper(cfg.LogLevel, logging.ShipperConfig{ + Backend: cfg.LogShipper.Backend, + URL: cfg.LogShipper.URL, + BatchSize: cfg.LogShipper.BatchSize, + FlushInterval: cfg.LogShipper.FlushInterval, + Service: "scanner", + Username: cfg.LogShipper.Username, + Password: cfg.LogShipper.Password, + }) + defer logging.Shutdown() + slog.Info("Starting ATCR scanner", "hold_url", cfg.Hold.URL, "workers", cfg.Scanner.Workers, diff --git a/scanner/internal/scan/grype.go b/scanner/internal/scan/grype.go index 8df8d14..956b60f 100644 --- a/scanner/internal/scan/grype.go +++ b/scanner/internal/scan/grype.go @@ -125,11 +125,11 @@ func scanVulnerabilities(ctx context.Context, s *sbom.SBOM, vulnDBPath string) ( "low", summary.Low, "total", summary.Total) - report := map[string]interface{}{ + report := map[string]any{ "matches": allMatches.Sorted(), "source": s.Source, "distro": s.Artifacts.LinuxDistribution, - "descriptor": map[string]interface{}{ + "descriptor": map[string]any{ "name": "grype", "version": "v0.107.1", }, diff --git a/scanner/internal/scan/worker.go b/scanner/internal/scan/worker.go index b5f35d1..4b140e4 100644 --- a/scanner/internal/scan/worker.go +++ b/scanner/internal/scan/worker.go @@ -9,6 +9,7 @@ import ( "log/slog" "os" "runtime" + "strings" "sync" "time" @@ -151,6 +152,37 @@ var unscannableConfigTypes = map[string]bool{ "application/vnd.dsse.envelope.v1+json": true, // DSSE envelopes (SLSA) } +// skipReason reports why a job cannot be scanned, or "" when it can be. +func skipReason(job *scanner.ScanJob) string { + if unscannableConfigTypes[job.Config.MediaType] { + return fmt.Sprintf("unscannable artifact type %s", job.Config.MediaType) + } + + // A buildx attestation manifest carries an ordinary image config with a + // single in-toto or DSSE payload as its layer, so the config media type + // alone does not identify it. buildOCILayout drops every non-tar layer, + // which would hand Syft an image with nothing in it. + if len(job.Layers) > 0 && !hasScannableLayer(job.Layers) { + return fmt.Sprintf("no scannable layers (%s)", job.Layers[0].MediaType) + } + + return "" +} + +// hasScannableLayer mirrors the layer filter in buildOCILayout: anything that +// is not a tar of some flavour is not something Syft can read. +func hasScannableLayer(layers []scanner.BlobDescriptor) bool { + for _, layer := range layers { + if layer.Digest == "" { + continue + } + if layer.MediaType == "" || strings.Contains(layer.MediaType, "tar") { + return true + } + } + return false +} + func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*scanner.ScanResult, error) { startTime := time.Now() @@ -158,8 +190,8 @@ func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*sc // Returning *SkipError tells the worker dispatch loop to send a "skipped" // message rather than an "error" — the hold marks these records as // permanently skipped and won't retry them on the rescan interval. - if unscannableConfigTypes[job.Config.MediaType] { - return nil, &SkipError{Reason: fmt.Sprintf("unscannable artifact type %s", job.Config.MediaType)} + if reason := skipReason(job); reason != "" { + return nil, &SkipError{Reason: reason} } // Ensure tmp dir exists diff --git a/scanner/internal/scan/worker_skip_test.go b/scanner/internal/scan/worker_skip_test.go new file mode 100644 index 0000000..b6e3e14 --- /dev/null +++ b/scanner/internal/scan/worker_skip_test.go @@ -0,0 +1,83 @@ +package scan + +import ( + "testing" + + scanner "atcr.io/scanner" +) + +// TestSkipReason covers the artifact shapes the scanner must refuse before it +// spends a download on them. The attestation case is the one that reached +// production: an in-toto SLSA provenance manifest carries an ordinary image +// config, so the config media type check alone waved it through and Syft was +// handed an OCI layout with no layers in it. +func TestSkipReason(t *testing.T) { + tests := []struct { + name string + configType string + layerTypes []string + wantSkip bool + }{ + { + name: "container image", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.oci.image.layer.v1.tar+gzip"}, + }, + { + name: "docker image", + configType: "application/vnd.docker.container.image.v1+json", + layerTypes: []string{"application/vnd.docker.image.rootfs.diff.tar.gzip"}, + }, + { + name: "layer media type absent", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{""}, + }, + { + name: "helm chart", + configType: "application/vnd.cncf.helm.config.v1+json", + layerTypes: []string{"application/vnd.cncf.helm.chart.content.v1.tar+gzip"}, + wantSkip: true, + }, + { + name: "in-toto attestation with an image config", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.in-toto+json"}, + wantSkip: true, + }, + { + name: "dsse envelope layer", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.dsse.envelope.v1+json"}, + wantSkip: true, + }, + { + name: "mixed layers keep the scannable one", + configType: "application/vnd.oci.image.config.v1+json", + layerTypes: []string{"application/vnd.in-toto+json", "application/vnd.oci.image.layer.v1.tar"}, + }, + { + name: "no layers at all is left to the pipeline", + configType: "application/vnd.oci.image.config.v1+json", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + job := &scanner.ScanJob{ + Config: scanner.BlobDescriptor{MediaType: tt.configType, Digest: "sha256:config"}, + } + for i, mt := range tt.layerTypes { + job.Layers = append(job.Layers, scanner.BlobDescriptor{ + MediaType: mt, + Digest: "sha256:layer" + string(rune('a'+i)), + }) + } + + reason := skipReason(job) + if got := reason != ""; got != tt.wantSkip { + t.Errorf("skipReason = %q, wantSkip=%v", reason, tt.wantSkip) + } + }) + } +} diff --git a/scanner/types.go b/scanner/types.go index 03e4fd2..4bf9000 100644 --- a/scanner/types.go +++ b/scanner/types.go @@ -6,16 +6,16 @@ import "encoding/json" // ScanJob represents a vulnerability scanning job received from the hold service type ScanJob struct { - Seq int64 `json:"seq"` - ManifestDigest string `json:"manifestDigest"` - Repository string `json:"repository"` - Tag string `json:"tag"` - UserDID string `json:"userDid"` - UserHandle string `json:"userHandle"` - HoldDID string `json:"holdDid"` - HoldEndpoint string `json:"holdEndpoint"` - Tier string `json:"tier"` - Config BlobDescriptor `json:"config"` + Seq int64 `json:"seq"` + ManifestDigest string `json:"manifestDigest"` + Repository string `json:"repository"` + Tag string `json:"tag"` + UserDID string `json:"userDid"` + UserHandle string `json:"userHandle"` + HoldDID string `json:"holdDid"` + HoldEndpoint string `json:"holdEndpoint"` + Tier string `json:"tier"` + Config BlobDescriptor `json:"config"` Layers []BlobDescriptor `json:"layers"` }