From 9900211fa430f1424e33b936e00e0b6d3cba0f1b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 19:16:18 +0000 Subject: [PATCH] Confine publish archive reads --- docs/internal/stage-publish.md | 20 +- docs/operations.md | 11 +- docs/roadmap/implementation.md | 2 +- internal/adapters/storage/fake.go | 29 +- internal/adapters/storage/object_store.go | 7 + internal/adapters/storage/s3_backend.go | 29 +- internal/app/commands_test.go | 18 +- internal/app/post_publish_cleanup_test.go | 8 + internal/app/runner.go | 1 + internal/artifacts/store.go | 1 + internal/fileops/confined_read.go | 78 ++++++ internal/manifest/manifest.go | 1 + internal/manifest/store.go | 23 +- internal/stage/extract_test.go | 4 +- internal/stage/placeholders_test.go | 2 +- internal/stage/publish.go | 321 ++++++++++++++-------- internal/stage/publish_test.go | 251 ++++++++++++++++- internal/stage/run_local.go | 1 + 18 files changed, 655 insertions(+), 152 deletions(-) create mode 100644 internal/fileops/confined_read.go diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 9293e08..685a034 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -26,10 +26,14 @@ Exact remote placement and the operator workflow belong in - stage can self-skip when publish disabled or run upload disabled. - validates prerequisite stage success and object-store availability. -- collects a deterministic run file list plus run `manifest.json`, excluding - `audio/**` and the run-local `extract/notarius-output/**` staging bundle. -- keeps run-local Notarius receipt and stderr diagnostics eligible for the run - archive. +- derives a deterministic run-archive allowlist from the validated run + `manifest.json`: declared run-local outputs, logs, generated configs, and the + manifest itself. Unlisted workspace files are not archive candidates. +- opens each archive candidate beneath its archive root without following + symlinked ancestors or leaf entries, verifies that it is a regular file and + checks a declared checksum when present, then streams the opened descriptor. +- derives the durable previous-cache archive from its validated manifest using + the same confinement and regular-file checks. - resolves publish output sources through runtime artifact catalog and manifest-aware resolution. - publishes extraction lanes only through explicit configured output rules; neither run-local nor durable Notarius bundles are scanned or uploaded wholesale. @@ -53,9 +57,11 @@ Includes counts/lists for: ## Invariants - `current/run_id.txt` is the remote commit marker and is written last. -- run upload excludes `audio/**` and `extract/notarius-output/**`. -- `extract/notarius.receipt.json` and `extract/notarius.stderr.log` remain - eligible run-record diagnostics. +- run and previous uploads contain only manifest-declared regular files opened + from verified descriptors; symlinks, special files, replacement races, and + undeclared entries are rejected or ignored before uploads begin. +- run-local diagnostics, including Notarius receipt and stderr files, are + archived only when recorded by the run manifest. - publish locks are not overridden by `--force`. The commit boundary and cleanup gate are normative architecture invariants; see diff --git a/docs/operations.md b/docs/operations.md index 721b88a..1d5be24 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -160,11 +160,12 @@ Run-local diagnostics are: - `runs/{run_id}/extract/notarius.stderr.log` - `runs/{run_id}/extract/notarius-output/` before durable promotion -The run-record upload excludes the complete -`extract/notarius-output/**` subtree. The receipt and stderr files remain -eligible run-record diagnostics. The durable bundle is never scanned for -implicit publication; only lanes named by explicit `pipeline.publish.outputs` -rules are uploaded. +The run-record upload is an allowlist derived from the validated run manifest, +not a workspace scan. Each declared source is opened without following +symlinked ancestors or the leaf, verified as a regular file, and streamed from +that verified descriptor. Unlisted files and unsafe entries are never uploaded. +The durable bundle is never scanned for implicit publication; only lanes named +by explicit `pipeline.publish.outputs` rules are uploaded. To intentionally replace the current extraction result, run: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 3895e8b..aa3512e 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -25,7 +25,7 @@ All stages are pending when this plan is created. | 7 | Bound and verify external result acquisition | RSK-013, TST-007 | Completed | | 8 | Terminate owned subprocess trees | RSK-011 | Completed | | 9 | Redact and cap subprocess diagnostics | RSK-012 | Completed | -| 10 | Confine publish archive reads | COR-005 | Pending | +| 10 | Confine publish archive reads | COR-005 | Completed | | 11 | Make manifest and run identity singular | COR-001, TST-006 | Pending | | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Pending | | 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Pending | diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index 30187c2..bce4ceb 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -142,7 +142,34 @@ func (f *FakeBackend) Upload(ctx context.Context, localPath, key string, opts Up return ObjectInfo{}, fmt.Errorf("upload object: key is required") } - data, err := os.ReadFile(localPath) + file, err := os.Open(localPath) + if err != nil { + return ObjectInfo{}, fmt.Errorf("upload object %q from %q: %w", key, localPath, err) + } + defer file.Close() + return f.uploadReader(ctx, file, key, opts, localPath) +} + +// UploadReader stores content provided by a caller-owned reader. +func (f *FakeBackend) UploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions) (ObjectInfo, error) { + return f.uploadReader(ctx, source, key, opts, "reader") +} + +func (f *FakeBackend) uploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions, localPath string) (ObjectInfo, error) { + if err := ctx.Err(); err != nil { + return ObjectInfo{}, err + } + if f.UploadErr != nil { + return ObjectInfo{}, f.UploadErr + } + if source == nil { + return ObjectInfo{}, fmt.Errorf("upload object: source is required") + } + if strings.TrimSpace(key) == "" { + return ObjectInfo{}, fmt.Errorf("upload object: key is required") + } + + data, err := io.ReadAll(source) if err != nil { return ObjectInfo{}, fmt.Errorf("upload object %q from %q: %w", key, localPath, err) } diff --git a/internal/adapters/storage/object_store.go b/internal/adapters/storage/object_store.go index 13ee97a..16cbb4b 100644 --- a/internal/adapters/storage/object_store.go +++ b/internal/adapters/storage/object_store.go @@ -2,9 +2,16 @@ package storage import ( "context" + "io" "time" ) +// ReaderUploader streams caller-owned, already-opened content to object storage. +// Callers retain source-selection and filesystem-confinement policy. +type ReaderUploader interface { + UploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions) (ObjectInfo, error) +} + // ObjectStore is a remote object storage boundary used by prepare, restore, and publish work. // // Key invariant: diff --git a/internal/adapters/storage/s3_backend.go b/internal/adapters/storage/s3_backend.go index bf517a2..3e16543 100644 --- a/internal/adapters/storage/s3_backend.go +++ b/internal/adapters/storage/s3_backend.go @@ -214,11 +214,27 @@ func (b *S3Backend) Upload(ctx context.Context, localPath, key string, opts Uplo if err != nil { return ObjectInfo{}, fmt.Errorf("upload object %q from %q: stat local file: %w", normalizedKey, localPath, err) } + return b.uploadReader(ctx, file, key, opts, stat.Size()) +} + +// UploadReader sends caller-owned content to key. +func (b *S3Backend) UploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions) (ObjectInfo, error) { + return b.uploadReader(ctx, source, key, opts, 0) +} + +func (b *S3Backend) uploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions, size int64) (ObjectInfo, error) { + normalizedKey := normalizeObjectKey(key) + if source == nil { + return ObjectInfo{}, fmt.Errorf("upload object: source is required") + } + if normalizedKey == "" { + return ObjectInfo{}, fmt.Errorf("upload object: key is required") + } input := &s3.PutObjectInput{ Bucket: &b.bucket, Key: &normalizedKey, - Body: file, + Body: source, Metadata: copyMetadata(opts.Metadata), } if strings.TrimSpace(opts.ContentType) != "" { @@ -228,14 +244,17 @@ func (b *S3Backend) Upload(ctx context.Context, localPath, key string, opts Uplo resp, err := b.client.PutObject(ctx, input) if err != nil { - return ObjectInfo{}, fmt.Errorf("upload object %q from %q: %w", normalizedKey, localPath, err) + return ObjectInfo{}, fmt.Errorf("upload object %q: %w", normalizedKey, err) } - return ObjectInfo{ + info := ObjectInfo{ Key: normalizedKey, - Size: stat.Size(), ETag: strings.Trim(valueOrEmpty(resp.ETag), "\""), - }, nil + } + if size > 0 { + info.Size = size + } + return info, nil } // Exists checks whether one object key exists. diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index a1da021..4382545 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -476,9 +476,6 @@ func writeValidConfigFiles(t *testing.T, workspaceRoot string, transcribeURL ... seriatimBinary := writeSeriatimAppTestWrapper(t) scriptoriumBinary := writeScriptoriumAppTestWrapper(t) auditaBinary := writeAuditaAppTestWrapper(t) - t.Setenv("GO_WANT_APP_SERIATIM_HELPER", "1") - t.Setenv("GO_WANT_APP_SCRIPTORIUM_HELPER", "1") - t.Setenv("GO_WANT_APP_AUDITA_HELPER", "1") t.Setenv("AUDITA_LLM_API_KEY", "test-audita-key") t.Setenv("PATH", filepath.Dir(scriptoriumBinary)+string(os.PathListSeparator)+os.Getenv("PATH")) @@ -623,7 +620,7 @@ func writeScriptoriumAppTestWrapper(t *testing.T) string { } func TestScriptoriumAppHelper(t *testing.T) { - if os.Getenv("GO_WANT_APP_SCRIPTORIUM_HELPER") != "1" { + if !appHelperInvocation() { return } @@ -663,7 +660,7 @@ func TestScriptoriumAppHelper(t *testing.T) { } func TestSeriatimAppHelper(t *testing.T) { - if os.Getenv("GO_WANT_APP_SERIATIM_HELPER") != "1" { + if !appHelperInvocation() { return } @@ -725,7 +722,7 @@ func writeAuditaAppTestWrapper(t *testing.T) string { } func TestAuditaAppHelper(t *testing.T) { - if os.Getenv("GO_WANT_APP_AUDITA_HELPER") != "1" { + if !appHelperInvocation() { return } @@ -779,6 +776,15 @@ func TestAuditaAppHelper(t *testing.T) { os.Exit(0) } +func appHelperInvocation() bool { + for _, arg := range os.Args { + if arg == "--" { + return true + } + } + return false +} + func appSeriatimFlagValue(args []string, name string) string { for i := 0; i < len(args)-1; i++ { if args[i] == name { diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index f11fd83..76d8df7 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -3,6 +3,7 @@ package app import ( "context" "errors" + "io" "os" "path/filepath" "strings" @@ -402,6 +403,13 @@ func (s *failKeyStore) Upload(ctx context.Context, localPath, key string, opts s return s.delegate.Upload(ctx, localPath, key, opts) } +func (s *failKeyStore) UploadReader(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { + if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + return storage.ObjectInfo{}, errors.New("forced upload failure") + } + return s.delegate.UploadReader(ctx, source, key, opts) +} + func (s *failKeyStore) Exists(ctx context.Context, key string) (bool, error) { return s.delegate.Exists(ctx, key) } diff --git a/internal/app/runner.go b/internal/app/runner.go index 55f436f..49a5f18 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -483,6 +483,7 @@ func mapResultOutputs(stageName string, result *stage.StageResult, runID string) Kind: kind, SourceID: sourceID, LocalPath: localPath, + ArchivePath: ref.ArchivePath, Contract: cloneContractMetadata(ref.Contract), ExternalProvenance: cloneExternalProvenance(ref.ExternalProvenance), ProducerRunID: runID, diff --git a/internal/artifacts/store.go b/internal/artifacts/store.go index d2d2fd2..2572ced 100644 --- a/internal/artifacts/store.go +++ b/internal/artifacts/store.go @@ -15,6 +15,7 @@ type Ref struct { SessionID string RelativePath string AbsolutePath string + ArchivePath string RemoteKey string Checksum string Contract *artifactmodel.ContractMetadata diff --git a/internal/fileops/confined_read.go b/internal/fileops/confined_read.go new file mode 100644 index 0000000..7d08463 --- /dev/null +++ b/internal/fileops/confined_read.go @@ -0,0 +1,78 @@ +package fileops + +import ( + "fmt" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +// OpenConfinedRegularFile opens a declared file beneath root without following +// symlinked ancestors or leaf entries. The returned descriptor remains valid if +// the pathname is later replaced. +func OpenConfinedRegularFile(rootPath, relativePath string) (*os.File, error) { + rootPath = filepath.Clean(strings.TrimSpace(rootPath)) + if rootPath == "" || rootPath == "." { + return nil, fmt.Errorf("source root is required") + } + relativePath, err := pathsafe.NormalizeRelativeDestination(relativePath) + if err != nil { + return nil, fmt.Errorf("source path: %w", err) + } + + root, err := openConfinedDirectory(rootPath) + if err != nil { + return nil, fmt.Errorf("open source root: %w", err) + } + defer root.Close() + + parts := strings.Split(filepath.FromSlash(relativePath), string(filepath.Separator)) + parent := root + for _, part := range parts[:len(parts)-1] { + child, err := openConfinedChild(parent, part, false, 0) + if err != nil { + if parent != root { + _ = parent.Close() + } + return nil, fmt.Errorf("open source ancestor %q: %w", part, err) + } + if parent != root { + _ = parent.Close() + } + parent = child + } + if parent != root { + defer parent.Close() + } + + name := parts[len(parts)-1] + declared, err := parent.Lstat(name) + if err != nil { + return nil, fmt.Errorf("inspect source %q: %w", relativePath, err) + } + if declared.Mode()&os.ModeSymlink != 0 || !declared.Mode().IsRegular() { + return nil, fmt.Errorf("source %q is not a regular file", relativePath) + } + + file, err := parent.Open(name) + if err != nil { + return nil, fmt.Errorf("open source %q: %w", relativePath, err) + } + opened, err := file.Stat() + if err != nil { + _ = file.Close() + return nil, fmt.Errorf("inspect opened source %q: %w", relativePath, err) + } + current, err := parent.Lstat(name) + if err != nil || current.Mode()&os.ModeSymlink != 0 || !current.Mode().IsRegular() || + !opened.Mode().IsRegular() || !os.SameFile(opened, declared) || !os.SameFile(opened, current) { + _ = file.Close() + if err != nil { + return nil, fmt.Errorf("reinspect source %q: %w", relativePath, err) + } + return nil, fmt.Errorf("source %q changed while being opened", relativePath) + } + return file, nil +} diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index f9c9d9a..ac60b4c 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -33,6 +33,7 @@ type ArtifactRecord struct { Kind string `json:"kind"` SourceID string `json:"source_id,omitempty"` LocalPath string `json:"local_path"` + ArchivePath string `json:"archive_path,omitempty"` Contract *artifactmodel.ContractMetadata `json:"contract,omitempty"` ExternalProvenance *artifactmodel.ExternalProvenance `json:"external_provenance,omitempty"` // ProducerRunID identifies the run that produced this durable artifact. diff --git a/internal/manifest/store.go b/internal/manifest/store.go index 9c28752..da9f19e 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -149,18 +149,35 @@ func (s *LocalStore) LoadRun(ctx context.Context, path string) (*RunManifest, er return nil, fmt.Errorf("load run manifest: path is required") } - data, err := os.ReadFile(path) + file, err := os.Open(path) if err != nil { return nil, fmt.Errorf("load run manifest %q: %w", path, err) } + defer file.Close() + return s.LoadRunReader(ctx, file) +} + +// LoadRunReader reads and validates a run manifest from a caller-owned reader. +func (s *LocalStore) LoadRunReader(ctx context.Context, source io.Reader) (*RunManifest, error) { + if err := checkContext(ctx); err != nil { + return nil, err + } + if source == nil { + return nil, fmt.Errorf("load run manifest: source is required") + } + + data, err := io.ReadAll(source) + if err != nil { + return nil, fmt.Errorf("read run manifest: %w", err) + } var m RunManifest if err := json.Unmarshal(data, &m); err != nil { - return nil, fmt.Errorf("decode run manifest %q: %w", path, err) + return nil, fmt.Errorf("decode run manifest: %w", err) } if err := validateLoadedRunManifest(&m); err != nil { - return nil, fmt.Errorf("run manifest %q invalid: %w", path, err) + return nil, fmt.Errorf("run manifest invalid: %w", err) } normalizeRunManifest(&m) diff --git a/internal/stage/extract_test.go b/internal/stage/extract_test.go index 718e8a7..83313e8 100644 --- a/internal/stage/extract_test.go +++ b/internal/stage/extract_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "os" "path/filepath" "reflect" @@ -318,10 +319,9 @@ func TestExtractStageAdapterResultIsImmediatelyReusableAndCatalogVisible(t *test if err := os.WriteFile(receiptFixture, receipt, 0o644); err != nil { t.Fatalf("WriteFile(receipt fixture) error = %v", err) } - if err := os.WriteFile(env.Config.Pipeline.Notarius.Binary, []byte("#!/bin/sh\ncat \"$NARRATIO_NOTARIUS_RECEIPT_FIXTURE\"\n"), 0o755); err != nil { + if err := os.WriteFile(env.Config.Pipeline.Notarius.Binary, []byte(fmt.Sprintf("#!/bin/sh\ncat %q\n", receiptFixture)), 0o755); err != nil { t.Fatalf("WriteFile(notarius helper) error = %v", err) } - t.Setenv("NARRATIO_NOTARIUS_RECEIPT_FIXTURE", receiptFixture) env.Notarius = notarius.NewSubprocessRunner() result, err := (extractStage{}).Run(context.Background(), env, m) diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 9dcce14..7c96b0f 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -125,7 +125,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { t.Fatalf("mkdir workdir inputs: %v", err) } writeStageTestFile(t, filepath.Join(m.LocalWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n") - writeStageTestFile(t, filepath.Join(m.LocalWorkDir, "manifest.json"), "{}\n") + writePublishRunManifest(t, m.LocalWorkDir, m.SessionID, m.Campaign, m.RunID, nil) for _, s := range stages { result, err := s.Run(context.Background(), env, m) if err != nil { diff --git a/internal/stage/publish.go b/internal/stage/publish.go index 907b94c..702b049 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -2,10 +2,12 @@ package stage import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" - "io/fs" + "io" "os" "path/filepath" "sort" @@ -16,13 +18,16 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" ) type publishStage struct{} type publishUploadFile struct { - RelativePath string - LocalPath string + RelativePath string + SourceRelativePath string + Checksum string } var publishPrerequisiteStages = []string{ @@ -85,11 +90,11 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("publish: resolve run root: %w", err) } - runRootInfo, err := os.Stat(runRoot) + runRootInfo, err := os.Lstat(runRoot) if err != nil { return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err) } - if !runRootInfo.IsDir() { + if runRootInfo.Mode()&os.ModeSymlink != 0 || !runRootInfo.IsDir() { return nil, fmt.Errorf("publish: run root %q is not a directory", runRoot) } @@ -110,20 +115,45 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("publish: run id is required") } - manifestSource, err := resolvePublishRunManifestSource(runRoot) + runManifestFile, err := fileops.OpenConfinedRegularFile(runRoot, "manifest.json") if err != nil { - return nil, fmt.Errorf("publish: resolve run manifest source: %w", err) + return nil, fmt.Errorf("publish: open run manifest: %w", err) + } + defer runManifestFile.Close() + runManifest, err := (&manifest.LocalStore{}).LoadRunReader(ctx, runManifestFile) + if err != nil { + return nil, fmt.Errorf("publish: load run manifest: %w", err) + } + if _, err := runManifestFile.Seek(0, io.SeekStart); err != nil { + return nil, fmt.Errorf("publish: rewind run manifest: %w", err) } - runFiles, err := collectPublishRunFiles(runRoot, manifestSource) + runFiles, err := collectPublishRunFiles(runRoot, runManifest) if err != nil { return nil, fmt.Errorf("publish: collect run files: %w", err) } + runSources, err := openPublishRunFiles(runRoot, runFiles, runManifestFile) + if err != nil { + return nil, fmt.Errorf("publish: verify run files: %w", err) + } + defer closePublishSources(runSources) + uploader, ok := env.ObjectStore.(storage.ReaderUploader) + if !ok { + return nil, fmt.Errorf("publish: object store does not support verified source uploads") + } sessionPaths := publishSessionPaths(env, m) - previousFiles, err := collectPublishPreviousFiles(sessionPaths.PreviousDir) + previousFiles, previousManifestFile, err := collectPublishPreviousFiles(ctx, sessionPaths.PreviousDir) if err != nil { return nil, fmt.Errorf("publish: collect previous files: %w", err) } + if previousManifestFile != nil { + defer previousManifestFile.Close() + } + previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles, previousManifestFile) + if err != nil { + return nil, fmt.Errorf("publish: verify previous files: %w", err) + } + defer closePublishSources(previousSources) runtimeCatalog, err := buildPublishRuntimeArtifactCatalog( sessionPaths, m, @@ -146,9 +176,9 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S return nil, fmt.Errorf("publish: resolve publish output rules: %w", err) } runUploaded := make([]string, 0, len(runFiles)) - for _, file := range runFiles { + for index, file := range runFiles { key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath) - if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil { + if _, err := uploader.UploadReader(ctx, runSources[index], key, storage.UploadOptions{}); err != nil { return nil, fmt.Errorf("publish: upload run file %q to %q: %w", file.RelativePath, key, err) } runUploaded = append(runUploaded, file.RelativePath) @@ -164,9 +194,9 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } previousUploaded := make([]string, 0, len(previousFiles)) - for _, file := range previousFiles { + for index, file := range previousFiles { key := artifacts.S3PublishedOutputKey(sessionPrefix, file.RelativePath) - if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil { + if _, err := uploader.UploadReader(ctx, previousSources[index], key, storage.UploadOptions{}); err != nil { return nil, fmt.Errorf("publish: upload previous file %q to %q: %w", file.RelativePath, key, err) } previousUploaded = append(previousUploaded, file.RelativePath) @@ -567,131 +597,206 @@ func resolveConfiguredArtifactLocalPath(paths artifacts.SessionPaths, configured return filepath.Join(paths.Root, rel), nil } -func collectPublishRunFiles(runRoot, manifestPath string) ([]publishUploadFile, error) { - files := make([]publishUploadFile, 0, 64) - err := filepath.WalkDir(runRoot, func(path string, d fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr - } - if path == runRoot { +func collectPublishRunFiles(runRoot string, runManifest *manifest.RunManifest) ([]publishUploadFile, error) { + if runManifest == nil { + return nil, fmt.Errorf("run manifest is required") + } + files := map[string]publishUploadFile{ + "manifest.json": {RelativePath: "manifest.json", SourceRelativePath: "manifest.json"}, + } + add := func(path, checksum string) error { + if strings.TrimSpace(path) == "" { return nil } - rel, err := filepath.Rel(runRoot, path) + rel, err := pathsafe.SlashRelativeFromRoot(runRoot, path) if err != nil { - return fmt.Errorf("relative path from %q to %q: %w", runRoot, path, err) + return nil } - rel = filepath.ToSlash(rel) - if publishRunPathExcluded(rel) { - if d.IsDir() { - return filepath.SkipDir + candidate := publishUploadFile{RelativePath: rel, SourceRelativePath: rel, Checksum: strings.TrimSpace(checksum)} + if prior, exists := files[rel]; exists && prior.Checksum != "" && candidate.Checksum != "" && prior.Checksum != candidate.Checksum { + return fmt.Errorf("source %q has conflicting declared checksums", rel) + } + if prior, exists := files[rel]; exists && prior.Checksum != "" { + candidate.Checksum = prior.Checksum + } + files[rel] = candidate + return nil + } + for _, record := range runManifest.Stages { + if record == nil { + continue + } + for _, output := range record.Outputs { + if err := add(output.ArchivePath, output.Checksum); err != nil { + return nil, err + } + if output.ArchivePath == "" { + if err := add(output.LocalPath, output.Checksum); err != nil { + return nil, err + } } - return nil } - if d.IsDir() { - return nil + for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { + if err := add(path, ""); err != nil { + return nil, err + } } - files = append(files, publishUploadFile{ - RelativePath: rel, - LocalPath: path, - }) - return nil - }) - if err != nil { - return nil, fmt.Errorf("walk %q: %w", runRoot, err) } - - manifestInfo, err := os.Stat(manifestPath) - if err != nil { - if os.IsNotExist(err) { - return nil, fmt.Errorf("manifest.json not found (checked path %q)", manifestPath) - } - return nil, fmt.Errorf("stat %q: %w", manifestPath, err) - } - if manifestInfo.IsDir() { - return nil, fmt.Errorf("manifest path %q is a directory", manifestPath) - } - files = append(files, publishUploadFile{ - RelativePath: "manifest.json", - LocalPath: manifestPath, - }) - seen := map[string]publishUploadFile{} + out := make([]publishUploadFile, 0, len(files)) for _, file := range files { - seen[file.RelativePath] = file + out = append(out, file) } - files = files[:0] - for _, file := range seen { - files = append(files, file) - } - - sort.Slice(files, func(i, j int) bool { - return files[i].RelativePath < files[j].RelativePath - }) - return files, nil + sort.Slice(out, func(i, j int) bool { return out[i].RelativePath < out[j].RelativePath }) + return out, nil } -func publishRunPathExcluded(rel string) bool { - return rel == "audio" || strings.HasPrefix(rel, "audio/") || - rel == "extract/notarius-output" || strings.HasPrefix(rel, "extract/notarius-output/") -} - -func collectPublishPreviousFiles(previousDir string) ([]publishUploadFile, error) { +func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]publishUploadFile, *os.File, error) { previousDir = filepath.Clean(strings.TrimSpace(previousDir)) - if previousDir == "" { - return nil, fmt.Errorf("previous directory is required") + if previousDir == "" || previousDir == "." { + return nil, nil, fmt.Errorf("previous directory is required") + } + info, err := os.Lstat(previousDir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil, nil } - info, err := os.Stat(previousDir) if err != nil { - if os.IsNotExist(err) { - return nil, nil - } - return nil, fmt.Errorf("stat %q: %w", previousDir, err) + return nil, nil, fmt.Errorf("inspect previous directory: %w", err) } - if !info.IsDir() { - return nil, fmt.Errorf("previous path %q is not a directory", previousDir) + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return nil, nil, fmt.Errorf("previous directory is not a regular directory") } - files := make([]publishUploadFile, 0, 16) - err = filepath.WalkDir(previousDir, func(path string, d fs.DirEntry, walkErr error) error { - if walkErr != nil { - return walkErr + manifestFile, err := fileops.OpenConfinedRegularFile(previousDir, "manifest.json") + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil, nil } - if d.IsDir() { + return nil, nil, fmt.Errorf("open previous manifest: %w", err) + } + previousManifest, err := (&manifest.LocalStore{}).LoadReader(ctx, manifestFile) + if err != nil { + _ = manifestFile.Close() + return nil, nil, fmt.Errorf("load previous manifest: %w", err) + } + if _, err := manifestFile.Seek(0, io.SeekStart); err != nil { + _ = manifestFile.Close() + return nil, nil, fmt.Errorf("rewind previous manifest: %w", err) + } + + files := map[string]publishUploadFile{ + "manifest.json": {RelativePath: filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, "manifest.json")), SourceRelativePath: "manifest.json"}, + } + add := func(path, checksum string) error { + if strings.TrimSpace(path) == "" { return nil } - rel, err := filepath.Rel(previousDir, path) + rel, err := pathsafe.SlashRelativeFromRoot(previousDir, path) if err != nil { - return fmt.Errorf("relative path from %q to %q: %w", previousDir, path, err) + return nil } - rel = filepath.ToSlash(rel) - files = append(files, publishUploadFile{ - RelativePath: filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, rel)), - LocalPath: path, - }) + archivePath := filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, rel)) + candidate := publishUploadFile{RelativePath: archivePath, SourceRelativePath: rel, Checksum: strings.TrimSpace(checksum)} + if prior, exists := files[rel]; exists && prior.Checksum != "" && candidate.Checksum != "" && prior.Checksum != candidate.Checksum { + return fmt.Errorf("source %q has conflicting declared checksums", rel) + } + if prior, exists := files[rel]; exists && prior.Checksum != "" { + candidate.Checksum = prior.Checksum + } + files[rel] = candidate return nil - }) - if err != nil { - return nil, fmt.Errorf("walk %q: %w", previousDir, err) } - - sort.Slice(files, func(i, j int) bool { - return files[i].RelativePath < files[j].RelativePath - }) - return files, nil + for _, artifact := range previousManifest.Artifacts { + if err := add(artifact.ArchivePath, artifact.Checksum); err != nil { + _ = manifestFile.Close() + return nil, nil, err + } + if artifact.ArchivePath == "" { + if err := add(artifact.LocalPath, artifact.Checksum); err != nil { + _ = manifestFile.Close() + return nil, nil, err + } + } + } + for _, record := range previousManifest.Stages { + if record == nil { + continue + } + for _, artifact := range record.Outputs { + if err := add(artifact.ArchivePath, artifact.Checksum); err != nil { + _ = manifestFile.Close() + return nil, nil, err + } + if artifact.ArchivePath == "" { + if err := add(artifact.LocalPath, artifact.Checksum); err != nil { + _ = manifestFile.Close() + return nil, nil, err + } + } + } + for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { + if err := add(path, ""); err != nil { + _ = manifestFile.Close() + return nil, nil, err + } + } + } + out := make([]publishUploadFile, 0, len(files)) + for _, file := range files { + out = append(out, file) + } + sort.Slice(out, func(i, j int) bool { return out[i].RelativePath < out[j].RelativePath }) + return out, manifestFile, nil } -func resolvePublishRunManifestSource(runRoot string) (string, error) { - path := filepath.Join(filepath.Clean(runRoot), "manifest.json") - info, err := os.Stat(path) - if err != nil { - if os.IsNotExist(err) { - return "", fmt.Errorf("manifest.json not found in run root %q", runRoot) +func openPublishRunFiles(runRoot string, files []publishUploadFile, manifestFile *os.File) ([]*os.File, error) { + sources := make([]*os.File, 0, len(files)) + for _, candidate := range files { + if manifestFile == nil { + closePublishSources(sources) + return nil, fmt.Errorf("archive manifest is required") } - return "", fmt.Errorf("stat %q: %w", path, err) + file := manifestFile + if candidate.SourceRelativePath != "manifest.json" { + var err error + file, err = fileops.OpenConfinedRegularFile(runRoot, candidate.SourceRelativePath) + if err != nil { + closePublishSources(sources) + return nil, err + } + } + if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil { + if file != manifestFile { + _ = file.Close() + } + closePublishSources(sources) + return nil, err + } + sources = append(sources, file) } - if info.IsDir() { - return "", fmt.Errorf("manifest path %q is a directory", path) + return sources, nil +} + +func verifyPublishSourceChecksum(file *os.File, relativePath, expected string) error { + if strings.TrimSpace(expected) == "" { + return nil + } + hash := sha256.New() + if _, err := io.Copy(hash, file); err != nil { + return fmt.Errorf("read source %q for checksum: %w", relativePath, err) + } + if _, err := file.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("rewind source %q after checksum: %w", relativePath, err) + } + if actual := hex.EncodeToString(hash.Sum(nil)); !strings.EqualFold(actual, expected) { + return fmt.Errorf("source %q does not match its declared checksum", relativePath) + } + return nil +} + +func closePublishSources(sources []*os.File) { + for _, source := range sources { + _ = source.Close() } - return path, nil } func directoryExists(path string) (bool, error) { diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index db4539f..0470bc7 100644 --- a/internal/stage/publish_test.go +++ b/internal/stage/publish_test.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "errors" + "io" "os" "path/filepath" "reflect" + "runtime" "sort" "strings" "testing" @@ -16,6 +18,7 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) @@ -96,10 +99,6 @@ func TestPublishUploadsRunRecordPublishedOutputsAndCurrentPointer(t *testing.T) sessionPrefix := m.S3SessionPrefix wantRunUploads := []string{ "analyze/outputs/artifacts/session_recap.md", - "archive/notarius-output/keep.json", - "extract/notarius-output-copy/keep.json", - "extract/notarius.receipt.json", - "extract/notarius.stderr.log", "merge/config/seriatim.generated.yml", "prepare/inputs/session.yml", "prepare/outputs/audio/speaker.flac", @@ -189,6 +188,7 @@ func TestPublishUploadsPreviousCacheWhenPresent(t *testing.T) { ) writeStageTestFile(t, filepath.Join(sessionRoot, "previous", "manifest.json"), "{\"session_id\":\"2026-04-12\"}\n") writeStageTestFile(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n") + writePublishPreviousManifest(t, filepath.Join(sessionRoot, "previous"), "2026-04-12", filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md")) result, err := publishStage{}.Run(context.Background(), env, m) if err != nil { @@ -220,6 +220,178 @@ func TestPublishToleratesMissingPreviousCache(t *testing.T) { } } +func TestPublishRejectsDeclaredLeafSymlinkBeforeUploading(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation requires privileges on Windows") + } + env, m, runRoot := publishFixture(t) + sentinel := filepath.Join(t.TempDir(), "outside.txt") + writeStageTestFile(t, sentinel, "outside secret\n") + target := filepath.Join(runRoot, "logs", "audita.stderr.log") + if err := os.Remove(target); err != nil { + t.Fatal(err) + } + if err := os.Symlink(sentinel, target); err != nil { + t.Fatal(err) + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), `source "logs/audita.stderr.log" is not a regular file`) { + t.Fatalf("Run() error = %v, want rejected symlink", err) + } + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside secret") +} + +func TestPublishRejectsSymlinkedRunManifestBeforeUploading(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation requires privileges on Windows") + } + env, m, runRoot := publishFixture(t) + sentinel := filepath.Join(t.TempDir(), "run-manifest.json") + writeStageTestFile(t, sentinel, `{"session_id":"outside"}`) + manifestPath := filepath.Join(runRoot, "manifest.json") + if err := os.Remove(manifestPath); err != nil { + t.Fatal(err) + } + if err := os.Symlink(sentinel, manifestPath); err != nil { + t.Fatal(err) + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), `source "manifest.json" is not a regular file`) { + t.Fatalf("Run() error = %v, want rejected run manifest symlink", err) + } + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside") +} + +func TestPublishRejectsSymlinkedSourceAncestorBeforeUploading(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation requires privileges on Windows") + } + env, m, runRoot := publishFixture(t) + outside := t.TempDir() + writeStageTestFile(t, filepath.Join(outside, "audita.stderr.log"), "outside secret\n") + logsDir := filepath.Join(runRoot, "logs") + if err := os.Rename(logsDir, logsDir+"-original"); err != nil { + t.Fatal(err) + } + if err := os.Symlink(outside, logsDir); err != nil { + t.Fatal(err) + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "source ancestor") { + t.Fatalf("Run() error = %v, want rejected symlinked ancestor", err) + } + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside secret") +} + +func TestPublishRejectsDeclaredSourceReplacement(t *testing.T) { + _, m, runRoot := publishFixture(t) + target := filepath.Join(runRoot, "logs", "audita.stderr.log") + checksum, err := artifacts.SHA256File(target) + if err != nil { + t.Fatal(err) + } + now := time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC) + runManifest := manifest.NewRun(m.SessionID, m.Campaign, m.RunID, false, nil, now) + runManifest.MarkStageSucceeded("archive", now, []manifest.ArtifactRecord{{LocalPath: target, Checksum: checksum}}) + files, err := collectPublishRunFiles(runRoot, runManifest) + if err != nil { + t.Fatal(err) + } + writeStageTestFile(t, target, "replacement\n") + manifestFile, err := fileops.OpenConfinedRegularFile(runRoot, "manifest.json") + if err != nil { + t.Fatal(err) + } + defer manifestFile.Close() + + _, err = openPublishRunFiles(runRoot, files, manifestFile) + if err == nil || !strings.Contains(err.Error(), `source "logs/audita.stderr.log" does not match its declared checksum`) { + t.Fatalf("openPublishRunFiles() error = %v, want replacement rejection", err) + } +} + +func TestPublishRejectsDeclaredDirectoryBeforeUploading(t *testing.T) { + env, m, runRoot := publishFixture(t) + target := filepath.Join(runRoot, "logs", "audita.stderr.log") + if err := os.Remove(target); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(target, 0o755); err != nil { + t.Fatal(err) + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), `source "logs/audita.stderr.log" is not a regular file`) { + t.Fatalf("Run() error = %v, want rejected directory", err) + } + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "") +} + +func TestPublishDoesNotUploadUndeclaredRunFile(t *testing.T) { + env, m, runRoot := publishFixture(t) + writeStageTestFile(t, filepath.Join(runRoot, "unlisted.txt"), "outside of manifest\n") + + if _, err := (publishStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + fake := env.ObjectStore.(*storage.FakeBackend) + if _, exists := fake.Objects[m.S3RunPrefix+"unlisted.txt"]; exists { + t.Fatal("undeclared run file was uploaded") + } + for _, object := range fake.Objects { + if strings.Contains(string(object.Data), "outside of manifest") { + t.Fatalf("undeclared bytes were uploaded in %q", object.Key) + } + } +} + +func TestPublishRejectsDeclaredPreviousSymlinkBeforeUploading(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation requires privileges on Windows") + } + env, m, _ := publishFixture(t) + previousDir := filepath.Join( + artifacts.SessionWorkDirForCampaign( + env.Config.Pipeline.Workspace.Root, + env.Config.Session.Campaign, + env.Config.Session.SessionID, + ), + config.PathPreviousDirSegment, + ) + target := filepath.Join(previousDir, "artifacts", "session_recap.md") + sentinel := filepath.Join(t.TempDir(), "outside.md") + writeStageTestFile(t, sentinel, "outside previous secret\n") + writeStageTestFile(t, target, "previous recap\n") + writePublishPreviousManifest(t, previousDir, m.SessionID, target) + if err := os.Remove(target); err != nil { + t.Fatal(err) + } + if err := os.Symlink(sentinel, target); err != nil { + t.Fatal(err) + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), `source "artifacts/session_recap.md" is not a regular file`) { + t.Fatalf("Run() error = %v, want rejected previous symlink", err) + } + assertNoPublishUploads(t, env.ObjectStore.(*storage.FakeBackend), "outside previous secret") +} + +func assertNoPublishUploads(t *testing.T, fake *storage.FakeBackend, forbidden string) { + t.Helper() + if len(fake.Uploads) != 0 { + t.Fatalf("unexpected uploads: %#v", fake.Uploads) + } + for _, object := range fake.Objects { + if forbidden == "" || strings.Contains(string(object.Data), forbidden) { + t.Fatalf("unexpected uploaded object %q", object.Key) + } + } +} + func TestPublishUsesCustomOutputRules(t *testing.T) { env, m, _ := publishFixture(t) env.Config.Pipeline.Publish.Outputs = []config.PublishOutputRule{ @@ -694,16 +866,30 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "final.trimmed.json"), "{\"segments\":[]}\n") writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") - writeStageTestFile(t, filepath.Join(runRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n") - writeStageTestFile(t, filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac"), "flac\n") - writeStageTestFile(t, filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") - writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "final.trimmed.json"), "{\"segments\":[]}\n") - writeStageTestFile(t, filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") - writeStageTestFile(t, filepath.Join(runRoot, "polish", "reports", "audita.report.json"), "{}\n") - writeStageTestFile(t, filepath.Join(runRoot, "merge", "config", "seriatim.generated.yml"), "key: value\n") - writeStageTestFile(t, filepath.Join(runRoot, "logs", "audita.stderr.log"), "stderr\n") + runArtifacts := []string{ + "prepare/inputs/session.yml", + "prepare/outputs/audio/speaker.flac", + "transcribe/outputs/transcripts/raw/speaker.json", + "trim/outputs/transcripts/final.trimmed.json", + "analyze/outputs/artifacts/session_recap.md", + "polish/reports/audita.report.json", + "merge/config/seriatim.generated.yml", + "logs/audita.stderr.log", + } + for rel, contents := range map[string]string{ + "prepare/inputs/session.yml": "session_id: 2026-04-19\n", + "prepare/outputs/audio/speaker.flac": "flac\n", + "transcribe/outputs/transcripts/raw/speaker.json": "{}\n", + "trim/outputs/transcripts/final.trimmed.json": "{\"segments\":[]}\n", + "analyze/outputs/artifacts/session_recap.md": "# recap\n", + "polish/reports/audita.report.json": "{}\n", + "merge/config/seriatim.generated.yml": "key: value\n", + "logs/audita.stderr.log": "stderr\n", + } { + writeStageTestFile(t, filepath.Join(runRoot, filepath.FromSlash(rel)), contents) + } writeStageTestFile(t, filepath.Join(runRoot, "audio", "speaker.flac"), "flac") - writeStageTestFile(t, filepath.Join(runRoot, "manifest.json"), "{}\n") + writePublishRunManifest(t, runRoot, sessionID, campaign, runID, runArtifacts) m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC)) m.Campaign = campaign @@ -754,6 +940,38 @@ func publishFixture(t *testing.T) (*Env, *manifest.Manifest, string) { return env, m, runRoot } +func writePublishRunManifest(t *testing.T, runRoot, sessionID, campaign, runID string, archivePaths []string) { + t.Helper() + now := time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC) + runManifest := manifest.NewRun(sessionID, campaign, runID, false, nil, now) + outputs := make([]manifest.ArtifactRecord, 0, len(archivePaths)) + for _, rel := range archivePaths { + outputs = append(outputs, manifest.ArtifactRecord{LocalPath: filepath.Join(runRoot, filepath.FromSlash(rel))}) + } + runManifest.MarkStageSucceeded("archive", now, outputs) + runManifest.MarkSucceeded(now) + data, err := json.Marshal(runManifest) + if err != nil { + t.Fatal(err) + } + writeStageTestFile(t, filepath.Join(runRoot, "manifest.json"), string(data)) +} + +func writePublishPreviousManifest(t *testing.T, previousDir, sessionID string, archivePaths ...string) { + t.Helper() + now := time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC) + previous := manifest.New(sessionID, now) + previous.Artifacts = make([]manifest.ArtifactRecord, 0, len(archivePaths)) + for _, path := range archivePaths { + previous.Artifacts = append(previous.Artifacts, manifest.ArtifactRecord{LocalPath: path}) + } + data, err := json.Marshal(previous) + if err != nil { + t.Fatal(err) + } + writeStageTestFile(t, filepath.Join(previousDir, "manifest.json"), string(data)) +} + func configurePublishExtractionFixture(t *testing.T, env *Env, m *manifest.Manifest) string { t.Helper() env.Config.Pipeline.Notarius = &config.NotariusConfig{ @@ -824,6 +1042,13 @@ func (s *publishedOutputFailingStore) Upload(ctx context.Context, localPath, key return s.delegate.Upload(ctx, localPath, key, opts) } +func (s *publishedOutputFailingStore) UploadReader(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { + if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + return storage.ObjectInfo{}, errors.New("forced upload failure") + } + return s.delegate.UploadReader(ctx, source, key, opts) +} + func (s *publishedOutputFailingStore) Exists(ctx context.Context, key string) (bool, error) { return s.delegate.Exists(ctx, key) } diff --git a/internal/stage/run_local.go b/internal/stage/run_local.go index 1e7c3e4..48971c1 100644 --- a/internal/stage/run_local.go +++ b/internal/stage/run_local.go @@ -141,6 +141,7 @@ func materializeRunLocalOutput( return artifacts.Ref{}, fmt.Errorf("checksum materialized output %q: %w", canonicalPath, err) } ref.AbsolutePath = canonicalPath + ref.ArchivePath = srcPath ref.Checksum = checksum return ref, nil }