CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(bsr): Add ChunkScanner

Browse Source 
The scanner can be used to read any BSR data file. When creating a new
scanner, it will first verify the BSR magic string. Individual Chunks
can then be read back by calling Scan. The scanner will compute a
sha256sum while it is reading chunks. If the scanner was created using
the WithSha256Sum option, it will compare the computed checksum with the
one provided via the option once the scanner encounters an end chunk
(DONE) or an io.EOF error from the reader. If the checksums do not
match, and error will be returned from Scan. The current sha256sum can
be returned from the scanner using the Sum method.

There is also a convenience function, ChunkWalk that can be used to call
a function on each Chunk returned by a scanner.
This commit is contained in:
Timothy Messier committed 2023-05-31 12:18:20 +00:00
1 parent f983a05a3e
commit 5d0a9b706e
8 files changed
+665 -2
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/bsr/chunk_scanner.go
+120
View File
@@ -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)
}
}
}
internal/bsr/chunk_scanner_test.go
+445
View File
@@ -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)
})
}
}
internal/bsr/decode.go
+3 -2
View File
@@ -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)
}
internal/bsr/errors.go
+6
View File
@@ -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")
)
internal/bsr/magic.go
+29
View File
@@ -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
}
internal/bsr/magic_test.go
+43
View File
@@ -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())
})
}
}
internal/bsr/options.go
+9
View File
@@ -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
}
}
internal/bsr/options_test.go
+10
View File
@@ -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)
})
}