Confine publish archive reads

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) {