vaultik/internal/storage/faultstore/faultstore.go

// Package faultstore provides a storage.Storer wrapper that injects
// faults on demand, so tests can reproduce the failure modes a real
// backend exhibits: an upload that fails partway, a backend that reports
// success while storing nothing, and reads that return corrupt or
// truncated bytes. It is the seam called for by the fault-injection
// tests (sneak/vaultik issue 72) and is meant to be reused by future
// tests rather than re-implemented per case.
//
// The wrapper delegates every method to the inner Storer. Two hooks
// change that: OnPut decides the fate of each write, and OnGet decides
// how each read's bytes are returned. Both are keyed by the object key,
// so a test can fault only blobs, only metadata, or a single object.
package faultstore

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"

	"sneak.berlin/go/vaultik/internal/storage"
)

// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")

// PutAction is the disposition OnPut assigns to a write.
type PutAction int

const (
	// PutNormal writes through to the inner Storer.
	PutNormal PutAction = iota
	// PutFail reads part of the stream, then fails without storing the
	// object — a network upload that dies partway through.
	PutFail
	// PutSwallow reports success but stores nothing — a backend that
	// lies about durability.
	PutSwallow
)

// GetFault is how OnGet chooses to damage a read.
type GetFault int

const (
	// GetNormal returns the stored bytes unchanged.
	GetNormal GetFault = iota
	// GetCorrupt flips a byte so the returned object no longer matches
	// what was stored.
	GetCorrupt
	// GetTruncate returns a short read: the object's bytes cut off
	// before the end.
	GetTruncate
)

// Storer wraps an inner storage.Storer with fault-injection hooks. A
// zero-valued hook means "no fault": construct with New and set only the
// hook a test needs.
type Storer struct {
	inner storage.Storer

	// OnPut, when set, is consulted before every Put and
	// PutWithProgress with the object key.
	OnPut func(key string) PutAction

	// OnGet, when set, is consulted for every Get with the object key
	// and damages the returned bytes accordingly.
	OnGet func(key string) GetFault
}

// New wraps inner. inner must be non-nil.
func New(inner storage.Storer) *Storer {
	return &Storer{inner: inner}
}

// midStreamBytes is how far a PutFail reads before failing, enough to be
// past the start of any real blob without depending on the blob's size.
const midStreamBytes = 512

// Put stores data unless OnPut faults the write.
func (f *Storer) Put(ctx context.Context, key string, data io.Reader) error {
	handled, err := f.injectPut(key, data)
	if handled {
		return err
	}

	return f.inner.Put(ctx, key, data)
}

// PutWithProgress stores data unless OnPut faults the write.
func (f *Storer) PutWithProgress(
	ctx context.Context, key string, data io.Reader,
	size int64, progress storage.ProgressCallback,
) error {
	handled, err := f.injectPut(key, data)
	if handled {
		return err
	}

	return f.inner.PutWithProgress(ctx, key, data, size, progress)
}

// Get retrieves data, damaging it if OnGet faults the read.
func (f *Storer) Get(ctx context.Context, key string) (io.ReadCloser, error) {
	rc, err := f.inner.Get(ctx, key)
	if err != nil {
		return nil, err
	}

	fault := GetNormal
	if f.OnGet != nil {
		fault = f.OnGet(key)
	}

	if fault == GetNormal {
		return rc, nil
	}

	data, err := io.ReadAll(rc)
	_ = rc.Close()

	if err != nil {
		return nil, err
	}

	return io.NopCloser(bytes.NewReader(damage(fault, data))), nil
}

// damage returns a faulted copy of the stored bytes. GetCorrupt flips a
// byte in the middle so decryption authentication fails; GetTruncate
// drops the final byte so the read ends short. Both are no-ops on empty
// input, which cannot be damaged into something distinguishable.
func damage(fault GetFault, data []byte) []byte {
	out := make([]byte, len(data))
	copy(out, data)

	if len(out) == 0 {
		return out
	}

	switch fault {
	case GetCorrupt:
		out[len(out)/2] ^= 0xff
	case GetTruncate:
		out = out[:len(out)-1]
	case GetNormal:
	}

	return out
}

// Stat delegates unchanged.
func (f *Storer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, error) {
	return f.inner.Stat(ctx, key)
}

// Delete delegates unchanged.
func (f *Storer) Delete(ctx context.Context, key string) error {
	return f.inner.Delete(ctx, key)
}

// List delegates unchanged.
func (f *Storer) List(ctx context.Context, prefix string) ([]string, error) {
	return f.inner.List(ctx, prefix)
}

// ListStream delegates unchanged.
func (f *Storer) ListStream(
	ctx context.Context, prefix string,
) <-chan storage.ObjectInfo {
	return f.inner.ListStream(ctx, prefix)
}

// Info delegates unchanged.
func (f *Storer) Info() storage.Info {
	return f.inner.Info()
}

func (f *Storer) putAction(key string) PutAction {
	if f.OnPut == nil {
		return PutNormal
	}

	return f.OnPut(key)
}

// injectPut handles the non-normal write dispositions. It reports
// whether it handled the write and, if so, with what error.
func (f *Storer) injectPut(key string, data io.Reader) (bool, error) {
	switch f.putAction(key) {
	case PutFail:
		// Consume part of the stream so the failure lands mid-transfer,
		// the way a dropped connection would, then error without
		// storing anything.
		_, _ = io.CopyN(io.Discard, data, midStreamBytes)

		return true, fmt.Errorf("%w for %q", ErrInjectedUpload, key)
	case PutSwallow:
		// A lying backend still drains the request body, then keeps
		// nothing.
		_, _ = io.Copy(io.Discard, data)

		return true, nil
	case PutNormal:
		return false, nil
	default:
		return false, nil
	}
}