From dc21e714a553a92d155f4cd345936579c45cd026 Mon Sep 17 00:00:00 2001
From: Damian Debkowski <damian.debkowski@hashicorp.com>
Date: Thu, 16 May 2024 00:09:32 -0700
Subject: [PATCH] feat(db): tracking storage bucket by worker

---
 .../oss/postgres/75/01_storage_bucket.up.sql  |   1 +
 ..._find_session_recordings_for_delete.up.sql |   1 +
 .../88/01_storage_bucket_credential.up.sql    | 233 ++++++++++++++++++
 ...ker_storage_bucket_credential_state.up.sql |  77 ++++++
 ...3_update_storage_bucket_credentials.up.sql |  32 +++
 .../migrations/oss/postgres_88_01_test.go     | 135 ++++++++++
 .../db/sqltest/initdb.d/01_colors_persona.sql |  12 +-
 .../storage/storage_bucket_credential.sql     |  46 ++++
 .../storage/valid_storage_bucket_added.sql    |  15 +-
 9 files changed, 545 insertions(+), 7 deletions(-)
 create mode 100644 internal/db/schema/migrations/oss/postgres/88/01_storage_bucket_credential.up.sql
 create mode 100644 internal/db/schema/migrations/oss/postgres/88/02_worker_storage_bucket_credential_state.up.sql
 create mode 100644 internal/db/schema/migrations/oss/postgres/88/03_update_storage_bucket_credentials.up.sql
 create mode 100644 internal/db/schema/migrations/oss/postgres_88_01_test.go
 create mode 100644 internal/db/sqltest/tests/storage/storage_bucket_credential.sql

diff --git a/internal/db/schema/migrations/oss/postgres/75/01_storage_bucket.up.sql b/internal/db/schema/migrations/oss/postgres/75/01_storage_bucket.up.sql
index 255e41e778..9b731ec85b 100644
--- a/internal/db/schema/migrations/oss/postgres/75/01_storage_bucket.up.sql
+++ b/internal/db/schema/migrations/oss/postgres/75/01_storage_bucket.up.sql
@@ -9,6 +9,7 @@ alter table storage_plugin_storage_bucket
   drop constraint if exists secrets_hmac_must_not_be_empty,
   alter column secrets_hmac drop not null;
 
+-- Replaced in 88/01_storage_bucket_credential.up.sql
 create view storage_plugin_storage_bucket_with_secret as
   select
     spsb.public_id,
diff --git a/internal/db/schema/migrations/oss/postgres/82/04_find_session_recordings_for_delete.up.sql b/internal/db/schema/migrations/oss/postgres/82/04_find_session_recordings_for_delete.up.sql
index e1ff5300ad..590d36a14e 100644
--- a/internal/db/schema/migrations/oss/postgres/82/04_find_session_recordings_for_delete.up.sql
+++ b/internal/db/schema/migrations/oss/postgres/82/04_find_session_recordings_for_delete.up.sql
@@ -3,6 +3,7 @@
 
 begin;
 
+  -- Replaced in 88/01_storage_bucket_credential.up.sql
   create view find_session_recordings_for_delete as
     select
       -- fields for session recordings
diff --git a/internal/db/schema/migrations/oss/postgres/88/01_storage_bucket_credential.up.sql b/internal/db/schema/migrations/oss/postgres/88/01_storage_bucket_credential.up.sql
new file mode 100644
index 0000000000..0f9bd3f87e
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres/88/01_storage_bucket_credential.up.sql
@@ -0,0 +1,233 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+
+  create table storage_bucket_credential (
+    private_id wt_private_id primary key,
+    storage_bucket_id wt_public_id not null
+    constraint storage_plugin_storage_bucket_fkey
+      references storage_plugin_storage_bucket(public_id)
+      on delete cascade
+      on update cascade
+      deferrable initially deferred
+    constraint storage_bucket_credential_storage_bucket_id_uq
+      unique
+  );
+  comment on table storage_bucket_credential is
+    'storage bucket credential contains entries that represent an abstract storage bucket credential.';
+
+  create trigger immutable_columns before update on storage_bucket_credential
+    for each row execute procedure immutable_columns('private_id', 'storage_bucket_id');
+
+  create table storage_bucket_credential_managed_secret (
+    private_id wt_private_id primary key default wt_url_safe_id()
+    constraint storage_bucket_credential_fkey
+      references storage_bucket_credential(private_id)
+      on delete cascade
+      on update cascade,
+    storage_bucket_id wt_public_id not null,
+    secrets_encrypted bytea not null
+    constraint secrets_must_not_be_empty
+      check(length(secrets_encrypted) > 0),
+    key_id wt_public_id not null
+    constraint kms_data_key_version_fkey
+      references kms_data_key_version(private_id)
+      on delete cascade
+      on update cascade
+  );
+  comment on table storage_bucket_credential_managed_secret is
+    'storage bucket credential managed secret contains entries that represent an storage bucket credential subtype.';
+
+  create trigger immutable_columns before update on storage_bucket_credential_managed_secret
+    for each row execute procedure immutable_columns('private_id', 'storage_bucket_id');
+
+  create table storage_bucket_credential_environmental (
+    private_id wt_private_id primary key default wt_url_safe_id()
+    constraint storage_bucket_credential_fkey
+      references storage_bucket_credential(private_id)
+      on delete cascade
+      on update cascade,
+    storage_bucket_id wt_public_id not null
+  );
+  comment on table storage_bucket_credential_environmental is
+    'storage bucket credential environmental contains entries that represent an storage bucket credential subtype.';
+
+  create trigger immutable_columns before update on storage_bucket_credential_environmental
+    for each row execute procedure immutable_columns('private_id', 'storage_bucket_id');
+
+  create function insert_storage_bucket_credential_subtype() returns trigger
+  as $$
+  begin
+    insert into storage_bucket_credential
+      (private_id, storage_bucket_id)
+    values
+      (new.private_id, new.storage_bucket_id);
+    return new;
+  end;
+  $$ language plpgsql;
+
+  create trigger insert_storage_bucket_credential_subtype before insert on storage_bucket_credential_environmental
+    for each row execute procedure insert_storage_bucket_credential_subtype();
+
+  create trigger insert_storage_bucket_credential_subtype before insert on storage_bucket_credential_managed_secret
+    for each row execute procedure insert_storage_bucket_credential_subtype();
+
+  create function delete_storage_bucket_credential_subtype() returns trigger
+  as $$
+  begin
+    delete from storage_bucket_credential
+    where private_id = old.private_id;
+    return null; -- result is ignored since this is an after trigger
+  end;
+  $$ language plpgsql;
+
+  create trigger delete_storage_bucket_credential_subtype after delete on storage_bucket_credential_environmental
+    for each row execute procedure delete_storage_bucket_credential_subtype();
+
+  create trigger delete_storage_bucket_credential_subtype after delete on storage_bucket_credential_managed_secret
+    for each row execute procedure delete_storage_bucket_credential_subtype();
+
+  -- migrate secrets from storage_plugin_storage_bucket_secret to storage_bucket_credential_managed_secret
+  insert into storage_bucket_credential_managed_secret
+    (storage_bucket_id, secrets_encrypted, key_id)
+  select storage_bucket_id, secrets_encrypted, key_id
+    from storage_plugin_storage_bucket_secret;
+
+  -- create storage_bucket_credential_environmental when storage_plugin_storage_bucket_secret does not exist
+  insert into storage_bucket_credential_environmental
+    (storage_bucket_id)
+  select storage_bucket.public_id
+    from (
+      select public_id
+        from storage_plugin_storage_bucket
+       where public_id not in (select storage_bucket_id from storage_plugin_storage_bucket_secret)
+    ) storage_bucket;
+
+  -- temporarily set the new storage_bucket_credential_id column to a text type
+  -- so that we can update the value later 
+  alter table storage_plugin_storage_bucket
+    add column storage_bucket_credential_id text
+  ;
+
+  -- set storage_bucket_credential_id to the expected value
+  update storage_plugin_storage_bucket
+    set (storage_bucket_credential_id) = (
+      select private_id
+        from storage_bucket_credential
+       where storage_plugin_storage_bucket.public_id = storage_bucket_credential.storage_bucket_id
+  );
+
+  -- update storage_bucket_credential_id column to the expected wt_private_id type
+  alter table storage_plugin_storage_bucket
+    alter column storage_bucket_credential_id type wt_private_id
+  ;
+
+  -- enforce foreign key reference constaint for storage_bucket_credential_id column
+  alter table storage_plugin_storage_bucket
+    add constraint storage_bucket_credential_id_fkey
+      foreign key (storage_bucket_credential_id)
+      references storage_bucket_credential(private_id)
+      on update cascade
+      deferrable initially deferred
+  ;
+
+  -- Replaces view from 75/01_storage_bucket.up.sql
+  drop view storage_plugin_storage_bucket_with_secret;
+  create view storage_plugin_storage_bucket_with_secret as
+  select
+    spsb.public_id,
+    spsb.scope_id,
+    spsb.name,
+    spsb.description,
+    spsb.create_time,
+    spsb.update_time,
+    spsb.version,
+    spsb.plugin_id,
+    spsb.bucket_name,
+    spsb.bucket_prefix,
+    spsb.worker_filter,
+    spsb.attributes,
+    spsb.secrets_hmac,
+    sbcms.secrets_encrypted,
+    sbcms.key_id,
+    spsb.storage_bucket_credential_id
+  from storage_plugin_storage_bucket spsb
+  left join storage_bucket_credential_managed_secret sbcms
+    on sbcms.storage_bucket_id = spsb.public_id;
+  comment on view storage_plugin_storage_bucket_with_secret is
+    'storage_plugin_storage_bucket_with_secret is a view where each row contains a storage bucket. '
+    'Encrypted secret and hmac value is not returned if a secret is not associated with the storage bucket.';
+
+  -- Replaces view from 82/04_find_session_recordings_for_delete.up.sql
+  drop view find_session_recordings_for_delete;
+  create view find_session_recordings_for_delete as
+    select
+      -- fields for session recordings
+      rs.public_id,
+      rs.storage_bucket_id,
+
+      -- fields for storage buckets. note this is ALL storage bucket fields
+      sb.scope_id    as storage_bucket_scope_id,
+      sb.name        as storage_bucket_name,
+      sb.description as storage_bucket_description,
+      sb.create_time as storage_bucket_create_time,
+      sb.update_time as storage_bucket_update_time,
+      sb.version     as storage_bucket_version,
+      sb.plugin_id,
+      sb.bucket_name,
+      sb.bucket_prefix,
+      sb.worker_filter,
+      sb.attributes,
+      sb.secrets_hmac,
+      sb.storage_bucket_credential_id,
+
+      -- fields for storage bucket secrets
+      sbcms.secrets_encrypted,
+      sbcms.key_id,
+
+      -- fields for storage bucket plugins
+      plg.scope_id    as plugin_scope_id,
+      plg.name        as plugin_name,
+      plg.description as plugin_description
+
+    from recording_session rs
+      left join storage_plugin_storage_bucket sb
+        on sb.public_id = rs.storage_bucket_id
+      left join storage_bucket_credential_managed_secret sbcms
+        on sbcms.storage_bucket_id = sb.public_id
+      left join plugin plg
+        on plg.public_id = sb.plugin_id
+    where rs.delete_after < now() or rs.delete_time < now()
+    order by rs.delete_time desc, rs.delete_after desc;
+  comment on view find_session_recordings_for_delete is
+    'find_session_recordings_for_delete is used by the delete_session_recording job to find all '
+    'session recordings that need to be automatically deleted along with their storage buckets.';
+
+  -- Drops table from 71/03_storage_bucket.up.sql
+  drop table storage_plugin_storage_bucket_secret;
+  
+  create view storage_bucket_credential_all_subtypes as
+  select
+    private_id,
+    storage_bucket_id,
+    key_id,
+    secrets_encrypted,
+    'managed_secret' as type
+  from
+    storage_bucket_credential_managed_secret
+  union  
+  select
+    private_id,
+    storage_bucket_id,
+    '' as key_id, -- key_id is not applicable to environmental subtype
+    ''::bytea as secrets_encrypted, -- secrets_encrypted is not applicable to environmental subtype
+    'environmental' as type
+  from 
+    storage_bucket_credential_environmental;
+  comment on view storage_bucket_credential_all_subtypes is
+    'storage_bucket_credential_all_subtypes is a view that contains all storage bucket credential '
+    'subtypes. There are two subtypes: environmental & managed secret. Columns that are not applicable '
+    'to the given subtype will have an empty value by default, not null.';
+
+commit;
diff --git a/internal/db/schema/migrations/oss/postgres/88/02_worker_storage_bucket_credential_state.up.sql b/internal/db/schema/migrations/oss/postgres/88/02_worker_storage_bucket_credential_state.up.sql
new file mode 100644
index 0000000000..78fbacdf80
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres/88/02_worker_storage_bucket_credential_state.up.sql
@@ -0,0 +1,77 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+
+  create table worker_storage_bucket_credential_permission_type_enm (
+    type text primary key
+      constraint only_predefined_permission_types_allowed
+      check (
+        type in (
+          'read',
+          'write',
+          'delete'
+        )
+      )
+  );
+  comment on table worker_storage_bucket_credential_permission_type_enm is
+    'worker_storage_bucket_credential_permission_type_enm is an enumeration table for storage bucket credential permission types.';
+
+  insert into worker_storage_bucket_credential_permission_type_enm (type)
+  values
+    ('read'),
+    ('write'),
+    ('delete');
+
+  create table worker_storage_bucket_credential_state_enm (
+    state text primary key
+      constraint only_predefined_state_types_allowed
+      check (
+        state in (
+          'ok',
+          'error',
+          'unknown'
+        )
+      )
+  );
+  comment on table worker_storage_bucket_credential_state_enm is
+    'worker_storage_bucket_credential_state_enm is an enumeration table for storage bucket credential state types.';
+
+  insert into worker_storage_bucket_credential_state_enm (state)
+  values
+    ('ok'),
+    ('error'),
+    ('unknown');
+
+  create table worker_storage_bucket_credential_state (
+    worker_id wt_public_id
+      constraint server_worker_fkey
+        references server_worker(public_id)
+        on delete cascade
+        on update cascade,
+    storage_bucket_credential_id wt_private_id
+      constraint storage_bucket_credential_id_fkey
+        references storage_bucket_credential(private_id)
+        on delete cascade
+        on update cascade,
+    permission_type text not null
+      constraint worker_storage_bucket_credential_permission_type_enm_fkey
+        references worker_storage_bucket_credential_permission_type_enm(type)
+        on delete restrict
+        on update cascade,
+    state text not null
+      constraint worker_storage_bucket_credential_state_enm_fkey
+        references worker_storage_bucket_credential_state_enm(state)
+        on delete restrict
+        on update cascade,
+    error_details text,
+    checked_at wt_timestamp,
+    primary key (worker_id, storage_bucket_credential_id, permission_type)
+  );
+  comment on table worker_storage_bucket_credential_state is
+    'worker storage bucket credential state contains entries that represent an association between a worker and storage bucket credential.';
+
+  create trigger immutable_columns before update on worker_storage_bucket_credential_state
+    for each row execute procedure immutable_columns('worker_id', 'storage_bucket_credential_id');
+
+commit;
diff --git a/internal/db/schema/migrations/oss/postgres/88/03_update_storage_bucket_credentials.up.sql b/internal/db/schema/migrations/oss/postgres/88/03_update_storage_bucket_credentials.up.sql
new file mode 100644
index 0000000000..ada39df4f4
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres/88/03_update_storage_bucket_credentials.up.sql
@@ -0,0 +1,32 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+
+  create view update_worker_storage_bucket_credential as
+    select distinct
+      sb.scope_id as storage_bucket_scope_id,
+      sb.name as storage_bucket_name,
+      sb.description as storage_bucket_description,
+      sb.bucket_name as storage_bucket_bucket_name,
+      sb.bucket_prefix as storage_bucket_bucket_prefix,
+      sb.worker_filter as storage_bucket_worker_filter,
+      sb.attributes as storage_bucket_attributes,
+      sb.plugin_id as plugin_id,
+      pl.name as plugin_name,
+      pl.description as plugin_description,
+      sbc.storage_bucket_id as storage_bucket_id,
+      sbcms.secrets_encrypted as ct_secrets,
+      sbcms.key_id as key_id
+
+    from storage_bucket_credential sbc
+    join storage_plugin_storage_bucket sb
+      on sbc.storage_bucket_id = sb.public_id
+    join plugin pl
+      on sb.plugin_id = pl.public_id
+    left join storage_bucket_credential_managed_secret sbcms
+      on  sbc.private_id = sbcms.private_id;
+  comment on view update_worker_storage_bucket_credential is
+    'update_worker_storage_bucket_credential is used find workers using storage bucket credentials that need to be updated to the latest version.';
+
+commit;
\ No newline at end of file
diff --git a/internal/db/schema/migrations/oss/postgres_88_01_test.go b/internal/db/schema/migrations/oss/postgres_88_01_test.go
new file mode 100644
index 0000000000..6b62b269f8
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres_88_01_test.go
@@ -0,0 +1,135 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package oss_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/common"
+	"github.com/hashicorp/boundary/internal/db/schema"
+	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/stretchr/testify/require"
+)
+
+// Test_MigrationStorageBucketCredential: in this test we will create a storage bucket
+// that utilizes an encrypted secret stored within the storage_plugin_storage_bucket_secret
+// table. Post migration, the entry from the storage_plugin_storage_bucket_secret should
+// be moved to the storage_bucket_credential_managed_secret table. This test will also
+// validate that storage_plugin_storage_bucket entries that do not have encrypted secrets,
+// will create entries in the storage_bucket_credential_environmental table.
+func Test_MigrationStorageBucketCredential(t *testing.T) {
+	require := require.New(t)
+	dialect := dbtest.Postgres
+	ctx := context.Background()
+
+	priorMigration := 87001
+	currentMigration := 88001
+
+	c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1))
+	require.NoError(err)
+	t.Cleanup(func() {
+		require.NoError(c())
+	})
+	d, err := common.SqlOpen(dialect, u)
+	require.NoError(err)
+
+	// migration to the prior migration (before the one we want to test)
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": priorMigration}),
+	))
+	require.NoError(err)
+
+	_, err = m.ApplyMigrations(ctx)
+	require.NoError(err)
+	state, err := m.CurrentState(ctx)
+	require.NoError(err)
+	want := &schema.State{
+		Initialized: true,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   priorMigration,
+				DatabaseSchemaVersion: priorMigration,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
+	}
+	require.Equal(want, state)
+
+	// Seed the data
+	pluginId := "plg________1"
+	insertPlugin := "insert into plugin (public_id, scope_id) values ('plg________1', 'global')"
+	_, err = d.ExecContext(ctx, insertPlugin)
+	require.NoError(err)
+
+	insertPluginSupported := "insert into plugin_storage_supported (public_id) values ('plg________1')"
+	_, err = d.ExecContext(ctx, insertPluginSupported)
+	require.NoError(err)
+
+	envStorageBucketId := "sb_________1"
+	managedSecretStorageBucketId := "sb_________2"
+	insertStorageBucket := "insert into storage_plugin_storage_bucket (public_id, scope_id, name, version, plugin_id, bucket_name, worker_filter) values ($1, 'global', $2, 1, $3, 'bucket', 'pki')"
+	_, err = d.ExecContext(ctx, insertStorageBucket, envStorageBucketId, "env bucket", pluginId)
+	require.NoError(err)
+	_, err = d.ExecContext(ctx, insertStorageBucket, managedSecretStorageBucketId, "managed secret bucket", pluginId)
+	require.NoError(err)
+
+	insertRootKey := "insert into kms_root_key (scope_id, private_id) values ('global', 'krk_____test')"
+	_, err = d.ExecContext(ctx, insertRootKey)
+	require.NoError(err)
+
+	insertRootKeyVersion := "insert into kms_root_key_version (root_key_id, private_id, key) values ('krk_____test', 'krkv____test', '_______test'::bytea)"
+	_, err = d.ExecContext(ctx, insertRootKeyVersion)
+	require.NoError(err)
+
+	insertKmsDataKey := "insert into kms_data_key (root_key_id, private_id, purpose) values ('krk_____test', 'kdk_____test', 'database')"
+	_, err = d.ExecContext(ctx, insertKmsDataKey)
+	require.NoError(err)
+
+	insertKmsDataKeyVersion := "insert into kms_data_key_version (root_key_version_id, data_key_id, private_id, key) values ('krkv____test', 'kdk_____test', 'kdkv____test', '_______test'::bytea)"
+	_, err = d.ExecContext(ctx, insertKmsDataKeyVersion)
+	require.NoError(err)
+
+	insertStorageBucketSecret := "insert into storage_plugin_storage_bucket_secret (storage_bucket_id, secrets_encrypted, key_id) values ($1, $2, $3)"
+	_, err = d.ExecContext(ctx, insertStorageBucketSecret, managedSecretStorageBucketId, []byte("hello_world"), "kdkv____test")
+	require.NoError(err)
+
+	// now we're ready for the migration we want to test.
+	m, err = schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": currentMigration}),
+	))
+	require.NoError(err)
+
+	_, err = m.ApplyMigrations(ctx)
+	require.NoError(err)
+	state, err = m.CurrentState(ctx)
+	require.NoError(err)
+
+	want = &schema.State{
+		Initialized: true,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   currentMigration,
+				DatabaseSchemaVersion: currentMigration,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
+	}
+	require.Equal(want, state)
+
+	// Assert that 'sb_________1' has a storage_bucket_credential_environmental entry
+	var count int
+	row := d.QueryRowContext(ctx, "select count(*) from storage_bucket_credential_environmental where storage_bucket_id = 'sb_________1'")
+	require.NoError(row.Err())
+	require.NoError(row.Scan(&count))
+	require.Equal(1, count)
+
+	// Assert that 'sb_________2' has a storage_bucket_credential_managed_secret entry
+	row = d.QueryRowContext(ctx, "select count(*) from storage_bucket_credential_managed_secret where storage_bucket_id = 'sb_________2'")
+	require.NoError(row.Err())
+	require.NoError(row.Scan(&count))
+	require.Equal(1, count)
+}
diff --git a/internal/db/sqltest/initdb.d/01_colors_persona.sql b/internal/db/sqltest/initdb.d/01_colors_persona.sql
index 8a49c75daa..fba8d7d947 100644
--- a/internal/db/sqltest/initdb.d/01_colors_persona.sql
+++ b/internal/db/sqltest/initdb.d/01_colors_persona.sql
@@ -401,11 +401,17 @@ begin;
   values
     ('pl__plg___sb');
 
+  insert into storage_bucket_credential_environmental
+    (private_id, storage_bucket_id)
+  values
+    ('sbc___global', 'sb____global'),
+    ('sbc___colors', 'sb____colors');
+
   insert into storage_plugin_storage_bucket
-    (plugin_id,      scope_id,       public_id,      bucket_name,             worker_filter,        secrets_hmac)
+    (plugin_id,      scope_id,       public_id,      bucket_name,             worker_filter,        secrets_hmac, storage_bucket_credential_id)
   values
-    ('pl__plg___sb', 'global',       'sb____global', 'Global Storage Bucket', 'test worker filter', '\xdeadbeef'),
-    ('pl__plg___sb', 'o_____colors', 'sb____colors', 'Colors Storage Bucket', 'test worker filter', '\xdeadbeef');
+    ('pl__plg___sb', 'global',       'sb____global', 'Global Storage Bucket', 'test worker filter', '\xdeadbeef', 'sbc___global'),
+    ('pl__plg___sb', 'o_____colors', 'sb____colors', 'Colors Storage Bucket', 'test worker filter', '\xdeadbeef', 'sbc___colors');
 
   insert into target_tcp
     (project_id,     public_id,      name)
diff --git a/internal/db/sqltest/tests/storage/storage_bucket_credential.sql b/internal/db/sqltest/tests/storage/storage_bucket_credential.sql
new file mode 100644
index 0000000000..877983347e
--- /dev/null
+++ b/internal/db/sqltest/tests/storage/storage_bucket_credential.sql
@@ -0,0 +1,46 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+    select plan(5);
+    select wtt_load('widgets', 'iam', 'kms');
+
+    -- fail to create storage bucket without storage bucket credential
+    prepare missing_storage_bucket_credential as
+    insert into storage_plugin_storage_bucket
+      (public_id,      scope_id, plugin_id,      bucket_name,        worker_filter)
+    values
+      ('sb_________1', 'global', 'pl__plg___sb', 'test bucket environmental', 'test worker filter');
+    select throws_ok('missing_storage_bucket_credential', '23502');
+
+    -- fail to create duplicate storage bucket credential environmental  
+    prepare insert_duplicate_env_credential as
+    insert into storage_bucket_credential_environmental
+      (private_id, storage_bucket_id)
+    values
+      ('sbc___global', 'sb____global');
+    select throws_ok('insert_duplicate_env_credential', '23505');
+
+    -- create a managed secret SBC
+    insert into storage_bucket_credential_managed_secret
+      (private_id, storage_bucket_id, secrets_encrypted, key_id)
+    values
+      ('sbc________1', 'sb_________1', 'secret'::bytea, 'kdkv___widget');
+    select is(count(*), 1::bigint) from storage_bucket_credential where private_id = 'sbc________1';
+
+    insert into storage_plugin_storage_bucket
+      (public_id,      scope_id, plugin_id,      bucket_name,        worker_filter,        secrets_hmac, storage_bucket_credential_id)
+    values
+      ('sb_________1', 'global', 'pl__plg___sb', 'test bucket name', 'test worker filter', '\xdeadbeef', 'sbc________1');
+    select is(count(*), 1::bigint) from storage_plugin_storage_bucket where public_id = 'sb_________1';
+
+    -- fail to create duplicate storage bucket credential managed secret
+    prepare insert_duplicate_managed_secret_credential as
+    insert into storage_bucket_credential_managed_secret
+      (private_id, storage_bucket_id, secrets_encrypted, key_id)
+    values
+      ('sbc________1', 'sb_________1', 'secret'::bytea, 'kdkv___widget');
+    select throws_ok('insert_duplicate_managed_secret_credential', '23505');
+
+    select * from finish();
+rollback;
diff --git a/internal/db/sqltest/tests/storage/valid_storage_bucket_added.sql b/internal/db/sqltest/tests/storage/valid_storage_bucket_added.sql
index caa882fb14..95babc297e 100644
--- a/internal/db/sqltest/tests/storage/valid_storage_bucket_added.sql
+++ b/internal/db/sqltest/tests/storage/valid_storage_bucket_added.sql
@@ -7,19 +7,26 @@ begin;
     -- Ensure test data is correct
     select is(count(*), 2::bigint) from storage_plugin_storage_bucket;
 
+    -- insert storage bucket credential
+    insert into storage_bucket_credential_environmental
+      (private_id, storage_bucket_id)
+    values
+      ('sbc________1', 'sb_________1'),
+      ('sbc________2', 'sb_________2');
+
     -- insert global storage bucket
     insert into storage_plugin_storage_bucket
-      (public_id,      scope_id, plugin_id,      bucket_name,        worker_filter,        secrets_hmac)
+      (public_id,      scope_id, plugin_id,      bucket_name,        worker_filter,        secrets_hmac, storage_bucket_credential_id)
     values
-      ('sb_________1', 'global', 'pl__plg___sb', 'test bucket name', 'test worker filter', '\xdeadbeef');
+      ('sb_________1', 'global', 'pl__plg___sb', 'test bucket name', 'test worker filter', '\xdeadbeef', 'sbc________1');
 
     select is(count(*), 1::bigint) from storage_plugin_storage_bucket where public_id = 'sb_________1';
 
     -- insert org storage bucket
     insert into storage_plugin_storage_bucket
-      (public_id,      scope_id,       plugin_id,      bucket_name,        worker_filter,        secrets_hmac)
+      (public_id,      scope_id,       plugin_id,      bucket_name,        worker_filter,        secrets_hmac, storage_bucket_credential_id)
     values
-      ('sb_________2', 'o_____colors', 'pl__plg___sb', 'test bucket name', 'test worker filter', '\xdeadbeef');
+      ('sb_________2', 'o_____colors', 'pl__plg___sb', 'test bucket name', 'test worker filter', '\xdeadbeef', 'sbc________2');
 
     select is(count(*), 1::bigint) from storage_plugin_storage_bucket where public_id = 'sb_________2';