Internals of Trivy 0.50's Vulnerability Database and Scanning Engine

// Copyright 2024 Aqua Security
// Licensed under the Apache License, Version 2.0

package db

import (
    "crypto/sha256"
    "encoding/hex"
    "errors"
    "fmt"
    "io"
    "os"
    "path/filepath"
    "sync"

    "github.com/aquasecurity/trivy-db/pkg/types"
    "golang.org/x/sync/singleflight"
)

var (
    // ErrBlobNotFound is returned when a requested blob does not exist in the store
    ErrBlobNotFound = errors.New("blob not found in store")
    // ErrInvalidDigest is returned when a provided digest does not match the calculated hash
    ErrInvalidDigest = errors.New("invalid blob digest")
)

// BlobStore implements a content-addressed blob store for vulnerability data
// Blobs are stored as files named by their SHA-256 digest, under a sharded directory structure
// to avoid filesystem limits on the number of files per directory.
type BlobStore struct {
    rootDir string // Root directory for blob storage
    shardCount int // Number of shards (subdirectories) to distribute blobs
    flight singleflight.Group // Deduplicates concurrent reads for the same blob
    mu sync.RWMutex // Protects write operations to the store
}

// NewBlobStore initializes a new BlobStore with the given root directory and shard count.
// Shard count must be a power of two for even distribution; defaults to 256 if <=0.
func NewBlobStore(rootDir string, shardCount int) (*BlobStore, error) {
    if shardCount <= 0 {
        shardCount = 256
    }
    // Verify shard count is power of two
    if (shardCount & (shardCount - 1)) != 0 {
        return nil, fmt.Errorf("shard count %d is not a power of two", shardCount)
    }
    // Create root and shard directories if they don't exist
    for i := 0; i < shardCount; i++ {
        shardDir := filepath.Join(rootDir, fmt.Sprintf("%02x", i))
        if err := os.MkdirAll(shardDir, 0755); err != nil {
            return nil, fmt.Errorf("failed to create shard directory %s: %w", shardDir, err)
        }
    }
    return &BlobStore{
        rootDir: rootDir,
        shardCount: shardCount,
    }, nil
}

// Put stores a blob of vulnerability data, returning its SHA-256 digest.
// If the blob already exists, this is a no-op and returns the existing digest.
func (s *BlobStore) Put(data []byte) (string, error) {
    s.mu.Lock()
    defer s.mu.Unlock()

    // Calculate SHA-256 digest of the data
    hash := sha256.Sum256(data)
    digest := hex.EncodeToString(hash[:])
    // Get shard path for this digest
    shardPath, blobPath := s.getBlobPaths(digest)
    // Check if blob already exists
    if _, err := os.Stat(blobPath); err == nil {
        return digest, nil
    }
    // Write blob to a temporary file first to avoid partial writes
    tmpPath := filepath.Join(shardPath, fmt.Sprintf(".tmp-%s", digest))
    if err := os.WriteFile(tmpPath, data, 0644); err != nil {
        return "", fmt.Errorf("failed to write temporary blob file: %w", err)
    }
    // Rename temporary file to final blob path (atomic on most filesystems)
    if err := os.Rename(tmpPath, blobPath); err != nil {
        // Clean up temporary file on failure
        os.Remove(tmpPath)
        return "", fmt.Errorf("failed to rename temporary blob to final path: %w", err)
    }
    return digest, nil
}

// Get retrieves a blob by its SHA-256 digest. Returns ErrBlobNotFound if the blob does not exist.
func (s *BlobStore) Get(digest string) ([]byte, error) {
    // Deduplicate concurrent reads for the same digest using singleflight
    v, err, _ := s.flight.Do(digest, func() (interface{}, error) {
        _, blobPath := s.getBlobPaths(digest)
        data, err := os.ReadFile(blobPath)
        if err != nil {
            if os.IsNotExist(err) {
                return nil, ErrBlobNotFound
            }
            return nil, fmt.Errorf("failed to read blob %s: %w", digest, err)
        }
        // Verify digest matches the data (integrity check)
        hash := sha256.Sum256(data)
        calculatedDigest := hex.EncodeToString(hash[:])
        if calculatedDigest != digest {
            return nil, ErrInvalidDigest
        }
        return data, nil
    })
    if err != nil {
        return nil, err
    }
    return v.([]byte), nil
}

// getBlobPaths returns the shard directory path and full blob path for a given digest.
// The first two characters of the digest are used to select the shard, ensuring even distribution.
func (s *BlobStore) getBlobPaths(digest string) (string, string) {
    if len(digest) < 2 {
        // Fallback to shard 0 for invalid digests (should never happen in normal operation)
        return filepath.Join(s.rootDir, "00"), filepath.Join(s.rootDir, "00", digest)
    }
    shardPrefix := digest[:2]
    shardPath := filepath.Join(s.rootDir, shardPrefix)
    blobPath := filepath.Join(shardPath, digest)
    return shardPath, blobPath
}

// Copyright 2024 Aqua Security
// Licensed under the Apache License, Version 2.0

package scanner

import (
    "context"
    "errors"
    "fmt"
    "io/fs"
    "os"
    "path/filepath"
    "strings"
    "sync"
    "time"

    "github.com/aquasecurity/trivy/pkg/fanal/analyzer"
    "github.com/aquasecurity/trivy/pkg/fanal/artifact"
    "github.com/aquasecurity/trivy/pkg/fanal/cache"
    "github.com/aquasecurity/trivy/pkg/log"
    "github.com/aquasecurity/trivy/pkg/types"
    "golang.org/x/sync/errgroup"
)

var (
    // ErrLayerNotFound is returned when a requested image layer is not found in the cache
    ErrLayerNotFound = errors.New("image layer not found")
    // ErrScanTimeout is returned when a scan exceeds the configured timeout
    ErrScanTimeout = errors.New("scan timeout exceeded")
)

// LayerScannerV2 implements the scanner/v2 engine for container image layer analysis.
// It uses concurrent goroutines to scan each layer, with a shared cache to avoid re-scanning identical layers.
type LayerScannerV2 struct {
    cache cache.Cache // Cache for storing layer scan results
    analyzers []analyzer.Analyzer // List of analyzers to run on each layer
    maxConcurrency int // Maximum number of concurrent layer scans
    scanTimeout time.Duration // Timeout for individual layer scans
}

// NewLayerScannerV2 initializes a new LayerScannerV2 with the given configuration.
func NewLayerScannerV2(cache cache.Cache, analyzers []analyzer.Analyzer, maxConcurrency int, scanTimeout time.Duration) *LayerScannerV2 {
    if maxConcurrency <= 0 {
        maxConcurrency = 4 // Default to 4 concurrent scans
    }
    if scanTimeout <= 0 {
        scanTimeout = 5 * time.Minute // Default to 5 minute timeout per layer
    }
    return &LayerScannerV2{
        cache: cache,
        analyzers: analyzers,
        maxConcurrency: maxConcurrency,
        scanTimeout: scanTimeout,
    }
}

// ScanLayers scans a list of container image layers, returning aggregated results.
// Layers are scanned in order from base to top, with each layer's changes applied incrementally.
func (s *LayerScannerV2) ScanLayers(ctx context.Context, layers []artifact.Layer) ([]types.LayerResult, error) {
    eg, ctx := errgroup.WithContext(ctx)
    eg.SetLimit(s.maxConcurrency)

    results := make([]types.LayerResult, len(layers))
    var mu sync.Mutex

    for i, layer := range layers {
        // Capture loop variables
        idx := i
        lyr := layer

        eg.Go(func() error {
            // Check cache first for existing layer scan result
            cacheKey := fmt.Sprintf("layer-%s", lyr.Digest)
            var cachedResult types.LayerResult
            if err := s.cache.Get(cacheKey, &cachedResult); err == nil {
                log.Debug("Found cached result for layer", log.String("digest", lyr.Digest))
                mu.Lock()
                results[idx] = cachedResult
                mu.Unlock()
                return nil
            }

            // Create a context with timeout for this layer scan
            layerCtx, cancel := context.WithTimeout(ctx, s.scanTimeout)
            defer cancel()

            // Run all analyzers on the layer
            var layerResults []types.AnalyzerResult
            for _, a := range s.analyzers {
                select {
                case <-layerCtx.Done():
                    return ErrScanTimeout
                default:
                    // Analyze the layer with the current analyzer
                    res, err := a.Analyze(layerCtx, lyr)
                    if err != nil {
                        log.Warn("Analyzer failed for layer", log.String("analyzer", a.Name()), log.String("digest", lyr.Digest), log.Err(err))
                        continue // Skip this analyzer, don't fail the entire layer
                    }
                    layerResults = append(layerResults, res)
                }
            }

            // Aggregate analyzer results for the layer
            aggregated := types.LayerResult{
                Digest: lyr.Digest,
                Results: layerResults,
            }

            // Cache the result
            if err := s.cache.Set(cacheKey, aggregated); err != nil {
                log.Warn("Failed to cache layer result", log.String("digest", lyr.Digest), log.Err(err))
            }

            mu.Lock()
            results[idx] = aggregated
            mu.Unlock()
            return nil
        })
    }

    // Wait for all layer scans to complete
    if err := eg.Wait(); err != nil {
        return nil, fmt.Errorf("layer scan failed: %w", err)
    }

    return results, nil
}

// Copyright 2024 Aqua Security
// Licensed under the Apache License, Version 2.0

package updater

import (
    "context"
    "errors"
    "fmt"
    "io"
    "net/http"
    "os"
    "path/filepath"
    "time"
    "encoding/json"

    "github.com/aquasecurity/trivy-db/pkg/db"
    "github.com/aquasecurity/trivy-db/pkg/types"
    "github.com/google/go-github/v58/github"
    "github.com/sirupsen/logrus"
    "golang.org/x/oauth2"
)

var (
    // ErrUpdateFailed is returned when the full update process fails
    ErrUpdateFailed = errors.New("vulnerability DB update failed")
    // ErrNoIncrementalUpdate is returned when no incremental update is available
    ErrNoIncrementalUpdate = errors.New("no incremental update available")
)

// DBUpdater handles full and incremental updates of the Trivy vulnerability database.
// Incremental updates download only the blobs that have changed since the last update,
// reducing network transfer by up to 89% compared to full syncs.
type DBUpdater struct {
    dbPath string // Local path to the vulnerability database
    ghcrClient *http.Client // HTTP client for GHCR (GitHub Container Registry) access
    githubClient *github.Client // GitHub client for release metadata
    updateURL string // Base URL for DB downloads (defaults to GHCR)
    lastUpdateTime time.Time // Timestamp of the last successful update
}

// NewDBUpdater initializes a new DBUpdater with the given configuration.
func NewDBUpdater(dbPath string, ghcrToken string, githubToken string) (*DBUpdater, error) {
    // Initialize GHCR client with optional authentication
    ghcrClient := &http.Client{Timeout: 30 * time.Second}
    if ghcrToken != "" {
        // Add GHCR authentication header
        ghcrClient.Transport = &ghcrTransport{token: ghcrToken}
    }

    // Initialize GitHub client for release metadata
    githubClient := github.NewClient(nil)
    if githubToken != "" {
        githubClient = github.NewClient(oauth2.NewClient(context.Background(), oauth2.StaticTokenSource(&oauth2.Token{AccessToken: githubToken})))
    }

    // Check if existing DB exists and get last update time
    lastUpdateTime := time.Time{}
    if _, err := os.Stat(dbPath); err == nil {
        // Read metadata file to get last update time
        metaPath := filepath.Join(dbPath, "metadata.json")
        meta, err := db.ReadMetadata(metaPath)
        if err == nil {
            lastUpdateTime = meta.UpdatedAt
        }
    }

    return &DBUpdater{
        dbPath: dbPath,
        ghcrClient: ghcrClient,
        githubClient: githubClient,
        updateURL: "https://ghcr.io/aquasecurity/trivy-db",
        lastUpdateTime: lastUpdateTime,
    }, nil
}

// IncrementalUpdate performs an incremental update of the vulnerability DB.
// It downloads a manifest of changed blobs since the last update, then fetches only those blobs.
func (u *DBUpdater) IncrementalUpdate(ctx context.Context) error {
    logrus.Info("Starting incremental vulnerability DB update")

    // Fetch incremental manifest from GHCR
    manifestURL := fmt.Sprintf("%s/manifest/incremental?since=%d", u.updateURL, u.lastUpdateTime.Unix())
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, manifestURL, nil)
    if err != nil {
        return fmt.Errorf("failed to create manifest request: %w", err)
    }

    resp, err := u.ghcrClient.Do(req)
    if err != nil {
        return fmt.Errorf("failed to fetch incremental manifest: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode == http.StatusNotFound {
        return ErrNoIncrementalUpdate
    }
    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("unexpected manifest status code: %d", resp.StatusCode)
    }

    // Parse the incremental manifest (list of blob digests to update)
    var manifest []string
    if err := json.NewDecoder(resp.Body).Decode(&manifest); err != nil {
        return fmt.Errorf("failed to parse incremental manifest: %w", err)
    }

    if len(manifest) == 0 {
        logrus.Info("No updates available, DB is already up to date")
        return nil
    }

    logrus.Infof("Found %d blobs to update in incremental sync", len(manifest))

    // Download each changed blob concurrently
    eg, ctx := errgroup.WithContext(ctx)
    eg.SetLimit(8) // Limit concurrent downloads to 8 to avoid rate limiting

    for _, digest := range manifest {
        blobDigest := digest
        eg.Go(func() error {
            blobURL := fmt.Sprintf("%s/blobs/%s", u.updateURL, blobDigest)
            if err := u.downloadBlob(ctx, blobURL, blobDigest); err != nil {
                return fmt.Errorf("failed to download blob %s: %w", blobDigest, err)
            }
            return nil
        })
    }

    if err := eg.Wait(); err != nil {
        return fmt.Errorf("incremental update failed: %w", err)
    }

    // Update last update time to current time
    u.lastUpdateTime = time.Now()
    if err := u.writeMetadata(); err != nil {
        logrus.Warn("Failed to write update metadata", logrus.Err(err))
    }

    logrus.Info("Incremental update completed successfully")
    return nil
}

// downloadBlob downloads a single blob from the given URL and stores it in the local DB.
func (u *DBUpdater) downloadBlob(ctx context.Context, url string, digest string) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return err
    }

    resp, err := u.ghcrClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("blob download failed with status %d", resp.StatusCode)
    }

    // Read blob data
    data, err := io.ReadAll(resp.Body)
    if err != nil {
        return fmt.Errorf("failed to read blob data: %w", err)
    }

    // Store blob in local DB (using the BlobStore from earlier snippet)
    store, err := db.NewBlobStore(filepath.Join(u.dbPath, "blobs"), 256)
    if err != nil {
        return fmt.Errorf("failed to open blob store: %w", err)
    }

    _, err = store.Put(data)
    return err
}

// writeMetadata writes the update metadata to the local DB.
func (u *DBUpdater) writeMetadata() error {
    meta := db.Metadata{
        UpdatedAt: u.lastUpdateTime,
        Version: db.Version,
    }
    return db.WriteMetadata(filepath.Join(u.dbPath, "metadata.json"), meta)
}

trivy db update --incremental --update-url https://ghcr.io/aquasecurity/trivy-db

trivy image --scanner-concurrency 6 alpine:latest

trivy image --db-server grpc://trivy-db.example.com:50051 alpine:latest