diff --git a/internal/framework/pipeline/handoff.go b/internal/framework/pipeline/handoff.go index b457a1f..fa8c8a9 100644 --- a/internal/framework/pipeline/handoff.go +++ b/internal/framework/pipeline/handoff.go @@ -243,6 +243,7 @@ type generatedReferenceFingerprintIdentity struct { SchemaDigest string `json:"schema_digest"` MediaType string `json:"media_type"` ContentDigest string `json:"content_digest"` + SizeBytes int64 `json:"size_bytes"` } // generatedReferenceDependencies returns the canonical semantic dependency @@ -271,6 +272,7 @@ func generatedReferenceDependencies(set contracts.ReferenceSet) []CheckpointFing SchemaDigest: contracts.DigestArtifactSchema(item.ArtifactSchema), MediaType: item.MediaType, ContentDigest: item.Digest, + SizeBytes: item.SizeBytes, } encoded, err := json.Marshal(identity) if err != nil { diff --git a/internal/framework/pipeline/handoff_test.go b/internal/framework/pipeline/handoff_test.go index f3a152b..bc0138d 100644 --- a/internal/framework/pipeline/handoff_test.go +++ b/internal/framework/pipeline/handoff_test.go @@ -185,6 +185,27 @@ func TestGeneratedReferenceFingerprintChangesWithCanonicalContent(t *testing.T) } } +func TestGeneratedReferenceFingerprintChangesWithCanonicalSize(t *testing.T) { + input, step, producerOutput := handoffFixture(t, codecNotes{Items: []string{"first"}}) + sets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{producerOutput}) + if err != nil { + t.Fatalf("build reference set: %v", err) + } + key := keyForReferenceTarget(step.lanes[0].resolved.ExtractReferences) + firstDeps := generatedReferenceDependencies(sets[key]) + changed := CloneReferenceSet(sets[key]) + item := changed.Slots["producer-output"].Items[0] + item.SizeBytes++ + changed.Slots["producer-output"] = contracts.ResolvedReferenceSlot{ + Slot: changed.Slots["producer-output"].Slot, + Items: []contracts.ReferenceItem{item}, + } + secondDeps := generatedReferenceDependencies(changed) + if reflect.DeepEqual(firstDeps, secondDeps) { + t.Fatalf("generated dependencies = %#v, want size-sensitive fingerprint", firstDeps) + } +} + func TestRunnerHandsOffAcceptedNormalizedOutputBeforeConsumerLanes(t *testing.T) { input, _, _ := handoffFixture(t, codecNotes{Items: []string{"first"}}) prepared := input.Prepared diff --git a/internal/framework/pipeline/references.go b/internal/framework/pipeline/references.go index dfed190..4f2f49f 100644 --- a/internal/framework/pipeline/references.go +++ b/internal/framework/pipeline/references.go @@ -3,7 +3,10 @@ package pipeline import ( "crypto/sha256" "encoding/hex" + "errors" "fmt" + "io" + "math" "mime" "net/url" "os" @@ -124,10 +127,30 @@ func materializeReferenceTarget( if err != nil { return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q: %w", referenceTargetContext(pipelineID, target), slotName, binding.Source, err) } - content, err := os.ReadFile(path) + file, err := os.Open(path) if err != nil { return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err) } + regularFileSize := int64(0) + if info, statErr := file.Stat(); statErr == nil && info.Mode().IsRegular() { + regularFileSize = info.Size() + } + content, err := readReferenceContent(file, slot.MaxBytes) + closeErr := file.Close() + if err == nil && closeErr != nil { + err = closeErr + } + if err != nil { + var sizeErr *referenceSizeLimitError + if errors.As(err, &sizeErr) { + sizeBytes := sizeErr.SizeBytes + if regularFileSize > sizeBytes { + sizeBytes = regularFileSize + } + return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, sizeBytes, slot.MaxBytes) + } + return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err) + } if !utf8.Valid(content) { return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q must be UTF-8 text", referenceTargetContext(pipelineID, target), slotName, path) } @@ -135,9 +158,6 @@ func materializeReferenceTarget( if !referenceMediaTypeAccepted(mediaType, slot.AcceptedMediaTypes) { return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q media type %q is not accepted", referenceTargetContext(pipelineID, target), slotName, path, mediaType) } - if slot.MaxBytes > 0 && int64(len(content)) > slot.MaxBytes { - return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, len(content), slot.MaxBytes) - } if len(content) == 0 { warnings = append(warnings, contracts.Warning{ Scope: referenceWarningScope(pipelineID, target, slotName), @@ -163,6 +183,32 @@ func materializeReferenceTarget( return set, warnings, nil } +type referenceSizeLimitError struct { + SizeBytes int64 + MaxBytes int64 +} + +func (err *referenceSizeLimitError) Error() string { + return fmt.Sprintf("reference is %d bytes, limit %d", err.SizeBytes, err.MaxBytes) +} + +func readReferenceContent(reader io.Reader, maxBytes int64) ([]byte, error) { + if maxBytes <= 0 { + return io.ReadAll(reader) + } + if maxBytes == math.MaxInt64 { + return nil, fmt.Errorf("maximum reference size %d cannot be safely bounded", maxBytes) + } + content, err := io.ReadAll(io.LimitReader(reader, maxBytes+1)) + if err != nil { + return nil, err + } + if int64(len(content)) > maxBytes { + return nil, &referenceSizeLimitError{SizeBytes: int64(len(content)), MaxBytes: maxBytes} + } + return content, nil +} + func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) { switch target.Stage { case StageChunk: diff --git a/internal/framework/pipeline/references_test.go b/internal/framework/pipeline/references_test.go index c07719f..887cc23 100644 --- a/internal/framework/pipeline/references_test.go +++ b/internal/framework/pipeline/references_test.go @@ -2,6 +2,9 @@ package pipeline import ( "encoding/json" + "errors" + "io" + "math" "os" "path/filepath" "reflect" @@ -12,6 +15,61 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) +type countingReader struct { + content []byte + consumed int +} + +func (reader *countingReader) Read(buffer []byte) (int, error) { + if len(reader.content) == 0 { + return 0, io.EOF + } + read := copy(buffer, reader.content) + reader.content = reader.content[read:] + reader.consumed += read + return read, nil +} + +func TestReadReferenceContentBoundsPositiveMaximum(t *testing.T) { + tests := []struct { + name string + content string + maxBytes int64 + wantContent string + wantConsumed int + wantLimitErr bool + wantReadError bool + }{ + {name: "exact maximum", content: "abc", maxBytes: 3, wantContent: "abc", wantConsumed: 3}, + {name: "oversized", content: "abcdef", maxBytes: 3, wantConsumed: 4, wantLimitErr: true}, + {name: "unbounded", content: "abcdef", maxBytes: 0, wantContent: "abcdef", wantConsumed: 6}, + {name: "negative maximum remains unbounded", content: "abcdef", maxBytes: -1, wantContent: "abcdef", wantConsumed: 6}, + {name: "overflowing maximum", content: "abcdef", maxBytes: math.MaxInt64, wantConsumed: 0, wantReadError: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + reader := &countingReader{content: []byte(tt.content)} + content, err := readReferenceContent(reader, tt.maxBytes) + if reader.consumed != tt.wantConsumed { + t.Fatalf("reader consumed %d bytes, want %d", reader.consumed, tt.wantConsumed) + } + if got := string(content); got != tt.wantContent { + t.Fatalf("content = %q, want %q", got, tt.wantContent) + } + var sizeErr *referenceSizeLimitError + if errors.As(err, &sizeErr) != tt.wantLimitErr { + t.Fatalf("error = %v, want limit error=%t", err, tt.wantLimitErr) + } + if (err != nil) != (tt.wantLimitErr || tt.wantReadError) { + t.Fatalf("error = %v, want error=%t", err, tt.wantLimitErr || tt.wantReadError) + } + if sizeErr != nil && (sizeErr.SizeBytes != 4 || sizeErr.MaxBytes != 3) { + t.Fatalf("size error = %#v, want observed size 4 and limit 3", sizeErr) + } + }) + } +} + func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) { configDir := t.TempDir() workingDir := t.TempDir()