Reject incomplete package streams in Crossplane Runtime parser (#1151)

pkg/xpkg/parser/parser.go

@@ -115,7 +115,7 @@ func (p *PackageParser) Parse(_ context.Context, reader io.ReadCloser) (*Package
 	for {
 		content, err := yr.Read()
 		if err != nil && !errors.Is(err, io.EOF) {
-			return pkg, err
+			return pkg, annotateErr(err, reader)
 		}
 
 		if errors.Is(err, io.EOF) {

pkg/xpkg/parser/parser_test.go

@@ -19,6 +19,9 @@ package parser
 import (
 	"bytes"
 	"context"
+	"errors"
+	"fmt"
+	"io"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -220,6 +223,83 @@ func TestParser(t *testing.T) {
 	}
 }
 
+func TestParserRejectsIncompleteDocument(t *testing.T) {
+	objScheme := runtime.NewScheme()
+	metaScheme := runtime.NewScheme()
+	_ = apiextensions.AddToScheme(objScheme)
+	_ = appsv1.AddToScheme(metaScheme)
+	errUnexpectedEOF := errors.New("unexpected EOF")
+
+	cases := map[string]struct {
+		reason string
+		reader io.ReadCloser
+		want   error
+	}{
+		"MalformedYAML": {
+			reason: "should reject malformed YAML",
+			reader: io.NopCloser(bytes.NewReader([]byte(`apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: "test`))),
+		},
+		"AnnotatedUnexpectedEOF": {
+			reason: "should preserve the reader annotation on stream errors",
+			reader: &annotatedReadCloser{
+				ReadCloser: &errorReadCloser{
+					reader: bytes.NewReader([]byte(`apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: test`)),
+					err: errUnexpectedEOF,
+				},
+				annotation: "package.yaml",
+			},
+			want: fmt.Errorf("package.yaml: %w", errUnexpectedEOF),
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, err := New(metaScheme, objScheme).Parse(context.Background(), tc.reader)
+			if tc.want == nil {
+				if err == nil {
+					t.Errorf("%s: Parse(...): expected error, got nil", tc.reason)
+				}
+				return
+			}
+
+			if diff := cmp.Diff(errUnexpectedEOF, errors.Unwrap(err), cmpopts.EquateErrors()); diff != "" {
+				t.Errorf("%s: Parse(...): -want cause, +got cause:\n%s", tc.reason, diff)
+			}
+			if diff := cmp.Diff(tc.want.Error(), err.Error()); diff != "" {
+				t.Errorf("%s: Parse(...): -want err, +got err:\n%s", tc.reason, diff)
+			}
+		})
+	}
+}
+
+type annotatedReadCloser struct {
+	io.ReadCloser
+	annotation any
+}
+
+func (r *annotatedReadCloser) Annotate() any { return r.annotation }
+
+type errorReadCloser struct {
+	reader *bytes.Reader
+	err    error
+}
+
+func (r *errorReadCloser) Read(p []byte) (int, error) {
+	n, err := r.reader.Read(p)
+	if err == io.EOF {
+		return n, r.err
+	}
+	return n, err
+}
+
+func (r *errorReadCloser) Close() error { return nil }
+
 func TestCleanYAML(t *testing.T) {
 	type args struct {
 		in []byte