feat(bsr): Define container and the required fs interfaces

Co-authored-by: Irena Rindos <irenarindos@users.noreply.github.com>

internal/bsr/bsr.go

@@ -3,12 +3,261 @@
 
 package bsr
 
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/checksum"
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	"github.com/hashicorp/boundary/internal/storage"
+)
+
+const (
+	bsrFile        = "%s.bsr"
+	connectionFile = "%s.connection"
+	channelFile    = "%s.channel"
+	messagesFile   = "messages-%s.data"
+	requestsFile   = "requests-%s.data"
+)
+
 // SessionMeta contains metadata about a session in a BSR.
 type SessionMeta struct {
 	Id string
 }
 
+// Session is the top level container in a bsr that contains the files for
+// a recorded session.
+type Session struct {
+	*container
+	multiplexed bool
+
+	Meta *SessionMeta
+}
+
+// NewSession creates a Session container for a given session id.
+func NewSession(ctx context.Context, meta *SessionMeta, f storage.FS, keys *kms.Keys, options ...Option) (*Session, error) {
+	const op = "bsr.NewSession"
+
+	switch {
+	case is.Nil(meta):
+		return nil, fmt.Errorf("%s: missing session meta: %w", op, ErrInvalidParameter)
+	case meta.Id == "":
+		return nil, fmt.Errorf("%s: missing session id: %w", op, ErrInvalidParameter)
+	case is.Nil(f):
+		return nil, fmt.Errorf("%s: missing storage fs: %w", op, ErrInvalidParameter)
+	case is.Nil(keys):
+		return nil, fmt.Errorf("%s: missing kms keys: %w", op, ErrInvalidParameter)
+	}
+
+	opts := getOpts(options...)
+
+	c, err := f.New(ctx, fmt.Sprintf(bsrFile, meta.Id))
+	if err != nil {
+		return nil, err
+	}
+
+	nc, err := newContainer(ctx, sessionContainer, c, keys)
+	if err != nil {
+		return nil, err
+	}
+	_, err = nc.WriteMeta(ctx, "id", meta.Id)
+	if err != nil {
+		return nil, err
+	}
+	return &Session{
+		container:   nc,
+		multiplexed: opts.withSupportsMultiplex,
+		Meta:        meta,
+	}, nil
+}
+
+// NewConnection creates a Connection container for a given connection id.
+func (s *Session) NewConnection(ctx context.Context, meta *ConnectionMeta) (*Connection, error) {
+	const op = "bsr.(Session).NewConnection"
+
+	switch {
+	case is.Nil(meta):
+		return nil, fmt.Errorf("%s: missing connection meta: %w", op, ErrInvalidParameter)
+	case meta.Id == "":
+		return nil, fmt.Errorf("%s: missing connection id: %w", op, ErrInvalidParameter)
+	}
+
+	name := fmt.Sprintf(connectionFile, meta.Id)
+	sc, err := s.container.container.SubContainer(ctx, name)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := s.WriteMeta(ctx, "connection", name); err != nil {
+		return nil, err
+	}
+
+	nc, err := newContainer(ctx, connectionContainer, sc, s.keys)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := nc.WriteMeta(ctx, "id", meta.Id); err != nil {
+		return nil, err
+	}
+	return &Connection{
+		container:   nc,
+		multiplexed: s.multiplexed,
+		Meta:        meta,
+	}, nil
+}
+
+// Close closes the Session container.
+func (s *Session) Close(ctx context.Context) error {
+	return s.container.close(ctx)
+}
+
 // ConnectionMeta contains metadata about a connection in a BSR.
 type ConnectionMeta struct {
 	Id string
 }
+
+// Connection is a container in a bsr for a specific connection in a session
+// container. It contains the files for the recorded connection.
+type Connection struct {
+	*container
+	multiplexed bool
+
+	Meta *ConnectionMeta
+}
+
+// NewChannel creates a Channel container for a given channel id.
+func (c *Connection) NewChannel(ctx context.Context, meta *ChannelMeta) (*Channel, error) {
+	const op = "bsr.(Connection).NewChannel"
+
+	if !c.multiplexed {
+		return nil, fmt.Errorf("%s: connection cannot make channels: %w", op, ErrNotSupported)
+	}
+
+	switch {
+	case is.Nil(meta):
+		return nil, fmt.Errorf("%s: missing channel meta: %w", op, ErrInvalidParameter)
+	case meta.Id == "":
+		return nil, fmt.Errorf("%s: missing channel id: %w", op, ErrInvalidParameter)
+	}
+
+	name := fmt.Sprintf(channelFile, meta.Id)
+	sc, err := c.container.container.SubContainer(ctx, name)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := c.WriteMeta(ctx, "channel", name); err != nil {
+		return nil, err
+	}
+	nc, err := newContainer(ctx, channelContainer, sc, c.keys)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := nc.WriteMeta(ctx, "id", meta.Id); err != nil {
+		return nil, err
+	}
+	return &Channel{
+		container: nc,
+		Meta:      meta,
+	}, nil
+}
+
+// NewMessagesWriter creates a writer for recording channel messages.
+func (c *Connection) NewMessagesWriter(ctx context.Context, dir Direction) (io.Writer, error) {
+	const op = "bsr.(Connection).NewMessagesWriter"
+
+	switch {
+	case !ValidDirection(dir):
+		return nil, fmt.Errorf("%s: invalid direction: %w", op, ErrInvalidParameter)
+	}
+
+	messagesName := fmt.Sprintf(messagesFile, dir.String())
+	c.container.WriteMeta(ctx, "messages", dir.String())
+	m, err := c.container.create(ctx, messagesName)
+	if err != nil {
+		return nil, err
+	}
+
+	return checksum.NewFile(ctx, m, c.checksums)
+}
+
+// NewRequestsWriter creates a writer for recording connection requests.
+func (c *Connection) NewRequestsWriter(ctx context.Context, dir Direction) (io.Writer, error) {
+	const op = "bsr.(Connection).NewRequestsWriter"
+
+	switch {
+	case !ValidDirection(dir):
+		return nil, fmt.Errorf("%s: invalid direction: %w", op, ErrInvalidParameter)
+	}
+
+	requestName := fmt.Sprintf(requestsFile, dir.String())
+	c.container.WriteMeta(ctx, "requests", dir.String())
+	m, err := c.container.create(ctx, requestName)
+	if err != nil {
+		return nil, err
+	}
+
+	return checksum.NewFile(ctx, m, c.checksums)
+}
+
+// Close closes the Connection container.
+func (c *Connection) Close(ctx context.Context) error {
+	return c.container.close(ctx)
+}
+
+// ChannelMeta contains metadata about a channel in a BSR.
+type ChannelMeta struct {
+	Id   string
+	Type string
+}
+
+// Channel is a container in a bsr for a specific channel in a session
+// container. It contains the files for the recorded channel.
+type Channel struct {
+	*container
+
+	Meta *ChannelMeta
+}
+
+// Close closes the Channel container.
+func (c *Channel) Close(ctx context.Context) error {
+	return c.container.close(ctx)
+}
+
+// NewMessagesWriter creates a writer for recording channel messages.
+func (c *Channel) NewMessagesWriter(ctx context.Context, dir Direction) (io.Writer, error) {
+	const op = "bsr.(Channel).NewMessagesWriter"
+
+	switch {
+	case !ValidDirection(dir):
+		return nil, fmt.Errorf("%s: invalid direction: %w", op, ErrInvalidParameter)
+	}
+
+	messagesName := fmt.Sprintf(messagesFile, dir.String())
+	c.container.WriteMeta(ctx, "messages", dir.String())
+	m, err := c.container.create(ctx, messagesName)
+	if err != nil {
+		return nil, err
+	}
+
+	return checksum.NewFile(ctx, m, c.checksums)
+}
+
+// NewRequestsWriter creates a writer for recording channel requests.
+func (c *Channel) NewRequestsWriter(ctx context.Context, dir Direction) (io.Writer, error) {
+	const op = "bsr.(Channel).NewRequestsWriter"
+
+	switch {
+	case !ValidDirection(dir):
+		return nil, fmt.Errorf("%s: invalid direction: %w", op, ErrInvalidParameter)
+	}
+
+	requestName := fmt.Sprintf(requestsFile, dir.String())
+	c.container.WriteMeta(ctx, "requests", dir.String())
+	m, err := c.container.create(ctx, requestName)
+	if err != nil {
+		return nil, err
+	}
+
+	return checksum.NewFile(ctx, m, c.checksums)
+}

internal/bsr/bsr_test.go

@@ -0,0 +1,575 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/hashicorp/boundary/internal/bsr"
+	"github.com/hashicorp/boundary/internal/bsr/internal/fstest"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	"github.com/hashicorp/boundary/internal/storage"
+	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+func assertContainer(ctx context.Context, t *testing.T, path, state string, typ string, fs *fstest.MemContainer, keys *kms.Keys) {
+	t.Helper()
+
+	td := filepath.Join("testdata", t.Name(), state, path)
+
+	// meta
+	wantMeta, err := os.ReadFile(filepath.Join(td, string(typ)+".meta"))
+	require.NoError(t, err, "unable to find test data file")
+	meta, ok := fs.Files[string(typ)+".meta"]
+	require.True(t, ok, "container is missing meta file")
+	assert.Equal(t, string(wantMeta), meta.Buf.String())
+
+	// summary
+	wantSummary, err := os.ReadFile(filepath.Join(td, string(typ)+".summary"))
+	require.NoError(t, err, "unable to find test data file")
+	summary, ok := fs.Files[string(typ)+".summary"]
+	require.True(t, ok, "container is missing summary file")
+	assert.Equal(t, string(wantSummary), summary.Buf.String())
+
+	// SHA256SUM checksums
+	wantChecksums, err := os.ReadFile(filepath.Join(td, "SHA256SUM"))
+	require.NoError(t, err, "unable to find test data file")
+	checksums, ok := fs.Files["SHA256SUM"]
+	require.True(t, ok, "container is missing checksums file")
+	assert.Equal(t, string(wantChecksums), checksums.Buf.String())
+
+	// SHA256SUM.sig signature file
+	sig, ok := fs.Files["SHA256SUM.sig"]
+	require.True(t, ok, "container is missing sig file")
+	switch state {
+	case "closed":
+		want, err := keys.SignWithPrivKey(ctx, wantChecksums)
+		require.NoError(t, err)
+
+		got := &wrapping.SigInfo{}
+		err = proto.Unmarshal(sig.Buf.Bytes(), got)
+		require.NoError(t, err)
+
+		assert.Empty(t,
+			cmp.Diff(
+				want,
+				got,
+				cmpopts.IgnoreUnexported(wrapping.SigInfo{}, wrapping.KeyInfo{}),
+			),
+		)
+	default:
+		assert.Equal(t, "", sig.Buf.String())
+	}
+
+	// journal
+	wantJournal, err := os.ReadFile(filepath.Join(td, ".journal"))
+	require.NoError(t, err, "unable to find test data file")
+	journal, ok := fs.Files[".journal"]
+	require.True(t, ok, "container is missing journal file")
+	assert.Equal(t, string(wantJournal), journal.Buf.String())
+}
+
+type connection struct {
+	mem  *fstest.MemContainer
+	conn *bsr.Connection
+	id   string
+
+	channels []*channel
+	files    []*file
+}
+
+type channel struct {
+	mem     *fstest.MemContainer
+	channel *bsr.Channel
+	id      string
+
+	files []*file
+}
+
+type file struct {
+	mem  *fstest.MemFile
+	file io.Writer
+}
+
+type createConn struct {
+	id       string
+	channels []createChannel
+	files    []createFile
+}
+
+type createChannel struct {
+	id    string
+	files []createFile
+}
+
+type createFile struct {
+	typ string
+	dir bsr.Direction
+}
+
+func TestBsr(t *testing.T) {
+	ctx := context.Background()
+
+	cases := []struct {
+		name  string
+		id    string
+		opts  []bsr.Option
+		c     *fstest.MemFS
+		keys  *kms.Keys
+		conns []createConn
+	}{
+		{
+			"session_not_multiplexed",
+			"session_123456789",
+			[]bsr.Option{},
+			fstest.NewMemFS(),
+			func() *kms.Keys {
+				keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session_123456789")
+				require.NoError(t, err)
+				return keys
+			}(),
+			[]createConn{
+				{
+					"conn_1",
+					nil,
+					[]createFile{
+						{"messages", bsr.Inbound},
+						{"messages", bsr.Outbound},
+						{"requests", bsr.Inbound},
+						{"requests", bsr.Outbound},
+					},
+				},
+			},
+		},
+		{
+			"session_multiplexed",
+			"session_123456789",
+			[]bsr.Option{bsr.WithSupportsMultiplex(true)},
+			fstest.NewMemFS(),
+			func() *kms.Keys {
+				keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session_123456789")
+				require.NoError(t, err)
+				return keys
+			}(),
+			[]createConn{
+				{
+					"conn_1",
+					[]createChannel{
+						{
+							"chan_1",
+							[]createFile{
+								{"messages", bsr.Inbound},
+								{"requests", bsr.Inbound},
+								{"messages", bsr.Outbound},
+								{"requests", bsr.Outbound},
+							},
+						},
+					},
+					[]createFile{
+						{"requests", bsr.Inbound},
+						{"requests", bsr.Outbound},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: tc.id}, tc.c, tc.keys, tc.opts...)
+			require.NoError(t, err)
+			require.NotNil(t, s)
+
+			sContainer, ok := tc.c.Containers[tc.id+".bsr"]
+			require.True(t, ok)
+
+			assertContainer(ctx, t, "", "opened", "session", sContainer, tc.keys)
+
+			createdConnections := make([]*connection, 0)
+
+			// create all the things
+			for _, conn := range tc.conns {
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: conn.id})
+				require.NoError(t, err)
+				require.NotNil(t, c)
+
+				cContainer, ok := sContainer.Sub[conn.id+".connection"]
+				require.True(t, ok)
+
+				assertContainer(ctx, t, conn.id, "opened", "connection", cContainer, tc.keys)
+
+				ff := make([]*file, 0, len(conn.files))
+				for _, f := range conn.files {
+					var w io.Writer
+					var err error
+					switch f.typ {
+					case "messages":
+						w, err = c.NewMessagesWriter(ctx, f.dir)
+					case "requests":
+						w, err = c.NewRequestsWriter(ctx, f.dir)
+					}
+					require.NoError(t, err)
+
+					fname := fmt.Sprintf("%s-%s.data", f.typ, f.dir.String())
+					memf, ok := cContainer.Files[fname]
+					require.True(t, ok, "file %s not in container %s", fname, cContainer.Name)
+
+					require.NoError(t, err)
+					ff = append(ff, &file{
+						mem:  memf,
+						file: w,
+					})
+				}
+
+				createdChannels := make([]*channel, 0, len(conn.channels))
+				for _, chann := range conn.channels {
+					ch, err := c.NewChannel(ctx, &bsr.ChannelMeta{Id: chann.id})
+					require.NoError(t, err)
+					require.NotNil(t, ch)
+
+					chContainer, ok := cContainer.Sub[chann.id+".channel"]
+					require.True(t, ok)
+
+					assertContainer(ctx, t, filepath.Join(conn.id, chann.id), "opened", "channel", chContainer, tc.keys)
+
+					ff := make([]*file, 0, len(chann.files))
+					for _, f := range chann.files {
+						var w io.Writer
+						var err error
+						switch f.typ {
+						case "messages":
+							w, err = ch.NewMessagesWriter(ctx, f.dir)
+						case "requests":
+							w, err = ch.NewRequestsWriter(ctx, f.dir)
+						}
+						require.NoError(t, err)
+
+						fname := fmt.Sprintf("%s-%s.data", f.typ, f.dir.String())
+						memf, ok := chContainer.Files[fname]
+						require.True(t, ok, "file %s not in container %s", fname, chContainer.Name)
+
+						require.NoError(t, err)
+						ff = append(ff, &file{
+							mem:  memf,
+							file: w,
+						})
+					}
+					createdChannels = append(createdChannels, &channel{
+						mem:     chContainer,
+						channel: ch,
+						id:      chann.id,
+						files:   ff,
+					})
+				}
+				createdConnections = append(createdConnections, &connection{
+					mem:      cContainer,
+					conn:     c,
+					id:       conn.id,
+					channels: createdChannels,
+					files:    ff,
+				})
+			}
+
+			// now close all the things that where created.
+			for _, conn := range createdConnections {
+				for _, channel := range conn.channels {
+					for _, f := range channel.files {
+						v, ok := f.file.(io.Closer)
+						require.True(t, ok, "file is not a io.Closer")
+						err = v.Close()
+						require.NoError(t, err)
+					}
+					err = channel.channel.Close(ctx)
+					require.NoError(t, err)
+
+					assertContainer(ctx, t, filepath.Join(conn.id, channel.id), "closed", "channel", channel.mem, tc.keys)
+				}
+
+				for _, f := range conn.files {
+					v, ok := f.file.(io.Closer)
+					require.True(t, ok, "file is not a io.Closer")
+					err = v.Close()
+					require.NoError(t, err)
+				}
+
+				err = conn.conn.Close(ctx)
+				require.NoError(t, err)
+				assertContainer(ctx, t, conn.id, "closed", "connection", conn.mem, tc.keys)
+			}
+
+			err = s.Close(ctx)
+			require.NoError(t, err)
+
+			assertContainer(ctx, t, "", "closed", "session", sContainer, tc.keys)
+		})
+	}
+}
+
+func TestNewSessionErrors(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	cases := []struct {
+		name      string
+		meta      *bsr.SessionMeta
+		f         storage.FS
+		keys      *kms.Keys
+		wantError error
+	}{
+		{
+			"nil-meta",
+			nil,
+			&fstest.MemFS{},
+			keys,
+			errors.New("bsr.NewSession: missing session meta: invalid parameter"),
+		},
+		{
+			"empty-session-id",
+			&bsr.SessionMeta{Id: ""},
+			&fstest.MemFS{},
+			keys,
+			errors.New("bsr.NewSession: missing session id: invalid parameter"),
+		},
+		{
+			"nil-fs",
+			&bsr.SessionMeta{Id: "session"},
+			nil,
+			keys,
+			errors.New("bsr.NewSession: missing storage fs: invalid parameter"),
+		},
+		{
+			"nil-keys",
+			&bsr.SessionMeta{Id: "session"},
+			&fstest.MemFS{},
+			nil,
+			errors.New("bsr.NewSession: missing kms keys: invalid parameter"),
+		},
+		{
+			"fs-new-error",
+			&bsr.SessionMeta{Id: "session"},
+			fstest.NewMemFS(fstest.WithNewFunc(func(_ context.Context, _ string) (storage.Container, error) {
+				return nil, errors.New("fs new error")
+			})),
+			keys,
+			errors.New("fs new error"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := bsr.NewSession(ctx, tc.meta, tc.f, tc.keys)
+			require.Error(t, err)
+			assert.EqualError(t, err, tc.wantError.Error())
+		})
+	}
+}
+
+func TestNewConnectionErrors(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	cases := []struct {
+		name      string
+		session   *bsr.Session
+		meta      *bsr.ConnectionMeta
+		wantError error
+	}{
+		{
+			"nil-meta",
+			func() *bsr.Session {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys)
+				require.NoError(t, err)
+				return s
+			}(),
+			nil,
+			errors.New("bsr.(Session).NewConnection: missing connection meta: invalid parameter"),
+		},
+		{
+			"empty-connection-id",
+			func() *bsr.Session {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys)
+				require.NoError(t, err)
+				return s
+			}(),
+			&bsr.ConnectionMeta{Id: ""},
+			errors.New("bsr.(Session).NewConnection: missing connection id: invalid parameter"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := tc.session.NewConnection(ctx, tc.meta)
+			require.Error(t, err)
+			assert.EqualError(t, err, tc.wantError.Error())
+		})
+	}
+}
+
+func TestNewChannelErrors(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	cases := []struct {
+		name       string
+		connection *bsr.Connection
+		meta       *bsr.ChannelMeta
+		wantError  error
+	}{
+		{
+			"nil-meta",
+			func() *bsr.Connection {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys, bsr.WithSupportsMultiplex(true))
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+				return c
+			}(),
+			nil,
+			errors.New("bsr.(Connection).NewChannel: missing channel meta: invalid parameter"),
+		},
+		{
+			"empty-connection-id",
+			func() *bsr.Connection {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys, bsr.WithSupportsMultiplex(true))
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+				return c
+			}(),
+			&bsr.ChannelMeta{Id: ""},
+			errors.New("bsr.(Connection).NewChannel: missing channel id: invalid parameter"),
+		},
+		{
+			"not-multiplexed",
+			func() *bsr.Connection {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys, bsr.WithSupportsMultiplex(false))
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+				return c
+			}(),
+			&bsr.ChannelMeta{Id: ""},
+			errors.New("bsr.(Connection).NewChannel: connection cannot make channels: not supported by protocol"),
+		},
+		{
+			"not-multiplexed-default",
+			func() *bsr.Connection {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys)
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+				return c
+			}(),
+			&bsr.ChannelMeta{Id: ""},
+			errors.New("bsr.(Connection).NewChannel: connection cannot make channels: not supported by protocol"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := tc.connection.NewChannel(ctx, tc.meta)
+			require.Error(t, err)
+			assert.EqualError(t, err, tc.wantError.Error())
+		})
+	}
+}
+
+func TestChannelNewMessagesWriterErrors(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	cases := []struct {
+		name      string
+		channel   *bsr.Channel
+		dir       bsr.Direction
+		wantError error
+	}{
+		{
+			"invalid-dir",
+			func() *bsr.Channel {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys, bsr.WithSupportsMultiplex(true))
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+
+				ch, err := c.NewChannel(ctx, &bsr.ChannelMeta{Id: "channel"})
+				require.NoError(t, err)
+				return ch
+			}(),
+			bsr.Direction(uint8(255)),
+			errors.New("bsr.(Channel).NewMessagesWriter: invalid direction: invalid parameter"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := tc.channel.NewMessagesWriter(ctx, tc.dir)
+			require.Error(t, err)
+			assert.EqualError(t, err, tc.wantError.Error())
+		})
+	}
+}
+
+func TestChannelNewRequestsWriterErrors(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	cases := []struct {
+		name      string
+		channel   *bsr.Channel
+		dir       bsr.Direction
+		wantError error
+	}{
+		{
+			"invalid-dir",
+			func() *bsr.Channel {
+				s, err := bsr.NewSession(ctx, &bsr.SessionMeta{Id: "session"}, &fstest.MemFS{}, keys, bsr.WithSupportsMultiplex(true))
+				require.NoError(t, err)
+
+				c, err := s.NewConnection(ctx, &bsr.ConnectionMeta{Id: "connection"})
+				require.NoError(t, err)
+
+				ch, err := c.NewChannel(ctx, &bsr.ChannelMeta{Id: "channel"})
+				require.NoError(t, err)
+				return ch
+			}(),
+			bsr.Direction(uint8(255)),
+			errors.New("bsr.(Channel).NewRequestsWriter: invalid direction: invalid parameter"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := tc.channel.NewRequestsWriter(ctx, tc.dir)
+			require.Error(t, err)
+			assert.EqualError(t, err, tc.wantError.Error())
+		})
+	}
+}

internal/bsr/byte_source.go

@@ -0,0 +1,35 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+// ByteSource identifies whether bytes are flowing from the user to target (client)
+// or target to user (server)
+type ByteSource uint8
+
+// ByteDirections
+const (
+	UnknownByteSource ByteSource = iota
+	Client
+	Server
+)
+
+func (d ByteSource) String() string {
+	switch d {
+	case Client:
+		return "client"
+	case Server:
+		return "server"
+	default:
+		return "unknown bytesource"
+	}
+}
+
+// ValidByteSource checks if a given ByteSource is valid.
+func ValidByteSource(d ByteSource) bool {
+	switch d {
+	case Client, Server:
+		return true
+	}
+	return false
+}

internal/bsr/byte_source_test.go

@@ -0,0 +1,83 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr_test
+
+import (
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/bsr"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidByteSource(t *testing.T) {
+	cases := []struct {
+		name string
+		in   bsr.ByteSource
+		want bool
+	}{
+		{
+			bsr.Client.String(),
+			bsr.Client,
+			true,
+		},
+		{
+			bsr.Server.String(),
+			bsr.Server,
+			true,
+		},
+		{
+			bsr.UnknownByteSource.String(),
+			bsr.UnknownByteSource,
+			false,
+		},
+		{
+			"something else",
+			bsr.ByteSource(255),
+			false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := bsr.ValidByteSource(tc.in)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}
+
+func TestByteSourceString(t *testing.T) {
+	cases := []struct {
+		name string
+		in   bsr.ByteSource
+		want string
+	}{
+		{
+			bsr.Client.String(),
+			bsr.Client,
+			"client",
+		},
+		{
+			bsr.Server.String(),
+			bsr.Server,
+			"server",
+		},
+		{
+			bsr.UnknownByteSource.String(),
+			bsr.UnknownByteSource,
+			"unknown bytesource",
+		},
+		{
+			"something else",
+			bsr.ByteSource(255),
+			"unknown bytesource",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := tc.in.String()
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}

internal/bsr/container.go

@@ -0,0 +1,189 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/checksum"
+	"github.com/hashicorp/boundary/internal/bsr/internal/journal"
+	"github.com/hashicorp/boundary/internal/bsr/internal/sign"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	"github.com/hashicorp/boundary/internal/storage"
+	"github.com/hashicorp/go-multierror"
+)
+
+const (
+	metaFile     = "%s.meta"
+	summaryFile  = "%s.summary"
+	checksumFile = "SHA256SUM"
+	sigFile      = "SHA256SUM.sig"
+	journalFile  = ".journal"
+)
+
+// ContainerType defines the type of container.
+type containerType string
+
+// Valid container types.
+const (
+	sessionContainer    containerType = "session"
+	connectionContainer containerType = "connection"
+	channelContainer    containerType = "channel"
+)
+
+// container contains a group of files in a BSR.
+// Each container has corresponding .meta, .summary, SHA256SUM, and SHA256SUM.sig files.
+type container struct {
+	container storage.Container
+	journal   *journal.Journal
+
+	metaName  string
+	sumName   string
+	meta      *checksum.File
+	sum       *checksum.File
+	checksums *sign.File
+	sigs      storage.File
+
+	keys *kms.Keys
+}
+
+// newContainer creates a container for the given type backed by the provide storage.Container.
+func newContainer(ctx context.Context, t containerType, c storage.Container, keys *kms.Keys) (*container, error) {
+	j, err := c.OpenFile(ctx, journalFile,
+		storage.WithCreateFile(),
+		storage.WithFileAccessMode(storage.WriteOnly),
+		storage.WithCloseSyncMode(storage.NoSync))
+	if err != nil {
+		return nil, err
+	}
+	jj, err := journal.New(ctx, j)
+	if err != nil {
+		return nil, err
+	}
+	cc := &container{
+		container: c,
+		journal:   jj,
+		keys:      keys,
+	}
+
+	cc.sigs, err = cc.create(ctx, sigFile)
+	if err != nil {
+		return nil, err
+	}
+
+	cs, err := cc.create(ctx, checksumFile)
+	if err != nil {
+		return nil, err
+	}
+	cc.checksums, err = sign.NewFile(ctx, cs, cc.sigs, keys)
+	if err != nil {
+		return nil, err
+	}
+
+	cc.metaName = fmt.Sprintf(metaFile, t)
+	meta, err := cc.create(ctx, cc.metaName)
+	if err != nil {
+		return nil, err
+	}
+	cc.meta, err = checksum.NewFile(ctx, meta, cc.checksums)
+	if err != nil {
+		return nil, err
+	}
+
+	cc.sumName = fmt.Sprintf(summaryFile, t)
+	sum, err := cc.create(ctx, cc.sumName)
+	if err != nil {
+		return nil, err
+	}
+	cc.sum, err = checksum.NewFile(ctx, sum, cc.checksums)
+	if err != nil {
+		return nil, err
+	}
+
+	return cc, nil
+}
+
+// Create creates a new file in the container for writing.
+func (c *container) create(ctx context.Context, s string) (storage.File, error) {
+	c.journal.Record("CREATING", s)
+	f, err := c.container.Create(ctx, s)
+	if err != nil {
+		return nil, err
+	}
+	jf, err := journal.NewFile(ctx, f, c.journal)
+	if err != nil {
+		return nil, err
+	}
+	defer c.journal.Record("CREATED", s)
+	return jf, nil
+}
+
+// writeMetaString writes a string to the containers meta file.
+func (c *container) writeMetaString(_ context.Context, s string) (int, error) {
+	return c.meta.WriteString(s)
+}
+
+// writeMetaLine writes a new line terminated line to the container's meta file.
+func (c *container) writeMetaLine(_ context.Context, s string) (int, error) {
+	return c.meta.WriteString(s + "\n")
+}
+
+// WriteMeta writes a new line terminated key : value pair to the container's meta file
+func (c *container) WriteMeta(_ context.Context, k, v string) (int, error) {
+	return c.meta.WriteString(fmt.Sprintf("%s: %s\n", k, v))
+}
+
+// writeSummaryString writes a string to the container's summary file.
+func (c *container) writeSummaryString(_ context.Context, s string) (int, error) {
+	return c.sum.WriteString(s)
+}
+
+// WriteSummary writes a new line terminated key : value pair to the container's summary file
+func (c *container) WriteSummary(_ context.Context, k, v string) (int, error) {
+	return c.sum.WriteString(fmt.Sprintf("%s: %s\n", k, v))
+}
+
+// WriteSummaryLine writes a new line terminated string to the container's summary file.
+func (c *container) WriteSummaryLine(_ context.Context, s string) (int, error) {
+	return c.sum.WriteString(s + "\n")
+}
+
+// WriteBinaryChecksum writes a checksum for a binary file to the checksum file.
+func (c *container) WriteBinaryChecksum(_ context.Context, sum []byte, fname string) (int, error) {
+	return c.checksums.WriteString(fmt.Sprintf("%x *%s\n", sum, fname))
+}
+
+// close closes a container, closing the underlying files in a container.
+func (c *container) close(_ context.Context) error {
+	const op = "bsr.(container).close"
+
+	var closeErrors *multierror.Error
+
+	if err := c.meta.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	if err := c.sum.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	if err := c.checksums.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	if err := c.sigs.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	if err := c.journal.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	if err := c.container.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	return closeErrors.ErrorOrNil()
+}

internal/bsr/errors.go

@@ -0,0 +1,18 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+import (
+	"errors"
+)
+
+// ErrUnknown represents an unknown error
+var ErrUnknown = errors.New("unknown error")
+
+// ErrInvalidParameter represents an invalid parameter error
+var 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")

internal/bsr/internal/checksum/checksum.go

@@ -0,0 +1,116 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package checksum provides a wrapper to compute a checksum on a writable file
+// while it is being written to, and record the final checksum when the file is
+// closed.
+package checksum
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+	"github.com/hashicorp/go-kms-wrapping/v2/extras/crypto"
+	"github.com/hashicorp/go-multierror"
+)
+
+const (
+	// A line for a checksum that can be read by sha256sum.
+	// Note that there are two spaces between the checksum and the file.
+	checksumLine = "%s  %s\n"
+)
+
+var errInvalidParameter = errors.New("invalid parameter")
+
+type writerFile interface {
+	fs.File
+	io.WriteCloser
+	io.StringWriter
+}
+
+// File is a writable file that will compute its checksum while it is written
+// to. When closed, it will write its checksum to the provided io.Writer.
+// The checksum is written in a format so that it can be verified by sha256sum.
+type File struct {
+	// ctx is used to provide a context to other functions when implementing
+	// std interfaces that to not take a context as an arg.
+	ctx context.Context
+
+	// Performs the checksuming, provides implementations for Write and
+	// WriteString
+	*crypto.Sha256SumWriter
+
+	checksumWriter io.Writer
+
+	underlying fs.File
+}
+
+// NewFile wraps the provided writerFile in a checksumed File. When the file is
+// closed, its checksum will be written to cs.
+func NewFile(ctx context.Context, f writerFile, cs io.Writer) (*File, error) {
+	const op = "checksum.NewFile"
+
+	switch {
+	case is.Nil(f):
+		return nil, fmt.Errorf("%s: missing writable file: %w", op, errInvalidParameter)
+	case is.Nil(cs):
+		return nil, fmt.Errorf("%s: missing checksum writer: %w", op, errInvalidParameter)
+	}
+
+	wrapped, err := crypto.NewSha256SumWriter(ctx, f)
+	if err != nil {
+		return nil, err
+	}
+	return &File{
+		ctx:             ctx,
+		checksumWriter:  cs,
+		Sha256SumWriter: wrapped,
+		underlying:      f,
+	}, nil
+}
+
+// Stat returns the fs.FileInfo for the underlying fs.File.
+func (f *File) Stat() (fs.FileInfo, error) {
+	return f.underlying.Stat()
+}
+
+// Read reads from the underlying fs.File.
+func (f *File) Read(b []byte) (int, error) {
+	return f.underlying.Read(b)
+}
+
+// Close closes the Sha256SumWriter, computes the checksum and writes it to
+// f.checksumWriter
+func (f *File) Close() error {
+	const op = "checksum.(File).Close"
+
+	var closeErrors *multierror.Error
+
+	// f.Sha256SumWriter will close f.underlying
+	if err := f.Sha256SumWriter.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	sum, err := f.Sha256SumWriter.Sum(f.ctx, crypto.WithHexEncoding(true))
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+		return closeErrors.ErrorOrNil()
+	}
+
+	s, err := f.Stat()
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+		return closeErrors.ErrorOrNil()
+	}
+
+	if _, err := f.checksumWriter.Write([]byte(fmt.Sprintf(checksumLine, sum, s.Name()))); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+	return closeErrors.ErrorOrNil()
+}
+
+var _ writerFile = (*File)(nil)

internal/bsr/internal/checksum/checksum_test.go

@@ -0,0 +1,157 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package checksum_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"io/fs"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/checksum"
+	"github.com/hashicorp/boundary/internal/bsr/internal/fstest"
+	"github.com/hashicorp/go-multierror"
+	"github.com/stretchr/testify/require"
+)
+
+type bufWriter interface {
+	io.Writer
+	String() string
+}
+
+type badWriter struct{}
+
+func (*badWriter) Write(_ []byte) (int, error) {
+	return 0, errors.New("write failed")
+}
+
+func (*badWriter) String() string {
+	return ""
+}
+
+func TestFile(t *testing.T) {
+	ctx := context.Background()
+
+	// testSum created via: echo -n "test-string" | sha256sum
+	const (
+		testSum    = "ffe65f1d98fafedea3514adc956c8ada5980c6c5d2552fd61f48401aefd5c00e"
+		testString = "test-string"
+	)
+
+	cases := []struct {
+		name                string
+		f                   *fstest.MemFile
+		cs                  bufWriter
+		write               string
+		want                string
+		wantNewErr          error
+		wantCloseErr        error
+		expectCheckSumOnErr bool
+	}{
+		{
+			"success",
+			fstest.NewMemFile("test", 0o664),
+			bytes.NewBuffer([]byte{}),
+			testString,
+			testSum + "  test\n",
+			nil,
+			nil,
+			false,
+		},
+		{
+			"nil-file",
+			nil,
+			bytes.NewBuffer([]byte{}),
+			testString,
+			testSum + "  test\n",
+			errors.New("checksum.NewFile: missing writable file: invalid parameter"),
+			nil,
+			false,
+		},
+		{
+			"nil-checksum-writer",
+			fstest.NewMemFile("test", 0o664),
+			nil,
+			testString,
+			testSum + "  test\n",
+			errors.New("checksum.NewFile: missing checksum writer: invalid parameter"),
+			nil,
+			false,
+		},
+		{
+			"close-error",
+			fstest.NewMemFile("test", 0o664, fstest.WithCloseFunc(func() error { return errors.New("close error") })),
+			bytes.NewBuffer([]byte{}),
+			testString,
+			testSum + "  test\n",
+			nil,
+			func() error {
+				var wantErrors *multierror.Error
+				wantErrors = multierror.Append(wantErrors, errors.New("checksum.(File).Close: crypto.(Sha256SumWriter).Close: close error"))
+				return wantErrors.ErrorOrNil()
+			}(),
+			true,
+		},
+		{
+			"stat-error",
+			fstest.NewMemFile("test", 0o664, fstest.WithStatFunc(func() (fs.FileInfo, error) { return nil, errors.New("stat error") })),
+			bytes.NewBuffer([]byte{}),
+			testString,
+			testSum + "  test\n",
+			nil,
+			func() error {
+				var wantErrors *multierror.Error
+				wantErrors = multierror.Append(wantErrors, errors.New("checksum.(File).Close: stat error"))
+				return wantErrors.ErrorOrNil()
+			}(),
+			false,
+		},
+		{
+			"write-error",
+			fstest.NewMemFile("test", 0o664),
+			&badWriter{},
+			testString,
+			testSum + "  test\n",
+			nil,
+			func() error {
+				var wantErrors *multierror.Error
+				wantErrors = multierror.Append(wantErrors, errors.New("checksum.(File).Close: write failed"))
+				return wantErrors.ErrorOrNil()
+			}(),
+			false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			f, err := checksum.NewFile(ctx, tc.f, tc.cs)
+			if tc.wantNewErr != nil {
+				require.EqualError(t, tc.wantNewErr, err.Error())
+				return
+			}
+			require.NoError(t, err)
+
+			_, err = f.WriteString(tc.write)
+			require.NoError(t, err)
+
+			err = f.Close()
+			if tc.wantCloseErr != nil {
+				require.EqualError(t, err, tc.wantCloseErr.Error())
+
+				if tc.expectCheckSumOnErr {
+					require.Equal(t, tc.want, tc.cs.String())
+				} else {
+					require.Equal(t, "", tc.cs.String())
+				}
+				return
+			}
+			require.NoError(t, err)
+
+			require.True(t, tc.f.Closed)
+			require.Equal(t, tc.want, tc.cs.String())
+		})
+	}
+}

internal/bsr/internal/fs/fs.go

@@ -1,71 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: MPL-2.0
-
-// Package fs defines the interfaces bsr needs with a storage system.
-package fs
-
-import (
-	"context"
-	"io"
-	"io/fs"
-)
-
-// FS is a filesystem for creating or reading BSRs.
-type FS interface {
-	New(context.Context, string) (Container, error)
-}
-
-// SyncMode is use to determine how a file is synced when closed.
-type SyncMode uint8
-
-// Valid sync modes.
-const (
-	Asynchronous SyncMode = iota
-	Synchronous
-	NoSync
-)
-
-// GetOpts - iterate the inbound Options and return a struct.
-func GetOpts(opt ...Option) Options {
-	opts := getDefaultOptions()
-	for _, o := range opt {
-		o(&opts)
-	}
-	return opts
-}
-
-func getDefaultOptions() Options {
-	return Options{
-		WithCloseSyncMode: Asynchronous,
-	}
-}
-
-// Options are fs options.
-type Options struct {
-	WithCloseSyncMode SyncMode
-}
-
-// Option is an fs option.
-type Option func(*Options)
-
-// WithCloseSyncMode sets how a file is synced when closed.
-func WithCloseSyncMode(m SyncMode) Option {
-	return func(o *Options) {
-		o.WithCloseSyncMode = m
-	}
-}
-
-// A Container is a filesystem abstraction that can create files or other containers.
-type Container interface {
-	io.Closer
-	Create(context.Context, string) (WriterFile, error)
-	OpenFile(context.Context, string, ...Option) (WriterFile, error)
-	SubContainer(context.Context, string) (Container, error)
-}
-
-// WriterFile is a file that can be written to.
-type WriterFile interface {
-	fs.File
-	io.Writer
-	io.StringWriter
-}

internal/bsr/internal/fstest/fs.go

@@ -0,0 +1,213 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package fstest provides test implementations of a the fs interfaces.
+package fstest
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"time"
+
+	sfs "io/fs"
+
+	"github.com/hashicorp/boundary/internal/storage"
+)
+
+// MemFS is a storage.FS that only resides in memory.
+type MemFS struct {
+	Containers map[string]*MemContainer
+
+	newFunc NewFunc
+}
+
+// NewMemFS creates a MemFS. It supports WithNewFunc.
+func NewMemFS(options ...Option) *MemFS {
+	opts := getOpts(options...)
+
+	return &MemFS{
+		Containers: make(map[string]*MemContainer),
+		newFunc:    opts.withNewFunc,
+	}
+}
+
+// New creates a storage.Container in the MemFS.
+func (m *MemFS) New(ctx context.Context, n string) (storage.Container, error) {
+	if m.newFunc != nil {
+		return m.newFunc(ctx, n)
+	}
+	if m.Containers == nil {
+		m.Containers = make(map[string]*MemContainer)
+	}
+	if _, exists := m.Containers[n]; exists {
+		return nil, fmt.Errorf("container %s already exists", n)
+	}
+	c := &MemContainer{
+		Name:  n,
+		Sub:   make(map[string]*MemContainer),
+		Files: make(map[string]*MemFile),
+	}
+	m.Containers[n] = c
+	return c, nil
+}
+
+// MemContainer is a storage.Container that resides in memory.
+type MemContainer struct {
+	Name string
+
+	Sub   map[string]*MemContainer
+	Files map[string]*MemFile
+
+	Closed bool
+}
+
+// Close closes the container.
+func (m *MemContainer) Close() error {
+	m.Closed = true
+	return nil
+}
+
+// Create makes a new storage.File in the container.
+func (m *MemContainer) Create(_ context.Context, n string) (storage.File, error) {
+	if m.Closed {
+		return nil, fmt.Errorf("create on closed container")
+	}
+	if _, exists := m.Files[n]; exists {
+		return nil, fmt.Errorf("file %s already exists", n)
+	}
+	f := &MemFile{
+		name: n,
+		Buf:  bytes.NewBuffer([]byte{}),
+		mode: sfs.ModeAppend,
+	}
+	m.Files[n] = f
+	return f, nil
+}
+
+// OpenFile creates a storage.File in the container using the provided options
+// It supports WithCloseSyncMode.
+func (m *MemContainer) OpenFile(_ context.Context, n string, option ...storage.Option) (storage.File, error) {
+	opts := storage.GetOpts(option...)
+	if m.Closed {
+		return nil, fmt.Errorf("create on closed container")
+	}
+	if _, exists := m.Files[n]; exists {
+		return nil, fmt.Errorf("file %s already exists", n)
+	}
+
+	var mode sfs.FileMode
+
+	switch opts.WithCloseSyncMode {
+	case storage.NoSync:
+		mode = sfs.ModeAppend | sfs.ModeTemporary
+	default:
+		mode = sfs.ModeAppend
+	}
+
+	f := &MemFile{
+		name: n,
+		Buf:  bytes.NewBuffer([]byte{}),
+		mode: mode,
+	}
+	m.Files[n] = f
+	return f, nil
+}
+
+// SubContainer creates a new storage.Container in the container.
+func (m *MemContainer) SubContainer(_ context.Context, n string, _ ...storage.Option) (storage.Container, error) {
+	if _, exists := m.Sub[n]; exists {
+		return nil, fmt.Errorf("container %s already exists", n)
+	}
+	c := &MemContainer{
+		Name:  n,
+		Sub:   make(map[string]*MemContainer),
+		Files: make(map[string]*MemFile),
+	}
+	m.Sub[n] = c
+	return c, nil
+}
+
+// memFileInfo implements storage.FileInfo
+type memFileInfo struct {
+	name string
+	size int64
+	mode sfs.FileMode
+	mod  time.Time
+}
+
+func (m *memFileInfo) Name() string       { return m.name }
+func (m *memFileInfo) Size() int64        { return m.size }
+func (m *memFileInfo) Mode() sfs.FileMode { return m.mode }
+func (m *memFileInfo) ModTime() time.Time { return m.mod }
+func (m *memFileInfo) IsDir() bool        { return false }
+func (m *memFileInfo) Sys() any           { return nil }
+
+// MemFile is a storage.File that resides in memory.
+type MemFile struct {
+	name    string
+	Buf     *bytes.Buffer
+	mode    sfs.FileMode
+	modtime time.Time
+
+	statFunc  StatFunc
+	closeFunc CloseFunc
+
+	Closed bool
+}
+
+// NewMemFile creates a MemFile. It supports WithCloseFunc and WithStatFunc
+// to customize the behavior of Close and Stat.
+func NewMemFile(n string, mode sfs.FileMode, options ...Option) *MemFile {
+	opts := getOpts(options...)
+
+	return &MemFile{
+		name:      n,
+		Buf:       bytes.NewBuffer([]byte{}),
+		mode:      mode,
+		statFunc:  opts.withStatFunc,
+		closeFunc: opts.withCloseFunc,
+	}
+}
+
+// Stat returns the FileInfo for the file.
+func (m *MemFile) Stat() (sfs.FileInfo, error) {
+	if m.statFunc != nil {
+		return m.statFunc()
+	}
+	return &memFileInfo{
+		name: m.name,
+		size: int64(m.Buf.Len()),
+		mode: m.mode,
+		mod:  m.modtime,
+	}, nil
+}
+
+func (m *MemFile) Read(p []byte) (int, error) {
+	return m.Buf.Read(p)
+}
+
+// Close closes the file.
+func (m *MemFile) Close() error {
+	if m.closeFunc != nil {
+		return m.closeFunc()
+	}
+	m.Closed = true
+	return nil
+}
+
+// WriteString implements io.StringWriter.
+func (m *MemFile) WriteString(s string) (n int, err error) {
+	return m.Write([]byte(s))
+}
+
+func (m *MemFile) Write(p []byte) (n int, err error) {
+	if m.Closed {
+		return 0, fmt.Errorf("write on closed file")
+	}
+
+	defer func() {
+		m.modtime = time.Now()
+	}()
+	return m.Buf.Write(p)
+}

internal/bsr/internal/fstest/options.go

@@ -0,0 +1,71 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package fstest
+
+import (
+	"context"
+	"io/fs"
+
+	"github.com/hashicorp/boundary/internal/storage"
+)
+
+// getOpts - iterate the inbound Options and return a struct
+func getOpts(opt ...Option) options {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		o(&opts)
+	}
+	return opts
+}
+
+// Option - how Options are passed as arguments
+type Option func(*options)
+
+// NewFunc is a function called in New.
+// It can be used to simulate an error during New.
+type NewFunc func(context.Context, string) (storage.Container, error)
+
+// CloseFunc is a function called at close.
+// It can be used to simulate an error during Close.
+type CloseFunc func() error
+
+// StatFunc is a function called for Stat.
+// It can be used to simulate an error on calls to Stat
+type StatFunc func() (fs.FileInfo, error)
+
+// options = how options are represented
+type options struct {
+	withNewFunc   NewFunc
+	withCloseFunc CloseFunc
+	withStatFunc  StatFunc
+}
+
+func getDefaultOptions() options {
+	return options{
+		withNewFunc:   nil,
+		withCloseFunc: nil,
+		withStatFunc:  nil,
+	}
+}
+
+// WithNewFunc is used to provide a custom new function.
+func WithNewFunc(f NewFunc) Option {
+	return func(o *options) {
+		o.withNewFunc = f
+	}
+}
+
+// WithCloseFunc is used to provide a custom close function.
+func WithCloseFunc(f CloseFunc) Option {
+	return func(o *options) {
+		o.withCloseFunc = f
+	}
+}
+
+// WithStatFunc is used to provide a custom Stat function.
+func WithStatFunc(f StatFunc) Option {
+	return func(o *options) {
+		o.withStatFunc = f
+	}
+}

internal/bsr/internal/journal/journal.go

@@ -0,0 +1,105 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package journal provides a writer that uses a journal file to aide in recovery.
+package journal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+)
+
+var errInvalidParameter = errors.New("invalid parameter")
+
+type writerFile interface {
+	fs.File
+	io.WriteCloser
+	io.StringWriter
+}
+
+// Journal is used to record meta data about the operations that will be and
+// have been performed on bsr containers and files. This can be used to aide in
+// recovery if a system crashes while writing a bsr.
+type Journal struct {
+	io.Writer
+}
+
+// New creates a Journal that is written to the provided io.Writer.
+func New(_ context.Context, w io.Writer) (*Journal, error) {
+	const op = "journal.New"
+
+	switch {
+	case is.Nil(w):
+		return nil, fmt.Errorf("%s: missing writer: %w", op, errInvalidParameter)
+	}
+	return &Journal{
+		Writer: w,
+	}, nil
+}
+
+// Close optionally closes the underlying io.Writer if it also implements
+// io.Closer.
+func (j *Journal) Close() error {
+	var i interface{} = j.Writer
+	v, ok := i.(io.WriteCloser)
+	if ok {
+		if err := v.Close(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Record writes the entry to the journal.
+func (j *Journal) Record(op, f string) error {
+	_, err := j.Write([]byte(fmt.Sprintf("%s %s\n", op, f)))
+	return err
+}
+
+// File is a writable file that will update a Journal as it closed.
+type File struct {
+	j *Journal
+	writerFile
+}
+
+// NewFile wraps the provided writable file in a journaled File. When the file
+// is closed, the journal will be written to.
+func NewFile(_ context.Context, f writerFile, j *Journal) (*File, error) {
+	const op = "journal.NewFile"
+	switch {
+	case is.Nil(f):
+		return nil, fmt.Errorf("%s: missing writable file: %w", op, errInvalidParameter)
+	case is.Nil(j):
+		return nil, fmt.Errorf("%s: missing journal: %w", op, errInvalidParameter)
+	}
+
+	return &File{
+		j:          j,
+		writerFile: f,
+	}, nil
+}
+
+// Close closes the underlying file, writing to the journal prior to and after
+// closing.
+func (f *File) Close() error {
+	const op = "journal.(File).Close"
+
+	s, err := f.Stat()
+	if err != nil {
+		return fmt.Errorf("%s: %w", op, err)
+	}
+	if err := f.j.Record("CLOSING", s.Name()); err != nil {
+		return fmt.Errorf("%s: %w", op, err)
+	}
+	if err := f.writerFile.Close(); err != nil {
+		return fmt.Errorf("%s: %w", op, err)
+	}
+	return f.j.Record("CLOSED", s.Name())
+}
+
+var _ writerFile = (*File)(nil)

internal/bsr/internal/journal/journal_test.go

@@ -0,0 +1,140 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package journal_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"io/fs"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/fstest"
+	"github.com/hashicorp/boundary/internal/bsr/internal/journal"
+	"github.com/stretchr/testify/require"
+)
+
+type bufWriter interface {
+	io.Writer
+	String() string
+}
+
+type badWriter struct{}
+
+func (*badWriter) Write(_ []byte) (int, error) {
+	return 0, errors.New("write failed")
+}
+
+func (*badWriter) String() string {
+	return ""
+}
+
+func TestFile(t *testing.T) {
+	ctx := context.Background()
+
+	const (
+		testString = "test-string"
+	)
+
+	cases := []struct {
+		name           string
+		f              *fstest.MemFile
+		buf            bufWriter
+		newJournalFunc func(w bufWriter) *journal.Journal
+		write          string
+		want           string
+		wantNewErr     error
+		wantCloseErr   error
+	}{
+		{
+			"success",
+			fstest.NewMemFile("test", 0o664),
+			bytes.NewBuffer([]byte{}),
+			func(w bufWriter) *journal.Journal { ww, err := journal.New(ctx, w); require.NoError(t, err); return ww },
+			testString,
+			"CLOSING test\nCLOSED test\n",
+			nil,
+			nil,
+		},
+		{
+			"nil-file",
+			nil,
+			bytes.NewBuffer([]byte{}),
+			func(w bufWriter) *journal.Journal { ww, err := journal.New(ctx, w); require.NoError(t, err); return ww },
+			testString,
+			"",
+			errors.New("journal.NewFile: missing writable file: invalid parameter"),
+			nil,
+		},
+		{
+			"nil-journal-writer",
+			fstest.NewMemFile("test", 0o664),
+			nil,
+			func(w bufWriter) *journal.Journal { return nil },
+			testString,
+			"",
+			errors.New("journal.NewFile: missing journal: invalid parameter"),
+			nil,
+		},
+		{
+			"close-error",
+			fstest.NewMemFile("test", 0o664, fstest.WithCloseFunc(func() error { return errors.New("close error") })),
+			bytes.NewBuffer([]byte{}),
+			func(w bufWriter) *journal.Journal { ww, err := journal.New(ctx, w); require.NoError(t, err); return ww },
+			testString,
+			"CLOSING test\n",
+			nil,
+			errors.New("journal.(File).Close: close error"),
+		},
+		{
+			"stat-error",
+			fstest.NewMemFile("test", 0o664, fstest.WithStatFunc(func() (fs.FileInfo, error) { return nil, errors.New("stat error") })),
+			bytes.NewBuffer([]byte{}),
+			func(w bufWriter) *journal.Journal { ww, err := journal.New(ctx, w); require.NoError(t, err); return ww },
+			testString,
+			"",
+			nil,
+			errors.New("journal.(File).Close: stat error"),
+		},
+		{
+			"write-error",
+			fstest.NewMemFile("test", 0o664),
+			&badWriter{},
+			func(w bufWriter) *journal.Journal { ww, err := journal.New(ctx, w); require.NoError(t, err); return ww },
+			testString,
+			"",
+			nil,
+			errors.New("journal.(File).Close: write failed"),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			j := tc.newJournalFunc(tc.buf)
+
+			f, err := journal.NewFile(ctx, tc.f, j)
+			if tc.wantNewErr != nil {
+				require.EqualError(t, tc.wantNewErr, err.Error())
+				return
+			}
+			require.NoError(t, err)
+
+			_, err = f.WriteString(tc.write)
+			require.NoError(t, err)
+
+			err = f.Close()
+			if tc.wantCloseErr != nil {
+				require.EqualError(t, err, tc.wantCloseErr.Error())
+
+				require.Equal(t, tc.want, tc.buf.String())
+				return
+			}
+			require.NoError(t, err)
+
+			require.True(t, tc.f.Closed)
+			require.Equal(t, tc.want, tc.buf.String())
+		})
+	}
+}

internal/bsr/internal/sign/sign.go

@@ -0,0 +1,169 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+// Package sign provides wrappers to compute a signature of data written to an
+// io.Writer
+package sign
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+
+	"github.com/hashicorp/boundary/internal/bsr/internal/is"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/hashicorp/go-multierror"
+	"google.golang.org/protobuf/proto"
+)
+
+var errInvalidParameter = errors.New("invalid parameter")
+
+// Writer is a wrapper that will compute a signature of all the data
+// written to its writer.
+type Writer struct {
+	keys *kms.Keys
+
+	buf *bytes.Buffer
+	w   io.Writer
+	tee io.Writer
+}
+
+// NewWriter returns a Writer that wraps an io.Writer.
+func NewWriter(_ context.Context, w io.Writer, keys *kms.Keys) (*Writer, error) {
+	const op = "sign.NewWriter"
+	switch {
+	case is.Nil(w):
+		return nil, fmt.Errorf("%s: missing sign writer: %w", op, errInvalidParameter)
+	case is.Nil(keys):
+		return nil, fmt.Errorf("%s: missing keys: %w", op, errInvalidParameter)
+	}
+
+	var buf bytes.Buffer
+	tee := io.MultiWriter(w, &buf)
+	return &Writer{
+		buf:  &buf,
+		tee:  tee,
+		keys: keys,
+		w:    w,
+	}, nil
+}
+
+func (w *Writer) Write(b []byte) (int, error) {
+	return w.tee.Write(b)
+}
+
+// WriteString implements the io.StringWriter WriteString method.
+func (w *Writer) WriteString(s string) (int, error) {
+	return w.Write([]byte(s))
+}
+
+// Close implements the io.Closer method.
+func (w *Writer) Close() error {
+	const op = "sign.(Writer).Close"
+	var i interface{} = w.w
+	v, ok := i.(io.WriteCloser)
+	if ok {
+		if err := v.Close(); err != nil {
+			return fmt.Errorf("%s: %w", op, err)
+		}
+	}
+	return nil
+}
+
+// Sign returns the signature of the data written to the writer.
+func (w *Writer) Sign(ctx context.Context) (*wrapping.SigInfo, error) {
+	sig, err := w.keys.SignWithPrivKey(ctx, w.buf.Bytes())
+	if err != nil {
+		return nil, err
+	}
+	return sig, nil
+}
+
+type writerFile interface {
+	fs.File
+	io.WriteCloser
+	io.StringWriter
+}
+
+// File is a writable file that will compute it's signature while it is written
+// to. When closed, it will write the signature to the provided io.Writer.
+type File struct {
+	ctx             context.Context
+	signatureWriter io.Writer
+
+	// Performs the signing, and provides implementations for Write and
+	// WriteString.
+	*Writer
+
+	underlying fs.File
+}
+
+// Stat returns the fs.FileInfo for the underlying fs.File.
+func (f *File) Stat() (fs.FileInfo, error) {
+	return f.underlying.Stat()
+}
+
+// Read reads from the underlying fs.File.
+func (f *File) Read(b []byte) (int, error) {
+	return f.underlying.Read(b)
+}
+
+// NewFile wraps the provided writerFile in a signed File. When the file is
+// closed, its signature is written to w.
+func NewFile(ctx context.Context, f writerFile, w io.Writer, keys *kms.Keys) (*File, error) {
+	const op = "sign.NewFile"
+
+	switch {
+	case is.Nil(f):
+		return nil, fmt.Errorf("%s: missing writable file: %w", op, errInvalidParameter)
+	case is.Nil(w):
+		return nil, fmt.Errorf("%s: missing sign writer: %w", op, errInvalidParameter)
+	case is.Nil(keys):
+		return nil, fmt.Errorf("%s: missing keys: %w", op, errInvalidParameter)
+	}
+
+	sw, err := NewWriter(ctx, f, keys)
+	if err != nil {
+		return nil, err
+	}
+	return &File{
+		ctx:             ctx,
+		signatureWriter: w,
+		Writer:          sw,
+		underlying:      f,
+	}, nil
+}
+
+// Close closes the wrapped writer and writes the SigInfo to f.signatureWriter.
+func (f *File) Close() error {
+	const op = "sign.(File).Close"
+
+	var closeErrors *multierror.Error
+	if err := f.Writer.Close(); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	sig, err := f.Writer.Sign(f.ctx)
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+		return closeErrors.ErrorOrNil()
+	}
+
+	b, err := proto.Marshal(sig)
+	if err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+		return closeErrors.ErrorOrNil()
+	}
+
+	if _, err := f.signatureWriter.Write(b); err != nil {
+		closeErrors = multierror.Append(closeErrors, fmt.Errorf("%s: %w", op, err))
+	}
+
+	return closeErrors.ErrorOrNil()
+}
+
+var _ writerFile = (*File)(nil)

internal/bsr/internal/sign/sign_test.go

@@ -0,0 +1,191 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package sign_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/hashicorp/boundary/internal/bsr/internal/fstest"
+	"github.com/hashicorp/boundary/internal/bsr/internal/sign"
+	"github.com/hashicorp/boundary/internal/bsr/kms"
+	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/hashicorp/go-multierror"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+type bufWriter interface {
+	io.Writer
+	Bytes() []byte
+}
+
+type badWriter struct{}
+
+func (*badWriter) Write(_ []byte) (int, error) {
+	return 0, errors.New("write failed")
+}
+
+func (*badWriter) Bytes() []byte {
+	return nil
+}
+
+func TestFile(t *testing.T) {
+	ctx := context.Background()
+
+	keys, err := kms.CreateKeys(ctx, kms.TestWrapper(t), "session")
+	require.NoError(t, err)
+
+	testSig := []byte("")
+	const (
+		testString = "test-string"
+	)
+
+	cases := []struct {
+		name               string
+		f                  *fstest.MemFile
+		sig                bufWriter
+		keys               *kms.Keys
+		write              string
+		want               []byte
+		wantNewErr         error
+		wantCloseErr       error
+		expectSigInfoOnErr bool
+	}{
+		{
+			"success",
+			fstest.NewMemFile("test", 0o664),
+			bytes.NewBuffer([]byte{}),
+			keys,
+			testString,
+			testSig,
+			nil,
+			nil,
+			false,
+		},
+		{
+			"nil-file",
+			nil,
+			bytes.NewBuffer([]byte{}),
+			keys,
+			testString,
+			testSig,
+			errors.New("sign.NewFile: missing writable file: invalid parameter"),
+			nil,
+			false,
+		},
+		{
+			"nil-sign-writer",
+			fstest.NewMemFile("test", 0o664),
+			nil,
+			keys,
+			testString,
+			testSig,
+			errors.New("sign.NewFile: missing sign writer: invalid parameter"),
+			nil,
+			false,
+		},
+		{
+			"nil-keys",
+			fstest.NewMemFile("test", 0o664),
+			bytes.NewBuffer([]byte{}),
+			nil,
+			testString,
+			testSig,
+			errors.New("sign.NewFile: missing keys: invalid parameter"),
+			nil,
+			false,
+		},
+		{
+			"close-error",
+			fstest.NewMemFile("test", 0o664, fstest.WithCloseFunc(func() error { return errors.New("close error") })),
+			bytes.NewBuffer([]byte{}),
+			keys,
+			testString,
+			testSig,
+			nil,
+			func() error {
+				var wantErrors *multierror.Error
+				wantErrors = multierror.Append(wantErrors, errors.New("sign.(File).Close: sign.(Writer).Close: close error"))
+				return wantErrors.ErrorOrNil()
+			}(),
+			true,
+		},
+		{
+			"write-error",
+			fstest.NewMemFile("test", 0o664),
+			&badWriter{},
+			keys,
+			testString,
+			testSig,
+			nil,
+			func() error {
+				var wantErrors *multierror.Error
+				wantErrors = multierror.Append(wantErrors, errors.New("sign.(File).Close: write failed"))
+				return wantErrors.ErrorOrNil()
+			}(),
+			false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			f, err := sign.NewFile(ctx, tc.f, tc.sig, tc.keys)
+			if tc.wantNewErr != nil {
+				require.EqualError(t, tc.wantNewErr, err.Error())
+				return
+			}
+			require.NoError(t, err)
+
+			_, err = f.WriteString(tc.write)
+			require.NoError(t, err)
+
+			err = f.Close()
+			if tc.wantCloseErr != nil {
+				require.EqualError(t, err, tc.wantCloseErr.Error())
+
+				if tc.expectSigInfoOnErr {
+					sigInfo := &wrapping.SigInfo{}
+					err = proto.Unmarshal(tc.sig.Bytes(), sigInfo)
+					require.NoError(t, err)
+					want, err := keys.SignWithPrivKey(ctx, []byte(tc.write))
+					require.NoError(t, err)
+					assert.Empty(t,
+						cmp.Diff(
+							want,
+							sigInfo,
+							cmpopts.IgnoreUnexported(wrapping.SigInfo{}, wrapping.KeyInfo{}),
+						),
+					)
+				} else {
+					require.Empty(t, tc.sig.Bytes())
+				}
+				return
+			}
+			require.NoError(t, err)
+
+			require.True(t, tc.f.Closed)
+
+			sigInfo := &wrapping.SigInfo{}
+			err = proto.Unmarshal(tc.sig.Bytes(), sigInfo)
+			require.NoError(t, err)
+
+			want, err := keys.SignWithPrivKey(ctx, []byte(tc.write))
+			require.NoError(t, err)
+			assert.Empty(t,
+				cmp.Diff(
+					want,
+					sigInfo,
+					cmpopts.IgnoreUnexported(wrapping.SigInfo{}, wrapping.KeyInfo{}),
+				),
+			)
+		})
+	}
+}

internal/bsr/options.go

@@ -0,0 +1,35 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package bsr
+
+// getOpts - iterate the inbound Options and return a struct
+func getOpts(opt ...Option) options {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		o(&opts)
+	}
+	return opts
+}
+
+// Option - how Options are passed as arguments
+type Option func(*options)
+
+// options = how options are represented
+type options struct {
+	withSupportsMultiplex bool
+}
+
+func getDefaultOptions() options {
+	return options{
+		withSupportsMultiplex: false,
+	}
+}
+
+// WithSupportsMultiplex is used indicate that a protocol supports multiplexing
+// and therefore a BSR can contain Channels.
+func WithSupportsMultiplex(b bool) Option {
+	return func(o *options) {
+		o.withSupportsMultiplex = b
+	}
+}

internal/bsr/testdata/TestBsr/session_multiplexed/closed/.journal

@@ -0,0 +1,16 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING session.meta
+CREATED session.meta
+CREATING session.summary
+CREATED session.summary
+CLOSING session.meta
+CLOSED session.meta
+CLOSING session.summary
+CLOSED session.summary
+CLOSING SHA256SUM
+CLOSED SHA256SUM
+CLOSING SHA256SUM.sig
+CLOSED SHA256SUM.sig

internal/bsr/testdata/TestBsr/session_multiplexed/closed/SHA256SUM

@@ -0,0 +1,2 @@
+ff2a532f46f3e2202e69b3aab760da95c290adf33a9a4b1c4a7e17e0a2671421  session.meta
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  session.summary

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/.journal

@@ -0,0 +1,24 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING connection.meta
+CREATED connection.meta
+CREATING connection.summary
+CREATED connection.summary
+CREATING requests-inbound.data
+CREATED requests-inbound.data
+CREATING requests-outbound.data
+CREATED requests-outbound.data
+CLOSING requests-inbound.data
+CLOSED requests-inbound.data
+CLOSING requests-outbound.data
+CLOSED requests-outbound.data
+CLOSING connection.meta
+CLOSED connection.meta
+CLOSING connection.summary
+CLOSED connection.summary
+CLOSING SHA256SUM
+CLOSED SHA256SUM
+CLOSING SHA256SUM.sig
+CLOSED SHA256SUM.sig

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/SHA256SUM

@@ -0,0 +1,4 @@
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-inbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-outbound.data
+e4df3504aaa6fee4426f67ba998cdd90f84c579d98db979e589d8ca2ffd1e733  connection.meta
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  connection.summary

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/chan_1/.journal

@@ -0,0 +1,32 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING channel.meta
+CREATED channel.meta
+CREATING channel.summary
+CREATED channel.summary
+CREATING messages-inbound.data
+CREATED messages-inbound.data
+CREATING requests-inbound.data
+CREATED requests-inbound.data
+CREATING messages-outbound.data
+CREATED messages-outbound.data
+CREATING requests-outbound.data
+CREATED requests-outbound.data
+CLOSING messages-inbound.data
+CLOSED messages-inbound.data
+CLOSING requests-inbound.data
+CLOSED requests-inbound.data
+CLOSING messages-outbound.data
+CLOSED messages-outbound.data
+CLOSING requests-outbound.data
+CLOSED requests-outbound.data
+CLOSING channel.meta
+CLOSED channel.meta
+CLOSING channel.summary
+CLOSED channel.summary
+CLOSING SHA256SUM
+CLOSED SHA256SUM
+CLOSING SHA256SUM.sig
+CLOSED SHA256SUM.sig

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/chan_1/SHA256SUM

@@ -0,0 +1,6 @@
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  messages-inbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-inbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  messages-outbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-outbound.data
+e30a207b53f444c7bd2d1e8cc102abf910dfafd70aaef7f24db00298832b74d3  channel.meta
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  channel.summary

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/chan_1/channel.meta

@@ -0,0 +1,5 @@
+id: chan_1
+messages: inbound
+requests: inbound
+messages: outbound
+requests: outbound

internal/bsr/testdata/TestBsr/session_multiplexed/closed/conn_1/connection.meta

@@ -0,0 +1,4 @@
+id: conn_1
+requests: inbound
+requests: outbound
+channel: chan_1.channel

internal/bsr/testdata/TestBsr/session_multiplexed/closed/session.meta

@@ -0,0 +1,2 @@
+id: session_123456789
+connection: conn_1.connection

internal/bsr/testdata/TestBsr/session_multiplexed/opened/.journal

@@ -0,0 +1,8 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING session.meta
+CREATED session.meta
+CREATING session.summary
+CREATED session.summary

internal/bsr/testdata/TestBsr/session_multiplexed/opened/conn_1/.journal

@@ -0,0 +1,8 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING connection.meta
+CREATED connection.meta
+CREATING connection.summary
+CREATED connection.summary

internal/bsr/testdata/TestBsr/session_multiplexed/opened/conn_1/chan_1/.journal

@@ -0,0 +1,8 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING channel.meta
+CREATED channel.meta
+CREATING channel.summary
+CREATED channel.summary

internal/bsr/testdata/TestBsr/session_multiplexed/opened/conn_1/chan_1/channel.meta

@@ -0,0 +1 @@
+id: chan_1

internal/bsr/testdata/TestBsr/session_multiplexed/opened/conn_1/connection.meta

@@ -0,0 +1 @@
+id: conn_1

internal/bsr/testdata/TestBsr/session_multiplexed/opened/session.meta

@@ -0,0 +1 @@
+id: session_123456789

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/.journal

@@ -0,0 +1,16 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING session.meta
+CREATED session.meta
+CREATING session.summary
+CREATED session.summary
+CLOSING session.meta
+CLOSED session.meta
+CLOSING session.summary
+CLOSED session.summary
+CLOSING SHA256SUM
+CLOSED SHA256SUM
+CLOSING SHA256SUM.sig
+CLOSED SHA256SUM.sig

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/SHA256SUM

@@ -0,0 +1,2 @@
+ff2a532f46f3e2202e69b3aab760da95c290adf33a9a4b1c4a7e17e0a2671421  session.meta
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  session.summary

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/conn_1/.journal

@@ -0,0 +1,32 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING connection.meta
+CREATED connection.meta
+CREATING connection.summary
+CREATED connection.summary
+CREATING messages-inbound.data
+CREATED messages-inbound.data
+CREATING messages-outbound.data
+CREATED messages-outbound.data
+CREATING requests-inbound.data
+CREATED requests-inbound.data
+CREATING requests-outbound.data
+CREATED requests-outbound.data
+CLOSING messages-inbound.data
+CLOSED messages-inbound.data
+CLOSING messages-outbound.data
+CLOSED messages-outbound.data
+CLOSING requests-inbound.data
+CLOSED requests-inbound.data
+CLOSING requests-outbound.data
+CLOSED requests-outbound.data
+CLOSING connection.meta
+CLOSED connection.meta
+CLOSING connection.summary
+CLOSED connection.summary
+CLOSING SHA256SUM
+CLOSED SHA256SUM
+CLOSING SHA256SUM.sig
+CLOSED SHA256SUM.sig

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/conn_1/SHA256SUM

@@ -0,0 +1,6 @@
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  messages-inbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  messages-outbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-inbound.data
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  requests-outbound.data
+03edd8ed2a4d678ea79d14538e504b6bb9aaec51c7e3ebf3a4891083d97ac7a9  connection.meta
+e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855  connection.summary

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/conn_1/connection.meta

@@ -0,0 +1,5 @@
+id: conn_1
+messages: inbound
+messages: outbound
+requests: inbound
+requests: outbound

internal/bsr/testdata/TestBsr/session_not_multiplexed/closed/session.meta

@@ -0,0 +1,2 @@
+id: session_123456789
+connection: conn_1.connection

internal/bsr/testdata/TestBsr/session_not_multiplexed/opened/.journal

@@ -0,0 +1,8 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING session.meta
+CREATED session.meta
+CREATING session.summary
+CREATED session.summary

internal/bsr/testdata/TestBsr/session_not_multiplexed/opened/conn_1/.journal

@@ -0,0 +1,8 @@
+CREATING SHA256SUM.sig
+CREATED SHA256SUM.sig
+CREATING SHA256SUM
+CREATED SHA256SUM
+CREATING connection.meta
+CREATED connection.meta
+CREATING connection.summary
+CREATED connection.summary

internal/bsr/testdata/TestBsr/session_not_multiplexed/opened/conn_1/connection.meta

@@ -0,0 +1 @@
+id: conn_1

internal/bsr/testdata/TestBsr/session_not_multiplexed/opened/session.meta

@@ -0,0 +1 @@
+id: session_123456789