actually add scanner implementation

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)
+}

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
+}

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"`
+}

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
+	}
+}

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[:]
+}

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
+}

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"
+}