Fix db migration that adds support for OIDC in the data warehouse (#1471

)

Four new columns were added to the `wh_user_dimension` table to support
OIDC. The data type for the columns was originally `wh_dim_text` which
is a domain type with a `not null` restriction. Adding the new columns
caused the migration to fail if the `wh_user_dimension` table had any
rows in it because the existing rows would violate the `not null`
restriction.

Closes #1469

internal/db/schema/migrations/postgres/14/01_wh_user_dimension_oidc.up.sql

@@ -1,10 +1,39 @@
 begin;
 
   alter table wh_user_dimension
-    add column auth_method_external_id  wh_dim_text,
-    add column auth_account_external_id wh_dim_text,
-    add column auth_account_full_name   wh_dim_text,
-    add column auth_account_email       wh_dim_text
+    add column auth_method_external_id  text,
+    add column auth_account_external_id text,
+    add column auth_account_full_name   text,
+    add column auth_account_email       text
+  ;
+
+  update wh_user_dimension
+  set auth_method_type =
+        case when auth_method_id like 'ampw_%' then 'password auth method'
+             when auth_method_id like 'amoidc_%' then 'oidc auth method'
+             else 'Unknown' end,
+      auth_account_type =
+        case when auth_account_id like 'acctpw_%' then 'password auth account'
+             when auth_account_id like 'acctoidc_%' then 'oidc auth account'
+             else 'Unknown' end,
+      auth_method_external_id =
+        case when auth_method_id like 'ampw_%' then 'Not Applicable'
+             else 'Unknown' end,
+      auth_account_external_id =
+        case when auth_method_id like 'ampw_%' then 'Not Applicable'
+             else 'Unknown' end,
+      auth_account_full_name =
+        case when auth_method_id like 'ampw_%' then 'Not Applicable'
+             else 'Unknown' end,
+      auth_account_email =
+        case when auth_method_id like 'ampw_%' then 'Not Applicable'
+             else 'Unknown' end;
+
+  alter table wh_user_dimension
+    alter column auth_method_external_id  type wh_dim_text,
+    alter column auth_account_external_id type wh_dim_text,
+    alter column auth_account_full_name   type wh_dim_text,
+    alter column auth_account_email       type wh_dim_text
   ;
 
   drop view whx_user_dimension_source;

internal/db/schema/migrations/postgres/14/warehouse_user_dim_test.go

@@ -0,0 +1,165 @@
+package migration
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/auth/oidc"
+	"github.com/hashicorp/boundary/internal/authtoken"
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/db/schema"
+	"github.com/hashicorp/boundary/internal/docker"
+	"github.com/hashicorp/boundary/internal/host/static"
+	"github.com/hashicorp/boundary/internal/iam"
+	"github.com/hashicorp/boundary/internal/kms"
+	"github.com/hashicorp/boundary/internal/session"
+	"github.com/hashicorp/boundary/internal/target"
+	wrapping "github.com/hashicorp/go-kms-wrapping"
+	"github.com/jinzhu/gorm"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigrations_UserDimension(t *testing.T) {
+	const (
+		priorMigration   = 13001
+		currentMigration = 14001
+	)
+
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+	ctx := context.Background()
+	dialect := "postgres"
+
+	c, u, _, err := docker.StartDbInDocker(dialect)
+	require.NoError(err)
+	t.Cleanup(func() {
+		require.NoError(c())
+	})
+	d, err := sql.Open(dialect, u)
+	require.NoError(err)
+
+	// migration to the prior migration (before the one we want to test)
+	oState := schema.TestCloneMigrationStates(t)
+	nState := schema.TestCreatePartialMigrationState(oState["postgres"], priorMigration)
+	oState["postgres"] = nState
+
+	m, err := schema.NewManager(ctx, dialect, d, schema.WithMigrationStates(oState))
+	require.NoError(err)
+
+	assert.NoError(m.RollForward(ctx))
+	state, err := m.CurrentState(ctx)
+	require.NoError(err)
+	assert.Equal(priorMigration, state.DatabaseSchemaVersion)
+	assert.False(state.Dirty)
+
+	// okay, now we can seed the database with test data
+	conn, err := gorm.Open(dialect, u)
+	require.NoError(err)
+
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+
+	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
+	require.NotNil(prj)
+	assert.NotEmpty(prj.GetPublicId())
+
+	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
+	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
+	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
+	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
+
+	tar := target.TestTcpTarget(t, conn, prj.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()}))
+	var sessions []*session.Session
+
+	kmsCache := kms.TestKms(t, conn, wrapper)
+	databaseWrapper, err := kmsCache.GetWrapper(ctx, org.GetPublicId(), kms.KeyPurposeDatabase)
+	require.NoError(err)
+
+	{
+		at := authtoken.TestAuthToken(t, conn, kmsCache, org.GetPublicId())
+		uId := at.GetIamUserId()
+
+		sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
+			UserId:      uId,
+			HostId:      h.GetPublicId(),
+			TargetId:    tar.GetPublicId(),
+			HostSetId:   hs.GetPublicId(),
+			AuthTokenId: at.GetPublicId(),
+			ScopeId:     prj.GetPublicId(),
+			Endpoint:    "tcp://127.0.0.1:22",
+		})
+		sessions = append(sessions, sess)
+	}
+
+	{
+		at := testOidcAuthToken(t, conn, kmsCache, databaseWrapper, org.GetPublicId())
+		uId := at.GetIamUserId()
+
+		sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
+			UserId:      uId,
+			HostId:      h.GetPublicId(),
+			TargetId:    tar.GetPublicId(),
+			HostSetId:   hs.GetPublicId(),
+			AuthTokenId: at.GetPublicId(),
+			ScopeId:     prj.GetPublicId(),
+			Endpoint:    "tcp://127.0.0.1:22",
+		})
+		sessions = append(sessions, sess)
+	}
+
+	sessionRepo, err := session.NewRepository(rw, rw, kmsCache)
+	require.NoError(err)
+
+	count, err := sessionRepo.TerminateCompletedSessions(ctx)
+	assert.NoError(err)
+	assert.Zero(count)
+
+	for _, sess := range sessions {
+		// call TerminateSession
+		_, err = sessionRepo.TerminateSession(ctx, sess.GetPublicId(), 1, session.ClosedByUser)
+		assert.NoError(err)
+	}
+
+	// now we're ready for the migration we want to test.
+	oState = schema.TestCloneMigrationStates(t)
+	nState = schema.TestCreatePartialMigrationState(oState["postgres"], currentMigration)
+	oState["postgres"] = nState
+
+	m, err = schema.NewManager(ctx, dialect, d, schema.WithMigrationStates(oState))
+	require.NoError(err)
+
+	assert.NoError(m.RollForward(ctx))
+	state, err = m.CurrentState(ctx)
+	require.NoError(err)
+	assert.Equal(currentMigration, state.DatabaseSchemaVersion)
+	assert.False(state.Dirty)
+}
+
+func testOidcAuthToken(t *testing.T, conn *gorm.DB, kms *kms.Kms, wrapper wrapping.Wrapper, scopeId string) *authtoken.AuthToken {
+	t.Helper()
+
+	authMethod := oidc.TestAuthMethod(
+		t, conn, wrapper, scopeId, oidc.ActivePrivateState,
+		"alice-rp", "fido",
+		oidc.WithIssuer(oidc.TestConvertToUrls(t, "https://www.alice.com")[0]),
+		oidc.WithSigningAlgs(oidc.RS256),
+		oidc.WithApiUrl(oidc.TestConvertToUrls(t, "https://www.alice.com/callback")[0]),
+	)
+	acct := oidc.TestAccount(t, conn, authMethod, "test-subject")
+
+	ctx := context.Background()
+	rw := db.New(conn)
+	iamRepo, err := iam.NewRepository(rw, rw, kms)
+	require.NoError(t, err)
+
+	u := iam.TestUser(t, iamRepo, scopeId, iam.WithAccountIds(acct.PublicId))
+
+	repo, err := authtoken.NewRepository(rw, rw, kms)
+	require.NoError(t, err)
+
+	at, err := repo.CreateAuthToken(ctx, u, acct.GetPublicId())
+	require.NoError(t, err)
+	return at
+}