Implement runtime.Framer for CBOR Sequences.

Kubernetes-commit: e2b36a0f0c3801085d765ad5155c8d08be9ed09c

Author: benluddy

pkg/runtime/serializer/cbor/framer.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cbor
+
+import (
+	"io"
+
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/fxamacker/cbor/v2"
+)
+
+// NewFramer returns a runtime.Framer based on RFC 8742 CBOR Sequences. Each frame contains exactly
+// one encoded CBOR data item.
+func NewFramer() runtime.Framer {
+	return framer{}
+}
+
+var _ runtime.Framer = framer{}
+
+type framer struct{}
+
+func (framer) NewFrameReader(rc io.ReadCloser) io.ReadCloser {
+	return &frameReader{
+		decoder: cbor.NewDecoder(rc),
+		closer:  rc,
+	}
+}
+
+func (framer) NewFrameWriter(w io.Writer) io.Writer {
+	// Each data item in a CBOR sequence is self-delimiting (like JSON objects).
+	return w
+}
+
+type frameReader struct {
+	decoder *cbor.Decoder
+	closer  io.Closer
+
+	overflow []byte
+}
+
+func (fr *frameReader) Read(dst []byte) (int, error) {
+	if len(fr.overflow) > 0 {
+		// We read a frame that was too large for the destination slice in a previous call
+		// to Read and have bytes left over.
+		n := copy(dst, fr.overflow)
+		if n < len(fr.overflow) {
+			fr.overflow = fr.overflow[n:]
+			return n, io.ErrShortBuffer
+		}
+		fr.overflow = nil
+		return n, nil
+	}
+
+	// The Reader contract allows implementations to use all of dst[0:len(dst)] as scratch
+	// space, even if n < len(dst), but it does not allow implementations to use
+	// dst[len(dst):cap(dst)]. Slicing it up-front allows us to append to it without worrying
+	// about overwriting dst[len(dst):cap(dst)].
+	m := cbor.RawMessage(dst[0:0:len(dst)])
+	if err := fr.decoder.Decode(&m); err != nil {
+		return 0, err
+	}
+
+	if len(m) > len(dst) {
+		// The frame was too big, m has a newly-allocated underlying array to accommodate
+		// it.
+		fr.overflow = m[len(dst):]
+		return copy(dst, m), io.ErrShortBuffer
+	}
+
+	return len(m), nil
+}
+
+func (fr *frameReader) Close() error {
+	return fr.closer.Close()
+}

pkg/runtime/serializer/cbor/framer_test.go

@@ -0,0 +1,147 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cbor_test
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/runtime/serializer/cbor"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+// TestFrameReaderReadError tests that the frame reader does not resume after encountering a
+// well-formedness error in the input stream. According to RFC 8742 Section 2.8: "[...] if any data
+// item in the sequence is not well formed, it is not possible to reliably decode the rest of the
+// sequence."
+func TestFrameReaderReadError(t *testing.T) {
+	input := []byte{
+		0xff, // ill-formed initial break
+		0xa0, // followed by well-formed empty map
+	}
+	fr := cbor.NewFramer().NewFrameReader(io.NopCloser(bytes.NewReader(input)))
+	for i := 0; i < 3; i++ {
+		n, err := fr.Read(nil)
+		if err == nil || errors.Is(err, io.ErrShortBuffer) {
+			t.Fatalf("expected a non-nil error other than io.ErrShortBuffer, got: %v", err)
+		}
+		if n != 0 {
+			t.Fatalf("expected 0 bytes read on error, got %d", n)
+		}
+	}
+}
+
+func TestFrameReaderRead(t *testing.T) {
+	type ChunkedFrame [][]byte
+
+	for _, tc := range []struct {
+		Name   string
+		Frames []ChunkedFrame
+	}{
+		{
+			Name: "consecutive frames",
+			Frames: []ChunkedFrame{
+				[][]byte{{0xa0}},
+				[][]byte{{0xa0}},
+			},
+		},
+		{
+			Name: "zero-length destination buffer",
+			Frames: []ChunkedFrame{
+				[][]byte{{}, {0xa0}},
+			},
+		},
+		{
+			Name: "overflow",
+			Frames: []ChunkedFrame{
+				[][]byte{
+					{0x43},
+					{'x'},
+					{'y', 'z'},
+				},
+				[][]byte{
+					{0xa1, 0x43, 'f', 'o', 'o'},
+					{'b'},
+					{'a', 'r'},
+				},
+			},
+		},
+	} {
+		t.Run(tc.Name, func(t *testing.T) {
+			var concatenation []byte
+			for _, f := range tc.Frames {
+				for _, c := range f {
+					concatenation = append(concatenation, c...)
+				}
+			}
+
+			fr := cbor.NewFramer().NewFrameReader(io.NopCloser(bytes.NewReader(concatenation)))
+
+			for _, frame := range tc.Frames {
+				var want, got []byte
+				for i, chunk := range frame {
+					dst := make([]byte, len(chunk), 2*len(chunk))
+					for i := len(dst); i < cap(dst); i++ {
+						dst[:cap(dst)][i] = 0xff
+					}
+					n, err := fr.Read(dst)
+					if n != len(chunk) {
+						t.Errorf("expected %d bytes read, got %d", len(chunk), n)
+					}
+					if i == len(frame)-1 && err != nil {
+						t.Errorf("unexpected non-nil error on last read of frame: %v", err)
+					} else if i < len(frame)-1 && !errors.Is(err, io.ErrShortBuffer) {
+						t.Errorf("expected io.ErrShortBuffer on all but the last read of a frame, got: %v", err)
+					}
+					for i := len(dst); i < cap(dst); i++ {
+						if dst[:cap(dst)][i] != 0xff {
+							t.Errorf("read mutated underlying array beyond slice length: %#v", dst[len(dst):cap(dst)])
+							break
+						}
+					}
+					want = append(want, chunk...)
+					got = append(got, dst...)
+				}
+				if diff := cmp.Diff(want, got); diff != "" {
+					t.Errorf("reassembled frame differs:\n%s", diff)
+				}
+			}
+		})
+	}
+}
+
+type fakeReadCloser struct {
+	err error
+}
+
+func (rc fakeReadCloser) Read(_ []byte) (int, error) {
+	return 0, nil
+}
+
+func (rc fakeReadCloser) Close() error {
+	return rc.err
+}
+
+func TestFrameReaderClose(t *testing.T) {
+	want := errors.New("test")
+	if got := cbor.NewFramer().NewFrameReader(fakeReadCloser{err: want}).Close(); !errors.Is(got, want) {
+		t.Errorf("got error %v, want %v", got, want)
+	}
+}