feat(bsr): Add ChunkDecoder

A ChunkDecoder can be used to BSR Chunks. Each call to Decode will read
the next chunk from the decoder's io.Reader. It handles decompressing
any compressed chunk data, and verifying the CRCs of each chunk.

internal/bsr/chunk_end.go

@@ -3,7 +3,12 @@
 
 package bsr
 
-import "context"
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+)
 
 // EndChunk identifies the end of the chunks in a BSR data file.
 // An EndChunk in a bsr data file is represented as:
@@ -37,3 +42,19 @@ func NewEnd(ctx context.Context, p Protocol, d Direction, t *Timestamp) (*EndChu
 		BaseChunk: bc,
 	}, nil
 }
+
+// DecodeEnd will decode an EndChunk.
+func DecodeEnd(_ context.Context, bc *BaseChunk, data []byte) (Chunk, error) {
+	const op = "bsr.DecodeEnd"
+
+	if is.Nil(bc) {
+		return nil, fmt.Errorf("%s: nil base chunk: %w", op, ErrInvalidParameter)
+	}
+	if bc.Type != ChunkEnd {
+		return nil, fmt.Errorf("%s: invalid chunk type %s", op, bc.Type)
+	}
+	if len(data) != 0 {
+		return nil, fmt.Errorf("%s: %w", op, ErrEndChunkNotEmpty)
+	}
+	return &EndChunk{BaseChunk: bc}, nil
+}

internal/bsr/chunk_end_test.go

@@ -126,3 +126,80 @@ func TestEndMarshalData(t *testing.T) {
 		})
 	}
 }
+
+func TestDecodeEnd(t *testing.T) {
+	ctx := context.Background()
+
+	ts := time.Date(2023, time.March, 16, 10, 47, 3, 14, time.UTC)
+
+	cases := []struct {
+		name    string
+		bc      *bsr.BaseChunk
+		encoded []byte
+		want    bsr.Chunk
+		wantErr error
+	}{
+		{
+			"header-no-compression",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkEnd,
+			},
+			[]byte(""),
+			&bsr.EndChunk{
+				BaseChunk: &bsr.BaseChunk{
+					Protocol:  "TEST",
+					Direction: bsr.Inbound,
+					Timestamp: bsr.NewTimestamp(ts),
+					Type:      bsr.ChunkEnd,
+				},
+			},
+			nil,
+		},
+		{
+			"header-wrong-type",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      "TEST",
+			},
+			[]byte(""),
+			nil,
+			errors.New("bsr.DecodeEnd: invalid chunk type TEST"),
+		},
+		{
+			"header-nil-base-chunk",
+			nil,
+			[]byte(""),
+			nil,
+			errors.New("bsr.DecodeEnd: nil base chunk: invalid parameter"),
+		},
+		{
+			"header-extra-data",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkEnd,
+			},
+			[]byte("foo"),
+			nil,
+			errors.New("bsr.DecodeEnd: end chunk not empty"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := bsr.DecodeEnd(ctx, tc.bc, tc.encoded)
+			if tc.wantErr != nil {
+				require.EqualError(t, err, tc.wantErr.Error())
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}

internal/bsr/chunk_header.go

@@ -6,6 +6,8 @@ package bsr
 import (
 	"context"
 	"fmt"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
 )
 
 // HeaderChunk is the first chunk in a BSR data file.
@@ -60,3 +62,26 @@ func NewHeader(ctx context.Context, p Protocol, d Direction, t *Timestamp, c Com
 		SessionId:   sessionId,
 	}, nil
 }
+
+// DecodeHeader will decode a Header chunk.
+func DecodeHeader(_ context.Context, bc *BaseChunk, data []byte) (Chunk, error) {
+	const op = "bsr.DecodeHeader"
+
+	if is.Nil(bc) {
+		return nil, fmt.Errorf("%s: nil base chunk: %w", op, ErrInvalidParameter)
+	}
+	if bc.Type != ChunkHeader {
+		return nil, fmt.Errorf("%s: invalid chunk type %s", op, bc.Type)
+	}
+
+	if uint32(len(data)) < compressionSize+encryptionSize {
+		return nil, fmt.Errorf("%s: not enough data", op)
+	}
+
+	h := &HeaderChunk{BaseChunk: bc}
+	h.Compression, data = Compression(data[:compressionSize][0]), data[compressionSize:]
+	h.Encryption, data = Encryption(data[:encryptionSize][0]), data[encryptionSize:]
+	h.SessionId = string(data)
+
+	return h, nil
+}

internal/bsr/chunk_header_test.go

@@ -190,3 +190,145 @@ func TestHeaderMarshalData(t *testing.T) {
 		})
 	}
 }
+
+func TestDecodeHeader(t *testing.T) {
+	ctx := context.Background()
+
+	ts := time.Date(2023, time.March, 16, 10, 47, 3, 14, time.UTC)
+
+	cases := []struct {
+		name    string
+		bc      *bsr.BaseChunk
+		encoded []byte
+		want    bsr.Chunk
+		wantErr error
+	}{
+		{
+			"header-no-compression",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkHeader,
+			},
+			[]byte(
+				"" + // so everything else aligns better
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"",
+			),
+			&bsr.HeaderChunk{
+				BaseChunk: &bsr.BaseChunk{
+					Protocol:  "TEST",
+					Direction: bsr.Inbound,
+					Timestamp: bsr.NewTimestamp(ts),
+					Type:      bsr.ChunkHeader,
+				},
+				Compression: bsr.NoCompression,
+				Encryption:  bsr.NoEncryption,
+				SessionId:   "sess_123456789",
+			},
+			nil,
+		},
+		{
+			"header-gzip-compression",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkHeader,
+			},
+			[]byte(
+				"" + // so everything else aligns better
+					"\x01" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"",
+			),
+			&bsr.HeaderChunk{
+				BaseChunk: &bsr.BaseChunk{
+					Protocol:  "TEST",
+					Direction: bsr.Inbound,
+					Timestamp: bsr.NewTimestamp(ts),
+					Type:      bsr.ChunkHeader,
+				},
+				Compression: bsr.GzipCompression,
+				Encryption:  bsr.NoEncryption,
+				SessionId:   "sess_123456789",
+			},
+			nil,
+		},
+		{
+			"header-wrong-type",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      "TEST",
+			},
+			[]byte(
+				"" + // so everything else aligns better
+					"\x01" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"",
+			),
+			nil,
+			errors.New("bsr.DecodeHeader: invalid chunk type TEST"),
+		},
+		{
+			"header-nil-base-chunk",
+			nil,
+			[]byte(
+				"" + // so everything else aligns better
+					"\x01" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"",
+			),
+			nil,
+			errors.New("bsr.DecodeHeader: nil base chunk: invalid parameter"),
+		},
+		{
+			"header-no-data",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkHeader,
+			},
+			nil,
+			nil,
+			errors.New("bsr.DecodeHeader: not enough data"),
+		},
+		{
+			"header-not-enough-data",
+			&bsr.BaseChunk{
+				Protocol:  "TEST",
+				Direction: bsr.Inbound,
+				Timestamp: bsr.NewTimestamp(ts),
+				Type:      bsr.ChunkHeader,
+			},
+			[]byte(
+				"" + // so everything else aligns better
+					"\x01" + // compression method
+					"",
+			),
+			nil,
+			errors.New("bsr.DecodeHeader: not enough data"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := bsr.DecodeHeader(ctx, tc.bc, tc.encoded)
+			if tc.wantErr != nil {
+				require.EqualError(t, err, tc.wantErr.Error())
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}

internal/bsr/compression.go

@@ -52,3 +52,15 @@ func (w *nullCompressionWriter) Close() error {
 func newNullCompressionWriter(b *bytes.Buffer) io.WriteCloser {
 	return &nullCompressionWriter{Buffer: b}
 }
+
+type nullCompressionReader struct {
+	*bytes.Buffer
+}
+
+func (w *nullCompressionReader) Close() error {
+	return nil
+}
+
+func newNullCompressionReader(b *bytes.Buffer) io.ReadCloser {
+	return &nullCompressionReader{Buffer: b}
+}

internal/bsr/decode.go

@@ -0,0 +1,208 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"io"
+	"log"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+)
+
+// DecodeChunkFunc is a function that given a BaseChunk and the data portion
+// of a chunk, will decode the data into a Chunk.
+type DecodeChunkFunc func(ctx context.Context, bc *BaseChunk, data []byte) (Chunk, error)
+
+type decodeFuncsRegistry map[Protocol]map[ChunkType]DecodeChunkFunc
+
+func (r decodeFuncsRegistry) Get(p Protocol, t ChunkType) (DecodeChunkFunc, bool) {
+	switch t {
+	case ChunkHeader:
+		return DecodeHeader, true
+	case ChunkEnd:
+		return DecodeEnd, true
+	default:
+		protocol, ok := r[p]
+		if !ok {
+			return nil, false
+		}
+		df, ok := protocol[t]
+		return df, ok
+	}
+}
+
+var chunkTypes decodeFuncsRegistry
+
+// RegisterChunkType registers a DecodeChunkFunc for the given Protocol and
+// ChunkType. A given Protocol and ChunkType can only have one decode function
+// registered.
+func RegisterChunkType(p Protocol, t ChunkType, df DecodeChunkFunc) error {
+	const op = "bsr.RegisterChunkType"
+
+	if chunkTypes == nil {
+		chunkTypes = make(map[Protocol]map[ChunkType]DecodeChunkFunc)
+	}
+
+	protocol, ok := chunkTypes[p]
+	if !ok {
+		protocol = make(map[ChunkType]DecodeChunkFunc)
+	}
+
+	_, ok = protocol[t]
+	if ok {
+		return fmt.Errorf("%s: %s %s: %w", op, p, t, ErrAlreadyRegistered)
+	}
+	protocol[t] = df
+	chunkTypes[p] = protocol
+	return nil
+}
+
+// ChunkDecoder is used to decode the data read from an io.Reader into Chunks.
+type ChunkDecoder struct {
+	r           io.Reader
+	compression Compression
+	encryption  Encryption
+
+	keys *kms.Keys
+}
+
+// NewChunkDecoder creates a ChunkDecoder that can decode the data read from
+// the given io.Reader. Supports the WithKeys option which will be used when
+// support for encrypted chunks is added.
+func NewChunkDecoder(_ context.Context, r io.Reader, options ...Option) (*ChunkDecoder, error) {
+	const op = "bsr.NewChunkDecoder"
+
+	if is.Nil(r) {
+		return nil, fmt.Errorf("%s: reader cannot be nil: %w", op, ErrInvalidParameter)
+	}
+
+	opts := getOpts(options...)
+
+	return &ChunkDecoder{
+		r:           r,
+		compression: NoCompression,
+		encryption:  NoEncryption,
+		keys:        opts.withKeys,
+	}, nil
+}
+
+// Decode will read from the io.Reader and return the next Chunk that it
+// decodes.  If the io.Reader reaches EOF, Decode will return an error of
+// io.EOF. Note that this is not a wrapped error so it can be checked for with
+// err == io.EOF like with most io.Readers. If there is an unexpected error
+// while decoding, such as an unsupported chunk type or corrupted data, an
+// ErrChunkDecode error will be returned. This will be a wrapped error and
+// should be checked for with errors.Is.
+func (d *ChunkDecoder) Decode(ctx context.Context) (Chunk, error) {
+	const op = "bsr.(ChunkDecoder).Decode"
+
+	var b *BaseChunk
+
+	buf := make([]byte, chunkBaseSize)
+	crcBuf := make([]byte, crcSize)
+
+	_, err := io.ReadAtLeast(d.r, buf, chunkBaseSize)
+	if err != nil {
+		if err == io.EOF {
+			return nil, err
+		}
+		return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode)
+	}
+
+	var length uint32
+	var protocol Protocol
+	var chunkType ChunkType
+	var direction Direction
+
+	crc := crc32.NewIEEE()
+
+	length, buf = binary.BigEndian.Uint32(buf[:lengthSize]), buf[lengthSize:]
+	databuf := make([]byte, length)
+	_, err = io.ReadAtLeast(d.r, databuf, int(length))
+	if err != nil {
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
+		}
+		return nil, fmt.Errorf("%s: %w: missing data: %w", op, err, ErrChunkDecode)
+	}
+	crc.Write(append(buf, databuf...))
+
+	_, err = io.ReadAtLeast(d.r, crcBuf, crcSize)
+	if err != nil {
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
+		}
+		return nil, fmt.Errorf("%s: %w: missing crc: %w", op, err, ErrChunkDecode)
+	}
+	if crc.Sum32() != binary.BigEndian.Uint32(crcBuf) {
+		log.Printf("%x", crc.Sum32())
+		return nil, fmt.Errorf("%s: chunk crc did not match: %w", op, ErrChunkDecode)
+	}
+
+	protocol, buf = Protocol(string(buf[:protocolSize])), buf[protocolSize:]
+	chunkType, buf = ChunkType(string(buf[:chunkTypeSize])), buf[chunkTypeSize:]
+	direction, buf = Direction(buf[:directionSize][0]), buf[directionSize:]
+
+	timestamp, err := decodeTimestamp(buf[:timestampSize])
+	if err != nil {
+		return nil, fmt.Errorf("%s: error parsing timestamp: %w: %w", op, err, ErrChunkDecode)
+	}
+	buf = buf[timestampSize:]
+	if len(buf) != 0 {
+		return nil, fmt.Errorf("%s: extra data in chunk: %w", op, ErrChunkDecode)
+	}
+
+	b, err = NewBaseChunk(ctx, protocol, direction, timestamp, chunkType)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode)
+	}
+
+	decompressBuf := bytes.NewBuffer(databuf)
+	var decompressor io.ReadCloser
+	switch chunkType {
+	// HEAD and END are never compressed
+	case ChunkHeader, ChunkEnd:
+		decompressor = newNullCompressionReader(decompressBuf)
+	default:
+		switch d.compression {
+		case GzipCompression:
+			decompressor, err = gzip.NewReader(decompressBuf)
+			if err != nil {
+				return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode)
+			}
+		default:
+			decompressor = newNullCompressionReader(decompressBuf)
+		}
+	}
+
+	decompressed, err := io.ReadAll(decompressor)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode)
+	}
+
+	df, ok := chunkTypes.Get(b.Protocol, b.Type)
+	if !ok {
+		return nil, fmt.Errorf("%s: unsupported chunk type %s for protocol %s: %w", op, b.Type, b.Protocol, ErrChunkDecode)
+	}
+
+	c, err := df(ctx, b, decompressed)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode)
+	}
+
+	switch cc := c.(type) {
+	case *HeaderChunk:
+		d.compression = cc.Compression
+		d.encryption = cc.Encryption
+	}
+
+	return c, nil
+}

internal/bsr/decode_test.go

@@ -0,0 +1,469 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/hashicorp/boundary/internal/bsr"
+)
+
+func init() {
+	bsr.RegisterChunkType("TEST", "TEST", func(_ context.Context, bc *bsr.BaseChunk, data []byte) (bsr.Chunk, error) {
+		return &testChunk{
+			BaseChunk: bc,
+			Data:      data,
+		}, nil
+	})
+	bsr.RegisterChunkType("TEST", "ERRR", func(_ context.Context, bc *bsr.BaseChunk, data []byte) (bsr.Chunk, error) {
+		return nil, errors.New("error in decode function")
+	})
+}
+
+func TestChunkDecoder(t *testing.T) {
+	ctx := context.Background()
+
+	ts := time.Date(2023, time.March, 16, 10, 47, 3, 14, time.UTC)
+
+	cases := []struct {
+		name    string
+		encoded []byte
+		want    []bsr.Chunk
+	}{
+		{
+			"header-no-compression",
+			[]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xbe\x4c\x7c\x20" + // crc
+					"",
+			),
+			[]bsr.Chunk{
+				&bsr.HeaderChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      bsr.ChunkHeader,
+					},
+					Compression: bsr.NoCompression,
+					Encryption:  bsr.NoEncryption,
+					SessionId:   "sess_123456789",
+				},
+			},
+		},
+		{
+			"header-end-no-compression",
+			[]byte(
+				"" + // header
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xbe\x4c\x7c\x20" + // crc
+					"" + // end
+					"\x00\x00\x00\x00" + // length
+					"TEST" + // protocol
+					"DONE" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x13" + // time nanoseconds
+					"\x50\x91\xfe\x72" + // crc
+					"",
+			),
+			[]bsr.Chunk{
+				&bsr.HeaderChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      bsr.ChunkHeader,
+					},
+					Compression: bsr.NoCompression,
+					Encryption:  bsr.NoEncryption,
+					SessionId:   "sess_123456789",
+				},
+				&bsr.EndChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts.Add(time.Nanosecond * 5)),
+						Type:      bsr.ChunkEnd,
+					},
+				},
+			},
+		},
+		{
+			"header-test-end-no-compression",
+			[]byte(
+				"" + // header
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xbe\x4c\x7c\x20" + // crc
+					"" + // test
+					"\x00\x00\x00\x03" + // length
+					"TEST" + // protocol
+					"TEST" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"foo" + // data
+					"\xa4\x6e\x48\x70" + // crc
+					"" + // end
+					"\x00\x00\x00\x00" + // length
+					"TEST" + // protocol
+					"DONE" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x13" + // time nanoseconds
+					"\x50\x91\xfe\x72" + // crc
+					"",
+			),
+			[]bsr.Chunk{
+				&bsr.HeaderChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      bsr.ChunkHeader,
+					},
+					Compression: bsr.NoCompression,
+					Encryption:  bsr.NoEncryption,
+					SessionId:   "sess_123456789",
+				},
+				&testChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      "TEST",
+					},
+					Data: []byte("foo"),
+				},
+				&bsr.EndChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts.Add(time.Nanosecond * 5)),
+						Type:      bsr.ChunkEnd,
+					},
+				},
+			},
+		},
+		{
+			"header-test-end-gzip-compression",
+			[]byte(
+				"" + // header
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x01" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\x10\x24\xed\xb1" + // crc
+					"" + // test
+					"\x00\x00\x00\x1b" + // length
+					"TEST" + // protocol
+					"TEST" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					gziped("foo") + // data
+					"\x29\x8e\x22\x12" + // crc
+					"" + // end
+					"\x00\x00\x00\x00" + // length
+					"TEST" + // protocol
+					"DONE" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x13" + // time nanoseconds
+					"\x50\x91\xfe\x72" + // crc
+					"",
+			),
+			[]bsr.Chunk{
+				&bsr.HeaderChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      bsr.ChunkHeader,
+					},
+					Compression: bsr.GzipCompression,
+					Encryption:  bsr.NoEncryption,
+					SessionId:   "sess_123456789",
+				},
+				&testChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts),
+						Type:      "TEST",
+					},
+					Data: []byte("foo"),
+				},
+				&bsr.EndChunk{
+					BaseChunk: &bsr.BaseChunk{
+						Protocol:  "TEST",
+						Direction: bsr.Inbound,
+						Timestamp: bsr.NewTimestamp(ts.Add(time.Nanosecond * 5)),
+						Type:      bsr.ChunkEnd,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.encoded)
+			dec, err := bsr.NewChunkDecoder(ctx, buf)
+			require.NoError(t, err)
+
+			got := make([]bsr.Chunk, 0, len(tc.want))
+			for {
+				c, err := dec.Decode(ctx)
+				if err == io.EOF {
+					break
+				}
+				require.NoError(t, err)
+				got = append(got, c)
+			}
+
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}
+
+func TestNewChunkDecoderErrors(t *testing.T) {
+	ctx := context.Background()
+	cases := []struct {
+		name string
+		r    io.Reader
+		want error
+	}{
+		{
+			"nil-reader",
+			nil,
+			errors.New("bsr.NewChunkDecoder: reader cannot be nil: invalid parameter"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := bsr.NewChunkDecoder(ctx, tc.r)
+			assert.EqualError(t, err, tc.want.Error())
+		})
+	}
+}
+
+func TestChunkDecoderDecodeErrors(t *testing.T) {
+	ctx := context.Background()
+	cases := []struct {
+		name string
+		r    io.Reader
+		want error
+	}{
+		{
+			"unknown-chunk-type",
+			bytes.NewBuffer([]byte(
+				"" + // test
+					"\x00\x00\x00\x03" + // length
+					"TEST" + // protocol
+					"UNKW" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"foo" + // data
+					"\x01\x96\x41\x8a" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unsupported chunk type UNKW for protocol TEST: error decoding chunk"),
+		},
+		{
+			"unknown-chunk-type-for-protocol",
+			bytes.NewBuffer([]byte(
+				"" + // test
+					"\x00\x00\x00\x03" + // length
+					"UNKW" + // protocol
+					"TEST" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"foo" + // data
+					"\x38\xbd\xb9\xee" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unsupported chunk type TEST for protocol UNKW: error decoding chunk"),
+		},
+		{
+			"chuck-missing-base-fields",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unexpected EOF: error decoding chunk"),
+		},
+		{
+			"chuck-missing-data",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unexpected EOF: missing data: error decoding chunk"),
+		},
+		{
+			"chuck-partial-missing-data",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unexpected EOF: missing data: error decoding chunk"),
+		},
+		{
+			"chuck-missing-crc",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unexpected EOF: missing crc: error decoding chunk"),
+		},
+		{
+			"chuck-partial-missing-crc",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xbe\x4c\x7c" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: unexpected EOF: missing crc: error decoding chunk"),
+		},
+		{
+			"chuck-crc-mismatch",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xbe\x4c\x7c\x00" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: chunk crc did not match: error decoding chunk"),
+		},
+		{
+			"chuck-invalid-direction",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x10" + // length
+					"TEST" + // protocol
+					"HEAD" + // type
+					"\x00" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"\x00" + // compression method
+					"\x00" + // encryption method
+					"sess_123456789" + // data
+					"\xdd\x4b\xa5\x04" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: bsr.NewBaseChunk: invalid direction: invalid parameter: error decoding chunk"),
+		},
+		{
+			"chuck-decode-function-error",
+			bytes.NewBuffer([]byte(
+				"" + // so everything else aligns better
+					"\x00\x00\x00\x03" + // length
+					"TEST" + // protocol
+					"ERRR" + // type
+					"\x01" + // direction
+					"\x00\x00\x00\x00\x64\x12\xf3\xa7" + // time seconds
+					"\x00\x00\x00\x0e" + // time nanoseconds
+					"foo" + // data
+					"\x30\xd5\x69\xbb" + // crc
+					"",
+			)),
+			errors.New("bsr.(ChunkDecoder).Decode: error in decode function: error decoding chunk"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			dec, err := bsr.NewChunkDecoder(ctx, tc.r)
+			require.NoError(t, err)
+
+			_, err = dec.Decode(ctx)
+			assert.EqualError(t, err, tc.want.Error())
+		})
+	}
+}

internal/bsr/errors.go

@@ -7,18 +7,29 @@ import (
 	"errors"
 )
 
-// ErrUnknown represents an unknown error
-var ErrUnknown = errors.New("unknown error")
+var (
+	// ErrUnknown represents an unknown error
+	ErrUnknown = errors.New("unknown error")
 
-// ErrInvalidParameter represents an invalid parameter error
-var ErrInvalidParameter = errors.New("invalid parameter")
+	// ErrInvalidParameter represents an invalid parameter error
+	ErrInvalidParameter = errors.New("invalid parameter")
 
-// ErrNotSupported represents an operation that is not supported for a
-// particular protocol.
-var ErrNotSupported = errors.New("not supported by protocol")
+	// ErrBsrKeyPersistenceFailure indicates a failure in persisting BSR encryption keys
+	ErrBsrKeyPersistenceFailure = errors.New("could not persist BSR keys")
 
-// ErrBsrKeyPersistenceFailure indicates a failure in persisting BSR encryption keys
-var ErrBsrKeyPersistenceFailure = errors.New("could not persist BSR keys")
+	// ErrSummaryUnavailable indicates a BSR summary is unavailable
+	ErrSummaryUnavailable = errors.New("summary not available")
 
-// ErrSummaryUnavailable indicates a BSR summary is unavailable
-var ErrSummaryUnavailable = errors.New("summary not available")
+	// ErrNotSupported represents an operation that is not supported for a
+	// particular protocol.
+	ErrNotSupported = errors.New("not supported by protocol")
+
+	// ErrAlreadyRegistered is an error with registering chunk decoder functions.
+	ErrAlreadyRegistered = errors.New("chunk type already registered")
+
+	// ErrEndChunkNotEmpty indicates a malformed END chunk.
+	ErrEndChunkNotEmpty = errors.New("end chunk not empty")
+
+	// ErrChunkDecode indicates an error when decoding a chunk.
+	ErrChunkDecode = errors.New("error decoding chunk")
+)

internal/bsr/options.go

@@ -3,6 +3,8 @@
 
 package bsr
 
+import "github.com/hashicorp/boundary/internal/bsr/kms"
+
 // getOpts - iterate the inbound Options and return a struct
 func getOpts(opt ...Option) options {
 	opts := getDefaultOptions()
@@ -18,11 +20,13 @@ type Option func(*options)
 // options = how options are represented
 type options struct {
 	withSupportsMultiplex bool
+	withKeys              *kms.Keys
 }
 
 func getDefaultOptions() options {
 	return options{
 		withSupportsMultiplex: false,
+		withKeys:              nil,
 	}
 }
 
@@ -33,3 +37,10 @@ func WithSupportsMultiplex(b bool) Option {
 		o.withSupportsMultiplex = b
 	}
 }
+
+// WithKeys is used to provide optional kms.Keys.
+func WithKeys(k *kms.Keys) Option {
+	return func(o *options) {
+		o.withKeys = k
+	}
+}

internal/bsr/options_test.go

@@ -0,0 +1,47 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_GetOpts(t *testing.T) {
+	t.Parallel()
+
+	t.Run("default", func(t *testing.T) {
+		assert := assert.New(t)
+		opts := getOpts()
+		testOpts := options{
+			withSupportsMultiplex: false,
+			withKeys:              nil,
+		}
+		assert.Equal(opts, testOpts)
+	})
+	t.Run("WithSupportsMultiplex", func(t *testing.T) {
+		assert := assert.New(t)
+		opts := getOpts(WithSupportsMultiplex(true))
+		testOpts := getDefaultOptions()
+		testOpts.withSupportsMultiplex = true
+		assert.Equal(opts, testOpts)
+	})
+	t.Run("WithKeys", func(t *testing.T) {
+		ctx := context.Background()
+		assert := assert.New(t)
+		require := require.New(t)
+
+		keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "foo")
+		require.NoError(err)
+
+		opts := getOpts(WithKeys(keys))
+		testOpts := getDefaultOptions()
+		testOpts.withKeys = keys
+		assert.Equal(opts, testOpts)
+	})
+}

internal/bsr/timestamp.go

@@ -5,6 +5,7 @@ package bsr
 
 import (
 	"encoding/binary"
+	"fmt"
 	"time"
 )
 
@@ -43,6 +44,24 @@ func (t *Timestamp) marshal() []byte {
 	return d
 }
 
+func decodeTimestamp(data []byte) (*Timestamp, error) {
+	const op = "bsr.decodeTimestamp"
+
+	var seconds uint64
+	var nanoseconds uint32
+
+	seconds, data = binary.BigEndian.Uint64(data[:secondSize]), data[secondSize:]
+	nanoseconds, data = binary.BigEndian.Uint32(data[:nanosecondSize]), data[nanosecondSize:]
+	if len(data) != 0 {
+		return nil, fmt.Errorf("%s: extra data", op)
+	}
+
+	tt := time.Unix(int64(seconds), int64(nanoseconds)).UTC()
+
+	t := Timestamp(tt)
+	return &t, nil
+}
+
 // AsTime returns a time.Time for a Timestamp.
 func (t *Timestamp) AsTime() time.Time {
 	return time.Time(*t)