Add storage backends and safety checks

This commit is contained in:
2026-05-31 02:00:24 +00:00
parent 29dbad2967
commit 3c2f36a6e5
10 changed files with 1582 additions and 0 deletions

37
docs/internal/storage.md Normal file
View File

@@ -0,0 +1,37 @@
# Storage
## Purpose
`internal/storage` defines backend-rooted logical file access for core packages. Callers use slash-separated paths relative to a configured backend root.
## Inputs and outputs
The storage interface supports byte reads, stream reads, byte writes, stream writes, exact metadata lookup, traversal, destination emptiness checks, and guarded managed deletion.
Entries report a logical path, type, and size when available. Entry types are `file`, `directory`, `symlink`, and `other`.
## Boundaries
Core packages should depend on `internal/storage`, not adapter packages. Adapter-specific path handling stays behind backend implementations.
The local adapter lives in `internal/adapters/local`. The fake backend lives in `internal/storage/fake` for tests and is not registered for runtime use.
## Paths
Logical file paths must be non-empty, relative, clean, slash-separated, and must not contain `.` or `..` segments or backslashes. Prefix paths follow the same rules, except an empty prefix means the backend root.
## Failure behavior
Storage errors use typed categories such as not found, already exists, invalid path, conflict, permission, temporary, unsupported, and unknown. Callers should use helper predicates rather than matching error strings.
## Deletion
Backends expose guarded managed deletion only. `DeleteManagedBundle` may delete listed managed outputs plus `.distributor.json`; it does not provide broad recursive deletion.
## Tests
Before changing storage behavior, inspect tests under:
- `internal/storage`
- `internal/storage/fake`
- `internal/adapters/local`

View File

@@ -0,0 +1,370 @@
package local
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
const backendName = "local"
type Backend struct {
root string
}
func New(root string) (*Backend, error) {
if root == "" {
return nil, storage.NewError(storage.OpOpenBackend, backendName, "", storage.ErrInvalidPath, nil)
}
absRoot, err := filepath.Abs(root)
if err != nil {
return nil, storage.NewError(storage.OpOpenBackend, backendName, root, storage.ErrInvalidPath, err)
}
return &Backend{root: filepath.Clean(absRoot)}, nil
}
func (b *Backend) ReadFile(ctx context.Context, path string) ([]byte, error) {
reader, err := b.OpenReader(ctx, path)
if err != nil {
return nil, err
}
defer reader.Close()
data, err := io.ReadAll(reader)
if err != nil {
return nil, storage.NewError(storage.OpReadFile, backendName, path, storage.ErrUnknown, err)
}
return data, nil
}
func (b *Backend) OpenReader(ctx context.Context, path string) (io.ReadCloser, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
nativePath, err := b.nativePath(path, false)
if err != nil {
return nil, err
}
if err := b.rejectSymlinkAncestors(nativePath, true); err != nil {
return nil, err
}
info, err := os.Lstat(nativePath)
if err != nil {
return nil, b.translateError(storage.OpOpenReader, path, err)
}
if !info.Mode().IsRegular() {
return nil, storage.NewError(storage.OpOpenReader, backendName, path, storage.ErrUnsupported, nil)
}
file, err := os.Open(nativePath)
if err != nil {
return nil, b.translateError(storage.OpOpenReader, path, err)
}
return file, nil
}
func (b *Backend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) {
opts.Size = int64(len(data))
opts.SizeKnown = true
return b.WriteFrom(ctx, path, bytes.NewReader(data), opts)
}
func (b *Backend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) {
if err := ctx.Err(); err != nil {
return storage.Entry{}, err
}
nativePath, err := b.nativePath(path, false)
if err != nil {
return storage.Entry{}, err
}
parent := filepath.Dir(nativePath)
if err := b.rejectSymlinkAncestors(parent, true); err != nil {
return storage.Entry{}, err
}
if info, err := os.Lstat(nativePath); err == nil {
if !opts.Overwrite {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrAlreadyExist, nil)
}
if !info.Mode().IsRegular() {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrConflict, nil)
}
} else if !errors.Is(err, fs.ErrNotExist) {
return storage.Entry{}, b.translateError(storage.OpWriteFrom, path, err)
}
if err := os.MkdirAll(parent, 0o755); err != nil {
return storage.Entry{}, b.translateError(storage.OpWriteFrom, path, err)
}
temp, err := os.CreateTemp(parent, ".distributor-write-*")
if err != nil {
return storage.Entry{}, b.translateError(storage.OpWriteFrom, path, err)
}
tempPath := temp.Name()
cleanup := true
defer func() {
if cleanup {
_ = os.Remove(tempPath)
}
}()
written, copyErr := io.Copy(temp, r)
closeErr := temp.Close()
if copyErr != nil {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrUnknown, copyErr)
}
if closeErr != nil {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrUnknown, closeErr)
}
if opts.SizeKnown && written != opts.Size {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrConflict, fmt.Errorf("stream size %d does not match expected size %d", written, opts.Size))
}
if err := os.Rename(tempPath, nativePath); err != nil {
return storage.Entry{}, b.translateError(storage.OpWriteFrom, path, err)
}
cleanup = false
return b.Stat(ctx, path)
}
func (b *Backend) Stat(ctx context.Context, path string) (storage.Entry, error) {
if err := ctx.Err(); err != nil {
return storage.Entry{}, err
}
nativePath, err := b.nativePath(path, true)
if err != nil {
return storage.Entry{}, err
}
info, err := os.Lstat(nativePath)
if err != nil {
return storage.Entry{}, b.translateError(storage.OpStat, path, err)
}
return entryFromInfo(path, info), nil
}
func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOptions, fn storage.WalkFunc) error {
if err := ctx.Err(); err != nil {
return err
}
nativePrefix, err := b.nativePath(prefix, true)
if err != nil {
return err
}
info, err := os.Lstat(nativePrefix)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
return nil
}
return b.translateError(storage.OpWalk, prefix, err)
}
visited := 0
emit := func(entry storage.Entry) error {
if err := ctx.Err(); err != nil {
return err
}
if opts.Limit > 0 && visited >= opts.Limit {
return storage.ErrStopWalk
}
visited++
if err := fn(entry); err != nil {
if errors.Is(err, storage.ErrStopWalk) {
return storage.ErrStopWalk
}
return storage.NewError(storage.OpWalk, backendName, entry.Path, storage.ErrUnknown, err)
}
return nil
}
if !info.IsDir() {
if err := emit(entryFromInfo(prefix, info)); errors.Is(err, storage.ErrStopWalk) {
return nil
} else if err != nil {
return err
}
return nil
}
walkErr := filepath.WalkDir(nativePrefix, func(nativePath string, dirEntry fs.DirEntry, err error) error {
if err != nil {
return b.translateError(storage.OpWalk, prefix, err)
}
if nativePath == nativePrefix {
return nil
}
relPath, err := filepath.Rel(b.root, nativePath)
if err != nil {
return b.translateError(storage.OpWalk, prefix, err)
}
logicalPath := filepath.ToSlash(relPath)
if !opts.Recursive && filepath.Dir(nativePath) != nativePrefix {
if dirEntry.IsDir() {
return filepath.SkipDir
}
return nil
}
info, err := dirEntry.Info()
if err != nil {
return b.translateError(storage.OpWalk, logicalPath, err)
}
return emit(entryFromInfo(logicalPath, info))
})
if errors.Is(walkErr, storage.ErrStopWalk) {
return nil
}
return walkErr
}
func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
found := false
err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
found = true
return storage.ErrStopWalk
})
if err != nil {
return false, err
}
return found, nil
}
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(bundlePath); err != nil {
return err
}
targets := make([]string, 0, len(managedOutputPaths)+1)
for _, outputPath := range managedOutputPaths {
target, err := storage.Join(bundlePath, outputPath)
if err != nil {
return err
}
targets = append(targets, target)
}
statePath, err := storage.StatePath(bundlePath)
if err != nil {
return err
}
targets = append(targets, statePath)
for _, logicalPath := range targets {
nativePath, err := b.nativePath(logicalPath, false)
if err != nil {
return err
}
if nativePath == b.root {
return storage.NewError(storage.OpDeleteManagedBundle, backendName, logicalPath, storage.ErrInvalidPath, nil)
}
info, err := os.Lstat(nativePath)
if err != nil {
if opts.IgnoreMissing && errors.Is(err, fs.ErrNotExist) {
continue
}
return b.translateError(storage.OpDeleteManagedBundle, logicalPath, err)
}
if info.IsDir() {
return storage.NewError(storage.OpDeleteManagedBundle, backendName, logicalPath, storage.ErrUnsupported, nil)
}
if err := os.Remove(nativePath); err != nil {
return b.translateError(storage.OpDeleteManagedBundle, logicalPath, err)
}
if opts.PruneEmptyDirs {
b.pruneEmptyParents(filepath.Dir(nativePath))
}
}
return nil
}
func (b *Backend) nativePath(logicalPath string, allowEmpty bool) (string, error) {
if logicalPath == "" {
if !allowEmpty {
return "", storage.NewError(storage.OpValidatePath, backendName, logicalPath, storage.ErrInvalidPath, nil)
}
return b.root, nil
}
if err := storage.ValidatePath(logicalPath); err != nil {
return "", err
}
nativePath := filepath.Clean(filepath.Join(b.root, filepath.FromSlash(logicalPath)))
rel, err := filepath.Rel(b.root, nativePath)
if err != nil {
return "", storage.NewError(storage.OpValidatePath, backendName, logicalPath, storage.ErrInvalidPath, err)
}
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) || filepath.IsAbs(rel) {
return "", storage.NewError(storage.OpValidatePath, backendName, logicalPath, storage.ErrInvalidPath, nil)
}
return nativePath, nil
}
func (b *Backend) rejectSymlinkAncestors(nativePath string, includeFinal bool) error {
rel, err := filepath.Rel(b.root, nativePath)
if err != nil {
return storage.NewError(storage.OpValidatePath, backendName, nativePath, storage.ErrInvalidPath, err)
}
if rel == "." {
return nil
}
segments := strings.Split(rel, string(filepath.Separator))
limit := len(segments)
if !includeFinal {
limit--
}
current := b.root
for i := 0; i < limit; i++ {
current = filepath.Join(current, segments[i])
info, err := os.Lstat(current)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
return nil
}
return b.translateError(storage.OpStat, filepath.ToSlash(filepath.Join(segments[:i+1]...)), err)
}
if info.Mode()&os.ModeSymlink != 0 {
return storage.NewError(storage.OpStat, backendName, filepath.ToSlash(filepath.Join(segments[:i+1]...)), storage.ErrUnsupported, nil)
}
}
return nil
}
func (b *Backend) pruneEmptyParents(start string) {
for current := start; current != b.root && strings.HasPrefix(current, b.root); current = filepath.Dir(current) {
err := os.Remove(current)
if err != nil {
return
}
}
}
func (b *Backend) translateError(op, path string, err error) error {
kind := storage.ErrUnknown
switch {
case errors.Is(err, fs.ErrNotExist):
kind = storage.ErrNotFound
case errors.Is(err, fs.ErrExist):
kind = storage.ErrAlreadyExist
case errors.Is(err, fs.ErrPermission):
kind = storage.ErrPermission
}
return storage.NewError(op, backendName, path, kind, err)
}
func entryFromInfo(path string, info fs.FileInfo) storage.Entry {
entryType := storage.EntryTypeOther
switch {
case info.Mode()&os.ModeSymlink != 0:
entryType = storage.EntryTypeSymlink
case info.Mode().IsRegular():
entryType = storage.EntryTypeFile
case info.IsDir():
entryType = storage.EntryTypeDirectory
}
return storage.Entry{
Path: path,
Type: entryType,
Size: info.Size(),
}
}

View File

@@ -0,0 +1,233 @@
package local
import (
"bytes"
"context"
"errors"
"io"
"os"
"path/filepath"
"reflect"
"testing"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
func TestBackendRejectsTraversal(t *testing.T) {
backend := newBackend(t)
_, err := backend.ReadFile(context.Background(), "../outside")
if !storage.IsInvalidPath(err) {
t.Fatalf("ReadFile traversal error = %v, want invalid path", err)
}
_, err = backend.WriteFile(context.Background(), "/absolute", []byte("data"), storage.WriteOptions{})
if !storage.IsInvalidPath(err) {
t.Fatalf("WriteFile absolute path error = %v, want invalid path", err)
}
}
func TestBackendReadWriteAndStream(t *testing.T) {
backend := newBackend(t)
entry, err := backend.WriteFile(context.Background(), "reports/report.md", []byte("hello"), storage.WriteOptions{})
if err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
if entry.Path != "reports/report.md" || entry.Type != storage.EntryTypeFile || entry.Size != 5 {
t.Fatalf("entry = %#v, want written file metadata", entry)
}
data, err := backend.ReadFile(context.Background(), "reports/report.md")
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
if string(data) != "hello" {
t.Fatalf("ReadFile() = %q, want hello", data)
}
reader, err := backend.OpenReader(context.Background(), "reports/report.md")
if err != nil {
t.Fatalf("OpenReader() error = %v", err)
}
streamed, err := io.ReadAll(reader)
closeErr := reader.Close()
if err != nil || closeErr != nil {
t.Fatalf("read stream error = %v close = %v", err, closeErr)
}
if !bytes.Equal(streamed, data) {
t.Fatalf("streamed = %q, want %q", streamed, data)
}
_, err = backend.WriteFile(context.Background(), "reports/report.md", []byte("again"), storage.WriteOptions{})
if !storage.IsAlreadyExists(err) {
t.Fatalf("WriteFile without overwrite error = %v, want already exists", err)
}
}
func TestBackendStatWalkAndList(t *testing.T) {
backend := newBackend(t)
mustWrite(t, backend, "b/two.txt", "2")
mustWrite(t, backend, "a/one.txt", "1")
entry, err := backend.Stat(context.Background(), "a/one.txt")
if err != nil {
t.Fatalf("Stat() error = %v", err)
}
if entry.Type != storage.EntryTypeFile || entry.Size != 1 {
t.Fatalf("entry = %#v, want file size 1", entry)
}
entries, err := storage.List(context.Background(), backend, "", storage.WalkOptions{Recursive: true})
if err != nil {
t.Fatalf("List() error = %v", err)
}
var paths []string
for _, entry := range entries {
paths = append(paths, entry.Path)
}
want := []string{"a", "a/one.txt", "b", "b/two.txt"}
if !reflect.DeepEqual(paths, want) {
t.Fatalf("paths = %v, want %v", paths, want)
}
entries, err = storage.List(context.Background(), backend, "", storage.WalkOptions{Recursive: false})
if err != nil {
t.Fatalf("List nonrecursive error = %v", err)
}
paths = paths[:0]
for _, entry := range entries {
paths = append(paths, entry.Path)
}
want = []string{"a", "b"}
if !reflect.DeepEqual(paths, want) {
t.Fatalf("nonrecursive paths = %v, want %v", paths, want)
}
}
func TestBackendSymlinkReportingAndReadRejection(t *testing.T) {
root := t.TempDir()
backend, err := New(root)
if err != nil {
t.Fatalf("New() error = %v", err)
}
if err := os.WriteFile(filepath.Join(root, "target.txt"), []byte("target"), 0o600); err != nil {
t.Fatalf("write target: %v", err)
}
if err := os.Symlink("target.txt", filepath.Join(root, "link.txt")); err != nil {
t.Fatalf("symlink: %v", err)
}
entry, err := backend.Stat(context.Background(), "link.txt")
if err != nil {
t.Fatalf("Stat() error = %v", err)
}
if entry.Type != storage.EntryTypeSymlink {
t.Fatalf("entry type = %s, want symlink", entry.Type)
}
_, err = backend.ReadFile(context.Background(), "link.txt")
if !storage.IsUnsupported(err) {
t.Fatalf("ReadFile symlink error = %v, want unsupported", err)
}
}
func TestBackendWriteFromSizeMismatchLeavesNoFinalFile(t *testing.T) {
backend := newBackend(t)
_, err := backend.WriteFrom(context.Background(), "out.txt", bytes.NewBufferString("short"), storage.WriteOptions{SizeKnown: true, Size: 99})
if !storage.IsConflict(err) {
t.Fatalf("WriteFrom size mismatch error = %v, want conflict", err)
}
_, err = backend.Stat(context.Background(), "out.txt")
if !storage.IsNotFound(err) {
t.Fatalf("Stat after failed write error = %v, want not found", err)
}
}
func TestBackendManagedDeletion(t *testing.T) {
backend := newBackend(t)
mustWrite(t, backend, "bundle/report.html", "html")
mustWrite(t, backend, "bundle/keep.txt", "keep")
mustWrite(t, backend, "bundle/.distributor.json", "{}")
err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
if err != nil {
t.Fatalf("DeleteManagedBundle() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.html"); !storage.IsNotFound(err) {
t.Fatalf("managed output stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/.distributor.json"); !storage.IsNotFound(err) {
t.Fatalf("state stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/keep.txt"); err != nil {
t.Fatalf("unlisted file stat error = %v", err)
}
if err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{""}, storage.DeleteOptions{}); !storage.IsInvalidPath(err) {
t.Fatalf("DeleteManagedBundle invalid output error = %v, want invalid path", err)
}
}
func TestBackendHasAny(t *testing.T) {
backend := newBackend(t)
found, err := backend.HasAny(context.Background(), "missing")
if err != nil {
t.Fatalf("HasAny missing error = %v", err)
}
if found {
t.Fatal("HasAny missing = true, want false")
}
mustWrite(t, backend, "bundle/report.md", "report")
found, err = backend.HasAny(context.Background(), "bundle")
if err != nil {
t.Fatalf("HasAny bundle error = %v", err)
}
if !found {
t.Fatal("HasAny bundle = false, want true")
}
found, err = backend.HasAny(context.Background(), "bund")
if err != nil {
t.Fatalf("HasAny sibling prefix error = %v", err)
}
if found {
t.Fatal("HasAny prefix sibling = true, want false")
}
}
func TestBackendWalkStops(t *testing.T) {
backend := newBackend(t)
mustWrite(t, backend, "a.txt", "a")
mustWrite(t, backend, "b.txt", "b")
visited := 0
err := backend.Walk(context.Background(), "", storage.WalkOptions{Recursive: true}, func(storage.Entry) error {
visited++
return storage.ErrStopWalk
})
if err != nil {
t.Fatalf("Walk() error = %v", err)
}
if visited != 1 {
t.Fatalf("visited = %d, want 1", visited)
}
errSentinel := errors.New("callback")
err = backend.Walk(context.Background(), "", storage.WalkOptions{Recursive: true}, func(storage.Entry) error {
return errSentinel
})
if !errors.Is(err, errSentinel) {
t.Fatalf("Walk callback error = %v, want sentinel", err)
}
}
func newBackend(t *testing.T) *Backend {
t.Helper()
backend, err := New(t.TempDir())
if err != nil {
t.Fatalf("New() error = %v", err)
}
return backend
}
func mustWrite(t *testing.T, backend *Backend, path, data string) {
t.Helper()
if _, err := backend.WriteFile(context.Background(), path, []byte(data), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile(%q) error = %v", path, err)
}
}

View File

@@ -0,0 +1,68 @@
package storage
import (
"context"
"errors"
"io"
)
type EntryType string
const (
EntryTypeFile EntryType = "file"
EntryTypeDirectory EntryType = "directory"
EntryTypeSymlink EntryType = "symlink"
EntryTypeOther EntryType = "other"
)
type Entry struct {
Path string
Type EntryType
Size int64
}
type Backend interface {
ReadFile(ctx context.Context, path string) ([]byte, error)
OpenReader(ctx context.Context, path string) (io.ReadCloser, error)
WriteFile(ctx context.Context, path string, data []byte, opts WriteOptions) (Entry, error)
WriteFrom(ctx context.Context, path string, r io.Reader, opts WriteOptions) (Entry, error)
Stat(ctx context.Context, path string) (Entry, error)
Walk(ctx context.Context, prefix string, opts WalkOptions, fn WalkFunc) error
HasAny(ctx context.Context, prefix string) (bool, error)
DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts DeleteOptions) error
}
type WalkOptions struct {
Recursive bool
Limit int
}
type WalkFunc func(Entry) error
var ErrStopWalk = errors.New("stop walk")
type WriteOptions struct {
ContentType string
Overwrite bool
PreferAtomic bool
Size int64
SizeKnown bool
}
type DeleteOptions struct {
IgnoreMissing bool
PruneEmptyDirs bool
}
func List(ctx context.Context, backend Backend, prefix string, opts WalkOptions) ([]Entry, error) {
var entries []Entry
err := backend.Walk(ctx, prefix, opts, func(entry Entry) error {
entries = append(entries, entry)
return nil
})
if err != nil {
return nil, err
}
SortEntries(entries)
return entries, nil
}

128
internal/storage/errors.go Normal file
View File

@@ -0,0 +1,128 @@
package storage
import (
"errors"
"fmt"
)
type ErrorKind string
const (
ErrNotFound ErrorKind = "not_found"
ErrAlreadyExist ErrorKind = "already_exists"
ErrNotEmpty ErrorKind = "not_empty"
ErrInvalidPath ErrorKind = "invalid_path"
ErrConflict ErrorKind = "conflict"
ErrPermission ErrorKind = "permission"
ErrTemporary ErrorKind = "temporary"
ErrUnsupported ErrorKind = "unsupported"
ErrUnknown ErrorKind = "unknown"
)
const (
OpValidatePath = "validate path"
OpReadFile = "read file"
OpOpenReader = "open reader"
OpWriteFile = "write file"
OpWriteFrom = "write stream"
OpStat = "stat"
OpWalk = "walk"
OpHasAny = "has any"
OpDeleteManagedBundle = "delete managed bundle"
OpRegisterBackend = "register backend"
OpOpenBackend = "open backend"
)
type Error struct {
Op string
Backend string
Path string
Kind ErrorKind
Err error
}
func NewError(op, backend, path string, kind ErrorKind, err error) *Error {
return &Error{
Op: op,
Backend: backend,
Path: path,
Kind: kind,
Err: err,
}
}
func (e *Error) Error() string {
if e == nil {
return "<nil>"
}
message := e.Op
if e.Backend != "" {
message += " " + e.Backend
}
if e.Path != "" {
message += " " + e.Path
}
if e.Kind != "" {
message += ": " + string(e.Kind)
}
if e.Err != nil {
message += ": " + e.Err.Error()
}
return message
}
func (e *Error) Unwrap() error {
if e == nil {
return nil
}
return e.Err
}
func (e *Error) Is(target error) bool {
var targetError *Error
if !errors.As(target, &targetError) {
return false
}
return targetError.Kind == "" || e.Kind == targetError.Kind
}
func Errorf(op, backend, path string, kind ErrorKind, format string, args ...any) *Error {
return NewError(op, backend, path, kind, fmt.Errorf(format, args...))
}
func IsKind(err error, kind ErrorKind) bool {
var storageErr *Error
return errors.As(err, &storageErr) && storageErr.Kind == kind
}
func IsNotFound(err error) bool {
return IsKind(err, ErrNotFound)
}
func IsAlreadyExists(err error) bool {
return IsKind(err, ErrAlreadyExist)
}
func IsNotEmpty(err error) bool {
return IsKind(err, ErrNotEmpty)
}
func IsInvalidPath(err error) bool {
return IsKind(err, ErrInvalidPath)
}
func IsConflict(err error) bool {
return IsKind(err, ErrConflict)
}
func IsPermission(err error) bool {
return IsKind(err, ErrPermission)
}
func IsTemporary(err error) bool {
return IsKind(err, ErrTemporary)
}
func IsUnsupported(err error) bool {
return IsKind(err, ErrUnsupported)
}

View File

@@ -0,0 +1,316 @@
package fake
import (
"bytes"
"context"
"errors"
"io"
"sort"
"strings"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
const backendName = "fake"
type Backend struct {
files map[string][]byte
dirs map[string]struct{}
symlinks map[string]struct{}
}
func New() *Backend {
return &Backend{
files: make(map[string][]byte),
dirs: map[string]struct{}{"": {}},
symlinks: make(map[string]struct{}),
}
}
func (b *Backend) AddDirectory(path string) error {
if err := storage.ValidatePrefix(path); err != nil {
return err
}
b.ensureParents(path)
b.dirs[path] = struct{}{}
return nil
}
func (b *Backend) AddSymlink(path string) error {
if err := storage.ValidatePath(path); err != nil {
return err
}
b.ensureParents(path)
delete(b.files, path)
delete(b.dirs, path)
b.symlinks[path] = struct{}{}
return nil
}
func (b *Backend) ReadFile(ctx context.Context, path string) ([]byte, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
if err := storage.ValidatePath(path); err != nil {
return nil, err
}
data, ok := b.files[path]
if !ok {
if b.exists(path) {
return nil, storage.NewError(storage.OpReadFile, backendName, path, storage.ErrUnsupported, nil)
}
return nil, storage.NewError(storage.OpReadFile, backendName, path, storage.ErrNotFound, nil)
}
return append([]byte(nil), data...), nil
}
func (b *Backend) OpenReader(ctx context.Context, path string) (io.ReadCloser, error) {
data, err := b.ReadFile(ctx, path)
if err != nil {
return nil, err
}
return io.NopCloser(bytes.NewReader(data)), nil
}
func (b *Backend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) {
opts.Size = int64(len(data))
opts.SizeKnown = true
return b.WriteFrom(ctx, path, bytes.NewReader(data), opts)
}
func (b *Backend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) {
if err := ctx.Err(); err != nil {
return storage.Entry{}, err
}
if err := storage.ValidatePath(path); err != nil {
return storage.Entry{}, err
}
if b.exists(path) && !opts.Overwrite {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrAlreadyExist, nil)
}
if _, ok := b.dirs[path]; ok {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrConflict, nil)
}
if _, ok := b.symlinks[path]; ok {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrConflict, nil)
}
data, err := io.ReadAll(r)
if err != nil {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrUnknown, err)
}
if opts.SizeKnown && int64(len(data)) != opts.Size {
return storage.Entry{}, storage.NewError(storage.OpWriteFrom, backendName, path, storage.ErrConflict, nil)
}
b.ensureParents(path)
b.files[path] = append([]byte(nil), data...)
delete(b.symlinks, path)
return storage.Entry{Path: path, Type: storage.EntryTypeFile, Size: int64(len(data))}, nil
}
func (b *Backend) Stat(ctx context.Context, path string) (storage.Entry, error) {
if err := ctx.Err(); err != nil {
return storage.Entry{}, err
}
if err := storage.ValidatePrefix(path); err != nil {
return storage.Entry{}, err
}
if data, ok := b.files[path]; ok {
return storage.Entry{Path: path, Type: storage.EntryTypeFile, Size: int64(len(data))}, nil
}
if _, ok := b.symlinks[path]; ok {
return storage.Entry{Path: path, Type: storage.EntryTypeSymlink}, nil
}
if _, ok := b.dirs[path]; ok {
return storage.Entry{Path: path, Type: storage.EntryTypeDirectory}, nil
}
return storage.Entry{}, storage.NewError(storage.OpStat, backendName, path, storage.ErrNotFound, nil)
}
func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOptions, fn storage.WalkFunc) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(prefix); err != nil {
return err
}
if entry, err := b.Stat(ctx, prefix); err == nil && entry.Type != storage.EntryTypeDirectory {
return emit(ctx, entry, opts, fn)
} else if err != nil && !storage.IsNotFound(err) {
return err
}
entries := b.entries()
visited := 0
for _, entry := range entries {
if entry.Path == "" || !entryBelow(prefix, entry.Path) {
continue
}
if !opts.Recursive && !isImmediateChild(prefix, entry.Path) {
continue
}
if opts.Limit > 0 && visited >= opts.Limit {
return nil
}
visited++
if err := ctx.Err(); err != nil {
return err
}
if err := fn(entry); err != nil {
if errors.Is(err, storage.ErrStopWalk) {
return nil
}
return storage.NewError(storage.OpWalk, backendName, entry.Path, storage.ErrUnknown, err)
}
}
return nil
}
func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
found := false
err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
found = true
return storage.ErrStopWalk
})
if err != nil {
return false, err
}
return found, nil
}
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(bundlePath); err != nil {
return err
}
targets := make([]string, 0, len(managedOutputPaths)+1)
for _, outputPath := range managedOutputPaths {
target, err := storage.Join(bundlePath, outputPath)
if err != nil {
return err
}
targets = append(targets, target)
}
statePath, err := storage.StatePath(bundlePath)
if err != nil {
return err
}
targets = append(targets, statePath)
for _, target := range targets {
if _, ok := b.dirs[target]; ok {
return storage.NewError(storage.OpDeleteManagedBundle, backendName, target, storage.ErrUnsupported, nil)
}
if !b.exists(target) {
if opts.IgnoreMissing {
continue
}
return storage.NewError(storage.OpDeleteManagedBundle, backendName, target, storage.ErrNotFound, nil)
}
delete(b.files, target)
delete(b.symlinks, target)
if opts.PruneEmptyDirs {
b.pruneEmptyParents(parentOf(target))
}
}
return nil
}
func (b *Backend) ensureParents(path string) {
parent := parentOf(path)
for parent != "" {
b.dirs[parent] = struct{}{}
parent = parentOf(parent)
}
b.dirs[""] = struct{}{}
}
func (b *Backend) pruneEmptyParents(path string) {
for path != "" {
if b.hasChild(path) {
return
}
delete(b.dirs, path)
path = parentOf(path)
}
}
func (b *Backend) hasChild(path string) bool {
for candidate := range b.files {
if entryBelow(path, candidate) {
return true
}
}
for candidate := range b.symlinks {
if entryBelow(path, candidate) {
return true
}
}
for candidate := range b.dirs {
if candidate != path && entryBelow(path, candidate) {
return true
}
}
return false
}
func (b *Backend) exists(path string) bool {
_, file := b.files[path]
_, dir := b.dirs[path]
_, symlink := b.symlinks[path]
return file || dir || symlink
}
func (b *Backend) entries() []storage.Entry {
entries := make([]storage.Entry, 0, len(b.files)+len(b.dirs)+len(b.symlinks))
for path, data := range b.files {
entries = append(entries, storage.Entry{Path: path, Type: storage.EntryTypeFile, Size: int64(len(data))})
}
for path := range b.dirs {
entries = append(entries, storage.Entry{Path: path, Type: storage.EntryTypeDirectory})
}
for path := range b.symlinks {
entries = append(entries, storage.Entry{Path: path, Type: storage.EntryTypeSymlink})
}
sort.Slice(entries, func(i, j int) bool {
return entries[i].Path < entries[j].Path
})
return entries
}
func emit(ctx context.Context, entry storage.Entry, opts storage.WalkOptions, fn storage.WalkFunc) error {
if opts.Limit > 0 && opts.Limit < 1 {
return nil
}
if err := ctx.Err(); err != nil {
return err
}
if err := fn(entry); err != nil && !errors.Is(err, storage.ErrStopWalk) {
return err
}
return nil
}
func entryBelow(prefix, path string) bool {
if prefix == "" {
return path != ""
}
return strings.HasPrefix(path, prefix+"/")
}
func isImmediateChild(prefix, path string) bool {
remainder := path
if prefix != "" {
remainder = strings.TrimPrefix(path, prefix+"/")
}
return !strings.Contains(remainder, "/")
}
func parentOf(path string) string {
index := strings.LastIndex(path, "/")
if index == -1 {
return ""
}
return path[:index]
}

View File

@@ -0,0 +1,187 @@
package fake
import (
"bytes"
"context"
"errors"
"io"
"reflect"
"testing"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
func TestBackendReadWriteAndStream(t *testing.T) {
backend := New()
entry, err := backend.WriteFile(context.Background(), "reports/report.md", []byte("hello"), storage.WriteOptions{})
if err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
if entry.Path != "reports/report.md" || entry.Type != storage.EntryTypeFile || entry.Size != 5 {
t.Fatalf("entry = %#v, want file metadata", entry)
}
data, err := backend.ReadFile(context.Background(), "reports/report.md")
if err != nil {
t.Fatalf("ReadFile() error = %v", err)
}
reader, err := backend.OpenReader(context.Background(), "reports/report.md")
if err != nil {
t.Fatalf("OpenReader() error = %v", err)
}
streamed, err := io.ReadAll(reader)
closeErr := reader.Close()
if err != nil || closeErr != nil {
t.Fatalf("stream read error = %v close = %v", err, closeErr)
}
if !bytes.Equal(data, streamed) {
t.Fatalf("streamed = %q, want %q", streamed, data)
}
_, err = backend.WriteFile(context.Background(), "reports/report.md", []byte("again"), storage.WriteOptions{})
if !storage.IsAlreadyExists(err) {
t.Fatalf("WriteFile existing error = %v, want already exists", err)
}
}
func TestBackendStatWalkAndList(t *testing.T) {
backend := New()
mustWrite(t, backend, "b/two.txt", "2")
mustWrite(t, backend, "a/one.txt", "1")
entry, err := backend.Stat(context.Background(), "a")
if err != nil {
t.Fatalf("Stat directory error = %v", err)
}
if entry.Type != storage.EntryTypeDirectory {
t.Fatalf("entry type = %s, want directory", entry.Type)
}
entries, err := storage.List(context.Background(), backend, "", storage.WalkOptions{Recursive: true})
if err != nil {
t.Fatalf("List() error = %v", err)
}
var paths []string
for _, entry := range entries {
paths = append(paths, entry.Path)
}
want := []string{"a", "a/one.txt", "b", "b/two.txt"}
if !reflect.DeepEqual(paths, want) {
t.Fatalf("paths = %v, want %v", paths, want)
}
}
func TestBackendRejectsInvalidPaths(t *testing.T) {
backend := New()
_, err := backend.WriteFile(context.Background(), "../outside", []byte("data"), storage.WriteOptions{})
if !storage.IsInvalidPath(err) {
t.Fatalf("WriteFile traversal error = %v, want invalid path", err)
}
_, err = backend.ReadFile(context.Background(), `bad\path`)
if !storage.IsInvalidPath(err) {
t.Fatalf("ReadFile backslash error = %v, want invalid path", err)
}
}
func TestBackendSymlinkReportingAndReadRejection(t *testing.T) {
backend := New()
if err := backend.AddSymlink("link.txt"); err != nil {
t.Fatalf("AddSymlink() error = %v", err)
}
entry, err := backend.Stat(context.Background(), "link.txt")
if err != nil {
t.Fatalf("Stat() error = %v", err)
}
if entry.Type != storage.EntryTypeSymlink {
t.Fatalf("entry type = %s, want symlink", entry.Type)
}
_, err = backend.ReadFile(context.Background(), "link.txt")
if !storage.IsUnsupported(err) {
t.Fatalf("ReadFile symlink error = %v, want unsupported", err)
}
}
func TestBackendWriteFromSizeMismatch(t *testing.T) {
backend := New()
_, err := backend.WriteFrom(context.Background(), "out.txt", bytes.NewBufferString("short"), storage.WriteOptions{SizeKnown: true, Size: 99})
if !storage.IsConflict(err) {
t.Fatalf("WriteFrom size mismatch error = %v, want conflict", err)
}
if _, err := backend.Stat(context.Background(), "out.txt"); !storage.IsNotFound(err) {
t.Fatalf("Stat after failed write error = %v, want not found", err)
}
}
func TestBackendManagedDeletion(t *testing.T) {
backend := New()
mustWrite(t, backend, "bundle/report.html", "html")
mustWrite(t, backend, "bundle/keep.txt", "keep")
mustWrite(t, backend, "bundle/.distributor.json", "{}")
err := backend.DeleteManagedBundle(context.Background(), "bundle", []string{"report.html"}, storage.DeleteOptions{PruneEmptyDirs: true})
if err != nil {
t.Fatalf("DeleteManagedBundle() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.html"); !storage.IsNotFound(err) {
t.Fatalf("managed output stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/.distributor.json"); !storage.IsNotFound(err) {
t.Fatalf("state stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/keep.txt"); err != nil {
t.Fatalf("unlisted file stat error = %v", err)
}
}
func TestBackendHasAnyAndWalkStop(t *testing.T) {
backend := New()
found, err := backend.HasAny(context.Background(), "missing")
if err != nil {
t.Fatalf("HasAny missing error = %v", err)
}
if found {
t.Fatal("HasAny missing = true, want false")
}
mustWrite(t, backend, "bundle/report.md", "report")
found, err = backend.HasAny(context.Background(), "bundle")
if err != nil {
t.Fatalf("HasAny bundle error = %v", err)
}
if !found {
t.Fatal("HasAny bundle = false, want true")
}
found, err = backend.HasAny(context.Background(), "bund")
if err != nil {
t.Fatalf("HasAny sibling prefix error = %v", err)
}
if found {
t.Fatal("HasAny sibling prefix = true, want false")
}
visited := 0
err = backend.Walk(context.Background(), "", storage.WalkOptions{Recursive: true}, func(storage.Entry) error {
visited++
return storage.ErrStopWalk
})
if err != nil {
t.Fatalf("Walk stop error = %v", err)
}
if visited != 1 {
t.Fatalf("visited = %d, want 1", visited)
}
errSentinel := errors.New("callback")
err = backend.Walk(context.Background(), "", storage.WalkOptions{Recursive: true}, func(storage.Entry) error {
return errSentinel
})
if !errors.Is(err, errSentinel) {
t.Fatalf("Walk callback error = %v, want sentinel", err)
}
}
func mustWrite(t *testing.T, backend *Backend, path, data string) {
t.Helper()
if _, err := backend.WriteFile(context.Background(), path, []byte(data), storage.WriteOptions{}); err != nil {
t.Fatalf("WriteFile(%q) error = %v", path, err)
}
}

64
internal/storage/path.go Normal file
View File

@@ -0,0 +1,64 @@
package storage
import (
"path"
"sort"
"strings"
)
const stateFileName = ".distributor.json"
func ValidatePath(value string) error {
if value == "" {
return NewError(OpValidatePath, "", value, ErrInvalidPath, nil)
}
return validateLogicalPath(value)
}
func ValidatePrefix(value string) error {
if value == "" {
return nil
}
return validateLogicalPath(value)
}
func Join(base, child string) (string, error) {
if err := ValidatePrefix(base); err != nil {
return "", err
}
if err := ValidatePath(child); err != nil {
return "", err
}
if base == "" {
return child, nil
}
return base + "/" + child, nil
}
func StatePath(bundlePath string) (string, error) {
if bundlePath == "" {
return stateFileName, nil
}
return Join(bundlePath, stateFileName)
}
func SortEntries(entries []Entry) {
sort.Slice(entries, func(i, j int) bool {
return entries[i].Path < entries[j].Path
})
}
func validateLogicalPath(value string) error {
if strings.Contains(value, "\\") || strings.HasPrefix(value, "/") {
return NewError(OpValidatePath, "", value, ErrInvalidPath, nil)
}
if path.Clean(value) != value {
return NewError(OpValidatePath, "", value, ErrInvalidPath, nil)
}
for _, segment := range strings.Split(value, "/") {
if segment == "" || segment == "." || segment == ".." {
return NewError(OpValidatePath, "", value, ErrInvalidPath, nil)
}
}
return nil
}

View File

@@ -0,0 +1,133 @@
package storage
import (
"context"
"errors"
"io"
"testing"
)
func TestValidatePath(t *testing.T) {
valid := []string{
"report.md",
"daily/report.md",
"a-b_1.2/report.html",
}
for _, path := range valid {
t.Run("valid "+path, func(t *testing.T) {
if err := ValidatePath(path); err != nil {
t.Fatalf("ValidatePath(%q) error = %v", path, err)
}
})
}
invalid := []string{
"",
"/absolute",
"../outside",
"nested/../outside",
"nested/./file",
"nested//file",
`nested\file`,
}
for _, path := range invalid {
t.Run("invalid "+path, func(t *testing.T) {
err := ValidatePath(path)
if !IsInvalidPath(err) {
t.Fatalf("ValidatePath(%q) error = %v, want invalid path", path, err)
}
})
}
}
func TestValidatePrefixAllowsRoot(t *testing.T) {
if err := ValidatePrefix(""); err != nil {
t.Fatalf("ValidatePrefix(\"\") error = %v", err)
}
if err := ValidatePrefix("a/.."); !IsInvalidPath(err) {
t.Fatalf("ValidatePrefix traversal error = %v, want invalid path", err)
}
}
func TestListSortsEntries(t *testing.T) {
backend := walkBackend{
entries: []Entry{
{Path: "z.txt", Type: EntryTypeFile},
{Path: "a.txt", Type: EntryTypeFile},
},
}
entries, err := List(context.Background(), backend, "", WalkOptions{})
if err != nil {
t.Fatalf("List() error = %v", err)
}
if got, want := []string{entries[0].Path, entries[1].Path}, []string{"a.txt", "z.txt"}; got[0] != want[0] || got[1] != want[1] {
t.Fatalf("paths = %v, want %v", got, want)
}
}
func TestTypedErrorPredicates(t *testing.T) {
err := NewError(OpReadFile, "test", "missing", ErrNotFound, errors.New("missing"))
if !IsNotFound(err) {
t.Fatalf("IsNotFound(%v) = false, want true", err)
}
if IsInvalidPath(err) {
t.Fatalf("IsInvalidPath(%v) = true, want false", err)
}
}
func TestRegistry(t *testing.T) {
registry := NewRegistry()
if err := registry.Register("test", func(context.Context, OpenConfig) (Backend, error) {
return walkBackend{}, nil
}); err != nil {
t.Fatalf("Register() error = %v", err)
}
if _, err := registry.Open(context.Background(), "test", nil); err != nil {
t.Fatalf("Open() error = %v", err)
}
if _, err := registry.Open(context.Background(), "missing", nil); !IsUnsupported(err) {
t.Fatalf("Open() error = %v, want unsupported", err)
}
}
type walkBackend struct {
entries []Entry
}
func (b walkBackend) ReadFile(context.Context, string) ([]byte, error) {
return nil, nil
}
func (b walkBackend) OpenReader(context.Context, string) (io.ReadCloser, error) {
return nil, nil
}
func (b walkBackend) WriteFile(context.Context, string, []byte, WriteOptions) (Entry, error) {
return Entry{}, nil
}
func (b walkBackend) WriteFrom(context.Context, string, io.Reader, WriteOptions) (Entry, error) {
return Entry{}, nil
}
func (b walkBackend) Stat(context.Context, string) (Entry, error) {
return Entry{}, nil
}
func (b walkBackend) Walk(_ context.Context, _ string, _ WalkOptions, fn WalkFunc) error {
for _, entry := range b.entries {
if err := fn(entry); err != nil {
return err
}
}
return nil
}
func (b walkBackend) HasAny(context.Context, string) (bool, error) {
return false, nil
}
func (b walkBackend) DeleteManagedBundle(context.Context, string, []string, DeleteOptions) error {
return nil
}

View File

@@ -0,0 +1,46 @@
package storage
import (
"context"
"sync"
)
type OpenConfig map[string]string
type Opener func(context.Context, OpenConfig) (Backend, error)
type Registry struct {
mu sync.RWMutex
openers map[string]Opener
}
func NewRegistry() *Registry {
return &Registry{openers: make(map[string]Opener)}
}
func (r *Registry) Register(name string, opener Opener) error {
if name == "" || opener == nil {
return NewError(OpRegisterBackend, name, "", ErrInvalidPath, nil)
}
r.mu.Lock()
defer r.mu.Unlock()
if _, exists := r.openers[name]; exists {
return NewError(OpRegisterBackend, name, "", ErrAlreadyExist, nil)
}
r.openers[name] = opener
return nil
}
func (r *Registry) Open(ctx context.Context, name string, cfg OpenConfig) (Backend, error) {
r.mu.RLock()
opener, ok := r.openers[name]
r.mu.RUnlock()
if !ok {
return nil, NewError(OpOpenBackend, name, "", ErrUnsupported, nil)
}
backend, err := opener(ctx, cfg)
if err != nil {
return nil, err
}
return backend, nil
}