vaultik/internal/blobgen/writer.go

// Package blobgen implements the blob data pipeline: streaming zstd
// compression, age encryption, and SHA256 content hashing for blob
// creation, plus the matching decrypt/decompress/verify reader.
package blobgen

import (
	"crypto/sha256"
	"errors"
	"fmt"
	"hash"
	"io"
	"runtime"

	"filippo.io/age"
	"github.com/klauspost/compress/zstd"
)

// DoubleSHA256 returns the double SHA-256 of content whose single SHA-256
// digest is sum: it hashes that digest once more. Stored objects are named by
// this second hash so that a name never reveals whether known content is
// present — an attacker who knows a plaintext, and thus its SHA-256, still
// cannot derive the stored name without hashing the digest again. Both a blob
// and the metadata database export are named this way.
func DoubleSHA256(sum []byte) []byte {
	h := sha256.Sum256(sum)

	return h[:]
}

// Zstd compression level bounds accepted by NewWriter.
const (
	minCompressionLevel = 1
	maxCompressionLevel = 19
)

// reservedCompressionCPUs is how many CPUs are left free of zstd
// compression work for I/O and hashing.
const reservedCompressionCPUs = 2

// ErrInvalidCompressionLevel is returned when the zstd compression level
// is outside the accepted 1-19 range.
var ErrInvalidCompressionLevel = errors.New(
	"invalid compression level: must be between 1 and 19")

// errInvalidRecipient is returned when a recipient string does not parse as
// an X25519 age1... public key. It omits the value, which can be sensitive.
var errInvalidRecipient = errors.New(
	"not a valid X25519 age1... recipient")

// Writer wraps compression and encryption with SHA256 hashing.
// Data flows: input -> tee(hasher, compressor -> encryptor -> destination)
// The hash is computed on the uncompressed input for deterministic content-addressing.
type Writer struct {
	teeWriter        io.Writer      // Tee to hasher and compressor
	compressor       *zstd.Encoder  // Compression layer
	encryptor        io.WriteCloser // Encryption layer
	hasher           hash.Hash      // SHA256 hasher (on uncompressed input)
	compressionLevel int
	bytesWritten     int64
}

// NewWriter creates a new Writer that compresses, encrypts, and hashes
// data. The hash is computed on the uncompressed input for deterministic
// content-addressing.
func NewWriter(
	w io.Writer, compressionLevel int, recipients []string,
) (*Writer, error) {
	// Validate compression level
	err := validateCompressionLevel(compressionLevel)
	if err != nil {
		return nil, err
	}

	// Create SHA256 hasher for the uncompressed input
	hasher := sha256.New()

	// Parse recipients
	var ageRecipients []age.Recipient

	for i, recipient := range recipients {
		// The recipient string can be sensitive (e.g. a secret key pasted by
		// mistake), so the error names its position, never its value.
		r, err := age.ParseX25519Recipient(recipient)
		if err != nil {
			return nil, fmt.Errorf("%w: recipient %d", errInvalidRecipient, i)
		}

		ageRecipients = append(ageRecipients, r)
	}

	// Create encryption writer that outputs to destination
	encWriter, err := age.Encrypt(w, ageRecipients...)
	if err != nil {
		return nil, fmt.Errorf("creating encryption writer: %w", err)
	}

	// Calculate compression concurrency: CPUs - 2, minimum 1
	concurrency := max(runtime.NumCPU()-reservedCompressionCPUs, 1)

	// Create compression writer with encryption as destination
	compressor, err := zstd.NewWriter(encWriter,
		zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(compressionLevel)),
		zstd.WithEncoderConcurrency(concurrency),
	)
	if err != nil {
		_ = encWriter.Close()

		return nil, fmt.Errorf("creating compression writer: %w", err)
	}

	// Create tee writer: input goes to both hasher and compressor
	teeWriter := io.MultiWriter(hasher, compressor)

	return &Writer{
		teeWriter:        teeWriter,
		compressor:       compressor,
		encryptor:        encWriter,
		hasher:           hasher,
		compressionLevel: compressionLevel,
	}, nil
}

// Write implements io.Writer
func (w *Writer) Write(p []byte) (int, error) {
	n, err := w.teeWriter.Write(p)
	w.bytesWritten += int64(n)

	return n, err
}

// Close closes all layers and returns any errors
func (w *Writer) Close() error {
	// Close compressor first
	err := w.compressor.Close()
	if err != nil {
		return fmt.Errorf("closing compressor: %w", err)
	}

	// Then close encryptor
	err = w.encryptor.Close()
	if err != nil {
		return fmt.Errorf("closing encryptor: %w", err)
	}

	return nil
}

// ContentID returns the double SHA-256 of the uncompressed input data: the
// name under which this content is stored. It is the second hash of the
// running SHA-256, via DoubleSHA256; see that function for why content is
// named this way rather than by its plain SHA-256.
func (w *Writer) ContentID() []byte {
	return DoubleSHA256(w.hasher.Sum(nil))
}

// BytesWritten returns the number of uncompressed bytes written
func (w *Writer) BytesWritten() int64 {
	return w.bytesWritten
}

func validateCompressionLevel(level int) error {
	// Zstd compression levels: 1-19 (default is 3)
	// SpeedFastest = 1, SpeedDefault = 3, SpeedBetterCompression = 7,
	// SpeedBestCompression = 11
	if level < minCompressionLevel || level > maxCompressionLevel {
		return fmt.Errorf("%w: got %d", ErrInvalidCompressionLevel, level)
	}

	return nil
}