sql: Add ownership concept

Added ownership when creating objects. Owners have ALL privilege on the object.

Currently, ownership cannot be changed, we will need to implement the
ALTER OWNER commands for all objects.

The privileges CREATE/DROP currently exist to alleviate missing privileges
from the lack of ownership, this PR does affect CREATE/DROP privileges.

Also added testuser2 certs to allow using testuser2 in logictests to test
inheritance between multiple roles.

Objects created before 20.2 will have not have ownership explicitly set,
however we have logic to check that ownerless objects before 20.2 have
admin as their owner if not a system object and node as an owner if it is
a system object.

Release note (sql change): Added "ownership" concept objects.
Objects must have an owner, all objects that do not have owners currently
will have admin set as the default owner except system objects.
System objects without owners will have node as their owner.
By default, owners are the creator of the object. Owners have all privileges
to the objects they own. Similarly, any roles that are members of the owner
role also have all privileges on the object.

pkg/ccl/backupccl/restore_job.go

@@ -449,7 +449,7 @@ func WriteDescriptors(
 			// the users on the restoring cluster match the ones that were on the
 			// cluster that was backed up. So we wipe the privileges on the database.
 			if descCoverage != tree.AllDescriptors {
-				desc.Privileges = sqlbase.NewDefaultPrivilegeDescriptor()
+				desc.Privileges = sqlbase.NewDefaultPrivilegeDescriptor(sqlbase.AdminRole)
 			}
 			wroteDBs[desc.GetID()] = desc
 			if err := catalogkv.WriteNewDescToBatch(ctx, false /* kvTrace */, settings, b, keys.SystemSQLCodec, desc.GetID(), desc); err != nil {
@@ -972,7 +972,7 @@ func createImportingDescriptors(
 	}
 	if details.DescriptorCoverage == tree.AllDescriptors {
 		databases = append(databases, sqlbase.NewInitialDatabaseDescriptor(
-			sqlbase.ID(tempSystemDBID), restoreTempSystemDB))
+			sqlbase.ID(tempSystemDBID), restoreTempSystemDB, sqlbase.AdminRole))
 	}
 
 	// We get the spans of the restoring tables _as they appear in the backup_,

pkg/ccl/backupccl/targets_test.go

@@ -46,7 +46,7 @@ func TestDescriptorsMatchingTargets(t *testing.T) {
 			return *desc.DescriptorProto()
 		}
 		mkDB := func(id sqlbase.ID, name string) sqlbase.Descriptor {
-			return *sqlbase.NewInitialDatabaseDescriptor(id, name).DescriptorProto()
+			return *sqlbase.NewInitialDatabaseDescriptor(id, name, sqlbase.AdminRole).DescriptorProto()
 		}
 		mkTyp := func(desc typDesc) sqlbase.Descriptor {
 			return *sqlbase.NewImmutableTypeDescriptor(desc).DescriptorProto()

pkg/ccl/importccl/import_table_creation.go

@@ -162,7 +162,7 @@ func MakeSimpleTableDescriptor(
 		keys.PublicSchemaID,
 		tableID,
 		hlc.Timestamp{WallTime: walltime},
-		sqlbase.NewDefaultPrivilegeDescriptor(),
+		sqlbase.NewDefaultPrivilegeDescriptor(sqlbase.AdminRole),
 		affected,
 		semaCtx,
 		&evalCtx,

pkg/ccl/importccl/load.go

@@ -210,8 +210,10 @@ func Load(
 			var txn *kv.Txn
 			// At this point the CREATE statements in the loaded SQL do not
 			// use the SERIAL type so we need not process SERIAL types here.
+
 			desc, err := sql.MakeTableDesc(ctx, txn, nil /* vt */, st, s, dbDesc.GetID(), keys.PublicSchemaID,
-				0 /* table ID */, ts, privs, affected, nil, evalCtx, evalCtx.SessionData, false /* temporary */)
+				0 /* table ID */, ts, privs, affected, nil, evalCtx, evalCtx.SessionData,
+				false /* temporary */)
 			if err != nil {
 				return backupccl.BackupManifest{}, errors.Wrap(err, "make table desc")
 			}

pkg/ccl/importccl/load_test.go

@@ -57,8 +57,8 @@ func TestGetDescriptorFromDB(t *testing.T) {
 	s, sqlDB, kvDB := serverutils.StartServer(t, params)
 	defer s.Stopper().Stop(ctx)
 
-	aliceDesc := sqlbase.NewInitialDatabaseDescriptor(10000, "alice")
-	bobDesc := sqlbase.NewInitialDatabaseDescriptor(9999, "bob")
+	aliceDesc := sqlbase.NewInitialDatabaseDescriptor(10000, "alice", sqlbase.AdminRole)
+	bobDesc := sqlbase.NewInitialDatabaseDescriptor(9999, "bob", sqlbase.AdminRole)
 
 	err := kvDB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
 		if err := txn.SetSystemConfigTrigger(true /* forSystemTenant */); err != nil {

pkg/ccl/importccl/read_import_mysql.go

@@ -392,9 +392,9 @@ func mysqlTableToCockroach(
 	}
 
 	var seqDesc *sqlbase.TableDescriptor
+	owner := sqlbase.AdminRole
 	// If we have an auto-increment seq, create it and increment the id.
 	if seqName != "" {
-		priv := sqlbase.NewDefaultPrivilegeDescriptor()
 		var opts tree.SequenceOptions
 		if startingValue != 0 {
 			opts = tree.SequenceOptions{{Name: tree.SeqOptStart, IntVal: &startingValue}}
@@ -404,6 +404,10 @@ func mysqlTableToCockroach(
 		var err error
 		if p != nil {
 			params := p.RunParams(ctx)
+			if params.SessionData() != nil {
+				owner = params.SessionData().User
+			}
+			priv := sqlbase.NewDefaultPrivilegeDescriptor(owner)
 			desc, err = sql.MakeSequenceTableDesc(
 				seqName,
 				opts,
@@ -416,6 +420,7 @@ func mysqlTableToCockroach(
 				&params,
 			)
 		} else {
+			priv := sqlbase.NewDefaultPrivilegeDescriptor(owner)
 			desc, err = sql.MakeSequenceTableDesc(
 				seqName,
 				opts,

pkg/ccl/importccl/read_import_pgdump.go

@@ -243,7 +243,7 @@ func readPostgresCreateTable(
 					keys.PublicSchemaID,
 					id,
 					hlc.Timestamp{WallTime: walltime},
-					sqlbase.NewDefaultPrivilegeDescriptor(),
+					sqlbase.NewDefaultPrivilegeDescriptor(params.SessionData().User),
 					false, /* temporary */
 					&params,
 				)

pkg/ccl/importccl/testutils_test.go

@@ -50,7 +50,7 @@ func descForTable(
 		name := parsed[0].AST.(*tree.CreateSequence).Name.String()
 
 		ts := hlc.Timestamp{WallTime: nanos}
-		priv := sqlbase.NewDefaultPrivilegeDescriptor()
+		priv := sqlbase.NewDefaultPrivilegeDescriptor(sqlbase.AdminRole)
 		desc, err := sql.MakeSequenceTableDesc(
 			name,
 			tree.SequenceOptions{},

pkg/kv/kvserver/client_rangefeed_test.go

@@ -65,7 +65,8 @@ func TestRangefeedWorksOnSystemRangesUnconditionally(t *testing.T) {
 		const junkDescriptorID = 42
 		require.GreaterOrEqual(t, keys.MaxReservedDescID, junkDescriptorID)
 		junkDescriptorKey := sqlbase.MakeDescMetadataKey(keys.SystemSQLCodec, junkDescriptorID)
-		junkDescriptor := sqlbase.NewInitialDatabaseDescriptor(junkDescriptorID, "junk")
+		junkDescriptor := sqlbase.NewInitialDatabaseDescriptor(
+			junkDescriptorID, "junk", sqlbase.AdminRole)
 		require.NoError(t, db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
 			if err := txn.SetSystemConfigTrigger(true /* forSystemTenant */); err != nil {
 				return err

pkg/kv/kvserver/gossip_test.go

@@ -219,13 +219,13 @@ func TestGossipAfterAbortOfSystemConfigTransactionAfterFailureDueToIntents(t *te
 	txB := db.NewTxn(ctx, "b")
 
 	require.NoError(t, txA.SetSystemConfigTrigger(true /* forSystemTenant */))
-	db1000 := sqlbase.NewInitialDatabaseDescriptor(1000, "1000")
+	db1000 := sqlbase.NewInitialDatabaseDescriptor(1000, "1000", sqlbase.AdminRole)
 	require.NoError(t, txA.Put(ctx,
 		keys.SystemSQLCodec.DescMetadataKey(1000),
 		db1000.DescriptorProto()))
 
 	require.NoError(t, txB.SetSystemConfigTrigger(true /* forSystemTenant */))
-	db2000 := sqlbase.NewInitialDatabaseDescriptor(2000, "2000")
+	db2000 := sqlbase.NewInitialDatabaseDescriptor(2000, "2000", sqlbase.AdminRole)
 	require.NoError(t, txB.Put(ctx,
 		keys.SystemSQLCodec.DescMetadataKey(2000),
 		db2000.DescriptorProto()))

pkg/kv/kvserver/reports/constraint_stats_report_test.go

@@ -758,7 +758,7 @@ func compileTestCase(tc baseReportTestCase) (compiledTestCase, error) {
 			}
 		}
 		sysCfgBuilder.addDBDesc(dbID,
-			sqlbase.NewInitialDatabaseDescriptor(sqlbase.ID(dbID), db.name))
+			sqlbase.NewInitialDatabaseDescriptor(sqlbase.ID(dbID), db.name, sqlbase.AdminRole))
 
 		for _, table := range db.tables {
 			tableID := objectCounter

pkg/security/certificate_manager_test.go

@@ -40,7 +40,7 @@ func TestManagerWithEmbedded(t *testing.T) {
 		t.Error("expected non-nil NodeCert")
 	}
 	clientCerts := cm.ClientCerts()
-	if a, e := len(clientCerts), 2; a != e {
+	if a, e := len(clientCerts), 3; a != e {
 		t.Errorf("expected %d client certs, found %d", e, a)
 	}
 
@@ -61,6 +61,9 @@ func TestManagerWithEmbedded(t *testing.T) {
 	if _, err := cm.GetClientTLSConfig("testuser"); err != nil {
 		t.Error(err)
 	}
+	if _, err := cm.GetClientTLSConfig("testuser2"); err != nil {
+		t.Error(err)
+	}
 	if _, err := cm.GetClientTLSConfig("my-random-user"); err == nil {
 		t.Error("unexpected success")
 	}

pkg/security/securitytest/.gitignore

@@ -1,2 +1,2 @@
 # Override root gitignore rule on *.test*
-!client.testuser.*
+!client.testuser*.*