eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Confine chunk plan storage to its cache root

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 02:03:58 +00:00
parent 8c59b6af14
commit 205e2a9908
2 changed files with 327 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
internal/framework/chunkplan/store.go
View File

@@ -2,8 +2,10 @@ package chunkplan
import (
"bytes"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"os"
@@ -16,9 +18,11 @@ import (
const SchemaVersion = pipeline.ChunkPlanSchemaVersion
const planFileName = "plan.json"
type filesystemStore struct {
root string
write func(string, []byte) error
write func(*os.Root, string, []byte) error
}
func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error) {
@@ -33,11 +37,43 @@ func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error) {
}
func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
target, err := s.planPath(sourceDigest)
digestDir, err := digestPathSegment(sourceDigest)
if err != nil {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, err
}
data, err := os.ReadFile(target)
root, err := s.openRoot(false)
if err != nil {
if os.IsNotExist(err) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
}
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("open chunk plan root: %w", err)
}
defer root.Close()
state, err := inspectDirectory(root, digestDir)
if err != nil {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan directory: %w", err)
}
if state == entryMissing {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
}
if state == entryRejected {
return invalidDecision()
}
target := planPath(digestDir)
state, err = inspectPlan(root, target)
if err != nil {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan file: %w", err)
}
if state == entryMissing {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
}
if state == entryRejected {
return invalidDecision()
}
data, err := root.ReadFile(target)
if err != nil {
if os.IsNotExist(err) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
@@ -66,7 +102,7 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
if err := validateRecord(record, record.SourceDigest); err != nil {
return fmt.Errorf("validate chunk plan record: %w", err)
}
target, err := s.planPath(record.SourceDigest)
digestDir, err := digestPathSegment(record.SourceDigest)
if err != nil {
return err
}
@@ -75,25 +111,134 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
return fmt.Errorf("encode chunk plan record: %w", err)
}
data = append(data, '\n')
root, err := s.openRoot(true)
if err != nil {
return fmt.Errorf("open chunk plan root: %w", err)
}
defer root.Close()
if err := ensureDirectory(root, digestDir); err != nil {
return fmt.Errorf("prepare chunk plan directory: %w", err)
}
target := planPath(digestDir)
state, err := inspectPlan(root, target)
if err != nil {
return fmt.Errorf("inspect chunk plan file: %w", err)
}
if state == entryRejected {
return fmt.Errorf("chunk plan file has an unsupported type")
}
writer := s.write
if writer == nil {
writer = writeAtomic
}
if err := writer(target, data); err != nil {
if err := writer(root, target, data); err != nil {
return fmt.Errorf("write chunk plan: %w", err)
}
return nil
}
func (s *filesystemStore) planPath(sourceDigest string) (string, error) {
func (s *filesystemStore) openRoot(create bool) (*os.Root, error) {
if s == nil || strings.TrimSpace(s.root) == "" {
return "", fmt.Errorf("chunk plan store must not be nil")
return nil, fmt.Errorf("chunk plan store must not be nil")
}
hexDigest, err := digestPathSegment(sourceDigest)
if create {
if err := os.MkdirAll(s.root, 0o700); err != nil {
return nil, err
}
}
root, err := os.OpenRoot(s.root)
if err != nil {
return "", err
return nil, err
}
return filepath.Join(s.root, hexDigest, "plan.json"), nil
if !create {
return root, nil
}
rootDirectory, err := root.Open(".")
if err != nil {
_ = root.Close()
return nil, err
}
defer rootDirectory.Close()
if err := rootDirectory.Chmod(0o700); err != nil {
_ = root.Close()
return nil, err
}
return root, nil
}
type entryState uint8
const (
entryPresent entryState = iota
entryMissing
entryRejected
)
func inspectDirectory(root *os.Root, digestDir string) (entryState, error) {
info, err := root.Lstat(digestDir)
if err != nil {
if os.IsNotExist(err) {
return entryMissing, nil
}
return entryPresent, err
}
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
return entryRejected, nil
}
return entryPresent, nil
}
func ensureDirectory(root *os.Root, digestDir string) error {
for {
state, err := inspectDirectory(root, digestDir)
if err != nil {
return err
}
switch state {
case entryRejected:
return fmt.Errorf("chunk plan directory has an unsupported type")
case entryMissing:
if err := root.Mkdir(digestDir, 0o700); err != nil && !errors.Is(err, os.ErrExist) {
return err
}
continue
}
directory, err := root.Open(digestDir)
if err != nil {
return err
}
info, statErr := directory.Stat()
if statErr == nil && !info.IsDir() {
statErr = fmt.Errorf("chunk plan directory has an unsupported type")
}
if statErr == nil {
statErr = directory.Chmod(0o700)
}
closeErr := directory.Close()
if statErr != nil {
return statErr
}
return closeErr
}
}
func inspectPlan(root *os.Root, target string) (entryState, error) {
info, err := root.Lstat(target)
if err != nil {
if os.IsNotExist(err) {
return entryMissing, nil
}
return entryPresent, err
}
if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
return entryRejected, nil
}
return entryPresent, nil
}
func planPath(digestDir string) string {
return digestDir + "/" + planFileName
}
func digestPathSegment(digest string) (string, error) {
@@ -184,33 +329,24 @@ type atomicWriteHooks struct {
BeforeRename func() error
}
func writeAtomic(target string, data []byte) error {
return writeAtomicWithHooks(target, data, atomicWriteHooks{})
func writeAtomic(root *os.Root, target string, data []byte) error {
return writeAtomicWithHooks(root, target, data, atomicWriteHooks{})
}
func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) error {
dir := filepath.Dir(target)
if err := os.MkdirAll(dir, 0o700); err != nil {
return err
}
if err := os.Chmod(dir, 0o700); err != nil {
return err
}
func writeAtomicWithHooks(root *os.Root, target string, data []byte, hooks atomicWriteHooks) error {
if hooks.BeforeCreateTemp != nil {
if err := hooks.BeforeCreateTemp(); err != nil {
return err
}
}
temp, err := os.CreateTemp(dir, ".plan.json.tmp-*")
temp, tempPath, err := createTemporaryFile(root, target)
if err != nil {
return err
}
tempPath := temp.Name()
removeTemp := true
defer func() {
if removeTemp {
_ = os.Remove(tempPath)
_ = root.Remove(tempPath)
}
}()
if err := temp.Chmod(0o600); err != nil {
@@ -233,9 +369,28 @@ func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) er
return err
}
}
if err := os.Rename(tempPath, target); err != nil {
if err := root.Rename(tempPath, target); err != nil {
return err
}
removeTemp = false
return nil
}
func createTemporaryFile(root *os.Root, target string) (*os.File, string, error) {
for attempt := 0; attempt < 32; attempt++ {
var suffix [16]byte
if _, err := rand.Read(suffix[:]); err != nil {
return nil, "", err
}
path := strings.TrimSuffix(target, planFileName) + ".plan.json.tmp-" + hex.EncodeToString(suffix[:])
file, err := root.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
if errors.Is(err, os.ErrExist) {
continue
}
if err != nil {
return nil, "", err
}
return file, path, nil
}
return nil, "", fmt.Errorf("create unique temporary chunk plan file")
}

151
internal/framework/chunkplan/store_test.go
View File

@@ -84,19 +84,118 @@ func TestFilesystemStorePermissions(t *testing.T) {
}
func TestFilesystemStoreMissingAndOperationalErrors(t *testing.T) {
root := t.TempDir()
root := filepath.Join(t.TempDir(), "plans")
store := newStore(t, root)
_, decision, err := store.Load(testSourceDigest)
if err != nil || decision.Status != pipeline.ChunkPlanMissing {
t.Fatalf("missing decision=%#v error=%v", decision, err)
}
if _, err := os.Stat(root); !os.IsNotExist(err) {
t.Fatalf("Load() created missing root: %v", err)
}
target := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json")
if err := os.MkdirAll(target, 0o700); err != nil {
t.Fatal(err)
}
if _, _, err := store.Load(testSourceDigest); err == nil {
t.Fatal("Load(plan.json directory) error = nil")
_, decision, err = store.Load(testSourceDigest)
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
t.Fatalf("Load(plan.json directory) decision=%#v error=%v", decision, err)
}
}
func TestFilesystemStoreRejectsSymlinkedEntries(t *testing.T) {
tests := []struct {
name string
setup func(t *testing.T, root, outside string)
}{
{
name: "digest directory",
setup: func(t *testing.T, root, outside string) {
t.Helper()
if err := os.MkdirAll(outside, 0o700); err != nil {
t.Fatal(err)
}
writeFile(t, filepath.Join(outside, "plan.json"), []byte("outside plan"), 0o640)
createSymlink(t, outside, filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:")))
},
},
{
name: "plan file",
setup: func(t *testing.T, root, outside string) {
t.Helper()
digestDir := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"))
if err := os.MkdirAll(digestDir, 0o700); err != nil {
t.Fatal(err)
}
writeFile(t, outside, []byte("outside plan"), 0o640)
createSymlink(t, outside, filepath.Join(digestDir, "plan.json"))
},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
root := filepath.Join(t.TempDir(), "plans")
if err := os.MkdirAll(root, 0o700); err != nil {
t.Fatal(err)
}
outside := filepath.Join(t.TempDir(), "outside")
tc.setup(t, root, outside)
before := readFile(t, outsidePlanPath(tc.name, outside))
beforeMode := fileMode(t, outsidePlanPath(tc.name, outside))
store := newStore(t, root)
_, decision, err := store.Load(testSourceDigest)
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
t.Fatalf("Load() decision=%#v error=%v", decision, err)
}
if err := store.Save(testRecord(t, 1)); err == nil {
t.Fatal("Save() error = nil")
}
outsidePlan := outsidePlanPath(tc.name, outside)
if got := readFile(t, outsidePlan); !bytes.Equal(got, before) {
t.Fatalf("outside content = %q, want %q", got, before)
}
if got := fileMode(t, outsidePlan); got != beforeMode {
t.Fatalf("outside mode = %04o, want %04o", got, beforeMode)
}
})
}
}
func TestFilesystemStoreRejectsUnexpectedEntryTypes(t *testing.T) {
tests := []struct {
name string
setup func(t *testing.T, root string)
}{
{
name: "digest file",
setup: func(t *testing.T, root string) {
writeFile(t, filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:")), []byte("not a directory"), 0o600)
},
},
{
name: "plan directory",
setup: func(t *testing.T, root string) {
if err := os.MkdirAll(filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json"), 0o700); err != nil {
t.Fatal(err)
}
},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
root := t.TempDir()
tc.setup(t, root)
store := newStore(t, root)
_, decision, err := store.Load(testSourceDigest)
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
t.Fatalf("Load() decision=%#v error=%v", decision, err)
}
if err := store.Save(testRecord(t, 1)); err == nil {
t.Fatal("Save() error = nil")
}
})
}
}
@@ -295,7 +394,9 @@ func TestFilesystemStoreInterruptedWritesPreservePreviousRecord(t *testing.T) {
{name: "before rename", hooks: atomicWriteHooks{BeforeRename: func() error { return errors.New("interrupted before rename") }}},
} {
t.Run(tc.name, func(t *testing.T) {
store.write = func(target string, data []byte) error { return writeAtomicWithHooks(target, data, tc.hooks) }
store.write = func(root *os.Root, target string, data []byte) error {
return writeAtomicWithHooks(root, target, data, tc.hooks)
}
if err := store.Save(testRecord(t, 2)); err == nil {
t.Fatal("Save() error = nil")
}
@@ -363,3 +464,45 @@ func assertNoTemps(t *testing.T, dir string) {
}
}
}
func createSymlink(t *testing.T, target, link string) {
t.Helper()
if err := os.Symlink(target, link); err != nil {
t.Skipf("create symlink: %v", err)
}
}
func writeFile(t *testing.T, path string, data []byte, mode os.FileMode) {
t.Helper()
if err := os.WriteFile(path, data, mode); err != nil {
t.Fatal(err)
}
if err := os.Chmod(path, mode); err != nil {
t.Fatal(err)
}
}
func readFile(t *testing.T, path string) []byte {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
return data
}
func fileMode(t *testing.T, path string) os.FileMode {
t.Helper()
info, err := os.Stat(path)
if err != nil {
t.Fatal(err)
}
return info.Mode().Perm()
}
func outsidePlanPath(name, outside string) string {
if name == "digest directory" {
return filepath.Join(outside, "plan.json")
}
return outside
}