cert: Use tenant scoped client cert for auth.

This PR extends the TLS code to use a tenant scoped
client cert to authenticate a client for specific tenant.

Release note (security update): We introduce a new
tenant scoped client certificate to authenticate a client
on a specific tenant. A tenant scoped client certificate contains the client
name within the CN and the tenant ID, to which the
certificate is being scoped to, as the SAN. The tenant ID
is embedded within the URI section with the format
"crdb://tenant/<tenant_id>". For example, a root
client certificate scoped to a tenant with ID 123
will contain "root" in the CN field and the URI
"crdb://tenant/123" in the URI section of the
certificate. This certificate will authorize the root
client on the tenant with the ID 123. It will result
in an authorization error if used to authenticate the
root client on any other tenant.

pkg/cli/client_url.go

@@ -16,7 +16,6 @@ import (
 	"path/filepath"
 
 	"github.com/cockroachdb/cockroach/pkg/cli/cliflags"
-	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server/pgurl"
@@ -360,7 +359,7 @@ func (cliCtx *cliContext) makeClientConnURL() (*pgurl.URL, error) {
 		userName = security.RootUserName()
 	}
 
-	sCtx := rpc.MakeSecurityContext(cliCtx.Config, security.CommandTLSSettings{}, roachpb.SystemTenantID)
+	sCtx := rpc.MakeSecurityContext(cliCtx.Config, security.CommandTLSSettings{}, cliCtx.tenantID)
 	if err := sCtx.LoadSecurityOptions(purl, userName); err != nil {
 		return nil, err
 	}

pkg/cli/context.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cli/clisqlshell"
 	"github.com/cockroachdb/cockroach/pkg/cli/democluster"
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/server/pgurl"
@@ -200,6 +201,10 @@ type cliContext struct {
 
 	// For `cockroach version --build-tag`.
 	showVersionUsingOnlyBuildTag bool
+
+	// tenantID indicates the tenant to run the CLI utility against.
+	// Default value is the system tenant.
+	tenantID roachpb.TenantID
 }
 
 // cliCtx captures the command-line parameters common to most CLI utilities.
@@ -233,6 +238,7 @@ func setCliContextDefaults() {
 	// TODO(knz): Deprecated in v21.1. Remove this.
 	cliCtx.deprecatedLogOverrides.reset()
 	cliCtx.showVersionUsingOnlyBuildTag = false
+	cliCtx.tenantID = roachpb.SystemTenantID
 }
 
 // sqlConnContext captures the connection configuration for all SQL

pkg/rpc/auth.go

@@ -116,7 +116,7 @@ func (a kvAuth) authenticate(ctx context.Context) (roachpb.TenantID, error) {
 		return roachpb.TenantID{}, errTLSInfoMissing
 	}
 
-	certUsers, err := security.GetCertificateUsers(&tlsInfo.State)
+	certUserScope, err := security.GetCertificateUserScope(&tlsInfo.State)
 	if err != nil {
 		return roachpb.TenantID{}, err
 	}
@@ -155,16 +155,35 @@ func (a kvAuth) authenticate(ctx context.Context) (roachpb.TenantID, error) {
 	//   gateway uses a connection dialed as the node user.
 	//
 	// In both cases, we must check that the client cert is either root
-	// or node.
+	// or node. We also need to check that the tenant scope for the cert
+	// is either the system tenant ID or matches the tenant ID of the server.
 
 	// TODO(benesch): the vast majority of RPCs should be limited to just
 	// NodeUser. This is not a security concern, as RootUser has access to
 	// read and write all data, merely good hygiene. For example, there is
 	// no reason to permit the root user to send raw Raft RPCs.
-	if !security.Contains(certUsers, security.NodeUser) &&
-		!security.Contains(certUsers, security.RootUser) {
-		return roachpb.TenantID{}, authErrorf("user %s is not allowed to perform this RPC", certUsers)
+	nodeUserScope, isNodeUser := certUserScope[security.NodeUser]
+	rootUserScope, isRootUser := certUserScope[security.RootUser]
+	if !isNodeUser && !isRootUser {
+		return roachpb.TenantID{}, authErrorf("user %s is not allowed to perform this RPC", clientCert.Subject.CommonName)
+	}
+	var userScope security.CertificateUserScope
+	if isNodeUser {
+		userScope = nodeUserScope
+	} else {
+		userScope = rootUserScope
+	}
+	_, isSystemTenant := userScope.TenantIDs[roachpb.SystemTenantID]
+	if isSystemTenant {
+		// Certificate scope includes system tenant, allow operation to proceed.
+		return roachpb.TenantID{}, nil
+	}
+	_, authorizedForCurrentTenant := userScope.TenantIDs[a.tenant.tenantID]
+	if authorizedForCurrentTenant {
+		// User to authorized to perform operations on current tenant, allow operation to proceed.
+		return roachpb.TenantID{}, nil
 	}
 
-	return roachpb.TenantID{}, nil
+	// User is not authorized on this particular tenant, return auth error.
+	return roachpb.TenantID{}, authErrorf("user %s is not allowed to perform RPC on tenant %d", userScope.Username, a.tenant.tenantID)
 }

pkg/rpc/auth_test.go

@@ -95,10 +95,11 @@ func TestTenantFromCert(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	correctOU := []string{security.TenantsOU}
 	for _, tc := range []struct {
-		ous        []string
-		commonName string
-		expTenID   roachpb.TenantID
-		expErr     string
+		ous         []string
+		commonName  string
+		expTenID    roachpb.TenantID
+		expErr      string
+		tenantScope []roachpb.TenantID
 	}{
 		{ous: correctOU, commonName: "10", expTenID: roachpb.MakeTenantID(10)},
 		{ous: correctOU, commonName: roachpb.MinTenantID.String(), expTenID: roachpb.MinTenantID},
@@ -109,9 +110,11 @@ func TestTenantFromCert(t *testing.T) {
 		{ous: correctOU, commonName: "1", expErr: `invalid tenant ID 1 in Common Name \(CN\)`},
 		{ous: correctOU, commonName: "root", expErr: `could not parse tenant ID from Common Name \(CN\)`},
 		{ous: correctOU, commonName: "other", expErr: `could not parse tenant ID from Common Name \(CN\)`},
-		{ous: []string{"foo"}, commonName: "other", expErr: `user \[other\] is not allowed to perform this RPC`},
-		{ous: nil, commonName: "other", expErr: `user \[other\] is not allowed to perform this RPC`},
+		{ous: []string{"foo"}, commonName: "other", expErr: `user other is not allowed to perform this RPC`},
+		{ous: nil, commonName: "other", expErr: `user other is not allowed to perform this RPC`},
 		{ous: append([]string{"foo"}, correctOU...), commonName: "other", expErr: `could not parse tenant ID from Common Name`},
+		{ous: nil, commonName: "root"},
+		{ous: nil, commonName: "root", tenantScope: []roachpb.TenantID{roachpb.MakeTenantID(10)}, expErr: "user root is not allowed to perform RPC on tenant {1}"},
 	} {
 		t.Run(tc.commonName, func(t *testing.T) {
 			cert := &x509.Certificate{
@@ -120,6 +123,11 @@ func TestTenantFromCert(t *testing.T) {
 					OrganizationalUnit: tc.ous,
 				},
 			}
+			if len(tc.tenantScope) > 0 {
+				tenantSANs, err := security.MakeTenantURISANs(tc.tenantScope)
+				require.NoError(t, err)
+				cert.URIs = append(cert.URIs, tenantSANs...)
+			}
 			tlsInfo := credentials.TLSInfo{
 				State: tls.ConnectionState{
 					PeerCertificates: []*x509.Certificate{cert},

pkg/rpc/pg.go

@@ -77,34 +77,23 @@ func (ctx *SecurityContext) LoadSecurityOptions(u *pgurl.URL, username security.
 		// (Re)populate the transport information.
 		u.WithTransport(pgurl.TransportTLS(tlsMode, caCertPath))
 
-		var missing bool // certs found on file system?
-		loader := security.GetAssetLoader()
-
 		// Fetch client certs, but don't fail if they're absent, we may be
 		// using a password.
 		certPath := ctx.ClientCertPath(username)
 		keyPath := ctx.ClientKeyPath(username)
-		_, err1 := loader.Stat(certPath)
-		_, err2 := loader.Stat(keyPath)
-		if err1 != nil || err2 != nil {
-			missing = true
-		}
+		certsAvailable := checkCertAndKeyAvailable(certPath, keyPath)
+
 		// If the command specifies user node, and we did not find
 		// client.node.crt, try with just node.crt.
-		if missing && username.IsNodeUser() {
-			missing = false
+		if !certsAvailable && username.IsNodeUser() {
 			certPath = ctx.NodeCertPath()
 			keyPath = ctx.NodeKeyPath()
-			_, err1 = loader.Stat(certPath)
-			_, err2 = loader.Stat(keyPath)
-			if err1 != nil || err2 != nil {
-				missing = true
-			}
+			certsAvailable = checkCertAndKeyAvailable(certPath, keyPath)
 		}
 
 		// If we found some certs, add them to the URL authentication
 		// method.
-		if !missing {
+		if certsAvailable {
 			pwEnabled, hasPw, pwd := u.GetAuthnPassword()
 			if !pwEnabled {
 				u.WithAuthn(pgurl.AuthnClientCert(certPath, keyPath))
@@ -130,3 +119,10 @@ func (ctx *SecurityContext) PGURL(user *url.Userinfo) (*pgurl.URL, error) {
 	}
 	return u, nil
 }
+
+func checkCertAndKeyAvailable(certPath string, keyPath string) bool {
+	loader := security.GetAssetLoader()
+	_, err1 := loader.Stat(certPath)
+	_, err2 := loader.Stat(keyPath)
+	return err1 == nil && err2 == nil
+}

pkg/security/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/clusterversion",
+        "//pkg/roachpb",
         "//pkg/server/telemetry",
         "//pkg/settings",
         "//pkg/settings/cluster",

pkg/security/auth.go

@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/errors"
 )
@@ -26,6 +27,18 @@ var certPrincipalMap struct {
 	m map[string]string
 }
 
+// CertificateUserScope indicates the scope of a user certificate i.e.
+// which tenants the user is allowed to authenticate on. A user can be
+// allowed to authenticate on multiple tenants. A certificate scope with
+// system tenant is treated as a global certificate which can authenticate
+// on any tenant.
+// TODO(rima): Should we remove this global authorization privilege for users
+// with system tenant scope?
+type CertificateUserScope struct {
+	Username  string
+	TenantIDs map[roachpb.TenantID]struct{}
+}
+
 // UserAuthHook authenticates a user based on their username and whether their
 // connection originates from a client or another node in the cluster. It
 // returns an optional func that is run at connection close.
@@ -81,8 +94,29 @@ func getCertificatePrincipals(cert *x509.Certificate) []string {
 	return results
 }
 
-// GetCertificateUsers extract the users from a client certificate.
-func GetCertificateUsers(tlsState *tls.ConnectionState) ([]string, error) {
+// getCertificateTenantScopes retrieves the tenant scopes from a certificate.
+func getCertificateTenantScopes(cert *x509.Certificate) (map[roachpb.TenantID]struct{}, error) {
+	tenantMap := make(map[roachpb.TenantID]struct{})
+	if len(cert.URIs) == 0 {
+		// No URI SAN, likely an old certificate. Populate tenant scope with default system tenant ID
+		// to generate a global client certificate
+		tenantMap[roachpb.SystemTenantID] = struct{}{}
+		return tenantMap, nil
+	}
+	for _, uri := range cert.URIs {
+		tenantID, err := ParseTenantURISAN(uri.String())
+		if err != nil {
+			return nil, err
+		}
+		tenantMap[tenantID] = struct{}{}
+	}
+	return tenantMap, nil
+}
+
+// GetCertificateUserScope extracts the users and tenant scopes from a client certificate.
+// We return it as a map from the username to the user scope to make certificate validation
+// based on username more efficient.
+func GetCertificateUserScope(tlsState *tls.ConnectionState) (userScopes map[string]CertificateUserScope, _ error) {
 	if tlsState == nil {
 		return nil, errors.Errorf("request is not using TLS")
 	}
@@ -93,7 +127,20 @@ func GetCertificateUsers(tlsState *tls.ConnectionState) ([]string, error) {
 	// any following certificates as intermediates. See:
 	// https://github.com/golang/go/blob/go1.8.1/src/crypto/tls/handshake_server.go#L723:L742
 	peerCert := tlsState.PeerCertificates[0]
-	return getCertificatePrincipals(peerCert), nil
+	users := getCertificatePrincipals(peerCert)
+	tenants, err := getCertificateTenantScopes(peerCert)
+	if err != nil {
+		return nil, err
+	}
+	userScopes = make(map[string]CertificateUserScope)
+	for _, user := range users {
+		scope := CertificateUserScope{
+			Username:  user,
+			TenantIDs: tenants,
+		}
+		userScopes[user] = scope
+	}
+	return userScopes, nil
 }
 
 // Contains returns true if the specified string is present in the given slice.
@@ -108,12 +155,13 @@ func Contains(sl []string, s string) bool {
 
 // UserAuthCertHook builds an authentication hook based on the security
 // mode and client certificate.
-func UserAuthCertHook(insecureMode bool, tlsState *tls.ConnectionState) (UserAuthHook, error) {
-	var certUsers []string
-
+func UserAuthCertHook(
+	insecureMode bool, tlsState *tls.ConnectionState, tenantID roachpb.TenantID,
+) (UserAuthHook, error) {
+	var certUserScope map[string]CertificateUserScope
 	if !insecureMode {
 		var err error
-		certUsers, err = GetCertificateUsers(tlsState)
+		certUserScope, err = GetCertificateUserScope(tlsState)
 		if err != nil {
 			return nil, err
 		}
@@ -141,12 +189,18 @@ func UserAuthCertHook(insecureMode bool, tlsState *tls.ConnectionState) (UserAut
 			return errors.Errorf("using tenant client certificate as user certificate is not allowed")
 		}
 
-		// The client certificate user must match the requested user.
-		if !Contains(certUsers, systemIdentity.Normalized()) {
-			return errors.Errorf("requested user is %s, but certificate is for %s", systemIdentity, certUsers)
+		userScope, ok := certUserScope[systemIdentity.Normalized()]
+		if !ok {
+			return errors.Errorf("requested user is %s is not authorized through this cert", systemIdentity)
 		}
-
-		return nil
+		// Verify either the tenantID or system tenant ID is contained in the userScope.
+		_, containsTenant := userScope.TenantIDs[tenantID]
+		_, containsSystem := userScope.TenantIDs[roachpb.SystemTenantID]
+		if containsTenant || containsSystem {
+			// Both user and tenant scope are satisfied by this certificate, allow authentication to succeed.
+			return nil
+		}
+		return errors.Errorf("request user %s is not authorized for tenant %d", systemIdentity, tenantID)
 	}, nil
 }