diff --git a/internal/bsr/chunk_end.go b/internal/bsr/chunk_end.go index a7b9fed6e..5dca3e536 100644 --- a/internal/bsr/chunk_end.go +++ b/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 +} diff --git a/internal/bsr/chunk_end_test.go b/internal/bsr/chunk_end_test.go index 6bbc19afc..6bfe787b9 100644 --- a/internal/bsr/chunk_end_test.go +++ b/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) + }) + } +} diff --git a/internal/bsr/chunk_header.go b/internal/bsr/chunk_header.go index 0a2383bb4..6664a2c3a 100644 --- a/internal/bsr/chunk_header.go +++ b/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 +} diff --git a/internal/bsr/chunk_header_test.go b/internal/bsr/chunk_header_test.go index ff0489a2b..4c689b5f0 100644 --- a/internal/bsr/chunk_header_test.go +++ b/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) + }) + } +} diff --git a/internal/bsr/compression.go b/internal/bsr/compression.go index 6e730c8ba..55bf1e3d0 100644 --- a/internal/bsr/compression.go +++ b/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} +} diff --git a/internal/bsr/decode.go b/internal/bsr/decode.go new file mode 100644 index 000000000..22dfc787f --- /dev/null +++ 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 +} diff --git a/internal/bsr/decode_test.go b/internal/bsr/decode_test.go new file mode 100644 index 000000000..8f8245c49 --- /dev/null +++ b/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()) + }) + } +} diff --git a/internal/bsr/errors.go b/internal/bsr/errors.go index 7299a431d..386122fca 100644 --- a/internal/bsr/errors.go +++ b/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") +) diff --git a/internal/bsr/options.go b/internal/bsr/options.go index cfcd434e3..b5b42b869 100644 --- a/internal/bsr/options.go +++ b/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 + } +} diff --git a/internal/bsr/options_test.go b/internal/bsr/options_test.go new file mode 100644 index 000000000..a9d6a1a2d --- /dev/null +++ b/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) + }) +} diff --git a/internal/bsr/timestamp.go b/internal/bsr/timestamp.go index 30bea681a..259e5de02 100644 --- a/internal/bsr/timestamp.go +++ b/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)