diff --git a/internal/bsr/chunk_scanner.go b/internal/bsr/chunk_scanner.go new file mode 100644 index 000000000..699c79850 --- /dev/null +++ b/internal/bsr/chunk_scanner.go @@ -0,0 +1,120 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package bsr + +import ( + "context" + "fmt" + "io" + + "github.com/hashicorp/boundary/internal/bsr/internal/is" + "github.com/hashicorp/go-kms-wrapping/v2/extras/crypto" +) + +// ChunkScanner can be used to read a Chunk at a time. +type ChunkScanner struct { + checksum []byte + reader *crypto.Sha256SumReader + chunkDecoder *ChunkDecoder +} + +// NewChunkScanner creates a ChunkScanner. The scanner will calculate a rolling +// sha256sum of all of the chunks that have been read. +// Supports the following options: +// - WithSha256Sum: This is used to provide an expected sha256sum. Once the +// scanner encounters an END chunk or an io.EOF error, it will compare the +// calculated sha256sum against this sum. If the sums do not match, ErrChecksum +// will be returned. +// +// Other options are passed through to the ChunkDecoder used by the scanner. +func NewChunkScanner(ctx context.Context, r io.Reader, options ...Option) (*ChunkScanner, error) { + const op = "bsr.NewChunkScanner" + + if is.Nil(r) { + return nil, fmt.Errorf("%s: reader is nil: %w", op, ErrInvalidParameter) + } + + sha256Reader, err := crypto.NewSha256SumReader(ctx, r) + if err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + + if err := ReadMagic(sha256Reader); err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + + cd, err := NewChunkDecoder(ctx, sha256Reader, options...) + if err != nil { + return nil, fmt.Errorf("%s: %w", op, err) + } + + opts := getOpts(options...) + + return &ChunkScanner{ + checksum: opts.withSha256Sum, + reader: sha256Reader, + chunkDecoder: cd, + }, nil +} + +// Scan reads the next Chunk from the reader. +// If the scanner was created using WithSha256Sum, the calculated sum will +// be compared to the provided sum when the scanner encounters either an END +// Chunk, or an io.EOF error. If the sums do not match, ErrChecksum will be +// returned. +func (cs *ChunkScanner) Scan(ctx context.Context) (Chunk, error) { + const op = "bsr.(ChunkScanner).Scan" + + c, err := cs.chunkDecoder.Decode(ctx) + if (err == io.EOF || + (err == nil && c.GetType() == ChunkEnd)) && + len(cs.checksum) > 0 { + sum, err := cs.Sum(ctx) + if err != nil { + return c, fmt.Errorf("%s: %w", op, err) + } + + if string(cs.checksum) != string(sum) { + return c, fmt.Errorf("%s: %w", op, ErrChecksum) + } + } + + return c, err +} + +// Close closes the scanner's reader. +func (cs *ChunkScanner) Close() error { + return cs.reader.Close() +} + +// Sum returns a hex encoded sha256sum of all of the chunks that have been +// scanned. +func (cs *ChunkScanner) Sum(ctx context.Context) ([]byte, error) { + return cs.reader.Sum(ctx, crypto.WithHexEncoding(true)) +} + +// ChunkReadFunc is a function that can be used by ChunkWalk to process a Chunk. +type ChunkReadFunc func(ctx context.Context, c Chunk) error + +// ChunkWalk will step through the chunks returned by the ChunkScanner and call +// the provided ChunkReadFunc f for each. If f returns an error or a non io.EOF +// error is returned from the scanner the walk will terminate early. Otherwise +// the walk will terminate once io.EOF is reached. +func ChunkWalk(ctx context.Context, s *ChunkScanner, f ChunkReadFunc) error { + const op = "bsr.ChunkWalk" + + for { + c, err := s.Scan(ctx) + if err == io.EOF { + return nil + } + if err != nil { + return fmt.Errorf("%s: %w", op, err) + } + + if ferr := f(ctx, c); ferr != nil { + return fmt.Errorf("%s: %w", op, ferr) + } + } +} diff --git a/internal/bsr/chunk_scanner_test.go b/internal/bsr/chunk_scanner_test.go new file mode 100644 index 000000000..5572caafe --- /dev/null +++ b/internal/bsr/chunk_scanner_test.go @@ -0,0 +1,445 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package bsr_test + +import ( + "bytes" + "context" + "fmt" + "io" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/bsr" + "github.com/hashicorp/go-kms-wrapping/v2/extras/crypto" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewChunkScanner(t *testing.T) { + ctx := context.Background() + cases := []struct { + name string + reader io.Reader + opts []bsr.Option + wantErr error + }{ + { + "valid", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "", + )), + nil, + nil, + }, + { + "valid-with-sha256sum", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "", + )), + []bsr.Option{bsr.WithSha256Sum([]byte("797cf38f8e3efa3da3ae95449466ffd62cfe084abd1c1134f9e74474e79570f0"))}, + nil, + }, + { + "nil-reader", + nil, + []bsr.Option{bsr.WithSha256Sum([]byte("797cf38f8e3efa3da3ae95449466ffd62cfe084abd1c1134f9e74474e79570f0"))}, + fmt.Errorf("bsr.NewChunkScanner: reader is nil: invalid parameter"), + }, + { + "no-magic", + bytes.NewBuffer([]byte("NOT_BSR_MAGIC")), + nil, + fmt.Errorf("bsr.NewChunkScanner: bsr.ReadMagic: invalid magic string"), + }, + { + "not-enough-data", + bytes.NewBuffer(bsr.Magic.Bytes()[:2]), + nil, + fmt.Errorf("bsr.NewChunkScanner: bsr.ReadMagic: crypto.(Sha256SumReader).Read: EOF"), + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := bsr.NewChunkScanner(ctx, tc.reader, tc.opts...) + if tc.wantErr != nil { + require.EqualError(t, err, tc.wantErr.Error()) + return + } + require.NoError(t, err) + }) + } +} + +func TestChunkScanner(t *testing.T) { + ctx := context.Background() + + ts := time.Date(2023, time.March, 16, 10, 47, 3, 14, time.UTC) + + cases := []struct { + name string + reader io.Reader + opts []bsr.Option + want bsr.Chunk + wantErr error + }{ + { + "header", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "\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 + "", + )), + nil, + &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, + }, + { + "checksum-eof", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "\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.Option{bsr.WithSha256Sum([]byte("foo"))}, + &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, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + scanner, err := bsr.NewChunkScanner(ctx, tc.reader, tc.opts...) + require.NoError(t, err) + + got, err := scanner.Scan(ctx) + if tc.wantErr != nil { + require.EqualError(t, err, tc.wantErr.Error()) + return + } + require.Equal(t, tc.want, got) + }) + } +} + +func TestChunkWalk(t *testing.T) { + ctx := context.Background() + + ts := time.Date(2023, time.March, 16, 10, 47, 3, 14, time.UTC) + + headerTest := []byte( + string(bsr.Magic) + + "\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 + "", + ) + headerTestEnd := append(headerTest, []byte( + ""+ // 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 + "", + )...) + + cases := []struct { + name string + reader io.Reader + opts []bsr.Option + f func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc + want []bsr.Chunk + wantErr error + }{ + { + "no-chunk", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "", + )), + nil, + func(t *testing.T, _ *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + assert.Fail(t, "read func should not be called") + return nil + } + }, + []bsr.Chunk{}, + nil, + }, + { + "header-test-end", + bytes.NewBuffer(headerTestEnd), + nil, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + *got = append(*got, c) + return nil + } + }, + []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, + }, + }, + }, + nil, + }, + { + "header-test-end-checksumed", + bytes.NewBuffer(headerTestEnd), + []bsr.Option{bsr.WithSha256Sum(func() []byte { + b, err := crypto.Sha256Sum(ctx, bytes.NewBuffer(headerTestEnd), crypto.WithHexEncoding(true)) + require.NoError(t, err) + return b + }())}, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + *got = append(*got, c) + return nil + } + }, + []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, + }, + }, + }, + nil, + }, + { + "header-test-no-end", + bytes.NewBuffer(headerTest), + nil, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + *got = append(*got, c) + return nil + } + }, + []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"), + }, + }, + nil, + }, + { + "header-test-no-end-checksumed", + bytes.NewBuffer(headerTest), + []bsr.Option{bsr.WithSha256Sum(func() []byte { + b, err := crypto.Sha256Sum(ctx, bytes.NewBuffer(headerTest), crypto.WithHexEncoding(true)) + require.NoError(t, err) + return b + }())}, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + *got = append(*got, c) + return nil + } + }, + []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"), + }, + }, + nil, + }, + { + "func-errors", + bytes.NewBuffer([]byte( + string(bsr.Magic) + + "\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 + "", + )), + nil, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + return fmt.Errorf("error in walk function") + } + }, + nil, + fmt.Errorf("bsr.ChunkWalk: error in walk function"), + }, + { + "incorrect-checksum", + bytes.NewBuffer(headerTestEnd), + []bsr.Option{bsr.WithSha256Sum([]byte("f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd2"))}, + func(t *testing.T, got *[]bsr.Chunk) bsr.ChunkReadFunc { + return func(_ context.Context, c bsr.Chunk) error { + *got = append(*got, c) + return nil + } + }, + nil, + fmt.Errorf("bsr.ChunkWalk: bsr.(ChunkScanner).Scan: computed checksum did NOT match"), + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + scanner, err := bsr.NewChunkScanner(ctx, tc.reader, tc.opts...) + require.NoError(t, err) + got := make([]bsr.Chunk, 0, len(tc.want)) + err = bsr.ChunkWalk(ctx, scanner, tc.f(t, &got)) + if tc.wantErr != nil { + require.EqualError(t, err, tc.wantErr.Error()) + return + } + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} diff --git a/internal/bsr/decode.go b/internal/bsr/decode.go index 22dfc787f..c0383d708 100644 --- a/internal/bsr/decode.go +++ b/internal/bsr/decode.go @@ -8,6 +8,7 @@ import ( "compress/gzip" "context" "encoding/binary" + "errors" "fmt" "hash/crc32" "io" @@ -111,8 +112,8 @@ func (d *ChunkDecoder) Decode(ctx context.Context) (Chunk, error) { _, err := io.ReadAtLeast(d.r, buf, chunkBaseSize) if err != nil { - if err == io.EOF { - return nil, err + if err == io.EOF || errors.Is(err, io.EOF) { + return nil, io.EOF } return nil, fmt.Errorf("%s: %w: %w", op, err, ErrChunkDecode) } diff --git a/internal/bsr/errors.go b/internal/bsr/errors.go index 386122fca..56923f06f 100644 --- a/internal/bsr/errors.go +++ b/internal/bsr/errors.go @@ -32,4 +32,10 @@ var ( // ErrChunkDecode indicates an error when decoding a chunk. ErrChunkDecode = errors.New("error decoding chunk") + + // ErrInvalidMagic is used when a binary bsr file is missing the magic string. + ErrInvalidMagic = errors.New("invalid magic string") + + // ErrChecksum indicates that a checksum did not match. + ErrChecksum = errors.New("computed checksum did NOT match") ) diff --git a/internal/bsr/magic.go b/internal/bsr/magic.go index e828b7ef2..f42379d05 100644 --- a/internal/bsr/magic.go +++ b/internal/bsr/magic.go @@ -3,6 +3,13 @@ package bsr +import ( + "fmt" + "io" + + "github.com/hashicorp/boundary/internal/bsr/internal/is" +) + const ( // Magic is the magic string / magic number / file signature used to // identify a BSR data file. @@ -19,3 +26,25 @@ type magic string func (s magic) Bytes() []byte { return []byte(s) } + +// ReadMagic attempts to read the magic string from the given io.Reader. +// If it is unable to read enough bytes, or if the magic string does not match +// an error is returned. +func ReadMagic(r io.Reader) error { + const op = "bsr.ReadMagic" + + if is.Nil(r) { + return fmt.Errorf("%s: reader is nil: %w", op, ErrInvalidParameter) + } + + m := make([]byte, magicSize) + _, err := io.ReadAtLeast(r, m, magicSize) + if err != nil { + return fmt.Errorf("%s: %w", op, err) + } + + if len(m) < magicSize || magic(m[:magicSize]) != Magic { + return fmt.Errorf("%s: %w", op, ErrInvalidMagic) + } + return nil +} diff --git a/internal/bsr/magic_test.go b/internal/bsr/magic_test.go index 5176f2557..949e37e5b 100644 --- a/internal/bsr/magic_test.go +++ b/internal/bsr/magic_test.go @@ -4,6 +4,9 @@ package bsr_test import ( + "bytes" + "errors" + "io" "testing" "github.com/hashicorp/boundary/internal/bsr" @@ -14,3 +17,43 @@ func TestMagic(t *testing.T) { assert.Equal(t, string(bsr.Magic), "\x89BSR\r\n\x1a\n") assert.Equal(t, bsr.Magic.Bytes(), []byte("\x89BSR\r\n\x1a\n")) } + +func TestReadMagic(t *testing.T) { + cases := []struct { + name string + r io.Reader + want error + }{ + { + "valid", + bytes.NewBuffer([]byte(bsr.Magic)), + nil, + }, + { + "nil-reader", + nil, + errors.New("bsr.ReadMagic: reader is nil: invalid parameter"), + }, + { + "not-enough-chars", + bytes.NewBuffer([]byte(bsr.Magic)[:len(bsr.Magic)-1]), + errors.New("bsr.ReadMagic: unexpected EOF"), + }, + { + "not-magic", + bytes.NewBuffer([]byte("notthemagic")), + errors.New("bsr.ReadMagic: invalid magic string"), + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := bsr.ReadMagic(tc.r) + if tc.want == nil { + assert.NoError(t, got) + return + } + assert.EqualError(t, got, tc.want.Error()) + }) + } +} diff --git a/internal/bsr/options.go b/internal/bsr/options.go index b5b42b869..f994dba3c 100644 --- a/internal/bsr/options.go +++ b/internal/bsr/options.go @@ -21,12 +21,14 @@ type Option func(*options) type options struct { withSupportsMultiplex bool withKeys *kms.Keys + withSha256Sum []byte } func getDefaultOptions() options { return options{ withSupportsMultiplex: false, withKeys: nil, + withSha256Sum: nil, } } @@ -44,3 +46,10 @@ func WithKeys(k *kms.Keys) Option { o.withKeys = k } } + +// WithSha256Sum is used to provide a hex encoded SHA256SUM. +func WithSha256Sum(b []byte) Option { + return func(o *options) { + o.withSha256Sum = b + } +} diff --git a/internal/bsr/options_test.go b/internal/bsr/options_test.go index a9d6a1a2d..95d7a810b 100644 --- a/internal/bsr/options_test.go +++ b/internal/bsr/options_test.go @@ -21,6 +21,7 @@ func Test_GetOpts(t *testing.T) { testOpts := options{ withSupportsMultiplex: false, withKeys: nil, + withSha256Sum: nil, } assert.Equal(opts, testOpts) }) @@ -44,4 +45,13 @@ func Test_GetOpts(t *testing.T) { testOpts.withKeys = keys assert.Equal(opts, testOpts) }) + t.Run("WithSha256Sum", func(t *testing.T) { + // echo "test" | sha256sum + sum := []byte(`f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd2`) + assert := assert.New(t) + opts := getOpts(WithSha256Sum(sum)) + testOpts := getDefaultOptions() + testOpts.withSha256Sum = sum + assert.Equal(opts, testOpts) + }) }