Bound remote asset content verification

internal/delivery/remote_assets.go

@@ -77,11 +77,11 @@ func inspectRemoteAssets(ctx context.Context, backend forge.Backend, releaseHand
 		case 0:
 			observation.State = AssetMissing
 		case 1:
-			digest, err := remoteAssetDigest(ctx, backend, matches[0].Handle)
+			digest, sizeMatches, err := remoteAssetDigest(ctx, backend, matches[0].Handle, item.Size)
 			if err != nil {
 				observation.State = AssetUnverifiable
 				observation.Err = err
-			} else if digest == item.SHA256 {
+			} else if sizeMatches && digest == item.SHA256 {
 				observation.State = AssetMatching
 			} else {
 				observation.State = AssetConflicting
@@ -99,18 +99,40 @@ func inspectRemoteAssets(ctx context.Context, backend forge.Backend, releaseHand
 	return inspection, nil
 }
 
-func remoteAssetDigest(ctx context.Context, backend forge.Backend, handle any) (string, error) {
+func remoteAssetDigest(ctx context.Context, backend forge.Backend, handle any, expectedSize int64) (string, bool, error) {
+	if err := ctx.Err(); err != nil {
+		return "", false, err
+	}
 	stream, err := backend.OpenAsset(ctx, handle)
 	if err != nil {
-		return "", err
+		return "", false, err
 	}
 	hasher := sha256.New()
-	_, readErr := io.Copy(hasher, stream)
-	closeErr := stream.Close()
-	if err := errors.Join(readErr, closeErr); err != nil {
-		return "", &forge.Failure{Kind: forge.FailureUnverifiable, Err: err}
+	read, readErr := io.Copy(hasher, io.LimitReader(stream, expectedSize))
+	excess := false
+	if readErr == nil && read == expectedSize {
+		var probe [1]byte
+		for attempts := 0; ; attempts++ {
+			var n int
+			n, readErr = stream.Read(probe[:])
+			excess = n != 0
+			if readErr != nil || excess {
+				if readErr == io.EOF {
+					readErr = nil
+				}
+				break
+			}
+			if attempts == 99 {
+				readErr = io.ErrNoProgress
+				break
+			}
+		}
 	}
-	return hex.EncodeToString(hasher.Sum(nil)), nil
+	closeErr := stream.Close()
+	if err := errors.Join(readErr, closeErr, ctx.Err()); err != nil {
+		return "", false, &forge.Failure{Kind: forge.FailureUnverifiable, Err: err}
+	}
+	return hex.EncodeToString(hasher.Sum(nil)), read == expectedSize && !excess, nil
 }
 
 func uploadMissingAssets(ctx context.Context, backend forge.Backend, releaseHandle any, inspection Inspection) (result UploadResult, err error) {

internal/delivery/remote_assets_test.go

@@ -53,6 +53,88 @@ func TestInspectRemoteAssetsComparesContentAndPreservesUnrelatedAssets(t *testin
 	}
 }
 
+type controlledAssetStream struct {
+	*bytes.Reader
+	readErr    error
+	errorAfter int
+	closeErr   error
+	cancel     context.CancelFunc
+	read       int
+	closed     bool
+}
+
+func (stream *controlledAssetStream) Read(p []byte) (int, error) {
+	n, err := stream.Reader.Read(p)
+	stream.read += n
+	if stream.cancel != nil {
+		stream.cancel()
+		stream.cancel = nil
+	}
+	if stream.readErr != nil && n > 0 && stream.read > stream.errorAfter {
+		err, stream.readErr = stream.readErr, nil
+	}
+	return n, err
+}
+
+func (stream *controlledAssetStream) Close() error {
+	stream.closed = true
+	return stream.closeErr
+}
+
+type controlledAssetBackend struct {
+	pageBackend
+	stream *controlledAssetStream
+}
+
+func (backend *controlledAssetBackend) ListAssets(context.Context, any) ([]forge.Asset, error) {
+	return []forge.Asset{{Filename: "asset.bin", Size: 999, Handle: "asset"}}, nil
+}
+
+func (backend *controlledAssetBackend) OpenAsset(context.Context, any) (io.ReadCloser, error) {
+	return backend.stream, nil
+}
+
+func TestInspectRemoteAssetBoundsContentAndPreservesErrors(t *testing.T) {
+	readErr := errors.New("read failed")
+	closeErr := errors.New("close failed")
+	for _, test := range []struct {
+		name, expected, actual string
+		readErr, closeErr      error
+		cancel                 bool
+		errorAfter             int
+		state                  AssetState
+	}{
+		{"empty", "", "", nil, nil, false, 0, AssetMatching},
+		{"shorter", "abcd", "abc", nil, nil, false, 0, AssetConflicting},
+		{"matching", "abcd", "abcd", nil, nil, false, 0, AssetMatching},
+		{"different", "abcd", "wxyz", nil, nil, false, 0, AssetConflicting},
+		{"oversized", "abcd", "abcdefghijk", nil, nil, false, 0, AssetConflicting},
+		{"empty oversized", "", "abc", nil, nil, false, 0, AssetConflicting},
+		{"data and error", "abcd", "abcd", readErr, nil, false, 0, AssetUnverifiable},
+		{"probe data and error", "abcd", "abcde", readErr, nil, false, 4, AssetUnverifiable},
+		{"close failure", "abcd", "abcdefgh", nil, closeErr, false, 0, AssetUnverifiable},
+		{"cancellation", "abcd", "abcd", nil, nil, true, 0, AssetUnverifiable},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			stream := &controlledAssetStream{Reader: bytes.NewReader([]byte(test.actual)), readErr: test.readErr, errorAfter: test.errorAfter, closeErr: test.closeErr}
+			if test.cancel {
+				stream.cancel = cancel
+			}
+			backend := &controlledAssetBackend{stream: stream}
+			inspection, err := inspectRemoteAssets(ctx, backend, "release", []artifact.ExpectedAsset{expectedAsset("asset.bin", []byte(test.expected))})
+			if err != nil || len(inspection.Assets) != 1 || inspection.Assets[0].State != test.state || !stream.closed || stream.read > len(test.expected)+1 {
+				t.Fatalf("inspection = %#v, error = %v, bytes = %d, closed = %t", inspection, err, stream.read, stream.closed)
+			}
+			cause := inspection.Assets[0].Err
+			if test.readErr != nil && !errors.Is(cause, test.readErr) || test.closeErr != nil && !errors.Is(cause, test.closeErr) || test.cancel && !errors.Is(cause, context.Canceled) {
+				t.Fatalf("inspection lost cause: %v", cause)
+			}
+		})
+	}
+}
+
 func TestInspectRemoteAssetsListingFailureDoesNotBecomeUnknownWrite(t *testing.T) {
 	assets := []artifact.ExpectedAsset{expectedAsset("a.bin", []byte("a")), expectedAsset("b.bin", []byte("b"))}
 	listErr := &forge.Failure{Kind: forge.FailureTransport, Err: errors.New("service unavailable")}
@@ -230,6 +312,11 @@ func TestUploadMissingAssetsReconcilesOnceAndKeepsLaterAssetsPending(t *testing.
 			followupAssets: []forge.Asset{{Filename: "a.bin", Handle: "a"}}, followupData: map[string][]byte{"a": []byte("x")},
 			wantState: Conflict, wantCategory: "conflicting",
 		},
+		{
+			name: "oversized bytes conflict", disposition: forge.WriteUncertain,
+			followupAssets: []forge.Asset{{Filename: "a.bin", Handle: "a"}}, followupData: map[string][]byte{"a": []byte("oversized")},
+			wantState: Conflict, wantCategory: "conflicting",
+		},
 		{
 			name: "missing after conflict remains conflict", disposition: forge.WriteConflict,
 			wantState: Conflict, wantCategory: "conflicting",