From 3b5b89b378a275aa6fafc3c0450410b5a8763aed Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Thu, 30 Oct 2025 21:47:50 -0500 Subject: [PATCH] actually add scanner implementation --- pkg/hold/scanner/extractor.go | 270 ++++++++++++++++++++++++++ pkg/hold/scanner/grype.go | 351 ++++++++++++++++++++++++++++++++++ pkg/hold/scanner/job.go | 67 +++++++ pkg/hold/scanner/queue.go | 226 ++++++++++++++++++++++ pkg/hold/scanner/storage.go | 123 ++++++++++++ pkg/hold/scanner/syft.go | 128 +++++++++++++ pkg/hold/scanner/worker.go | 116 +++++++++++ 7 files changed, 1281 insertions(+) create mode 100644 pkg/hold/scanner/extractor.go create mode 100644 pkg/hold/scanner/grype.go create mode 100644 pkg/hold/scanner/job.go create mode 100644 pkg/hold/scanner/queue.go create mode 100644 pkg/hold/scanner/storage.go create mode 100644 pkg/hold/scanner/syft.go create mode 100644 pkg/hold/scanner/worker.go diff --git a/pkg/hold/scanner/extractor.go b/pkg/hold/scanner/extractor.go new file mode 100644 index 0000000..bad3365 --- /dev/null +++ b/pkg/hold/scanner/extractor.go @@ -0,0 +1,270 @@ +package scanner + +import ( + "archive/tar" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" +) + +// extractLayers extracts all image layers from storage to a temporary directory +// Returns the directory path and a cleanup function +func (w *Worker) extractLayers(ctx context.Context, job *ScanJob) (string, func(), error) { + // Create temp directory for extraction + // Use the database directory as the base (since we're in a scratch container with no /tmp) + scanTmpBase := filepath.Join(w.config.Database.Path, "scanner-tmp") + if err := os.MkdirAll(scanTmpBase, 0755); err != nil { + return "", nil, fmt.Errorf("failed to create scanner temp base: %w", err) + } + + tmpDir, err := os.MkdirTemp(scanTmpBase, "scan-*") + if err != nil { + return "", nil, fmt.Errorf("failed to create temp directory: %w", err) + } + + cleanup := func() { + if err := os.RemoveAll(tmpDir); err != nil { + slog.Warn("Failed to clean up temp directory", "dir", tmpDir, "error", err) + } + } + + // Create image directory structure + imageDir := filepath.Join(tmpDir, "image") + if err := os.MkdirAll(imageDir, 0755); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to create image directory: %w", err) + } + + // Download and extract config blob + slog.Info("Downloading config blob", "digest", job.Config.Digest) + configPath := filepath.Join(imageDir, "config.json") + if err := w.downloadBlob(ctx, job.Config.Digest, configPath); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to download config blob: %w", err) + } + + // Validate config is valid JSON + configData, err := os.ReadFile(configPath) + if err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to read config: %w", err) + } + var configObj map[string]interface{} + if err := json.Unmarshal(configData, &configObj); err != nil { + cleanup() + return "", nil, fmt.Errorf("invalid config JSON: %w", err) + } + + // Create layers directory for extracted content + layersDir := filepath.Join(imageDir, "layers") + if err := os.MkdirAll(layersDir, 0755); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to create layers directory: %w", err) + } + + // Download and extract each layer in order (creating overlayfs-style filesystem) + rootfsDir := filepath.Join(imageDir, "rootfs") + if err := os.MkdirAll(rootfsDir, 0755); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to create rootfs directory: %w", err) + } + + for i, layer := range job.Layers { + slog.Info("Extracting layer", "index", i, "digest", layer.Digest, "size", layer.Size) + + // Download layer blob to temp file + layerPath := filepath.Join(layersDir, fmt.Sprintf("layer-%d.tar.gz", i)) + if err := w.downloadBlob(ctx, layer.Digest, layerPath); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to download layer %d: %w", i, err) + } + + // Extract layer on top of rootfs (overlayfs style) + if err := w.extractTarGz(layerPath, rootfsDir); err != nil { + cleanup() + return "", nil, fmt.Errorf("failed to extract layer %d: %w", i, err) + } + + // Remove layer tar.gz to save space + os.Remove(layerPath) + } + + // Check what was extracted + entries, err := os.ReadDir(rootfsDir) + if err != nil { + slog.Warn("Failed to read rootfs directory", "error", err) + } else { + slog.Info("Successfully extracted image", + "layers", len(job.Layers), + "rootfs", rootfsDir, + "topLevelEntries", len(entries), + "sampleEntries", func() []string { + var samples []string + for i, e := range entries { + if i >= 10 { + break + } + samples = append(samples, e.Name()) + } + return samples + }()) + } + + return rootfsDir, cleanup, nil +} + +// downloadBlob downloads a blob from storage to a local file +func (w *Worker) downloadBlob(ctx context.Context, digest, destPath string) error { + // Convert digest to storage path using distribution's sharding scheme + // Format: /docker/registry/v2/blobs/sha256/47/4734bc89.../data + // where 47 is the first 2 characters of the hash for directory sharding + blobPath := blobPathForDigest(digest) + + // Open blob from storage driver + reader, err := w.driver.Reader(ctx, blobPath, 0) + if err != nil { + return fmt.Errorf("failed to open blob %s: %w", digest, err) + } + defer reader.Close() + + // Create destination file + dest, err := os.Create(destPath) + if err != nil { + return fmt.Errorf("failed to create destination file: %w", err) + } + defer dest.Close() + + // Copy blob data to file + if _, err := io.Copy(dest, reader); err != nil { + return fmt.Errorf("failed to copy blob data: %w", err) + } + + return nil +} + +// extractTarGz extracts a tar.gz file to a destination directory (overlayfs style) +func (w *Worker) extractTarGz(tarGzPath, destDir string) error { + // Open tar.gz file + file, err := os.Open(tarGzPath) + if err != nil { + return fmt.Errorf("failed to open tar.gz: %w", err) + } + defer file.Close() + + // Create gzip reader + gzr, err := gzip.NewReader(file) + if err != nil { + return fmt.Errorf("failed to create gzip reader: %w", err) + } + defer gzr.Close() + + // Create tar reader + tr := tar.NewReader(gzr) + + // Extract each file + for { + header, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return fmt.Errorf("failed to read tar header: %w", err) + } + + // Build target path (clean to prevent path traversal) + target := filepath.Join(destDir, filepath.Clean(header.Name)) + + // Ensure target is within destDir (security check) + if !strings.HasPrefix(target, filepath.Clean(destDir)+string(os.PathSeparator)) { + slog.Warn("Skipping path outside destination", "path", header.Name) + continue + } + + switch header.Typeflag { + case tar.TypeDir: + // Create directory + if err := os.MkdirAll(target, os.FileMode(header.Mode)); err != nil { + return fmt.Errorf("failed to create directory %s: %w", target, err) + } + + case tar.TypeReg: + // Create parent directory + if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil { + return fmt.Errorf("failed to create parent directory: %w", err) + } + + // Create file + outFile, err := os.OpenFile(target, os.O_CREATE|os.O_RDWR|os.O_TRUNC, os.FileMode(header.Mode)) + if err != nil { + return fmt.Errorf("failed to create file %s: %w", target, err) + } + + // Copy file contents + if _, err := io.Copy(outFile, tr); err != nil { + outFile.Close() + return fmt.Errorf("failed to write file %s: %w", target, err) + } + outFile.Close() + + case tar.TypeSymlink: + // Create symlink + if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil { + return fmt.Errorf("failed to create parent directory for symlink: %w", err) + } + + // Remove existing file/symlink if it exists + os.Remove(target) + + if err := os.Symlink(header.Linkname, target); err != nil { + slog.Warn("Failed to create symlink", "target", target, "link", header.Linkname, "error", err) + } + + case tar.TypeLink: + // Create hard link + linkTarget := filepath.Join(destDir, filepath.Clean(header.Linkname)) + if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil { + return fmt.Errorf("failed to create parent directory for hardlink: %w", err) + } + + // Remove existing file if it exists + os.Remove(target) + + if err := os.Link(linkTarget, target); err != nil { + slog.Warn("Failed to create hardlink", "target", target, "link", linkTarget, "error", err) + } + + default: + slog.Debug("Skipping unsupported tar entry type", "type", header.Typeflag, "name", header.Name) + } + } + + return nil +} + +// blobPathForDigest converts a digest to a storage path using distribution's sharding scheme +// Format: /docker/registry/v2/blobs/sha256/47/4734bc89.../data +// where 47 is the first 2 characters of the hash for directory sharding +func blobPathForDigest(digest string) string { + // Split digest into algorithm and hash + parts := strings.SplitN(digest, ":", 2) + if len(parts) != 2 { + // Fallback for malformed digest + return fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) + } + + algorithm := parts[0] + hash := parts[1] + + // Use first 2 characters for sharding + if len(hash) < 2 { + return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/data", algorithm, hash) + } + + return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", algorithm, hash[:2], hash) +} diff --git a/pkg/hold/scanner/grype.go b/pkg/hold/scanner/grype.go new file mode 100644 index 0000000..0a0117a --- /dev/null +++ b/pkg/hold/scanner/grype.go @@ -0,0 +1,351 @@ +package scanner + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "log/slog" + "os" + "path/filepath" + "sync" + + "github.com/anchore/grype/grype" + "github.com/anchore/grype/grype/db/v6/distribution" + "github.com/anchore/grype/grype/db/v6/installation" + "github.com/anchore/grype/grype/distro" + "github.com/anchore/grype/grype/match" + "github.com/anchore/grype/grype/matcher" + "github.com/anchore/grype/grype/matcher/dotnet" + "github.com/anchore/grype/grype/matcher/golang" + "github.com/anchore/grype/grype/matcher/java" + "github.com/anchore/grype/grype/matcher/javascript" + "github.com/anchore/grype/grype/matcher/python" + "github.com/anchore/grype/grype/matcher/ruby" + "github.com/anchore/grype/grype/matcher/stock" + grypePkg "github.com/anchore/grype/grype/pkg" + "github.com/anchore/grype/grype/vulnerability" + "github.com/anchore/syft/syft/sbom" +) + +// Global vulnerability database (shared across workers) +var ( + vulnDB vulnerability.Provider + vulnDBLock sync.RWMutex +) + +// scanVulnerabilities scans an SBOM for vulnerabilities using Grype +// Returns vulnerability report JSON, digest, summary, and any error +func (w *Worker) scanVulnerabilities(ctx context.Context, s *sbom.SBOM) ([]byte, string, VulnerabilitySummary, error) { + slog.Info("Scanning for vulnerabilities with Grype") + + // Load vulnerability database (cached globally) + store, err := w.loadVulnDatabase(ctx) + if err != nil { + return nil, "", VulnerabilitySummary{}, fmt.Errorf("failed to load vulnerability database: %w", err) + } + + // Create package context from SBOM (need distro for synthesis) + var grypeDistro *distro.Distro + if s.Artifacts.LinuxDistribution != nil { + grypeDistro = distro.FromRelease(s.Artifacts.LinuxDistribution, nil) + if grypeDistro != nil { + slog.Info("Using distro for package synthesis", + "name", grypeDistro.Name(), + "version", grypeDistro.Version, + "type", grypeDistro.Type, + "codename", grypeDistro.Codename) + } + } + + // Convert Syft packages to Grype packages WITH distro info + synthesisConfig := grypePkg.SynthesisConfig{ + GenerateMissingCPEs: true, + Distro: grypePkg.DistroConfig{ + Override: grypeDistro, + }, + } + grypePackages := grypePkg.FromCollection(s.Artifacts.Packages, synthesisConfig) + + slog.Info("Converted packages for vulnerability scanning", + "syftPackages", s.Artifacts.Packages.PackageCount(), + "grypePackages", len(grypePackages), + "distro", func() string { + if s.Artifacts.LinuxDistribution != nil { + return fmt.Sprintf("%s %s", s.Artifacts.LinuxDistribution.Name, s.Artifacts.LinuxDistribution.Version) + } + return "none" + }()) + + // Create matchers + matchers := matcher.NewDefaultMatchers(matcher.Config{ + Java: java.MatcherConfig{}, + Ruby: ruby.MatcherConfig{}, + Python: python.MatcherConfig{}, + Dotnet: dotnet.MatcherConfig{}, + Javascript: javascript.MatcherConfig{}, + Golang: golang.MatcherConfig{}, + Stock: stock.MatcherConfig{}, + }) + + // Create package context with the same distro we used for synthesis + pkgContext := grypePkg.Context{ + Source: &s.Source, + Distro: grypeDistro, + } + + // Create vulnerability matcher + vulnerabilityMatcher := &grype.VulnerabilityMatcher{ + VulnerabilityProvider: store, + Matchers: matchers, + NormalizeByCVE: true, + } + + // Find vulnerabilities + slog.Info("Matching vulnerabilities", + "packages", len(grypePackages), + "distro", func() string { + if grypeDistro != nil { + return fmt.Sprintf("%s %s", grypeDistro.Name(), grypeDistro.Version) + } + return "none" + }()) + allMatches, _, err := vulnerabilityMatcher.FindMatches(grypePackages, pkgContext) + if err != nil { + return nil, "", VulnerabilitySummary{}, fmt.Errorf("failed to find vulnerabilities: %w", err) + } + + slog.Info("Vulnerability matching complete", + "totalMatches", allMatches.Count()) + + // If we found 0 matches, log some diagnostic info + if allMatches.Count() == 0 { + slog.Warn("No vulnerability matches found - this may indicate an issue", + "distro", func() string { + if grypeDistro != nil { + return fmt.Sprintf("%s %s", grypeDistro.Name(), grypeDistro.Version) + } + return "none" + }(), + "packages", len(grypePackages), + "databaseBuilt", func() string { + vulnDBLock.RLock() + defer vulnDBLock.RUnlock() + if vulnDB == nil { + return "not loaded" + } + // We can't easily get the build date here without exposing internal state + return "loaded" + }()) + } + + // Count vulnerabilities by severity + summary := w.countVulnerabilitiesBySeverity(*allMatches) + + slog.Info("Vulnerability scan complete", + "critical", summary.Critical, + "high", summary.High, + "medium", summary.Medium, + "low", summary.Low, + "total", summary.Total) + + // Create vulnerability report JSON + report := map[string]interface{}{ + "matches": allMatches.Sorted(), + "source": s.Source, + "distro": s.Artifacts.LinuxDistribution, + "descriptor": map[string]interface{}{ + "name": "grype", + "version": "v0.102.0", // TODO: Get actual Grype version + }, + "summary": summary, + } + + // Encode report to JSON + reportJSON, err := json.MarshalIndent(report, "", " ") + if err != nil { + return nil, "", VulnerabilitySummary{}, fmt.Errorf("failed to encode vulnerability report: %w", err) + } + + // Calculate digest + hash := sha256.Sum256(reportJSON) + digest := fmt.Sprintf("sha256:%x", hash) + + slog.Info("Vulnerability report generated", "size", len(reportJSON), "digest", digest) + + // Upload report blob to storage + if err := w.uploadBlob(ctx, digest, reportJSON); err != nil { + return nil, "", VulnerabilitySummary{}, fmt.Errorf("failed to upload vulnerability report: %w", err) + } + + return reportJSON, digest, summary, nil +} + +// loadVulnDatabase loads the Grype vulnerability database (with caching) +func (w *Worker) loadVulnDatabase(ctx context.Context) (vulnerability.Provider, error) { + // Check if database is already loaded + vulnDBLock.RLock() + if vulnDB != nil { + vulnDBLock.RUnlock() + return vulnDB, nil + } + vulnDBLock.RUnlock() + + // Acquire write lock to load database + vulnDBLock.Lock() + defer vulnDBLock.Unlock() + + // Check again (another goroutine might have loaded it) + if vulnDB != nil { + return vulnDB, nil + } + + slog.Info("Loading Grype vulnerability database", "path", w.config.Scanner.VulnDBPath) + + // Ensure database directory exists + if err := ensureDir(w.config.Scanner.VulnDBPath); err != nil { + return nil, fmt.Errorf("failed to create vulnerability database directory: %w", err) + } + + // Configure database distribution + distConfig := distribution.DefaultConfig() + + // Configure database installation + installConfig := installation.Config{ + DBRootDir: w.config.Scanner.VulnDBPath, + ValidateAge: true, + ValidateChecksum: true, + MaxAllowedBuiltAge: w.config.Scanner.VulnDBUpdateInterval, + } + + // Load database (should already be downloaded by initializeVulnDatabase) + store, status, err := grype.LoadVulnerabilityDB(distConfig, installConfig, false) + if err != nil { + return nil, fmt.Errorf("failed to load vulnerability database (status=%v): %w (hint: database may still be downloading)", status, err) + } + + slog.Info("Vulnerability database loaded", + "status", status, + "built", status.Built, + "location", status.Path, + "schemaVersion", status.SchemaVersion) + + // Check database file size to verify it has content + if stat, err := os.Stat(status.Path); err == nil { + slog.Info("Vulnerability database file stats", + "size", stat.Size(), + "sizeMB", stat.Size()/1024/1024) + } + + // Cache database globally + vulnDB = store + + slog.Info("Vulnerability database loaded successfully") + return vulnDB, nil +} + +// countVulnerabilitiesBySeverity counts vulnerabilities by severity level +func (w *Worker) countVulnerabilitiesBySeverity(matches match.Matches) VulnerabilitySummary { + summary := VulnerabilitySummary{} + + for m := range matches.Enumerate() { + summary.Total++ + + // Get severity from vulnerability metadata + if m.Vulnerability.Metadata != nil { + severity := m.Vulnerability.Metadata.Severity + switch severity { + case "Critical": + summary.Critical++ + case "High": + summary.High++ + case "Medium": + summary.Medium++ + case "Low": + summary.Low++ + } + } + } + + return summary +} + +// initializeVulnDatabase downloads and initializes the vulnerability database on startup +func (w *Worker) initializeVulnDatabase(ctx context.Context) error { + slog.Info("Initializing vulnerability database", "path", w.config.Scanner.VulnDBPath) + + // Ensure database directory exists + if err := ensureDir(w.config.Scanner.VulnDBPath); err != nil { + return fmt.Errorf("failed to create vulnerability database directory: %w", err) + } + + // Create temp directory for Grype downloads (scratch container has no /tmp) + tmpDir := filepath.Join(w.config.Database.Path, "tmp") + if err := ensureDir(tmpDir); err != nil { + return fmt.Errorf("failed to create temp directory: %w", err) + } + + // Set TMPDIR environment variable so Grype uses our temp directory + oldTmpDir := os.Getenv("TMPDIR") + os.Setenv("TMPDIR", tmpDir) + defer func() { + if oldTmpDir != "" { + os.Setenv("TMPDIR", oldTmpDir) + } else { + os.Unsetenv("TMPDIR") + } + }() + + // Configure database distribution + distConfig := distribution.DefaultConfig() + + // Configure database installation + installConfig := installation.Config{ + DBRootDir: w.config.Scanner.VulnDBPath, + ValidateAge: true, + ValidateChecksum: true, + MaxAllowedBuiltAge: w.config.Scanner.VulnDBUpdateInterval, + } + + // Create distribution client for downloading + downloader, err := distribution.NewClient(distConfig) + if err != nil { + return fmt.Errorf("failed to create database downloader: %w", err) + } + + // Create curator to manage database + curator, err := installation.NewCurator(installConfig, downloader) + if err != nil { + return fmt.Errorf("failed to create database curator: %w", err) + } + + // Check if database already exists + status := curator.Status() + if !status.Built.IsZero() && status.Error == nil { + slog.Info("Vulnerability database already exists", "built", status.Built, "schema", status.SchemaVersion) + return nil + } + + // Download database (this may take several minutes) + slog.Info("Downloading vulnerability database (this may take 5-10 minutes)...") + updated, err := curator.Update() + if err != nil { + return fmt.Errorf("failed to download vulnerability database: %w", err) + } + + if updated { + slog.Info("Vulnerability database downloaded successfully") + } else { + slog.Info("Vulnerability database is up to date") + } + + return nil +} + +// ensureDir creates a directory if it doesn't exist +func ensureDir(path string) error { + if err := os.MkdirAll(path, 0755); err != nil { + return fmt.Errorf("failed to create directory %s: %w", path, err) + } + return nil +} diff --git a/pkg/hold/scanner/job.go b/pkg/hold/scanner/job.go new file mode 100644 index 0000000..0e5f850 --- /dev/null +++ b/pkg/hold/scanner/job.go @@ -0,0 +1,67 @@ +package scanner + +import ( + "time" + + "atcr.io/pkg/atproto" +) + +// ScanJob represents a vulnerability scanning job for a container image +type ScanJob struct { + // ManifestDigest is the digest of the manifest to scan + ManifestDigest string + + // Repository is the repository name (e.g., "alice/myapp") + Repository string + + // Tag is the tag name (e.g., "latest") + Tag string + + // UserDID is the DID of the user who owns this image + UserDID string + + // UserHandle is the handle of the user (for display) + UserHandle string + + // Config is the image config blob descriptor + Config atproto.BlobReference + + // Layers are the image layer blob descriptors (in order) + Layers []atproto.BlobReference + + // EnqueuedAt is when this job was enqueued + EnqueuedAt time.Time +} + +// ScanResult represents the result of a vulnerability scan +type ScanResult struct { + // Job is the original scan job + Job *ScanJob + + // VulnerabilitiesJSON is the raw Grype JSON output + VulnerabilitiesJSON []byte + + // Summary contains vulnerability counts by severity + Summary VulnerabilitySummary + + // SBOMDigest is the digest of the SBOM blob (if SBOM was generated) + SBOMDigest string + + // VulnDigest is the digest of the vulnerability report blob + VulnDigest string + + // ScannedAt is when the scan completed + ScannedAt time.Time + + // ScannerVersion is the version of the scanner used + ScannerVersion string +} + +// VulnerabilitySummary contains counts of vulnerabilities by severity +type VulnerabilitySummary struct { + Critical int `json:"critical"` + High int `json:"high"` + Medium int `json:"medium"` + Low int `json:"low"` + Total int `json:"total"` +} diff --git a/pkg/hold/scanner/queue.go b/pkg/hold/scanner/queue.go new file mode 100644 index 0000000..e2f96f8 --- /dev/null +++ b/pkg/hold/scanner/queue.go @@ -0,0 +1,226 @@ +package scanner + +import ( + "context" + "fmt" + "log/slog" + "sync" + + "atcr.io/pkg/atproto" +) + +// Queue manages a pool of workers for scanning container images +type Queue struct { + jobs chan *ScanJob + results chan *ScanResult + workers int + wg sync.WaitGroup + ctx context.Context + cancel context.CancelFunc +} + +// NewQueue creates a new scanner queue with the specified number of workers +func NewQueue(workers int, bufferSize int) *Queue { + ctx, cancel := context.WithCancel(context.Background()) + + return &Queue{ + jobs: make(chan *ScanJob, bufferSize), + results: make(chan *ScanResult, bufferSize), + workers: workers, + ctx: ctx, + cancel: cancel, + } +} + +// Start starts the worker pool +// The workerFunc is called for each job to perform the actual scanning +func (q *Queue) Start(workerFunc func(context.Context, *ScanJob) (*ScanResult, error)) { + slog.Info("Starting scanner worker pool", "workers", q.workers) + + for i := 0; i < q.workers; i++ { + q.wg.Add(1) + go q.worker(i, workerFunc) + } + + // Start result handler goroutine + q.wg.Add(1) + go q.resultHandler() +} + +// worker processes jobs from the queue +func (q *Queue) worker(id int, workerFunc func(context.Context, *ScanJob) (*ScanResult, error)) { + defer q.wg.Done() + + slog.Info("Scanner worker started", "worker_id", id) + + for { + select { + case <-q.ctx.Done(): + slog.Info("Scanner worker shutting down", "worker_id", id) + return + + case job, ok := <-q.jobs: + if !ok { + slog.Info("Scanner worker: jobs channel closed", "worker_id", id) + return + } + + slog.Info("Scanner worker processing job", + "worker_id", id, + "repository", job.Repository, + "tag", job.Tag, + "digest", job.ManifestDigest) + + result, err := workerFunc(q.ctx, job) + if err != nil { + slog.Error("Scanner worker failed to process job", + "worker_id", id, + "repository", job.Repository, + "tag", job.Tag, + "error", err) + continue + } + + // Send result to results channel + select { + case q.results <- result: + slog.Info("Scanner worker completed job", + "worker_id", id, + "repository", job.Repository, + "tag", job.Tag, + "vulnerabilities", result.Summary.Total) + case <-q.ctx.Done(): + return + } + } + } +} + +// resultHandler processes scan results (for logging and metrics) +func (q *Queue) resultHandler() { + defer q.wg.Done() + + for { + select { + case <-q.ctx.Done(): + return + + case result, ok := <-q.results: + if !ok { + return + } + + // Log the result + slog.Info("Scan completed", + "repository", result.Job.Repository, + "tag", result.Job.Tag, + "digest", result.Job.ManifestDigest, + "critical", result.Summary.Critical, + "high", result.Summary.High, + "medium", result.Summary.Medium, + "low", result.Summary.Low, + "total", result.Summary.Total, + "scanner", result.ScannerVersion) + } + } +} + +// Enqueue adds a job to the queue +func (q *Queue) Enqueue(jobAny any) error { + // Type assert to ScanJob (can be map or struct from HandleNotifyManifest) + var job *ScanJob + + switch v := jobAny.(type) { + case *ScanJob: + job = v + case map[string]interface{}: + // Convert map to ScanJob (from HandleNotifyManifest) + job = &ScanJob{ + ManifestDigest: v["manifestDigest"].(string), + Repository: v["repository"].(string), + Tag: v["tag"].(string), + UserDID: v["userDID"].(string), + UserHandle: v["userHandle"].(string), + } + + // Parse config blob reference + if configMap, ok := v["config"].(map[string]interface{}); ok { + job.Config = atproto.BlobReference{ + Digest: configMap["digest"].(string), + Size: convertToInt64(configMap["size"]), + MediaType: configMap["mediaType"].(string), + } + } + + // Parse layers + if layersSlice, ok := v["layers"].([]interface{}); ok { + slog.Info("Parsing layers from scan job", + "layersFound", len(layersSlice)) + job.Layers = make([]atproto.BlobReference, len(layersSlice)) + for i, layerAny := range layersSlice { + if layerMap, ok := layerAny.(map[string]interface{}); ok { + job.Layers[i] = atproto.BlobReference{ + Digest: layerMap["digest"].(string), + Size: convertToInt64(layerMap["size"]), + MediaType: layerMap["mediaType"].(string), + } + } + } + } else { + slog.Warn("No layers found in scan job map", + "layersType", fmt.Sprintf("%T", v["layers"]), + "layersValue", v["layers"]) + } + default: + return fmt.Errorf("invalid job type: %T", jobAny) + } + + select { + case q.jobs <- job: + slog.Info("Enqueued scan job", + "repository", job.Repository, + "tag", job.Tag, + "digest", job.ManifestDigest) + return nil + case <-q.ctx.Done(): + return q.ctx.Err() + } +} + +// Shutdown gracefully shuts down the queue, waiting for all workers to finish +func (q *Queue) Shutdown() { + slog.Info("Shutting down scanner queue") + + // Close the jobs channel to signal no more jobs + close(q.jobs) + + // Wait for all workers to finish + q.wg.Wait() + + // Close results channel + close(q.results) + + // Cancel context + q.cancel() + + slog.Info("Scanner queue shut down complete") +} + +// Len returns the number of jobs currently in the queue +func (q *Queue) Len() int { + return len(q.jobs) +} + +// convertToInt64 converts an interface{} number to int64, handling both float64 and int64 +func convertToInt64(v interface{}) int64 { + switch n := v.(type) { + case float64: + return int64(n) + case int64: + return n + case int: + return int64(n) + default: + return 0 + } +} diff --git a/pkg/hold/scanner/storage.go b/pkg/hold/scanner/storage.go new file mode 100644 index 0000000..7781d8e --- /dev/null +++ b/pkg/hold/scanner/storage.go @@ -0,0 +1,123 @@ +package scanner + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "log/slog" + "time" + + "atcr.io/pkg/atproto" +) + +// storeResults uploads scan results and creates ORAS manifest records in the hold's PDS +func (w *Worker) storeResults(ctx context.Context, job *ScanJob, sbomDigest, vulnDigest string, vulnJSON []byte, summary VulnerabilitySummary) error { + if !w.config.Scanner.VulnEnabled { + slog.Info("Vulnerability scanning disabled, skipping result storage") + return nil + } + + slog.Info("Storing scan results as ORAS artifact", + "repository", job.Repository, + "subjectDigest", job.ManifestDigest, + "vulnDigest", vulnDigest) + + // Create ORAS manifest for vulnerability report + orasManifest := map[string]interface{}{ + "schemaVersion": 2, + "mediaType": "application/vnd.oci.image.manifest.v1+json", + "artifactType": "application/vnd.atcr.vulnerabilities+json", + "config": map[string]interface{}{ + "mediaType": "application/vnd.oci.empty.v1+json", + "digest": "sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a", // Empty JSON object + "size": 2, + }, + "subject": map[string]interface{}{ + "mediaType": "application/vnd.oci.image.manifest.v1+json", + "digest": job.ManifestDigest, + "size": 0, // We don't have the size, but it's optional + }, + "layers": []map[string]interface{}{ + { + "mediaType": "application/json", + "digest": vulnDigest, + "size": len(vulnJSON), + "annotations": map[string]string{ + "org.opencontainers.image.title": "vulnerability-report.json", + }, + }, + }, + "annotations": map[string]string{ + "io.atcr.vuln.critical": fmt.Sprintf("%d", summary.Critical), + "io.atcr.vuln.high": fmt.Sprintf("%d", summary.High), + "io.atcr.vuln.medium": fmt.Sprintf("%d", summary.Medium), + "io.atcr.vuln.low": fmt.Sprintf("%d", summary.Low), + "io.atcr.vuln.total": fmt.Sprintf("%d", summary.Total), + "io.atcr.vuln.scannedAt": time.Now().Format(time.RFC3339), + "io.atcr.vuln.scannerVersion": w.getScannerVersion(), + }, + } + + // Encode ORAS manifest to JSON + orasManifestJSON, err := json.Marshal(orasManifest) + if err != nil { + return fmt.Errorf("failed to encode ORAS manifest: %w", err) + } + + // Calculate ORAS manifest digest + orasDigest := fmt.Sprintf("sha256:%x", sha256Bytes(orasManifestJSON)) + + // Upload ORAS manifest blob to storage + if err := w.uploadBlob(ctx, orasDigest, orasManifestJSON); err != nil { + return fmt.Errorf("failed to upload ORAS manifest blob: %w", err) + } + + // Create manifest record in hold's PDS + if err := w.createManifestRecord(ctx, job, orasDigest, orasManifestJSON, summary); err != nil { + return fmt.Errorf("failed to create manifest record: %w", err) + } + + slog.Info("Successfully stored scan results", "orasDigest", orasDigest) + return nil +} + +// createManifestRecord creates an ORAS manifest record in the hold's PDS +func (w *Worker) createManifestRecord(ctx context.Context, job *ScanJob, orasDigest string, orasManifestJSON []byte, summary VulnerabilitySummary) error { + // Create ManifestRecord from ORAS manifest + record, err := atproto.NewManifestRecord(job.Repository, orasDigest, orasManifestJSON) + if err != nil { + return fmt.Errorf("failed to create manifest record: %w", err) + } + + // Set SBOM/vulnerability specific fields + record.OwnerDID = job.UserDID + record.ScannedAt = time.Now().Format(time.RFC3339) + record.ScannerVersion = w.getScannerVersion() + + // Add hold DID (this ORAS artifact is stored in the hold's PDS) + record.HoldDID = w.pds.DID() + + // Convert digest to record key (remove "sha256:" prefix) + rkey := orasDigest[len("sha256:"):] + + // Store record in hold's PDS + slog.Info("Creating manifest record in hold's PDS", + "collection", atproto.ManifestCollection, + "rkey", rkey, + "ownerDid", job.UserDID) + + _, _, err = w.pds.CreateManifestRecord(ctx, record, rkey) + if err != nil { + return fmt.Errorf("failed to put record in PDS: %w", err) + } + + slog.Info("Manifest record created successfully", "uri", fmt.Sprintf("at://%s/%s/%s", w.pds.DID(), atproto.ManifestCollection, rkey)) + return nil +} + +// sha256Bytes calculates SHA256 hash of byte slice +func sha256Bytes(data []byte) []byte { + hash := sha256.Sum256(data) + return hash[:] +} diff --git a/pkg/hold/scanner/syft.go b/pkg/hold/scanner/syft.go new file mode 100644 index 0000000..347bcd0 --- /dev/null +++ b/pkg/hold/scanner/syft.go @@ -0,0 +1,128 @@ +package scanner + +import ( + "context" + "crypto/sha256" + "fmt" + "log/slog" + "os" + + "github.com/anchore/syft/syft" + "github.com/anchore/syft/syft/format" + "github.com/anchore/syft/syft/format/spdxjson" + "github.com/anchore/syft/syft/sbom" + "github.com/anchore/syft/syft/source/directorysource" +) + +// generateSBOM generates an SBOM using Syft from an extracted image directory +// Returns the SBOM object, SBOM JSON, its digest, and any error +func (w *Worker) generateSBOM(ctx context.Context, imageDir string) (*sbom.SBOM, []byte, string, error) { + slog.Info("Generating SBOM with Syft", "imageDir", imageDir) + + // Check if directory exists and is accessible + entries, err := os.ReadDir(imageDir) + if err != nil { + return nil, nil, "", fmt.Errorf("failed to read image directory: %w", err) + } + slog.Info("Image directory contents", + "path", imageDir, + "entries", len(entries), + "sampleFiles", func() []string { + var samples []string + for i, e := range entries { + if i >= 20 { + break + } + samples = append(samples, e.Name()) + } + return samples + }()) + + // Create Syft source from directory + src, err := directorysource.NewFromPath(imageDir) + if err != nil { + return nil, nil, "", fmt.Errorf("failed to create Syft source: %w", err) + } + defer src.Close() + + // Generate SBOM + slog.Info("Running Syft cataloging") + sbomResult, err := syft.CreateSBOM(ctx, src, nil) + if err != nil { + return nil, nil, "", fmt.Errorf("failed to generate SBOM: %w", err) + } + + if sbomResult == nil { + return nil, nil, "", fmt.Errorf("Syft returned nil SBOM") + } + + slog.Info("SBOM generated", + "packages", sbomResult.Artifacts.Packages.PackageCount(), + "distro", func() string { + if sbomResult.Artifacts.LinuxDistribution != nil { + return fmt.Sprintf("%s %s", sbomResult.Artifacts.LinuxDistribution.Name, sbomResult.Artifacts.LinuxDistribution.Version) + } + return "none" + }()) + + // Encode SBOM to SPDX JSON format + encoder, err := spdxjson.NewFormatEncoderWithConfig(spdxjson.DefaultEncoderConfig()) + if err != nil { + return nil, nil, "", fmt.Errorf("failed to create SPDX encoder: %w", err) + } + + sbomJSON, err := format.Encode(*sbomResult, encoder) + if err != nil { + return nil, nil, "", fmt.Errorf("failed to encode SBOM to SPDX JSON: %w", err) + } + + // Calculate digest + hash := sha256.Sum256(sbomJSON) + digest := fmt.Sprintf("sha256:%x", hash) + + slog.Info("SBOM encoded", "format", "spdx-json", "size", len(sbomJSON), "digest", digest) + + // Upload SBOM blob to storage + if err := w.uploadBlob(ctx, digest, sbomJSON); err != nil { + return nil, nil, "", fmt.Errorf("failed to upload SBOM blob: %w", err) + } + + return sbomResult, sbomJSON, digest, nil +} + +// uploadBlob uploads a blob to storage +func (w *Worker) uploadBlob(ctx context.Context, digest string, data []byte) error { + // Convert digest to storage path (same format as distribution uses) + // Path format: /docker/registry/v2/blobs/sha256/ab/abcd1234.../data + algorithm := "sha256" + digestHex := digest[len("sha256:"):] + if len(digestHex) < 2 { + return fmt.Errorf("invalid digest: %s", digest) + } + + blobPath := fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", + algorithm, + digestHex[:2], + digestHex) + + slog.Info("Uploading blob to storage", "digest", digest, "size", len(data), "path", blobPath) + + // Write blob to storage + writer, err := w.driver.Writer(ctx, blobPath, false) + if err != nil { + return fmt.Errorf("failed to create storage writer: %w", err) + } + defer writer.Close() + + if _, err := writer.Write(data); err != nil { + writer.Cancel(ctx) + return fmt.Errorf("failed to write blob data: %w", err) + } + + if err := writer.Commit(ctx); err != nil { + return fmt.Errorf("failed to commit blob: %w", err) + } + + slog.Info("Successfully uploaded blob", "digest", digest) + return nil +} diff --git a/pkg/hold/scanner/worker.go b/pkg/hold/scanner/worker.go new file mode 100644 index 0000000..6f2a13f --- /dev/null +++ b/pkg/hold/scanner/worker.go @@ -0,0 +1,116 @@ +package scanner + +import ( + "context" + "fmt" + "log/slog" + "time" + + "atcr.io/pkg/hold" + "atcr.io/pkg/hold/pds" + "github.com/distribution/distribution/v3/registry/storage/driver" +) + +// Worker performs vulnerability scanning on container images +type Worker struct { + config *hold.Config + driver driver.StorageDriver + pds *pds.HoldPDS + queue *Queue +} + +// NewWorker creates a new scanner worker +func NewWorker(config *hold.Config, driver driver.StorageDriver, pds *pds.HoldPDS) *Worker { + return &Worker{ + config: config, + driver: driver, + pds: pds, + } +} + +// Start starts the worker pool and initializes vulnerability database +func (w *Worker) Start(queue *Queue) { + w.queue = queue + + // Initialize vulnerability database on startup if scanning is enabled + if w.config.Scanner.VulnEnabled { + go func() { + ctx := context.Background() + if err := w.initializeVulnDatabase(ctx); err != nil { + slog.Error("Failed to initialize vulnerability database", "error", err) + slog.Warn("Vulnerability scanning will be disabled until database is available") + } + }() + } + + queue.Start(w.processJob) +} + +// processJob processes a single scan job +func (w *Worker) processJob(ctx context.Context, job *ScanJob) (*ScanResult, error) { + slog.Info("Processing scan job", + "repository", job.Repository, + "tag", job.Tag, + "digest", job.ManifestDigest, + "layers", len(job.Layers)) + + startTime := time.Now() + + // Step 1: Extract image layers from storage + slog.Info("Extracting image layers", "repository", job.Repository) + imageDir, cleanup, err := w.extractLayers(ctx, job) + if err != nil { + return nil, fmt.Errorf("failed to extract layers: %w", err) + } + defer cleanup() + + // Step 2: Generate SBOM with Syft + slog.Info("Generating SBOM", "repository", job.Repository) + sbomResult, _, sbomDigest, err := w.generateSBOM(ctx, imageDir) + if err != nil { + return nil, fmt.Errorf("failed to generate SBOM: %w", err) + } + + // Step 3: Scan SBOM with Grype (if enabled) + var vulnJSON []byte + var vulnDigest string + var summary VulnerabilitySummary + + if w.config.Scanner.VulnEnabled { + slog.Info("Scanning for vulnerabilities", "repository", job.Repository) + vulnJSON, vulnDigest, summary, err = w.scanVulnerabilities(ctx, sbomResult) + if err != nil { + return nil, fmt.Errorf("failed to scan vulnerabilities: %w", err) + } + } + + // Step 4: Upload results to storage and create ORAS manifests + slog.Info("Storing scan results", "repository", job.Repository) + err = w.storeResults(ctx, job, sbomDigest, vulnDigest, vulnJSON, summary) + if err != nil { + return nil, fmt.Errorf("failed to store results: %w", err) + } + + duration := time.Since(startTime) + slog.Info("Scan job completed", + "repository", job.Repository, + "tag", job.Tag, + "duration", duration, + "vulnerabilities", summary.Total) + + return &ScanResult{ + Job: job, + VulnerabilitiesJSON: vulnJSON, + Summary: summary, + SBOMDigest: sbomDigest, + VulnDigest: vulnDigest, + ScannedAt: time.Now(), + ScannerVersion: w.getScannerVersion(), + }, nil +} + +// getScannerVersion returns the version string for the scanner +func (w *Worker) getScannerVersion() string { + // TODO: Get actual Syft and Grype versions dynamically + return "syft-v1.36.0/grype-v0.102.0" +}