Add immutable remote commit reader
This commit is contained in:
264
internal/artifacts/remote_commit.go
Normal file
264
internal/artifacts/remote_commit.go
Normal file
@@ -0,0 +1,264 @@
|
||||
package artifacts
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"path"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const RemoteCommitFormatVersion = 1
|
||||
|
||||
// RemoteArtifactType identifies the durable purpose of an object in a remote commit.
|
||||
type RemoteArtifactType string
|
||||
|
||||
const (
|
||||
RemoteArtifactTypeSessionManifest RemoteArtifactType = "session_manifest"
|
||||
RemoteArtifactTypeRunManifest RemoteArtifactType = "run_manifest"
|
||||
RemoteArtifactTypePublishedOutput RemoteArtifactType = "published_output"
|
||||
RemoteArtifactTypePreviousArtifact RemoteArtifactType = "previous_artifact"
|
||||
)
|
||||
|
||||
// RemoteArtifact maps one typed source to an immutable remote destination.
|
||||
type RemoteArtifact struct {
|
||||
Type RemoteArtifactType `json:"type"`
|
||||
Source string `json:"source"`
|
||||
DestinationKey string `json:"destination_key"`
|
||||
SHA256 string `json:"sha256"`
|
||||
Size int64 `json:"size"`
|
||||
Generation string `json:"generation"`
|
||||
}
|
||||
|
||||
// RemoteCommitManifest declares the complete immutable object set for one run.
|
||||
type RemoteCommitManifest struct {
|
||||
FormatVersion int `json:"format_version"`
|
||||
Campaign string `json:"campaign"`
|
||||
SessionID string `json:"session_id"`
|
||||
RunID string `json:"run_id"`
|
||||
Artifacts []RemoteArtifact `json:"artifacts"`
|
||||
}
|
||||
|
||||
// CurrentCommitPointer is the sole mutable selector for a remote commit manifest.
|
||||
type CurrentCommitPointer struct {
|
||||
FormatVersion int `json:"format_version"`
|
||||
Campaign string `json:"campaign"`
|
||||
SessionID string `json:"session_id"`
|
||||
RunID string `json:"run_id"`
|
||||
CommitKey string `json:"commit_key"`
|
||||
CommitSHA256 string `json:"commit_sha256"`
|
||||
CommitSize int64 `json:"commit_size"`
|
||||
CommitGeneration string `json:"commit_generation"`
|
||||
}
|
||||
|
||||
func (m RemoteCommitManifest) Validate() error {
|
||||
if m.FormatVersion != RemoteCommitFormatVersion {
|
||||
return fmt.Errorf("unsupported remote commit format version %d", m.FormatVersion)
|
||||
}
|
||||
if err := ValidateSessionIdentity(m.Campaign, m.SessionID); err != nil {
|
||||
return fmt.Errorf("remote commit identity: %w", err)
|
||||
}
|
||||
if err := ValidateRunIdentity(m.RunID); err != nil {
|
||||
return fmt.Errorf("remote commit run identity: %w", err)
|
||||
}
|
||||
if len(m.Artifacts) == 0 {
|
||||
return fmt.Errorf("remote commit artifacts are required")
|
||||
}
|
||||
|
||||
destinations := make(map[string]struct{}, len(m.Artifacts))
|
||||
for index, artifact := range m.Artifacts {
|
||||
if err := artifact.Validate(); err != nil {
|
||||
return fmt.Errorf("remote commit artifact %d: %w", index, err)
|
||||
}
|
||||
if _, exists := destinations[artifact.DestinationKey]; exists {
|
||||
return fmt.Errorf("remote commit declares duplicate destination %q", artifact.DestinationKey)
|
||||
}
|
||||
destinations[artifact.DestinationKey] = struct{}{}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m RemoteCommitManifest) ValidateForSessionPrefix(sessionPrefix string) error {
|
||||
if err := m.Validate(); err != nil {
|
||||
return err
|
||||
}
|
||||
runPrefix := S3RunPrefix(sessionPrefix, m.RunID)
|
||||
if runPrefix == "" {
|
||||
return fmt.Errorf("remote commit run prefix is required")
|
||||
}
|
||||
for _, artifact := range m.Artifacts {
|
||||
if !strings.HasPrefix(artifact.DestinationKey, runPrefix) {
|
||||
return fmt.Errorf("remote commit artifact destination %q is outside immutable run prefix %q", artifact.DestinationKey, runPrefix)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a RemoteArtifact) Validate() error {
|
||||
switch a.Type {
|
||||
case RemoteArtifactTypeSessionManifest, RemoteArtifactTypeRunManifest, RemoteArtifactTypePublishedOutput, RemoteArtifactTypePreviousArtifact:
|
||||
default:
|
||||
return fmt.Errorf("unsupported artifact type %q", a.Type)
|
||||
}
|
||||
if strings.TrimSpace(a.Source) == "" {
|
||||
return fmt.Errorf("source is required")
|
||||
}
|
||||
if err := validateRemoteObjectKey(a.DestinationKey); err != nil {
|
||||
return fmt.Errorf("destination key: %w", err)
|
||||
}
|
||||
if err := validateSHA256(a.SHA256); err != nil {
|
||||
return fmt.Errorf("sha256: %w", err)
|
||||
}
|
||||
if a.Size < 0 {
|
||||
return fmt.Errorf("size must not be negative")
|
||||
}
|
||||
if strings.TrimSpace(a.Generation) == "" {
|
||||
return fmt.Errorf("generation is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p CurrentCommitPointer) Validate() error {
|
||||
if p.FormatVersion != RemoteCommitFormatVersion {
|
||||
return fmt.Errorf("unsupported current commit pointer format version %d", p.FormatVersion)
|
||||
}
|
||||
if err := ValidateSessionIdentity(p.Campaign, p.SessionID); err != nil {
|
||||
return fmt.Errorf("current commit pointer identity: %w", err)
|
||||
}
|
||||
if err := ValidateRunIdentity(p.RunID); err != nil {
|
||||
return fmt.Errorf("current commit pointer run identity: %w", err)
|
||||
}
|
||||
if err := validateRemoteObjectKey(p.CommitKey); err != nil {
|
||||
return fmt.Errorf("current commit pointer key: %w", err)
|
||||
}
|
||||
if err := validateSHA256(p.CommitSHA256); err != nil {
|
||||
return fmt.Errorf("current commit pointer checksum: %w", err)
|
||||
}
|
||||
if p.CommitSize < 0 {
|
||||
return fmt.Errorf("current commit pointer size must not be negative")
|
||||
}
|
||||
if strings.TrimSpace(p.CommitGeneration) == "" {
|
||||
return fmt.Errorf("current commit pointer generation is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p CurrentCommitPointer) ValidateForSessionPrefix(sessionPrefix string) error {
|
||||
if err := p.Validate(); err != nil {
|
||||
return err
|
||||
}
|
||||
expectedKey := S3RunCommitKey(sessionPrefix, p.RunID)
|
||||
if p.CommitKey != expectedKey {
|
||||
return fmt.Errorf("current commit pointer key %q does not match canonical commit key %q", p.CommitKey, expectedKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// EncodeRemoteCommitManifest validates and serializes an immutable commit manifest.
|
||||
func EncodeRemoteCommitManifest(m RemoteCommitManifest) ([]byte, error) {
|
||||
if err := m.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
data, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("encode remote commit manifest: %w", err)
|
||||
}
|
||||
return append(data, '\n'), nil
|
||||
}
|
||||
|
||||
// DecodeRemoteCommitManifest strictly decodes a remote commit manifest.
|
||||
func DecodeRemoteCommitManifest(data []byte) (*RemoteCommitManifest, error) {
|
||||
var commit RemoteCommitManifest
|
||||
if err := decodeStrictJSON(data, &commit); err != nil {
|
||||
return nil, fmt.Errorf("decode remote commit manifest: %w", err)
|
||||
}
|
||||
if err := commit.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &commit, nil
|
||||
}
|
||||
|
||||
// EncodeCurrentCommitPointer validates and serializes a current commit pointer.
|
||||
func EncodeCurrentCommitPointer(p CurrentCommitPointer) ([]byte, error) {
|
||||
if err := p.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
data, err := json.Marshal(p)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("encode current commit pointer: %w", err)
|
||||
}
|
||||
return append(data, '\n'), nil
|
||||
}
|
||||
|
||||
// DecodeCurrentCommitPointer strictly decodes a current commit pointer.
|
||||
func DecodeCurrentCommitPointer(data []byte) (*CurrentCommitPointer, error) {
|
||||
var pointer CurrentCommitPointer
|
||||
if err := decodeStrictJSON(data, &pointer); err != nil {
|
||||
return nil, fmt.Errorf("decode current commit pointer: %w", err)
|
||||
}
|
||||
if err := pointer.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pointer, nil
|
||||
}
|
||||
|
||||
func (m RemoteCommitManifest) Artifact(artifactType RemoteArtifactType) (RemoteArtifact, bool) {
|
||||
var found RemoteArtifact
|
||||
for _, artifact := range m.Artifacts {
|
||||
if artifact.Type != artifactType {
|
||||
continue
|
||||
}
|
||||
if found.DestinationKey != "" {
|
||||
return RemoteArtifact{}, false
|
||||
}
|
||||
found = artifact
|
||||
}
|
||||
return found, found.DestinationKey != ""
|
||||
}
|
||||
|
||||
func remoteObjectSHA256(data []byte) string {
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func validateSHA256(value string) error {
|
||||
if len(value) != sha256.Size*2 || strings.ToLower(value) != value {
|
||||
return fmt.Errorf("must be a lowercase SHA-256 digest")
|
||||
}
|
||||
if _, err := hex.DecodeString(value); err != nil {
|
||||
return fmt.Errorf("must be hexadecimal: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRemoteObjectKey(value string) error {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return fmt.Errorf("is required")
|
||||
}
|
||||
if strings.TrimSpace(value) != value || strings.HasPrefix(value, "/") || strings.Contains(value, `\\`) {
|
||||
return fmt.Errorf("must be a clean relative object key")
|
||||
}
|
||||
if cleaned := path.Clean(value); cleaned != value || cleaned == "." || cleaned == ".." || strings.HasPrefix(cleaned, "../") {
|
||||
return fmt.Errorf("must be a clean relative object key")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func decodeStrictJSON(data []byte, destination any) error {
|
||||
decoder := json.NewDecoder(bytes.NewReader(data))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(destination); err != nil {
|
||||
return err
|
||||
}
|
||||
var trailing any
|
||||
if err := decoder.Decode(&trailing); err != io.EOF {
|
||||
if err == nil {
|
||||
return fmt.Errorf("multiple JSON values are not allowed")
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user