diff --git a/pkg/ccl/gssapiccl/BUILD.bazel b/pkg/ccl/gssapiccl/BUILD.bazel
index 7bad23863358..0c07ca8a687c 100644
--- a/pkg/ccl/gssapiccl/BUILD.bazel
+++ b/pkg/ccl/gssapiccl/BUILD.bazel
@@ -35,6 +35,7 @@ go_library(
             "//pkg/sql/pgwire/identmap",
             "//pkg/sql/sem/tree",
             "@com_github_cockroachdb_errors//:errors",
+            "@com_github_go_ldap_ldap_v3//:ldap",
         ],
         "//conditions:default": [],
     }),
diff --git a/pkg/ccl/gssapiccl/gssapi.go b/pkg/ccl/gssapiccl/gssapi.go
index bcee2625e39f..5d5c9f9524d4 100644
--- a/pkg/ccl/gssapiccl/gssapi.go
+++ b/pkg/ccl/gssapiccl/gssapi.go
@@ -28,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/hba"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/identmap"
 	"github.com/cockroachdb/errors"
+	"github.com/go-ldap/ldap/v3"
 )
 
 const (
@@ -82,7 +83,7 @@ func authGSS(
 	}
 
 	behaviors.SetAuthenticator(func(
-		_ context.Context, _ username.SQLUsername, _ bool, _ pgwire.PasswordRetrievalFn,
+		_ context.Context, _ username.SQLUsername, _ bool, _ pgwire.PasswordRetrievalFn, _ *ldap.DN,
 	) error {
 		// Enforce krb_realm option, if any.
 		if realms := entry.GetOptions("krb_realm"); len(realms) > 0 {
diff --git a/pkg/ccl/logictestccl/testdata/logic_test/subject b/pkg/ccl/logictestccl/testdata/logic_test/subject
index d8355e208e91..6d3f408dc203 100644
--- a/pkg/ccl/logictestccl/testdata/logic_test/subject
+++ b/pkg/ccl/logictestccl/testdata/logic_test/subject
@@ -8,14 +8,14 @@ statement error failed to parse distinguished name foo: DN ended with incomplete
 CREATE ROLE role_with_subject SUBJECT 'foo'
 
 statement ok
-CREATE ROLE role_with_subject SUBJECT 'foo=bar'
+CREATE ROLE role_with_subject SUBJECT 'CN=bar'
 
 query T
 SELECT value FROM system.role_options
 WHERE username = 'role_with_subject'
 AND option = 'SUBJECT'
 ----
-foo=bar
+CN=bar
 
 statement ok
 ALTER ROLE role_with_subject SUBJECT 'O=US, CN=role_with_subject'
@@ -60,8 +60,11 @@ AND option = 'SUBJECT'
 ----
 O=US\,A, CN=role_with_subject
 
-statement error SUBJECT CN must match "role_with_subject" but got "different_role"
+statement ok
 ALTER ROLE role_with_subject SUBJECT 'O=US\,A, CN=different_role'
 
-statement error SUBJECT must have only one CN attribute
+statement ok
 ALTER ROLE role_with_subject SUBJECT 'CN=role_with_subject, O=US\,A, CN=different_role'
+
+statement error SUBJECT contains illegal field type "OrgUnit", should be one of \["CN" "L" "ST" "O" "OU" "C" "STREET" "DC" "UID"\]
+ALTER ROLE role_with_subject SUBJECT 'CN=role_with_subject, O=US\,A, CN=different_role, OrgUnit=Marketing';
diff --git a/pkg/security/BUILD.bazel b/pkg/security/BUILD.bazel
index 374e98e6252c..1922fe447665 100644
--- a/pkg/security/BUILD.bazel
+++ b/pkg/security/BUILD.bazel
@@ -25,6 +25,7 @@ go_library(
     deps = [
         "//pkg/roachpb",
         "//pkg/security/certnames",
+        "//pkg/security/distinguishedname",
         "//pkg/security/password",
         "//pkg/security/securityassets",
         "//pkg/security/username",
@@ -48,6 +49,7 @@ go_library(
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_errors//oserror",
+        "@com_github_go_ldap_ldap_v3//:ldap",
         "@org_golang_x_crypto//bcrypt",
         "@org_golang_x_crypto//ocsp",
         "@org_golang_x_sync//errgroup",
@@ -82,6 +84,7 @@ go_test(
         "//pkg/roachpb",
         "//pkg/rpc",
         "//pkg/security/certnames",
+        "//pkg/security/distinguishedname",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/security/username",
@@ -100,6 +103,7 @@ go_test(
         "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_go_ldap_ldap_v3//:ldap",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_exp//rand",
     ] + select({
diff --git a/pkg/security/auth.go b/pkg/security/auth.go
index 16e6f62e5a8d..d98090658b35 100644
--- a/pkg/security/auth.go
+++ b/pkg/security/auth.go
@@ -18,11 +18,13 @@ import (
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/security/distinguishedname"
 	"github.com/cockroachdb/cockroach/pkg/security/password"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/errors"
+	"github.com/go-ldap/ldap/v3"
 )
 
 var certPrincipalMap struct {
@@ -30,11 +32,14 @@ var certPrincipalMap struct {
 	m map[string]string
 }
 
-// CertificateUserScope indicates the scope of a user certificate i.e.
-// which tenant the user is allowed to authenticate on. Older client certificates
+// CertificateUserScope indicates the scope of a user certificate i.e. which
+// tenant the user is allowed to authenticate on. Older client certificates
 // without a tenant scope are treated as global certificates which can
-// authenticate on any tenant strictly for backward compatibility with the
-// older certificates.
+// authenticate on any tenant strictly for backward compatibility with the older
+// certificates. A certificate must specify the SQL user name in either the CN
+// or a DNS SAN entry, so one certificate has multiple candidate usernames. The
+// GetCertificateUserScope function expands a cert into a set of "scopes" with
+// each possible username (and tenant ID).
 type CertificateUserScope struct {
 	Username string
 	TenantID roachpb.TenantID
@@ -100,7 +105,10 @@ func getCertificatePrincipals(cert *x509.Certificate) []string {
 	return results
 }
 
-// GetCertificateUserScope extracts the certificate scopes from a client certificate.
+// GetCertificateUserScope extracts the certificate scopes from a client
+// certificate. It tries to get CRDB prefixed SAN URIs and extracts tenantID and
+// user information. If there is no such URI, then it gets principal transformed
+// CN and SAN DNSNames with global scope.
 func GetCertificateUserScope(
 	peerCert *x509.Certificate,
 ) (userScopes []CertificateUserScope, _ error) {
@@ -148,6 +156,7 @@ func UserAuthCertHook(
 	tlsState *tls.ConnectionState,
 	tenantID roachpb.TenantID,
 	certManager *CertificateManager,
+	roleSubject *ldap.DN,
 ) (UserAuthHook, error) {
 	var certUserScope []CertificateUserScope
 	if !insecureMode {
@@ -190,7 +199,15 @@ func UserAuthCertHook(
 			return errors.Errorf("using tenant client certificate as user certificate is not allowed")
 		}
 
-		if ValidateUserScope(certUserScope, systemIdentity.Normalized(), tenantID) {
+		var certSubject *ldap.DN
+		if roleSubject != nil {
+			var err error
+			if certSubject, err = distinguishedname.ParseDNFromCertificate(peerCert); err != nil {
+				return errors.Errorf("could not parse certificate subject DN")
+			}
+		}
+
+		if ValidateUserScope(certUserScope, systemIdentity.Normalized(), tenantID, roleSubject, certSubject) {
 			if certManager != nil {
 				certManager.MaybeUpsertClientExpiration(
 					ctx,
@@ -200,8 +217,13 @@ func UserAuthCertHook(
 			}
 			return nil
 		}
-		return errors.WithDetailf(errors.Errorf("certificate authentication failed for user %q", systemIdentity),
-			"The client certificate is valid for %s.", FormatUserScopes(certUserScope))
+		return errors.WithDetailf(
+			errors.Errorf(
+				"certificate authentication failed for user %q (DN: %s)",
+				systemIdentity,
+				roleSubject,
+			),
+			"The client certificate (DN: %s) is valid for %s.", certSubject, FormatUserScopes(certUserScope))
 	}, nil
 }
 
@@ -294,17 +316,28 @@ func (i *PasswordUserAuthError) FormatError(p errors.Printer) error {
 	return i.err
 }
 
-// ValidateUserScope returns true if the user is a valid user for the tenant based on the certificate
-// user scope. It also returns true if the certificate is a global certificate. A client certificate
-// is considered global only when it doesn't contain a tenant SAN which is only possible for older
-// client certificates created prior to introducing tenant based scoping for the client.
+// ValidateUserScope returns true if the user is a valid user for the tenant
+// based on the certificate user scope. It also returns true if the certificate
+// is a global certificate. A client certificate is considered global only when
+// it doesn't contain a tenant SAN which is only possible for older client
+// certificates created prior to introducing tenant based scoping for the
+// client. Additionally, if subject role option is set for a user, we check if
+// certificate parsed subject DN matches the set subject.
 func ValidateUserScope(
-	certUserScope []CertificateUserScope, user string, tenantID roachpb.TenantID,
+	certUserScope []CertificateUserScope,
+	user string,
+	tenantID roachpb.TenantID,
+	roleSubject *ldap.DN,
+	certSubject *ldap.DN,
 ) bool {
+	// if subject role option is set, it must match the certificate subject
+	if roleSubject != nil && !roleSubject.Equal(certSubject) {
+		return false
+	}
 	for _, scope := range certUserScope {
 		if scope.Username == user {
-			// If username matches, allow authentication to succeed if the tenantID is a match
-			// or if the certificate scope is global.
+			// If username matches, allow authentication to succeed if
+			// the tenantID is a match or if the certificate scope is global.
 			if scope.TenantID == tenantID || scope.Global {
 				return true
 			}
diff --git a/pkg/security/auth_test.go b/pkg/security/auth_test.go
index a47ea1d9680f..ff6c50b67447 100644
--- a/pkg/security/auth_test.go
+++ b/pkg/security/auth_test.go
@@ -15,6 +15,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
+	"encoding/asn1"
 	"fmt"
 	"net/url"
 	"strings"
@@ -22,60 +23,76 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
+	"github.com/cockroachdb/cockroach/pkg/security/distinguishedname"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/go-ldap/ldap/v3"
 	"github.com/stretchr/testify/require"
 )
 
 // Construct a fake tls.ConnectionState object. The spec is a semicolon
 // separated list of peer certificate specifications. Each peer certificate
-// specification can have an optional OU in parenthesis followed by
-// a comma separated list of names where the first name is the
-// CommonName and the remaining names are SubjectAlternateNames.
-// The SubjectAlternateNames can go under DNSNames or URIs. To distinguish
-// the two, prefix the SAN with the type dns: or uri:. For example,
-// "foo" creates a single peer certificate with the CommonName "foo". The spec
-// "foo,dns:bar,dns:blah" creates a single peer certificate with the CommonName "foo" and a
-// DNSNames "bar" and "blah". "(Tenants)foo,dns:bar" creates a single
-// tenant client certificate with OU=Tenants, CN=foo and DNSName=bar.
-// A spec with "foo,dns:bar,uri:crdb://tenant/123" creates a single peer certificate
-// with CommonName foo, DNSName bar and URI set to crdb://tenant/123.
-// Contrast that with "foo;bar" which creates two peer certificates with the
-// CommonNames "foo" and "bar" respectively.
+// specification has a subject DN in parenthesis followed by a comma separated
+// list of SubjectAlternateNames. The SubjectAlternateNames can go under
+// DNSNames or URIs. To distinguish the two, prefix the SAN with the type dns:
+// or uri:. For example, "(CN=foo)" creates a single peer certificate with the
+// CommonName "foo". The spec "(CN=foo)dns:bar,dns:blah" creates a single peer
+// certificate with the CommonName "foo" and a DNSNames "bar" and "blah".
+// "(OU=Tenants,CN=foo)dns:bar" creates a single tenant client certificate with
+// OU=Tenants, CN=foo and DNSName=bar. A spec with
+// "(CN=foo)dns:bar,uri:crdb://tenant/123" creates a single peer certificate
+// with CommonName foo, DNSName bar and URI set to crdb://tenant/123. Contrast
+// that with "(CN=foo);(CN=bar)" which creates two peer certificates with the
+// CommonNames "foo" and "bar" respectively. To create a certificate with full
+// DN subject, required spec would be "(O=Cockroach,OU=Order Processing
+// Team,CN=foo),dns:bar" which creates a certificate with O=Cockroach,OU=Order
+// Processing Team,CN=foo, DNSName=bar.
 func makeFakeTLSState(t *testing.T, spec string) *tls.ConnectionState {
 	tls := &tls.ConnectionState{}
 	uriPrefix := "uri:"
 	dnsPrefix := "dns:"
 	if spec != "" {
 		for _, peerSpec := range strings.Split(spec, ";") {
-			var ou []string
+			subjectDN := [][]string{}
+			specSAN := ""
 			if strings.HasPrefix(peerSpec, "(") {
-				ouAndRest := strings.Split(peerSpec[1:], ")")
-				ou = ouAndRest[:1]
-				peerSpec = ouAndRest[1]
-			}
-			names := strings.Split(peerSpec, ",")
-			if len(names) == 0 {
-				continue
+				subjectDNAndRest := strings.Split(peerSpec[1:], ")")
+				fieldsSubjectDN := strings.Split(subjectDNAndRest[0], ",")
+				for _, field := range fieldsSubjectDN {
+					fieldKeyAndValue := strings.Split(field, "=")
+					subjectDN = append(subjectDN, []string{fieldKeyAndValue[0], fieldKeyAndValue[1]})
+				}
+				if len(subjectDNAndRest) == 2 {
+					specSAN = subjectDNAndRest[1]
+				}
 			}
 			peerCert := &x509.Certificate{}
-			peerCert.Subject = pkix.Name{
-				CommonName:         names[0],
-				OrganizationalUnit: ou,
+			RDNSeq, err := generateRDNSequenceFromSpecMap(subjectDN)
+			if err != nil {
+				t.Fatalf("unable to generate RDN Sequence from subjectDN spec, err: %v", err)
 			}
-			for i := 1; i < len(names); i++ {
-				if strings.HasPrefix(names[i], dnsPrefix) {
-					peerCert.DNSNames = append(peerCert.DNSNames, strings.TrimPrefix(names[i], dnsPrefix))
-				} else if strings.HasPrefix(names[i], uriPrefix) {
-					rawURI := strings.TrimPrefix(names[i], uriPrefix)
-					url, err := url.Parse(rawURI)
-					if err != nil {
-						t.Fatalf("unable to create tls spec due to invalid URI %s", rawURI)
+			peerCert.Subject.FillFromRDNSequence(&RDNSeq)
+			peerCert.RawSubject, err = asn1.Marshal(RDNSeq)
+			if err != nil {
+				t.Fatalf("unable to marshal subject, err: %v", err)
+			}
+
+			if len(specSAN) != 0 {
+				listSANs := strings.Split(specSAN, ",")
+				for i := 0; i < len(listSANs); i++ {
+					if strings.HasPrefix(listSANs[i], dnsPrefix) {
+						peerCert.DNSNames = append(peerCert.DNSNames, strings.TrimPrefix(listSANs[i], dnsPrefix))
+					} else if strings.HasPrefix(listSANs[i], uriPrefix) {
+						rawURI := strings.TrimPrefix(listSANs[i], uriPrefix)
+						uri, err := url.Parse(rawURI)
+						if err != nil {
+							t.Fatalf("unable to create tls spec due to invalid URI %s", rawURI)
+						}
+						peerCert.URIs = append(peerCert.URIs, uri)
+					} else {
+						t.Fatalf("subject altername names are expected to have uri: or dns: prefix")
 					}
-					peerCert.URIs = append(peerCert.URIs, url)
-				} else {
-					t.Fatalf("subject altername names are expected to have uri: or dns: prefix")
 				}
 			}
 			tls.PeerCertificates = append(tls.PeerCertificates, peerCert)
@@ -84,10 +101,64 @@ func makeFakeTLSState(t *testing.T, spec string) *tls.ConnectionState {
 	return tls
 }
 
+// generateRDNSequenceFromSpecMap takes a list subject DN fields and
+// corresponding values. It generates pkix.RDNSequence for these fields. The
+// returned  sequence could then used to generate cert.Subject and
+// cert.RawSubject for creating a mock crypto/x509 certificate object.
+func generateRDNSequenceFromSpecMap(
+	subjectSpecMap [][]string,
+) (RDNSeq pkix.RDNSequence, err error) {
+	var (
+		oidCountry            = []int{2, 5, 4, 6}
+		oidOrganization       = []int{2, 5, 4, 10}
+		oidOrganizationalUnit = []int{2, 5, 4, 11}
+		oidCommonName         = []int{2, 5, 4, 3}
+		oidLocality           = []int{2, 5, 4, 7}
+		oidProvince           = []int{2, 5, 4, 8}
+		oidStreetAddress      = []int{2, 5, 4, 9}
+		oidUID                = []int{0, 9, 2342, 19200300, 100, 1, 1}
+		oidDC                 = []int{0, 9, 2342, 19200300, 100, 1, 25}
+	)
+
+	for _, fieldAndValue := range subjectSpecMap {
+		field := fieldAndValue[0]
+		fieldValue := fieldAndValue[1]
+		var attrTypeAndValue pkix.AttributeTypeAndValue
+		switch field {
+		case "CN":
+			attrTypeAndValue.Type = oidCommonName
+		case "L":
+			attrTypeAndValue.Type = oidLocality
+		case "ST":
+			attrTypeAndValue.Type = oidProvince
+		case "O":
+			attrTypeAndValue.Type = oidOrganization
+		case "OU":
+			attrTypeAndValue.Type = oidOrganizationalUnit
+		case "C":
+			attrTypeAndValue.Type = oidCountry
+		case "STREET":
+			attrTypeAndValue.Type = oidStreetAddress
+		case "DC":
+			attrTypeAndValue.Type = oidDC
+		case "UID":
+			attrTypeAndValue.Type = oidUID
+		default:
+			return nil, fmt.Errorf("found unknown field value %q in spec map", field)
+		}
+		attrTypeAndValue.Value = fieldValue
+		RDNSeq = append(RDNSeq, pkix.RelativeDistinguishedNameSET{
+			attrTypeAndValue,
+		})
+	}
+
+	return RDNSeq, nil
+}
+
 func TestGetCertificateUserScope(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	t.Run("good request: single certificate", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo")
+		state := makeFakeTLSState(t, "(CN=foo)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -99,7 +170,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("request with multiple certs, but only one chain (eg: origin certs are client and CA)", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo;CA")
+		state := makeFakeTLSState(t, "(CN=foo);(CN=CA)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -111,7 +182,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("always use the first certificate", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo;bar")
+		state := makeFakeTLSState(t, "(CN=foo);(CN=bar)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -123,7 +194,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("extract all of the principals from the first certificate", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo,dns:bar,dns:blah;CA")
+		state := makeFakeTLSState(t, "(CN=foo)dns:bar,dns:blah;(CN=CA)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -134,7 +205,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("extracts username, tenantID from tenant URI SAN", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo,uri:crdb://tenant/123/user/foo;CA")
+		state := makeFakeTLSState(t, "(CN=foo)uri:crdb://tenant/123/user/foo;(CN=CA)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -147,7 +218,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("extracts tenant URI SAN even when multiple URIs, where one URI is not of CRBD format", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo,uri:mycompany:sv:rootclient:dev:usw1,uri:crdb://tenant/123/user/foo;CA")
+		state := makeFakeTLSState(t, "(CN=foo)uri:mycompany:sv:rootclient:dev:usw1,uri:crdb://tenant/123/user/foo;(CN=CA)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -160,7 +231,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("errors when tenant URI SAN is not of expected format, even if other URI SAN is provided", func(t *testing.T) {
-		state := makeFakeTLSState(t, "foo,uri:mycompany:sv:rootclient:dev:usw1,uri:crdb://tenant/bad/format/123;CA")
+		state := makeFakeTLSState(t, "(CN=foo)uri:mycompany:sv:rootclient:dev:usw1,uri:crdb://tenant/bad/format/123;(CN=CA)")
 		cert := state.PeerCertificates[0]
 		userScopes, err := security.GetCertificateUserScope(cert)
 		require.Nil(t, userScopes)
@@ -168,7 +239,7 @@ func TestGetCertificateUserScope(t *testing.T) {
 	})
 
 	t.Run("falls back to global client cert when crdb URI SAN scheme is not followed", func(t *testing.T) {
-		state := makeFakeTLSState(t, "sanuri,uri:mycompany:sv:rootclient:dev:usw1;CA")
+		state := makeFakeTLSState(t, "(CN=sanuri)uri:mycompany:sv:rootclient:dev:usw1;(CN=CA)")
 		cert := state.PeerCertificates[0]
 		if userScopes, err := security.GetCertificateUserScope(cert); err != nil {
 			t.Error(err)
@@ -213,23 +284,23 @@ func TestGetCertificateUsersMapped(t *testing.T) {
 		expected string
 	}{
 		// No mapping present.
-		{"foo", "", "foo"},
+		{"(CN=foo)", "", "foo"},
 		// The basic mapping case.
-		{"foo", "foo:bar", "bar"},
+		{"(CN=foo)", "foo:bar", "bar"},
 		// Identity mapping.
-		{"foo", "foo:foo", "foo"},
+		{"(CN=foo)", "foo:foo", "foo"},
 		// Mapping does not apply to cert principals.
-		{"foo", "bar:bar", "foo"},
+		{"(CN=foo)", "bar:bar", "foo"},
 		// The last mapping for a principal takes precedence.
-		{"foo", "foo:bar,foo:blah", "blah"},
+		{"(CN=foo)", "foo:bar,foo:blah", "blah"},
 		// First principal mapped, second principal unmapped.
-		{"foo,dns:bar", "foo:blah", "blah,bar"},
+		{"(CN=foo)dns:bar", "foo:blah", "blah,bar"},
 		// First principal unmapped, second principal mapped.
-		{"bar,dns:foo", "foo:blah", "bar,blah"},
+		{"(CN=bar)dns:foo", "foo:blah", "bar,blah"},
 		// Both principals mapped.
-		{"foo,dns:bar", "foo:bar,bar:foo", "bar,foo"},
+		{"(CN=foo)dns:bar", "foo:bar,bar:foo", "bar,foo"},
 		// Verify desired string splits.
-		{"foo:has:colon", "foo:has:colon:bar", "bar"},
+		{"(CN=foo:has:colon)", "foo:has:colon:bar", "bar"},
 	}
 	for _, c := range testCases {
 		t.Run("", func(t *testing.T) {
@@ -263,54 +334,84 @@ func TestAuthenticationHook(t *testing.T) {
 	fooUser := username.MakeSQLUsernameFromPreNormalizedString("foo")
 	barUser := username.MakeSQLUsernameFromPreNormalizedString("bar")
 	blahUser := username.MakeSQLUsernameFromPreNormalizedString("blah")
+	subjectDNString := "O=Cockroach,OU=Order Processing Team,UID=b8b40653-7f74-4f14-8a61-59f7f3b18184,CN=foo"
+	fieldMismatchSubjectDNString := "O=Cockroach,OU=Marketing Team,UID=b8b40653-7f74-4f14-8a61-59f7f3b18184,CN=foo"
+	subsetSubjectDNString := "O=Cockroach,OU=Order Processing Team,CN=foo"
+	fieldMismatchOnlyOnCommonNameString := "O=Cockroach,OU=Order Processing Team,UID=b8b40653-7f74-4f14-8a61-59f7f3b18184,CN=bar"
 
 	testCases := []struct {
-		insecure           bool
-		tlsSpec            string
-		username           username.SQLUsername
-		principalMap       string
-		buildHookSuccess   bool
-		publicHookSuccess  bool
-		privateHookSuccess bool
-		tenantID           roachpb.TenantID
-		expectedErr        string
+		insecure               bool
+		tlsSpec                string
+		username               username.SQLUsername
+		principalMap           string
+		buildHookSuccess       bool
+		publicHookSuccess      bool
+		privateHookSuccess     bool
+		tenantID               roachpb.TenantID
+		isSubjectRoleOptionSet bool
+		expectedErr            string
 	}{
 		// Insecure mode, empty username.
-		{true, "", username.SQLUsername{}, "", true, false, false, roachpb.SystemTenantID, `user is missing`},
+		{true, "", username.SQLUsername{}, "", true, false, false, roachpb.SystemTenantID, false, `user is missing`},
 		// Insecure mode, non-empty username.
-		{true, "", fooUser, "", true, true, false, roachpb.SystemTenantID, `user "foo" is not allowed`},
+		{true, "", fooUser, "", true, true, false, roachpb.SystemTenantID, false, `user "foo" is not allowed`},
 		// Secure mode, no TLS state.
-		{false, "", username.SQLUsername{}, "", false, false, false, roachpb.SystemTenantID, `no client certificates in request`},
+		{false, "", username.SQLUsername{}, "", false, false, false, roachpb.SystemTenantID, false, `no client certificates in request`},
 		// Secure mode, bad user.
-		{false, "foo", username.NodeUserName(), "", true, false, false, roachpb.SystemTenantID,
-			`certificate authentication failed for user "node"`},
+		{false, "(CN=foo)", username.NodeUserName(), "", true, false, false, roachpb.SystemTenantID,
+			false, `certificate authentication failed for user "node"`},
 		// Secure mode, node user.
-		{false, username.NodeUser, username.NodeUserName(), "", true, true, true, roachpb.SystemTenantID, ``},
+		{false, "(CN=node)", username.NodeUserName(), "", true, true, true, roachpb.SystemTenantID, false, ``},
 		// Secure mode, node cert and unrelated user.
-		{false, username.NodeUser, fooUser, "", true, false, false, roachpb.SystemTenantID,
-			`certificate authentication failed for user "foo"`},
+		{false, "(CN=node)", fooUser, "", true, false, false, roachpb.SystemTenantID,
+			false, `certificate authentication failed for user "foo"`},
 		// Secure mode, root user.
-		{false, username.RootUser, username.NodeUserName(), "", true, false, false, roachpb.SystemTenantID,
-			`certificate authentication failed for user "node"`},
+		{false, "(CN=root)", username.NodeUserName(), "", true, false, false, roachpb.SystemTenantID,
+			false, `certificate authentication failed for user "node"`},
 		// Secure mode, tenant cert, foo user.
-		{false, "(Tenants)foo", fooUser, "", true, false, false, roachpb.SystemTenantID,
-			`using tenant client certificate as user certificate is not allowed`},
+		{false, "(OU=Tenants,CN=foo)", fooUser, "", true, false, false, roachpb.SystemTenantID,
+			false, `using tenant client certificate as user certificate is not allowed`},
 		// Secure mode, multiple cert principals.
-		{false, "foo,dns:bar", fooUser, "", true, true, false, roachpb.SystemTenantID, `user "foo" is not allowed`},
-		{false, "foo,dns:bar", barUser, "", true, true, false, roachpb.SystemTenantID, `user "bar" is not allowed`},
+		{false, "(CN=foo)dns:bar", fooUser, "", true, true, false, roachpb.SystemTenantID, false, `user "foo" is not allowed`},
+		{false, "(CN=foo)dns:bar", barUser, "", true, true, false, roachpb.SystemTenantID, false, `user "bar" is not allowed`},
 		// Secure mode, principal map.
-		{false, "foo,dns:bar", blahUser, "foo:blah", true, true, false, roachpb.SystemTenantID, `user "blah" is not allowed`},
-		{false, "foo,dns:bar", blahUser, "bar:blah", true, true, false, roachpb.SystemTenantID, `user "blah" is not allowed`},
-		{false, "foo,uri:crdb://tenant/123/user/foo", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
-			`user "foo" is not allowed`},
-		{false, "foo,uri:crdb://tenant/123/user/foo", fooUser, "", true, false, false, roachpb.SystemTenantID,
-			`certificate authentication failed for user "foo"`},
-		{false, "foo", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
-			`user "foo" is not allowed`},
-		{false, "foo,uri:crdb://tenant/1/user/foo", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
-			`certificate authentication failed for user "foo"`},
-		{false, "foo,uri:crdb://tenant/123/user/foo", blahUser, "", true, false, false, roachpb.MustMakeTenantID(123),
-			`certificate authentication failed for user "blah"`},
+		{false, "(CN=foo)dns:bar", blahUser, "foo:blah", true, true, false, roachpb.SystemTenantID, false, `user "blah" is not allowed`},
+		{false, "(CN=foo)dns:bar", blahUser, "bar:blah", true, true, false, roachpb.SystemTenantID, false, `user "blah" is not allowed`},
+		{false, "(CN=foo)uri:crdb://tenant/123/user/foo", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
+			false, `user "foo" is not allowed`},
+		{false, "(CN=foo)uri:crdb://tenant/123/user/foo", fooUser, "", true, false, false, roachpb.SystemTenantID,
+			false, `certificate authentication failed for user "foo"`},
+		{false, "(CN=foo)", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
+			false, `user "foo" is not allowed`},
+		{false, "(CN=foo)uri:crdb://tenant/1/user/foo", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			false, `certificate authentication failed for user "foo"`},
+		{false, "(CN=foo)uri:crdb://tenant/123/user/foo", blahUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			false, `certificate authentication failed for user "blah"`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option not set.
+		{false, "(" + subjectDNString + ")", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
+			false, `user "foo" is not allowed`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option set matching TLS cert subject.
+		{false, "(" + subjectDNString + ")", fooUser, "", true, true, false, roachpb.MustMakeTenantID(123),
+			true, `user "foo" is not allowed`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option set, TLS cert DN empty.
+		{false, "(CN=foo)", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			true, `certificate authentication failed for user "foo"`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option set, TLS cert DN mismatches on OU field.
+		{false, "(" + fieldMismatchSubjectDNString + ")", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			true, `certificate authentication failed for user "foo"`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option set, TLS cert DN subset of role subject DN.
+		{false, "(" + subsetSubjectDNString + ")", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			true, `certificate authentication failed for user "foo"`},
+		// Secure mode, client cert having full DN, foo user with subject role
+		// option set mismatching TLS cert subject only on CN(required for
+		// matching) having DNS as foo.
+		{false, "(" + fieldMismatchOnlyOnCommonNameString + ")dns:foo", fooUser, "", true, false, false, roachpb.MustMakeTenantID(123),
+			true, `certificate authentication failed for user "foo"`},
 	}
 
 	ctx := context.Background()
@@ -321,11 +422,18 @@ func TestAuthenticationHook(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
+
+			var roleSubject *ldap.DN
+			if tc.isSubjectRoleOptionSet {
+				roleSubject, _ = distinguishedname.ParseDN(subjectDNString)
+			}
+
 			hook, err := security.UserAuthCertHook(
 				tc.insecure,
 				makeFakeTLSState(t, tc.tlsSpec),
 				tc.tenantID,
 				nil, /* certManager */
+				roleSubject,
 			)
 			if (err == nil) != tc.buildHookSuccess {
 				t.Fatalf("expected success=%t, got err=%v", tc.buildHookSuccess, err)
diff --git a/pkg/security/distinguishedname/BUILD.bazel b/pkg/security/distinguishedname/BUILD.bazel
index 5b19065bad44..61fdf59464c1 100644
--- a/pkg/security/distinguishedname/BUILD.bazel
+++ b/pkg/security/distinguishedname/BUILD.bazel
@@ -6,7 +6,6 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/security/distinguishedname",
     visibility = ["//visibility:public"],
     deps = [
-        "//pkg/security/username",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_go_ldap_ldap_v3//:ldap",
     ],
diff --git a/pkg/security/distinguishedname/parse.go b/pkg/security/distinguishedname/parse.go
index f1750b734106..dc28f464c19f 100644
--- a/pkg/security/distinguishedname/parse.go
+++ b/pkg/security/distinguishedname/parse.go
@@ -11,36 +11,28 @@
 package distinguishedname
 
 import (
-	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/asn1"
+	"slices"
+	"strings"
+
 	"github.com/cockroachdb/errors"
 	"github.com/go-ldap/ldap/v3"
 )
 
 // ValidateDN validates a distinguished name string to verify that it is
-// well-formed and valid for the given user.
-func ValidateDN(u username.SQLUsername, dnStr string) error {
-	if u.IsRootUser() {
-		return errors.Newf("role %q cannot have a SUBJECT", u)
-	}
+// well-formed and only contains attribute types defined in RFC4514.
+func ValidateDN(dnStr string) error {
 	dn, err := ParseDN(dnStr)
 	if err != nil {
 		return err
 	}
-	sawCN := false
+	attributeTypeList := []string{"CN", "L", "ST", "O", "OU", "C", "STREET", "DC", "UID"}
 	for _, rdn := range dn.RDNs {
 		for _, attr := range rdn.Attributes {
-			if attr.Type == "CN" {
-				if sawCN {
-					return errors.Newf("SUBJECT must have only one CN attribute")
-				}
-				sawCN = true
-				normalizedCN, err := username.MakeSQLUsernameFromUserInput(attr.Value, username.PurposeValidation)
-				if err != nil {
-					return err
-				}
-				if normalizedCN != u {
-					return errors.Newf("SUBJECT CN must match %q but got %q", u, attr.Value)
-				}
+			if !slices.Contains(attributeTypeList, strings.ToUpper(attr.Type)) {
+				return errors.Newf("SUBJECT contains illegal field type %q, should be one of %+q", attr.Type, attributeTypeList)
 			}
 		}
 	}
@@ -56,3 +48,47 @@ func ParseDN(dnStr string) (*ldap.DN, error) {
 	}
 	return dn, nil
 }
+
+// ParseDNFromCertificate parses the distinguished name for the subject from
+// X.509 certificate provided. It retains the sequence of fields as provided in
+// the certificate subject and also parses all fields mentioned in RFC4514 which
+// ldap/v3 library currently supports.
+func ParseDNFromCertificate(cert *x509.Certificate) (*ldap.DN, error) {
+	var RDNSeq pkix.RDNSequence
+	_, err := asn1.Unmarshal(cert.RawSubject, &RDNSeq)
+	if err != nil {
+		return nil, err
+	}
+
+	// This is required because RDNSeq.String() reverses the order of fields.
+	// The x509 library possibly intended to use cert.Subject.ToRDNSequence and
+	// RDNSequence.String() in succession which is done in the library function
+	// cert.Subject.String(). But since x509 is incapable of handling all fields
+	// defined in RFC 4514, we need to directly parse cert.RawSubject here.
+	slices.Reverse(RDNSeq)
+	subjectDN, err := ParseDN(RDNSeq.String())
+	if err != nil {
+		return nil, err
+	}
+
+	const (
+		// Go only parses a subset of the possible fields in a DN (golang/go#25667).
+		// We add the remaining ones defined in section 3 of RFC 4514
+		// (https://datatracker.ietf.org/doc/html/rfc4514#section-3)
+		encodedUserID          = "0.9.2342.19200300.100.1.1"
+		encodedDomainComponent = "0.9.2342.19200300.100.1.25"
+	)
+
+	for _, dn := range subjectDN.RDNs {
+		for _, attr := range dn.Attributes {
+			switch attr.Type {
+			case encodedUserID:
+				attr.Type = "UID"
+			case encodedDomainComponent:
+				attr.Type = "DC"
+			}
+		}
+	}
+
+	return subjectDN, nil
+}
diff --git a/pkg/sql/pgwire/BUILD.bazel b/pkg/sql/pgwire/BUILD.bazel
index 305f6d4eaf7e..ac6c55877bb6 100644
--- a/pkg/sql/pgwire/BUILD.bazel
+++ b/pkg/sql/pgwire/BUILD.bazel
@@ -84,6 +84,7 @@ go_library(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_go_ldap_ldap_v3//:ldap",
         "@com_github_lib_pq//oid",
         "@com_github_xdg_go_scram//:scram",
         "@io_opentelemetry_go_otel//attribute",
diff --git a/pkg/sql/pgwire/auth.go b/pkg/sql/pgwire/auth.go
index ae637754e388..224cede13ce2 100644
--- a/pkg/sql/pgwire/auth.go
+++ b/pkg/sql/pgwire/auth.go
@@ -148,7 +148,7 @@ func (c *conn) handleAuthentication(
 
 	// Check that the requested user exists and retrieve the hashed
 	// password in case password authentication is needed.
-	exists, canLoginSQL, _, canUseReplicationMode, isSuperuser, defaultSettings, _, pwRetrievalFn, err :=
+	exists, canLoginSQL, _, canUseReplicationMode, isSuperuser, defaultSettings, roleSubject, pwRetrievalFn, err :=
 		sql.GetUserSessionInitInfo(
 			ctx,
 			execCfg,
@@ -175,10 +175,13 @@ func (c *conn) handleAuthentication(
 		return connClose, c.sendError(ctx, pgerror.Newf(pgcode.InvalidAuthorizationSpecification, "%s does not have login privilege", dbUser))
 	}
 
-	// At this point, we know that the requested user exists and is
-	// allowed to log in. Now we can delegate to the selected AuthMethod
-	// implementation to complete the authentication.
-	if err := behaviors.Authenticate(ctx, systemIdentity, true /* public */, pwRetrievalFn); err != nil {
+	// At this point, we know that the requested user exists and is allowed to log
+	// in. Now we can delegate to the selected AuthMethod implementation to
+	// complete the authentication.
+	// TODO(souravcrl): Verify whether to use systemIdentity or dbUser here since
+	// systemIdentity refers to external system name, which is same as dbUser name
+	// incase ReplacementIdentity is not set.
+	if err := behaviors.Authenticate(ctx, systemIdentity, true /* public */, pwRetrievalFn, roleSubject); err != nil {
 		ac.LogAuthFailed(ctx, eventpb.AuthFailReason_UNKNOWN, err)
 		if pErr := (*security.PasswordUserAuthError)(nil); errors.As(err, &pErr) {
 			err = pgerror.WithCandidateCode(err, pgcode.InvalidPassword)
diff --git a/pkg/sql/pgwire/auth_behaviors.go b/pkg/sql/pgwire/auth_behaviors.go
index 90ce1659f47d..c9435678ad5b 100644
--- a/pkg/sql/pgwire/auth_behaviors.go
+++ b/pkg/sql/pgwire/auth_behaviors.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/errors"
+	"github.com/go-ldap/ldap/v3"
 )
 
 // AuthBehaviors encapsulates the per-connection behaviors that may be
@@ -51,9 +52,10 @@ func (b *AuthBehaviors) Authenticate(
 	systemIdentity username.SQLUsername,
 	clientConnection bool,
 	pwRetrieveFn PasswordRetrievalFn,
+	roleSubject *ldap.DN,
 ) error {
 	if found := b.authenticator; found != nil {
-		return found(ctx, systemIdentity, clientConnection, pwRetrieveFn)
+		return found(ctx, systemIdentity, clientConnection, pwRetrieveFn, roleSubject)
 	}
 	return errors.New("no Authenticator provided to AuthBehaviors")
 }
diff --git a/pkg/sql/pgwire/auth_methods.go b/pkg/sql/pgwire/auth_methods.go
index 177fec44b1c9..939a266e3cb8 100644
--- a/pkg/sql/pgwire/auth_methods.go
+++ b/pkg/sql/pgwire/auth_methods.go
@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
+	"github.com/go-ldap/ldap/v3"
 	"github.com/xdg-go/scram"
 )
 
@@ -123,6 +124,7 @@ func authPassword(
 		systemIdentity username.SQLUsername,
 		clientConnection bool,
 		pwRetrieveFn PasswordRetrievalFn,
+		_ *ldap.DN,
 	) error {
 		return passwordAuthenticator(ctx, systemIdentity, clientConnection, pwRetrieveFn, c, execCfg)
 	})
@@ -246,6 +248,7 @@ func authScram(
 		systemIdentity username.SQLUsername,
 		clientConnection bool,
 		pwRetrieveFn PasswordRetrievalFn,
+		_ *ldap.DN,
 	) error {
 		return scramAuthenticator(ctx, systemIdentity, clientConnection, pwRetrieveFn, c, execCfg)
 	})
@@ -423,6 +426,7 @@ func authCert(
 		systemIdentity username.SQLUsername,
 		clientConnection bool,
 		pwRetrieveFn PasswordRetrievalFn,
+		roleSubject *ldap.DN,
 	) error {
 		if len(tlsState.PeerCertificates) == 0 {
 			return errors.New("no TLS peer certificates, but required for auth")
@@ -442,6 +446,7 @@ func authCert(
 			&tlsState,
 			execCfg.RPCContext.TenantID,
 			cm,
+			roleSubject,
 		)
 		if err != nil {
 			return err
@@ -520,6 +525,7 @@ func authAutoSelectPasswordProtocol(
 		systemIdentity username.SQLUsername,
 		clientConnection bool,
 		pwRetrieveFn PasswordRetrievalFn,
+		_ *ldap.DN,
 	) error {
 		// Request information about the password hash.
 		expired, hashedPassword, pwRetrieveErr := pwRetrieveFn(ctx)
@@ -615,7 +621,7 @@ func authTrust(
 ) (*AuthBehaviors, error) {
 	b := &AuthBehaviors{}
 	b.SetRoleMapper(UseProvidedIdentity)
-	b.SetAuthenticator(func(_ context.Context, _ username.SQLUsername, _ bool, _ PasswordRetrievalFn) error {
+	b.SetAuthenticator(func(_ context.Context, _ username.SQLUsername, _ bool, _ PasswordRetrievalFn, _ *ldap.DN) error {
 		return nil
 	})
 	return b, nil
@@ -633,7 +639,7 @@ func authReject(
 ) (*AuthBehaviors, error) {
 	b := &AuthBehaviors{}
 	b.SetRoleMapper(UseProvidedIdentity)
-	b.SetAuthenticator(func(ctx context.Context, _ username.SQLUsername, _ bool, _ PasswordRetrievalFn) error {
+	b.SetAuthenticator(func(ctx context.Context, _ username.SQLUsername, _ bool, _ PasswordRetrievalFn, _ *ldap.DN) error {
 		err := errors.New("authentication rejected by configuration")
 		c.LogAuthFailed(ctx, eventpb.AuthFailReason_LOGIN_DISABLED, err)
 		return err
@@ -664,7 +670,7 @@ func authSessionRevivalToken(token []byte) AuthMethod {
 	) (*AuthBehaviors, error) {
 		b := &AuthBehaviors{}
 		b.SetRoleMapper(UseProvidedIdentity)
-		b.SetAuthenticator(func(ctx context.Context, user username.SQLUsername, _ bool, _ PasswordRetrievalFn) error {
+		b.SetAuthenticator(func(ctx context.Context, user username.SQLUsername, _ bool, _ PasswordRetrievalFn, _ *ldap.DN) error {
 			c.LogAuthInfof(ctx, "session revival token detected; attempting to use it")
 			if !sql.AllowSessionRevival.Get(&execCfg.Settings.SV) || execCfg.Codec.ForSystemTenant() {
 				return errors.New("session revival tokens are not supported on this cluster")
@@ -736,7 +742,7 @@ func authJwtToken(
 	}
 	b := &AuthBehaviors{}
 	b.SetRoleMapper(UseProvidedIdentity)
-	b.SetAuthenticator(func(ctx context.Context, user username.SQLUsername, clientConnection bool, pwRetrieveFn PasswordRetrievalFn) error {
+	b.SetAuthenticator(func(ctx context.Context, user username.SQLUsername, clientConnection bool, pwRetrieveFn PasswordRetrievalFn, _ *ldap.DN) error {
 		c.LogAuthInfof(ctx, "JWT token detected; attempting to use it")
 		if !clientConnection {
 			err := errors.New("JWT token authentication is only available for client connections")
diff --git a/pkg/sql/pgwire/authenticator.go b/pkg/sql/pgwire/authenticator.go
index 870dc1c49c8d..631043846e64 100644
--- a/pkg/sql/pgwire/authenticator.go
+++ b/pkg/sql/pgwire/authenticator.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/security/password"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/go-ldap/ldap/v3"
 )
 
 // Authenticator is a component of an AuthMethod that determines if the
@@ -25,6 +26,7 @@ type Authenticator = func(
 	systemIdentity username.SQLUsername,
 	clientConnection bool,
 	pwRetrieveFn PasswordRetrievalFn,
+	roleSubject *ldap.DN,
 ) error
 
 // PasswordRetrievalFn defines a method to retrieve a hashed password
diff --git a/pkg/sql/roleoption/role_option.go b/pkg/sql/roleoption/role_option.go
index 251e635b3845..769367d3800d 100644
--- a/pkg/sql/roleoption/role_option.go
+++ b/pkg/sql/roleoption/role_option.go
@@ -220,7 +220,10 @@ func MakeListFromKVOptions(
 				if err := base.CheckEnterpriseEnabled(settings, "SUBJECT role option"); err != nil {
 					return err
 				}
-				if err := distinguishedname.ValidateDN(u, s); err != nil {
+				if u.IsRootUser() {
+					return pgerror.Newf(pgcode.InvalidParameterValue, "role %q cannot have a SUBJECT", u)
+				}
+				if err := distinguishedname.ValidateDN(s); err != nil {
 					return pgerror.WithCandidateCode(err, pgcode.InvalidParameterValue)
 				}
 				return nil
diff --git a/pkg/sql/user.go b/pkg/sql/user.go
index 73c3ce989257..70004f5fa937 100644
--- a/pkg/sql/user.go
+++ b/pkg/sql/user.go
@@ -320,7 +320,7 @@ func retrieveAuthInfo(
 
 	// Use fully qualified table name to avoid looking up "".system.role_options.
 	const getLoginDependencies = `SELECT option, value FROM system.public.role_options ` +
-		`WHERE username=$1 AND option IN ('NOLOGIN', 'VALID UNTIL', 'NOSQLLOGIN', 'REPLICATION')`
+		`WHERE username=$1 AND option IN ('NOLOGIN', 'VALID UNTIL', 'NOSQLLOGIN', 'REPLICATION', 'SUBJECT')`
 
 	roleOptsIt, err := ie.QueryIteratorEx(
 		ctx, "get-login-dependencies", nil, /* txn */