diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 2166adb..8818871 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -26,6 +26,12 @@ Exact remote placement and the operator workflow belong in ## Key Behavior +`publish.go` retains the transaction-owning `Run` path. Source discovery, +manifest-derived archive metadata, confined opening, and checksum validation +live in `publish_sources.go`; artifact-catalog matching, selection, required +and optional rules, destination resolution, and collision inputs live in +`publish_outputs.go`. + - when publishing or run upload is disabled, completes successfully with no outputs and records explanatory metadata. This is not an explicit self-skip: the session manifest records success. Enablement and upload policy are @@ -115,6 +121,7 @@ The commit boundary and cleanup gate are normative architecture invariants; see - [Artifact Internals](artifacts.md) explains source resolution and current-state helpers. - Implementation and tests: `internal/stage/publish.go`, + `internal/stage/publish_sources.go`, `internal/stage/publish_outputs.go`, `internal/stage/publish_test.go`, `internal/stage/semantic_contracts_delivery.go`, `internal/app/operator_helpers_test.go`, and diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e1b0695..ad064ea 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -169,7 +169,7 @@ exact-key reads, and source-based inspection. Run ## Stage 4 — Separate Source And Output Planning -**Status:** Pending +**Status:** Completed ### Goal diff --git a/internal/stage/publish.go b/internal/stage/publish.go index ad35e7c..d72c75b 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -6,7 +6,6 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" - "errors" "fmt" "io" "os" @@ -15,19 +14,11 @@ import ( "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" - "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/manifest" - "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" ) type publishStage struct{} -type publishUploadFile struct { - RelativePath string - SourceRelativePath string - Checksum string -} var publishPrerequisiteStages = []string{ "prepare", @@ -259,36 +250,12 @@ func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publi return nil } -type publishOutput struct { - Source string - Dest string - Required bool - LocalPath string - Provenance string -} - type publishTransfer struct { Artifact artifacts.RemoteArtifact Source *os.File TemporaryPath string } -type publishLockedOutput struct { - Source string - Dest string - RemoteKey string - Reason string - Required bool - LocalPath string - Provenance string -} - -type publishSkippedUnselectedOutput struct { - Source string - Dest string - Required bool -} - func buildPublishTransferPlan( runPrefix string, runFiles []publishUploadFile, @@ -384,25 +351,6 @@ func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload) } -func openPublishOutputSource(localPath string) (*os.File, error) { - path := filepath.Clean(strings.TrimSpace(localPath)) - if path == "" || path == "." { - return nil, fmt.Errorf("local path is required") - } - info, err := os.Lstat(path) - if err != nil { - return nil, err - } - if !info.Mode().IsRegular() { - return nil, fmt.Errorf("source is not a regular file") - } - return os.Open(path) -} - -func publishOutputRemoteKey(runPrefix, destination string) string { - return artifacts.S3RunRelativeDestinationKey(runPrefix, filepath.ToSlash(filepath.Join("published", destination))) -} - func uploadVerifiedPublishTransfer( ctx context.Context, store storage.ObjectStore, @@ -555,22 +503,6 @@ func closePublishTransfers(transfers []publishTransfer) { } } -func publishUploadPaths(files []publishUploadFile) []string { - paths := make([]string, 0, len(files)) - for _, file := range files { - paths = append(paths, file.RelativePath) - } - return paths -} - -func publishOutputPaths(outputs []publishOutput) []string { - paths := make([]string, 0, len(outputs)) - for _, output := range outputs { - paths = append(paths, output.Dest) - } - return paths -} - func publishSHA256(data []byte) string { sum := sha256.Sum256(data) return hex.EncodeToString(sum[:]) @@ -652,408 +584,6 @@ func publishSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths return store.SessionPathsFor(campaign, sessionID) } -func resolvePublishOutputs( - paths artifacts.SessionPaths, - m *manifest.Manifest, - catalog *artifacts.ArtifactCatalog, - rules []config.PublishOutputRule, - locks []config.PublishLockRule, - selectedArtifactKeys []string, - runPrefix string, -) ([]publishOutput, []string, []publishSkippedUnselectedOutput, []publishLockedOutput, error) { - out := make([]publishOutput, 0, len(rules)) - skippedOptionalOutputs := make([]string, 0) - skippedUnselectedOutputs := make([]publishSkippedUnselectedOutput, 0) - lockedOutputs := make([]publishLockedOutput, 0) - lockSet := publishLockSet(locks) - selectedSet := publishSelectedArtifactSet(selectedArtifactKeys) - configuredOutputs := configuredOutputPathMapFromCatalog(catalog) - extractionOutputs := extractionOutputSetFromCatalog(catalog) - for _, rule := range rules { - source := strings.TrimSpace(rule.Source) - required := rule.Required == nil || *rule.Required - dest, err := resolvePublishOutputDest(rule, configuredOutputs, extractionOutputs) - if err != nil { - return nil, nil, nil, nil, fmt.Errorf("source %q: %w", source, err) - } - if len(selectedSet) > 0 { - if key, ok := artifactpolicy.ParseConfiguredSource(source); ok { - if _, selected := selectedSet[key]; !selected { - skippedUnselectedOutputs = append(skippedUnselectedOutputs, publishSkippedUnselectedOutput{ - Source: source, - Dest: dest, - Required: required, - }) - continue - } - } - } - lock, locked := lockSet[source] - resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog) - if err != nil { - if locked { - lockedOutputs = append(lockedOutputs, publishLockedOutput{ - Source: source, - Dest: dest, - RemoteKey: publishOutputRemoteKey(runPrefix, dest), - Reason: strings.TrimSpace(lock.Reason), - Required: required, - }) - continue - } - if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required { - skippedOptionalOutputs = append(skippedOptionalOutputs, dest) - continue - } - if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { - return nil, nil, nil, nil, fmt.Errorf("required output source unavailable: %q", source) - } - return nil, nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err) - } - if locked { - lockedOutputs = append(lockedOutputs, publishLockedOutput{ - Source: source, - Dest: dest, - RemoteKey: publishOutputRemoteKey(runPrefix, dest), - Reason: strings.TrimSpace(lock.Reason), - Required: required, - LocalPath: resolved.Path, - Provenance: resolved.Provenance, - }) - continue - } - out = append(out, publishOutput{ - Source: source, - Dest: dest, - Required: required, - LocalPath: resolved.Path, - Provenance: resolved.Provenance, - }) - } - return out, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, nil -} - -func publishSelectedArtifactSet(selected []string) map[string]struct{} { - if len(selected) == 0 { - return nil - } - out := make(map[string]struct{}, len(selected)) - for _, key := range selected { - trimmed := strings.TrimSpace(key) - if trimmed == "" { - continue - } - out[trimmed] = struct{}{} - } - return out -} - -func publishLockSet(locks []config.PublishLockRule) map[string]config.PublishLockRule { - out := make(map[string]config.PublishLockRule, len(locks)) - for _, lock := range locks { - source := strings.TrimSpace(lock.Source) - if source == "" { - continue - } - lock.Source = source - lock.Reason = strings.TrimSpace(lock.Reason) - out[source] = lock - } - return out -} - -func resolvePublishOutputDest(rule config.PublishOutputRule, configured map[string]string, extractions map[string]struct{}) (string, error) { - return artifactpolicy.ResolvePublishedDestinationWithExtractions(rule.Source, rule.Dest, configured, extractions) -} - -func configuredOutputPathMapFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]string { - out := map[string]string{} - if catalog == nil { - return out - } - for _, entry := range catalog.ListConfigured() { - if strings.TrimSpace(entry.ConfiguredKey) == "" { - continue - } - out[entry.ConfiguredKey] = strings.TrimSpace(entry.CanonicalRelPath) - } - return out -} - -func extractionOutputSetFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]struct{} { - out := map[string]struct{}{} - if catalog == nil { - return out - } - for _, entry := range catalog.ListExtraction() { - if strings.TrimSpace(entry.ExtractionKey) != "" { - out[entry.ExtractionKey] = struct{}{} - } - } - return out -} - -func buildPublishRuntimeArtifactCatalog( - paths artifacts.SessionPaths, - m *manifest.Manifest, - scriptoriumCfg *config.ScriptoriumConfig, - notariusCfg *config.NotariusConfig, -) (*artifacts.ArtifactCatalog, error) { - extractionDefinitions := artifacts.ExtractionDefinitionsFromConfig(notariusCfg) - configured := artifacts.ConfiguredArtifactDefinitions(nil) - if scriptoriumCfg != nil { - configured = artifacts.ConfiguredArtifactDefinitions(scriptoriumCfg.Artifacts) - } - effective, err := artifacts.ResolveEffectiveArtifactSet(configured, nil) - if err != nil { - return nil, err - } - catalog, err := artifacts.BootstrapRuntimeCatalog(configured, effective, extractionDefinitions) - if err != nil { - return nil, err - } - if notariusCfg != nil && notariusCfg.Enabled { - catalog.HydrateExtractionArtifacts(paths, m, extractionDefinitions) - } - catalog.HydrateAnalyzeArtifacts(paths, m, configured) - - return catalog, nil -} - -func collectPublishRunFiles(runRoot string, sessionManifest *manifest.Manifest) ([]publishUploadFile, error) { - if sessionManifest == nil { - return nil, fmt.Errorf("session manifest is required") - } - files := map[string]publishUploadFile{} - add := func(path, checksum string) error { - if strings.TrimSpace(path) == "" { - return nil - } - rel, err := pathsafe.SlashRelativeFromRoot(runRoot, path) - if err != nil { - return nil - } - 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 sessionManifest.Stages { - if record == nil { - continue - } - for _, output := range record.Outputs { - if output.ProducerRunID != "" && output.ProducerRunID != sessionManifest.RunID { - continue - } - 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 - } - } - } - for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { - if err := add(path, ""); err != nil { - return 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, nil -} - -func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]publishUploadFile, error) { - previousDir = filepath.Clean(strings.TrimSpace(previousDir)) - if previousDir == "" || previousDir == "." { - return nil, fmt.Errorf("previous directory is required") - } - info, err := os.Lstat(previousDir) - if errors.Is(err, os.ErrNotExist) { - return nil, nil - } - if err != nil { - return nil, fmt.Errorf("inspect previous directory: %w", err) - } - if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { - return nil, fmt.Errorf("previous directory is not a regular directory") - } - - manifestFile, err := openPublishOutputSource(filepath.Join(previousDir, "manifest.json")) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return nil, nil - } - return nil, fmt.Errorf("open previous manifest: %w", err) - } - defer manifestFile.Close() - previousManifest, err := (&manifest.LocalStore{}).LoadReader(ctx, manifestFile) - if err != nil { - return nil, fmt.Errorf("load 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 := pathsafe.SlashRelativeFromRoot(previousDir, path) - if err != nil { - return nil - } - 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 - } - for _, artifact := range previousManifest.Artifacts { - if err := add(artifact.ArchivePath, artifact.Checksum); err != nil { - return nil, err - } - if artifact.ArchivePath == "" { - if err := add(artifact.LocalPath, artifact.Checksum); err != nil { - return 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 { - return nil, err - } - if artifact.ArchivePath == "" { - if err := add(artifact.LocalPath, artifact.Checksum); err != nil { - return nil, err - } - } - } - for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { - if err := add(path, ""); err != nil { - return 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, nil -} - -func openPublishRunFiles(runRoot string, files []publishUploadFile) ([]*os.File, error) { - sources := make([]*os.File, 0, len(files)) - for _, candidate := range files { - file, err := openPublishSourceUnderRoot(runRoot, candidate.SourceRelativePath) - if err != nil { - closePublishSources(sources) - return nil, err - } - if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil { - _ = file.Close() - closePublishSources(sources) - return nil, err - } - sources = append(sources, file) - } - return sources, nil -} - -func openPublishSourceUnderRoot(rootPath, relativePath string) (*os.File, error) { - relativePath, err := pathsafe.NormalizeRelativeDestination(relativePath) - if err != nil { - return nil, fmt.Errorf("validate publish source path %q: %w", relativePath, err) - } - - root, err := os.OpenRoot(rootPath) - if err != nil { - return nil, fmt.Errorf("open publish source root %q: %w", rootPath, err) - } - openedRoots := []*os.Root{root} - defer func() { - for index := len(openedRoots) - 1; index >= 0; index-- { - _ = openedRoots[index].Close() - } - }() - - parts := strings.Split(relativePath, "/") - current := root - for _, part := range parts[:len(parts)-1] { - info, err := current.Lstat(part) - if err != nil { - return nil, fmt.Errorf("inspect publish source ancestor %q: %w", part, err) - } - if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { - return nil, fmt.Errorf("publish source ancestor %q is not a directory without symlinks", part) - } - next, err := current.OpenRoot(part) - if err != nil { - return nil, fmt.Errorf("open publish source ancestor %q: %w", part, err) - } - openedRoots = append(openedRoots, next) - current = next - } - - leaf := parts[len(parts)-1] - info, err := current.Lstat(leaf) - if err != nil { - return nil, fmt.Errorf("inspect publish source %q: %w", relativePath, err) - } - if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { - return nil, fmt.Errorf("publish source %q is not a regular file without symlinks", relativePath) - } - file, err := current.Open(leaf) - if err != nil { - return nil, fmt.Errorf("open publish source %q: %w", relativePath, err) - } - return file, 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() - } -} - func writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) { if m == nil { return "", fmt.Errorf("manifest is required") diff --git a/internal/stage/publish_outputs.go b/internal/stage/publish_outputs.go new file mode 100644 index 0000000..a0cfa3d --- /dev/null +++ b/internal/stage/publish_outputs.go @@ -0,0 +1,202 @@ +package stage + +import ( + "errors" + "fmt" + "path/filepath" + "strings" + + "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/manifest" +) + +type publishOutput struct { + Source string + Dest string + Required bool + LocalPath string + Provenance string +} + +type publishLockedOutput struct { + Source string + Dest string + RemoteKey string + Reason string + Required bool + LocalPath string + Provenance string +} + +type publishSkippedUnselectedOutput struct { + Source string + Dest string + Required bool +} + +func resolvePublishOutputs( + paths artifacts.SessionPaths, + m *manifest.Manifest, + catalog *artifacts.ArtifactCatalog, + rules []config.PublishOutputRule, + locks []config.PublishLockRule, + selectedArtifactKeys []string, + runPrefix string, +) ([]publishOutput, []string, []publishSkippedUnselectedOutput, []publishLockedOutput, error) { + out := make([]publishOutput, 0, len(rules)) + skippedOptionalOutputs := make([]string, 0) + skippedUnselectedOutputs := make([]publishSkippedUnselectedOutput, 0) + lockedOutputs := make([]publishLockedOutput, 0) + lockSet := publishLockSet(locks) + selectedSet := publishSelectedArtifactSet(selectedArtifactKeys) + configuredOutputs := configuredOutputPathMapFromCatalog(catalog) + extractionOutputs := extractionOutputSetFromCatalog(catalog) + for _, rule := range rules { + source := strings.TrimSpace(rule.Source) + required := rule.Required == nil || *rule.Required + dest, err := resolvePublishOutputDest(rule, configuredOutputs, extractionOutputs) + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("source %q: %w", source, err) + } + if len(selectedSet) > 0 { + if key, ok := artifactpolicy.ParseConfiguredSource(source); ok { + if _, selected := selectedSet[key]; !selected { + skippedUnselectedOutputs = append(skippedUnselectedOutputs, publishSkippedUnselectedOutput{ + Source: source, Dest: dest, Required: required, + }) + continue + } + } + } + lock, locked := lockSet[source] + resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog) + if err != nil { + if locked { + lockedOutputs = append(lockedOutputs, publishLockedOutput{ + Source: source, Dest: dest, RemoteKey: publishOutputRemoteKey(runPrefix, dest), + Reason: strings.TrimSpace(lock.Reason), Required: required, + }) + continue + } + if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required { + skippedOptionalOutputs = append(skippedOptionalOutputs, dest) + continue + } + if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { + return nil, nil, nil, nil, fmt.Errorf("required output source unavailable: %q", source) + } + return nil, nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err) + } + if locked { + lockedOutputs = append(lockedOutputs, publishLockedOutput{ + Source: source, Dest: dest, RemoteKey: publishOutputRemoteKey(runPrefix, dest), + Reason: strings.TrimSpace(lock.Reason), Required: required, + LocalPath: resolved.Path, Provenance: resolved.Provenance, + }) + continue + } + out = append(out, publishOutput{ + Source: source, Dest: dest, Required: required, + LocalPath: resolved.Path, Provenance: resolved.Provenance, + }) + } + return out, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, nil +} + +func publishSelectedArtifactSet(selected []string) map[string]struct{} { + if len(selected) == 0 { + return nil + } + out := make(map[string]struct{}, len(selected)) + for _, key := range selected { + trimmed := strings.TrimSpace(key) + if trimmed != "" { + out[trimmed] = struct{}{} + } + } + return out +} + +func publishLockSet(locks []config.PublishLockRule) map[string]config.PublishLockRule { + out := make(map[string]config.PublishLockRule, len(locks)) + for _, lock := range locks { + source := strings.TrimSpace(lock.Source) + if source == "" { + continue + } + lock.Source = source + lock.Reason = strings.TrimSpace(lock.Reason) + out[source] = lock + } + return out +} + +func resolvePublishOutputDest(rule config.PublishOutputRule, configured map[string]string, extractions map[string]struct{}) (string, error) { + return artifactpolicy.ResolvePublishedDestinationWithExtractions(rule.Source, rule.Dest, configured, extractions) +} + +func configuredOutputPathMapFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]string { + out := map[string]string{} + if catalog == nil { + return out + } + for _, entry := range catalog.ListConfigured() { + if strings.TrimSpace(entry.ConfiguredKey) != "" { + out[entry.ConfiguredKey] = strings.TrimSpace(entry.CanonicalRelPath) + } + } + return out +} + +func extractionOutputSetFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]struct{} { + out := map[string]struct{}{} + if catalog == nil { + return out + } + for _, entry := range catalog.ListExtraction() { + if strings.TrimSpace(entry.ExtractionKey) != "" { + out[entry.ExtractionKey] = struct{}{} + } + } + return out +} + +func buildPublishRuntimeArtifactCatalog( + paths artifacts.SessionPaths, + m *manifest.Manifest, + scriptoriumCfg *config.ScriptoriumConfig, + notariusCfg *config.NotariusConfig, +) (*artifacts.ArtifactCatalog, error) { + extractionDefinitions := artifacts.ExtractionDefinitionsFromConfig(notariusCfg) + configured := artifacts.ConfiguredArtifactDefinitions(nil) + if scriptoriumCfg != nil { + configured = artifacts.ConfiguredArtifactDefinitions(scriptoriumCfg.Artifacts) + } + effective, err := artifacts.ResolveEffectiveArtifactSet(configured, nil) + if err != nil { + return nil, err + } + catalog, err := artifacts.BootstrapRuntimeCatalog(configured, effective, extractionDefinitions) + if err != nil { + return nil, err + } + if notariusCfg != nil && notariusCfg.Enabled { + catalog.HydrateExtractionArtifacts(paths, m, extractionDefinitions) + } + catalog.HydrateAnalyzeArtifacts(paths, m, configured) + return catalog, nil +} + +func publishOutputRemoteKey(runPrefix, destination string) string { + return artifacts.S3RunRelativeDestinationKey(runPrefix, filepath.ToSlash(filepath.Join("published", destination))) +} + +func publishOutputPaths(outputs []publishOutput) []string { + paths := make([]string, 0, len(outputs)) + for _, output := range outputs { + paths = append(paths, output.Dest) + } + return paths +} diff --git a/internal/stage/publish_sources.go b/internal/stage/publish_sources.go new file mode 100644 index 0000000..50e5c65 --- /dev/null +++ b/internal/stage/publish_sources.go @@ -0,0 +1,281 @@ +package stage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +type publishUploadFile struct { + RelativePath string + SourceRelativePath string + Checksum string +} + +func collectPublishRunFiles(runRoot string, sessionManifest *manifest.Manifest) ([]publishUploadFile, error) { + if sessionManifest == nil { + return nil, fmt.Errorf("session manifest is required") + } + files := map[string]publishUploadFile{} + add := func(path, checksum string) error { + if strings.TrimSpace(path) == "" { + return nil + } + rel, err := pathsafe.SlashRelativeFromRoot(runRoot, path) + if err != nil { + return nil + } + 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 sessionManifest.Stages { + if record == nil { + continue + } + for _, output := range record.Outputs { + if output.ProducerRunID != "" && output.ProducerRunID != sessionManifest.RunID { + continue + } + 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 + } + } + } + for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { + if err := add(path, ""); err != nil { + return 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, nil +} + +func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]publishUploadFile, error) { + previousDir = filepath.Clean(strings.TrimSpace(previousDir)) + if previousDir == "" || previousDir == "." { + return nil, fmt.Errorf("previous directory is required") + } + info, err := os.Lstat(previousDir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("inspect previous directory: %w", err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return nil, fmt.Errorf("previous directory is not a regular directory") + } + + manifestFile, err := openPublishOutputSource(filepath.Join(previousDir, "manifest.json")) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, fmt.Errorf("open previous manifest: %w", err) + } + defer manifestFile.Close() + previousManifest, err := (&manifest.LocalStore{}).LoadReader(ctx, manifestFile) + if err != nil { + return nil, fmt.Errorf("load 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 := pathsafe.SlashRelativeFromRoot(previousDir, path) + if err != nil { + return nil + } + 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 + } + for _, artifact := range previousManifest.Artifacts { + if err := add(artifact.ArchivePath, artifact.Checksum); err != nil { + return nil, err + } + if artifact.ArchivePath == "" { + if err := add(artifact.LocalPath, artifact.Checksum); err != nil { + return 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 { + return nil, err + } + if artifact.ArchivePath == "" { + if err := add(artifact.LocalPath, artifact.Checksum); err != nil { + return nil, err + } + } + } + for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) { + if err := add(path, ""); err != nil { + return 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, nil +} + +func openPublishRunFiles(runRoot string, files []publishUploadFile) ([]*os.File, error) { + sources := make([]*os.File, 0, len(files)) + for _, candidate := range files { + file, err := openPublishSourceUnderRoot(runRoot, candidate.SourceRelativePath) + if err != nil { + closePublishSources(sources) + return nil, err + } + if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil { + _ = file.Close() + closePublishSources(sources) + return nil, err + } + sources = append(sources, file) + } + return sources, nil +} + +func openPublishSourceUnderRoot(rootPath, relativePath string) (*os.File, error) { + relativePath, err := pathsafe.NormalizeRelativeDestination(relativePath) + if err != nil { + return nil, fmt.Errorf("validate publish source path %q: %w", relativePath, err) + } + + root, err := os.OpenRoot(rootPath) + if err != nil { + return nil, fmt.Errorf("open publish source root %q: %w", rootPath, err) + } + openedRoots := []*os.Root{root} + defer func() { + for index := len(openedRoots) - 1; index >= 0; index-- { + _ = openedRoots[index].Close() + } + }() + + parts := strings.Split(relativePath, "/") + current := root + for _, part := range parts[:len(parts)-1] { + info, err := current.Lstat(part) + if err != nil { + return nil, fmt.Errorf("inspect publish source ancestor %q: %w", part, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return nil, fmt.Errorf("publish source ancestor %q is not a directory without symlinks", part) + } + next, err := current.OpenRoot(part) + if err != nil { + return nil, fmt.Errorf("open publish source ancestor %q: %w", part, err) + } + openedRoots = append(openedRoots, next) + current = next + } + + leaf := parts[len(parts)-1] + info, err := current.Lstat(leaf) + if err != nil { + return nil, fmt.Errorf("inspect publish source %q: %w", relativePath, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { + return nil, fmt.Errorf("publish source %q is not a regular file without symlinks", relativePath) + } + file, err := current.Open(leaf) + if err != nil { + return nil, fmt.Errorf("open publish source %q: %w", relativePath, err) + } + return file, nil +} + +func openPublishOutputSource(localPath string) (*os.File, error) { + path := filepath.Clean(strings.TrimSpace(localPath)) + if path == "" || path == "." { + return nil, fmt.Errorf("local path is required") + } + info, err := os.Lstat(path) + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() { + return nil, fmt.Errorf("source is not a regular file") + } + return os.Open(path) +} + +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() + } +} + +func publishUploadPaths(files []publishUploadFile) []string { + paths := make([]string, 0, len(files)) + for _, file := range files { + paths = append(paths, file.RelativePath) + } + return paths +}