Allow backends to extract credentials from payloads and trigger an authentication workflow (#23924)

* wip

* Work on the tuneable allowance and some bugs

* Call handleCancellableRequest instead, which gets the audit order more correct and includes the preauth response

* Get rid of no longer needed operation

* Phew, this wasn't necessary

* Add auth error handling by the backend, and fix a bug with handleInvalidCredentials

* Cleanup req/resp naming

* Use the new form, and data

* Discovered that tokens werent really being checked because isLoginRequest returns true for the re-request into the backend, when it shouldnt

* Add a few more checks in the delegated request handler for bad inputs

 - Protect the delegated handler from bad inputs from the backend such
   as an empty accessor, a path that isn't registered as a login request
 - Add similar protections for bad auth results as we do in the normal
   login request paths. Technically not 100% needed but if somehow the
   handleCancelableRequest doesn't use the handleLoginRequest code path
   we could get into trouble in the future
 - Add delegated-auth-accessors flag to the secrets tune command and
   api-docs

* Unit tests and some small fixes

* Remove transit preauth test, rely on unit tests

* Cleanup and add a little more commentary in tests

* Fix typos, add another failure use-case which we reference a disabled auth mount

* PR Feedback

 - Use router to lookup mount instead of defining a new lookup method
 - Enforce auth table types and namespace when mount is found
 - Define a type alias for the handleInvalidCreds
 - Fix typos/grammar
 - Clean up globals in test

* Additional PR feedback

 - Add test for delegated auth handler
 - Force batch token usage
 - Add a test to validate failures if a non-batch token is used
 - Check for Data member being nil in test cases

* Update failure error message around requiring batch tokens

* Trap MFA requests

* Reword some error messages

* Add test and fixes for delegated response wrapping

* Move MFA test to dedicated mount

 - If the delegated auth tests were running in parallel, the MFA test
   case might influence the other tests, so move the MFA to a dedicated
   mount

* PR feedback: use textproto.CanonicalMIMEHeaderKey

 - Change the X-Vault-Wrap-Ttl constant to X-Vault-Wrap-TTL
   and use textproto.CanonicalMIMEHeaderKey to format it
   within the delete call.
 - This protects the code around changes of the constant typing

* PR feedback

 - Append Error to RequestDelegatedAuth
 - Force error interface impl through explicit nil var assignment on
   RequestDelegatedAuthError
 - Clean up test factory and leverage NewTestSoloCluster
 - Leverage newer maps.Clone as this is 1.16 only

---------

Co-authored-by: Scott G. Miller <smiller@hashicorp.com>

api/sys_mounts.go

@@ -271,6 +271,7 @@ type MountConfigInput struct {
 	AllowedManagedKeys        []string                `json:"allowed_managed_keys,omitempty" mapstructure:"allowed_managed_keys"`
 	PluginVersion             string                  `json:"plugin_version,omitempty"`
 	UserLockoutConfig         *UserLockoutConfigInput `json:"user_lockout_config,omitempty"`
+	DelegatedAuthAccessors    []string                `json:"delegated_auth_accessors,omitempty" mapstructure:"delegated_auth_accessors"`
 	// Deprecated: This field will always be blank for newer server responses.
 	PluginName string `json:"plugin_name,omitempty" mapstructure:"plugin_name"`
 }
@@ -303,6 +304,8 @@ type MountConfigOutput struct {
 	TokenType                 string                   `json:"token_type,omitempty" mapstructure:"token_type"`
 	AllowedManagedKeys        []string                 `json:"allowed_managed_keys,omitempty" mapstructure:"allowed_managed_keys"`
 	UserLockoutConfig         *UserLockoutConfigOutput `json:"user_lockout_config,omitempty"`
+	DelegatedAuthAccessors    []string                 `json:"delegated_auth_accessors,omitempty" mapstructure:"delegated_auth_accessors"`
+
 	// Deprecated: This field will always be blank for newer server responses.
 	PluginName string `json:"plugin_name,omitempty" mapstructure:"plugin_name"`
 }

command/commands.go

@@ -161,6 +161,9 @@ const (
 	// flagNameLogLevel is used to specify the log level applied to logging
 	// Supported log levels: Trace, Debug, Error, Warn, Info
 	flagNameLogLevel = "log-level"
+	// flagNameDelegatedAuthAccessors allows operators to specify the allowed mount accessors a backend can delegate
+	// authentication
+	flagNameDelegatedAuthAccessors = "delegated-auth-accessors"
 )
 
 var (

command/secrets_tune.go

@@ -35,6 +35,7 @@ type SecretsTuneCommand struct {
 	flagVersion                   int
 	flagPluginVersion             string
 	flagAllowedManagedKeys        []string
+	flagDelegatedAuthAccessors    []string
 }
 
 func (c *SecretsTuneCommand) Synopsis() string {
@@ -158,6 +159,14 @@ func (c *SecretsTuneCommand) Flags() *FlagSets {
 			"the plugin catalog, and will not start running until the plugin is reloaded.",
 	})
 
+	f.StringSliceVar(&StringSliceVar{
+		Name:   flagNameDelegatedAuthAccessors,
+		Target: &c.flagDelegatedAuthAccessors,
+		Usage: "A list of permitted authentication accessors this backend can delegate authentication to. " +
+			"Note that multiple values may be specified by providing this option multiple times, " +
+			"each time with 1 accessor.",
+	})
+
 	return set
 }
 
@@ -242,6 +251,10 @@ func (c *SecretsTuneCommand) Run(args []string) int {
 		if fl.Name == flagNamePluginVersion {
 			mountConfigInput.PluginVersion = c.flagPluginVersion
 		}
+
+		if fl.Name == flagNameDelegatedAuthAccessors {
+			mountConfigInput.DelegatedAuthAccessors = c.flagDelegatedAuthAccessors
+		}
 	})
 
 	if err := client.Sys().TuneMount(mountPath, mountConfigInput); err != nil {

helper/testhelpers/testhelpers.go

@@ -811,9 +811,15 @@ func RetryUntil(t testing.T, timeout time.Duration, f func() error) {
 	t.Fatalf("did not complete before deadline, err: %v", err)
 }
 
-// CreateEntityAndAlias clones an existing client and creates an entity/alias.
+// CreateEntityAndAlias clones an existing client and creates an entity/alias, uses userpass mount path
 // It returns the cloned client, entityID, and aliasID.
 func CreateEntityAndAlias(t testing.T, client *api.Client, mountAccessor, entityName, aliasName string) (*api.Client, string, string) {
+	return CreateEntityAndAliasWithinMount(t, client, mountAccessor, "userpass", entityName, aliasName)
+}
+
+// CreateEntityAndAliasWithinMount clones an existing client and creates an entity/alias, within the specified mountPath
+// It returns the cloned client, entityID, and aliasID.
+func CreateEntityAndAliasWithinMount(t testing.T, client *api.Client, mountAccessor, mountPath, entityName, aliasName string) (*api.Client, string, string) {
 	t.Helper()
 	userClient, err := client.Clone()
 	if err != nil {
@@ -841,7 +847,8 @@ func CreateEntityAndAlias(t testing.T, client *api.Client, mountAccessor, entity
 	if aliasID == "" {
 		t.Fatal("Alias ID not present in response")
 	}
-	_, err = client.Logical().WriteWithContext(context.Background(), fmt.Sprintf("auth/userpass/users/%s", aliasName), map[string]interface{}{
+	path := fmt.Sprintf("auth/%s/users/%s", mountPath, aliasName)
+	_, err = client.Logical().WriteWithContext(context.Background(), path, map[string]interface{}{
 		"password": "testpassword",
 	})
 	if err != nil {

sdk/helper/consts/consts.go

@@ -19,6 +19,10 @@ const (
 	// SSRF protection.
 	RequestHeaderName = "X-Vault-Request"
 
+	// WrapTTLHeaderName is the name of the header containing a directive to
+	// wrap the response
+	WrapTTLHeaderName = "X-Vault-Wrap-TTL"
+
 	// PerformanceReplicationALPN is the negotiated protocol used for
 	// performance replication.
 	PerformanceReplicationALPN = "replication_v1"

sdk/logical/error.go

@@ -3,7 +3,10 @@
 
 package logical
 
-import "errors"
+import (
+	"context"
+	"errors"
+)
 
 var (
 	// ErrUnsupportedOperation is returned if the operation is not supported
@@ -61,6 +64,47 @@ var (
 	ErrPathFunctionalityRemoved = errors.New("functionality on this path has been removed")
 )
 
+type DelegatedAuthErrorHandler func(ctx context.Context, initiatingRequest, authRequest *Request, authResponse *Response, err error) (*Response, error)
+
+var _ error = &RequestDelegatedAuthError{}
+
+// RequestDelegatedAuthError Special error indicating the backend wants to delegate authentication elsewhere
+type RequestDelegatedAuthError struct {
+	mountAccessor string
+	path          string
+	data          map[string]interface{}
+	errHandler    DelegatedAuthErrorHandler
+}
+
+func NewDelegatedAuthenticationRequest(mountAccessor, path string, data map[string]interface{}, errHandler DelegatedAuthErrorHandler) *RequestDelegatedAuthError {
+	return &RequestDelegatedAuthError{
+		mountAccessor: mountAccessor,
+		path:          path,
+		data:          data,
+		errHandler:    errHandler,
+	}
+}
+
+func (d *RequestDelegatedAuthError) Error() string {
+	return "authentication delegation requested"
+}
+
+func (d *RequestDelegatedAuthError) MountAccessor() string {
+	return d.mountAccessor
+}
+
+func (d *RequestDelegatedAuthError) Path() string {
+	return d.path
+}
+
+func (d *RequestDelegatedAuthError) Data() map[string]interface{} {
+	return d.data
+}
+
+func (d *RequestDelegatedAuthError) AuthErrorHandler() DelegatedAuthErrorHandler {
+	return d.errHandler
+}
+
 type HTTPCodedError interface {
 	Error() string
 	Code() int

sdk/logical/request.go

@@ -56,6 +56,7 @@ const (
 	NoClientToken ClientTokenSource = iota
 	ClientTokenFromVaultHeader
 	ClientTokenFromAuthzHeader
+	ClientTokenFromInternalAuth
 )
 
 type WALState struct {

vault/external_tests/delegated_auth/delegated_auth_test.go

@@ -0,0 +1,530 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package delegated_auth
+
+import (
+	"context"
+	"fmt"
+	paths "path"
+	"testing"
+
+	"github.com/hashicorp/vault/api"
+	"github.com/hashicorp/vault/builtin/credential/userpass"
+	"github.com/hashicorp/vault/builtin/logical/totp"
+	"github.com/hashicorp/vault/helper/testhelpers"
+	"github.com/hashicorp/vault/helper/testhelpers/minimal"
+	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/hashicorp/vault/vault"
+	"github.com/stretchr/testify/require"
+)
+
+func buildDaError(defaults map[string]string, d *framework.FieldData) *logical.RequestDelegatedAuthError {
+	fieldDataOrDefault := func(field string, d *framework.FieldData) string {
+		if val, ok := d.GetOk(field); ok {
+			return val.(string)
+		}
+
+		return defaults[field]
+	}
+
+	accessor := fieldDataOrDefault("accessor", d)
+	path := fieldDataOrDefault("path", d)
+	username := fieldDataOrDefault("username", d)
+	password := fieldDataOrDefault("password", d)
+	var errorHandler logical.DelegatedAuthErrorHandler
+	if handleErrorRaw, ok := d.GetOk("handle_error"); ok {
+		if handleErrorRaw.(bool) {
+			errorHandler = func(ctx context.Context, initiatingRequest, authRequest *logical.Request, authResponse *logical.Response, err error) (*logical.Response, error) {
+				return logical.ErrorResponse(fmt.Sprintf("my custom handler: %v", err)), nil
+			}
+		}
+	}
+
+	loginPath := paths.Join(path, username)
+	data := map[string]interface{}{"password": password}
+
+	return logical.NewDelegatedAuthenticationRequest(accessor, loginPath, data, errorHandler)
+}
+
+func buildDelegatedAuthFactory(defaults map[string]string) logical.Factory {
+	opHandler := func(ctx context.Context, req *logical.Request, d *framework.FieldData) (*logical.Response, error) {
+		daError := buildDaError(defaults, d)
+		if req.ClientToken == "" || req.ClientTokenSource != logical.ClientTokenFromInternalAuth {
+			return nil, daError
+		}
+
+		if req.Operation == logical.ListOperation {
+			return logical.ListResponse([]string{"success", req.ClientToken}), nil
+		}
+
+		if d.Get("loop").(bool) {
+			return nil, daError
+		}
+
+		if d.Get("perform_write").(bool) {
+			entry, err := logical.StorageEntryJSON("test", map[string]string{"test": "value"})
+			if err != nil {
+				return nil, err
+			}
+			if err = req.Storage.Put(ctx, entry); err != nil {
+				return nil, err
+			}
+		}
+
+		return &logical.Response{
+			Data: map[string]interface{}{
+				"success": true,
+				"token":   req.ClientToken,
+			},
+		}, nil
+	}
+
+	return func(ctx context.Context, config *logical.BackendConfig) (logical.Backend, error) {
+		b := new(framework.Backend)
+		b.BackendType = logical.TypeLogical
+		b.Paths = []*framework.Path{
+			{
+				Pattern: "preauth-test/list/?",
+				Operations: map[logical.Operation]framework.OperationHandler{
+					logical.ListOperation: &framework.PathOperation{Callback: opHandler},
+				},
+			},
+			{
+				Pattern: "preauth-test",
+				Operations: map[logical.Operation]framework.OperationHandler{
+					logical.ReadOperation:   &framework.PathOperation{Callback: opHandler},
+					logical.PatchOperation:  &framework.PathOperation{Callback: opHandler},
+					logical.UpdateOperation: &framework.PathOperation{Callback: opHandler},
+					logical.DeleteOperation: &framework.PathOperation{Callback: opHandler},
+				},
+				Fields: map[string]*framework.FieldSchema{
+					"accessor":      {Type: framework.TypeString},
+					"path":          {Type: framework.TypeString},
+					"username":      {Type: framework.TypeString},
+					"password":      {Type: framework.TypeString},
+					"loop":          {Type: framework.TypeBool},
+					"perform_write": {Type: framework.TypeBool},
+					"handle_error":  {Type: framework.TypeBool},
+				},
+			},
+		}
+		b.PathsSpecial = &logical.Paths{Unauthenticated: []string{"preauth-test", "preauth-test/*"}}
+		err := b.Setup(ctx, config)
+		return b, err
+	}
+}
+
+func TestDelegatedAuth(t *testing.T) {
+	t.Parallel()
+
+	// A map of success values to be populated once and used in request
+	// operations that can't pass in values
+	delegatedReqDefaults := map[string]string{
+		"username": "allowed-est",
+		"password": "test",
+		"path":     "login",
+	}
+
+	delegatedAuthFactory := buildDelegatedAuthFactory(delegatedReqDefaults)
+	coreConfig := &vault.CoreConfig{
+		CredentialBackends: map[string]logical.Factory{
+			"userpass": userpass.Factory,
+		},
+		LogicalBackends: map[string]logical.Factory{
+			"delegateauthtest": delegatedAuthFactory,
+			"totp":             totp.Factory,
+		},
+	}
+
+	cluster := minimal.NewTestSoloCluster(t, coreConfig)
+	cluster.Start()
+	defer cluster.Cleanup()
+
+	client := testhelpers.WaitForActiveNode(t, cluster).Client
+
+	// Setup two users, one with an allowed policy, another without a policy within userpass
+	err := client.Sys().PutPolicy("allow-est",
+		`path "dat/preauth-test" { capabilities = ["read", "create", "update", "patch", "delete"] }
+               path "dat/preauth-test/*" { capabilities = ["read","list"] }`)
+	require.NoError(t, err, "Failed to write policy allow-est")
+
+	err = client.Sys().EnableAuthWithOptions("userpass", &api.EnableAuthOptions{
+		Type: "userpass",
+	})
+	require.NoError(t, err, "failed mounting userpass endpoint")
+
+	_, err = client.Logical().Write("auth/userpass/users/allowed-est", map[string]interface{}{
+		"password":   "test",
+		"policies":   "allow-est",
+		"token_type": "batch",
+	})
+	require.NoError(t, err, "failed to create allowed-est user")
+
+	_, err = client.Logical().Write("auth/userpass/users/not-allowed-est", map[string]interface{}{
+		"password":   "test",
+		"token_type": "batch",
+	})
+	require.NoError(t, err, "failed to create not-allowed-est user")
+
+	_, err = client.Logical().Write("auth/userpass/users/bad-token-type-est", map[string]interface{}{
+		"password":   "test",
+		"token_type": "service",
+	})
+	require.NoError(t, err, "failed to create bad-token-type-est user")
+
+	// Setup another auth mount so we can test multiple accessors in mount tuning works later
+	err = client.Sys().EnableAuthWithOptions("userpass2", &api.EnableAuthOptions{
+		Type: "userpass",
+	})
+	require.NoError(t, err, "failed mounting userpass2")
+
+	_, err = client.Logical().Write("auth/userpass2/users/allowed-est-2", map[string]interface{}{
+		"password":   "test",
+		"policies":   "allow-est",
+		"token_type": "batch",
+	})
+	require.NoError(t, err, "failed to create allowed-est-2 user")
+
+	// Setup a dedicated mount for MFA purposes
+	err = client.Sys().EnableAuthWithOptions("userpass-mfa", &api.EnableAuthOptions{
+		Type: "userpass",
+	})
+	require.NoError(t, err, "failed mounting userpass-mfa")
+
+	// Fetch the userpass auth accessors
+	resp, err := client.Logical().Read("/sys/mounts/auth/userpass")
+	require.NoError(t, err, "failed to query for mount accessor")
+	require.NotNil(t, resp, "received nil response from mount accessor query")
+	require.NotNil(t, resp.Data, "received response with nil Data")
+	require.NotEmpty(t, resp.Data["accessor"], "Accessor field was empty: %v", resp)
+	upAccessor := resp.Data["accessor"].(string)
+
+	resp, err = client.Logical().Read("/sys/mounts/auth/userpass2")
+	require.NoError(t, err, "failed to query for mount accessor for userpass2")
+	require.NotNil(t, resp, "received nil response from mount accessor query for userpass2")
+	require.NotNil(t, resp.Data, "received response with nil Data")
+	require.NotEmpty(t, resp.Data["accessor"], "Accessor field was empty: %v", resp)
+	upAccessor2 := resp.Data["accessor"].(string)
+
+	resp, err = client.Logical().Read("/sys/mounts/auth/userpass-mfa")
+	require.NoError(t, err, "failed to query for MFA mount accessor")
+	require.NotNil(t, resp, "received nil response from MFA mount accessor query")
+	require.NotNil(t, resp.Data, "received response with nil Data for MFA mount accessor query")
+	require.NotEmpty(t, resp.Data["accessor"], "MFA mount Accessor field was empty: %v", resp)
+	upAccessorMFA := resp.Data["accessor"].(string)
+
+	resp, err = client.Logical().Read("/sys/mounts/cubbyhole")
+	require.NoError(t, err, "failed to query for mount accessor for cubbyhole")
+	require.NotNil(t, resp, "received nil response from mount accessor query for cubbyhole")
+	require.NotNil(t, resp.Data, "received response with nil Data")
+	require.NotEmpty(t, resp.Data["accessor"], "Accessor field was empty: %v", resp)
+	cubbyAccessor := resp.Data["accessor"].(string)
+
+	// Set up our backend mount that will delegate its auth to the userpass mount
+	err = client.Sys().Mount("dat", &api.MountInput{
+		Type: "delegateauthtest",
+		Config: api.MountConfigInput{
+			DelegatedAuthAccessors: []string{
+				upAccessor, upAccessorMFA, "an-accessor-that-does-not-exist",
+				cubbyAccessor,
+			},
+		},
+	})
+	require.NoError(t, err, "failed mounting delegated auth endpoint")
+
+	delegatedReqDefaults["accessor"] = upAccessor
+
+	// We want a client without any previous tokens set to make sure we aren't using
+	// the other token.
+	clientNoToken, err := client.Clone()
+	require.NoError(t, err, "failed cloning client")
+	clientNoToken.ClearToken()
+
+	// Happy path test for the various operation types we want to support, make sure
+	// for each one that we don't error out and we get back a token value from the backend
+	// call.
+	for _, test := range []string{"delete", "read", "list", "write"} {
+		t.Run("op-"+test, func(st *testing.T) {
+			switch test {
+			case "delete":
+				resp, err = clientNoToken.Logical().Delete("dat/preauth-test")
+			case "read":
+				resp, err = clientNoToken.Logical().Read("dat/preauth-test")
+			case "list":
+				resp, err = clientNoToken.Logical().List("dat/preauth-test/list/")
+			case "write":
+				resp, err = clientNoToken.Logical().Write("dat/preauth-test", map[string]interface{}{
+					"accessor": delegatedReqDefaults["accessor"],
+					"path":     delegatedReqDefaults["path"],
+					"username": delegatedReqDefaults["username"],
+					"password": delegatedReqDefaults["password"],
+				})
+			}
+
+			require.NoErrorf(st, err, "failed making %s pre-auth call with allowed-est", test)
+			require.NotNilf(st, resp, "pre-auth %s call returned nil", test)
+			require.NotNil(t, resp.Data, "received response with nil Data")
+			if test != "list" {
+				require.Equalf(st, true, resp.Data["success"], "Got an incorrect response from %s call in success field", test)
+				require.NotEmptyf(st, resp.Data["token"], "no token returned by %s handler", test)
+			} else {
+				require.NotEmpty(st, resp.Data["keys"], "list operation did not contain keys in response")
+				keys := resp.Data["keys"].([]interface{})
+				require.Equal(st, 2, len(keys), "keys field did not contain expected 2 elements")
+				require.Equal(st, "success", keys[0], "the first keys field did not contain expected value")
+				require.NotEmpty(st, keys[1], "the second keys field did not contain a token")
+			}
+		})
+	}
+
+	// Test various failure scenarios
+	failureTests := []struct {
+		name          string
+		accessor      string
+		path          string
+		username      string
+		password      string
+		errorContains string
+		forceLoop     bool
+	}{
+		{
+			name:          "policy-denies-user",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "not-allowed-est",
+			password:      "test",
+			errorContains: "permission denied",
+		},
+		{
+			name:          "bad-password",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "allowed-est",
+			password:      "bad-password",
+			errorContains: "invalid credentials",
+		},
+		{
+			name:          "unknown-user",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "non-existant-user",
+			password:      "test",
+			errorContains: "invalid username or password",
+		},
+		{
+			name:          "missing-user",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "",
+			password:      "test",
+			errorContains: "was not considered a login request",
+		},
+		{
+			name:          "missing-password",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "allowed-est",
+			password:      "",
+			errorContains: "missing password",
+		},
+		{
+			name:          "bad-path-within-delegated-auth-error",
+			accessor:      upAccessor,
+			path:          "not-the-login-path",
+			username:      "allowed-est",
+			password:      "test",
+			errorContains: "was not considered a login request",
+		},
+		{
+			name:          "empty-path-within-delegated-auth-error",
+			accessor:      upAccessor,
+			path:          "",
+			username:      "allowed-est",
+			password:      "test",
+			errorContains: "was not considered a login request",
+		},
+		{
+			name:          "empty-accessor-within-delegated-auth-error",
+			accessor:      "",
+			path:          "login",
+			username:      "allowed-est",
+			password:      "test",
+			errorContains: "backend returned an invalid mount accessor",
+		},
+		{
+			name:          "accessor-does-not-exist-within-delegated-auth-error",
+			accessor:      "an-accessor-that-does-not-exist",
+			path:          "login",
+			username:      "allowed-est",
+			password:      "test",
+			errorContains: "requested delegate authentication accessor 'an-accessor-that-does-not-exist' was not found",
+		},
+		{
+			name:          "non-allowed-accessor-within-delegated-auth-error",
+			accessor:      upAccessor2,
+			path:          "login",
+			username:      "allowed-est-2",
+			password:      "test",
+			errorContains: fmt.Sprintf("delegated auth to accessor %s not permitted", upAccessor2),
+		},
+		{
+			name:          "force-constant-login-request-loop",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "allowed-est",
+			password:      "test",
+			forceLoop:     true,
+			errorContains: "delegated authentication requested but authentication token present",
+		},
+		{
+			name:          "non-auth-mount-accessor",
+			accessor:      cubbyAccessor,
+			path:          "login",
+			username:      "allowed-est",
+			password:      "test",
+			errorContains: fmt.Sprintf("requested delegate authentication mount '%s' was not an auth mount", cubbyAccessor),
+		},
+		{
+			name:          "fails-on-non-batch-token",
+			accessor:      upAccessor,
+			path:          "login",
+			username:      "bad-token-type-est",
+			password:      "test",
+			errorContains: "delegated auth requests must be configured to issue batch tokens",
+		},
+	}
+	for _, test := range failureTests {
+		t.Run(test.name, func(st *testing.T) {
+			resp, err = clientNoToken.Logical().Write("dat/preauth-test", map[string]interface{}{
+				"accessor": test.accessor,
+				"path":     test.path,
+				"username": test.username,
+				"password": test.password,
+				"loop":     test.forceLoop,
+			})
+			if test.errorContains != "" {
+				require.ErrorContains(st, err, test.errorContains,
+					"pre-auth call should have failed due to policy restriction got resp: %v err: %v", resp, err)
+			} else {
+				require.Error(st, err, "Expected failure got resp: %v err: %v", resp, err)
+			}
+		})
+	}
+
+	// Make sure we can add an accessor to the mount that previously failed above,
+	// and the request handling code does use both accessor values.
+	t.Run("multiple-accessors", func(st *testing.T) {
+		err = client.Sys().TuneMount("dat", api.MountConfigInput{DelegatedAuthAccessors: []string{upAccessor, upAccessor2, upAccessorMFA}})
+		require.NoError(t, err, "Failed to tune mount to update delegated auth accessors")
+
+		resp, err = clientNoToken.Logical().Write("dat/preauth-test", map[string]interface{}{
+			"accessor": upAccessor2,
+			"path":     "login",
+			"username": "allowed-est-2",
+			"password": "test",
+		})
+
+		require.NoError(st, err, "failed making pre-auth call with allowed-est-2")
+		require.NotNil(st, resp, "pre-auth %s call returned nil with allowed-est-2")
+		require.NotNil(t, resp.Data, "received response with nil Data")
+		require.Equal(st, true, resp.Data["success"], "Got an incorrect response from call in success field with allowed-est-2")
+		require.NotEmpty(st, resp.Data["token"], "no token returned with allowed-est-2 user")
+	})
+
+	// Test we can delegate a permission denied error back to the originating
+	// backend for processing/response to the client
+	t.Run("backend-handles-permission-denied", func(st *testing.T) {
+		resp, err = clientNoToken.Logical().Write("dat/preauth-test", map[string]interface{}{
+			"accessor":     upAccessor,
+			"path":         "login",
+			"username":     "allowed-est",
+			"password":     "test2",
+			"handle_error": true,
+		})
+
+		require.ErrorContains(st, err, "my custom handler: invalid credentials")
+	})
+
+	// Test we can delegate a permission denied error back to the originating
+	// backend for processing/response to the client
+	t.Run("mfa-request-is-denied", func(st *testing.T) {
+		// Mount the totp secrets engine
+		testhelpers.SetupTOTPMount(st, client)
+
+		// Create a test entity and alias
+		totpUser := "test-totp"
+		testhelpers.CreateEntityAndAliasWithinMount(st, client, upAccessorMFA, "userpass-mfa", "entity1", totpUser)
+
+		// Configure a default TOTP method
+		methodID := testhelpers.SetupTOTPMethod(st, client, map[string]interface{}{
+			"issuer":                  "yCorp",
+			"period":                  5,
+			"algorithm":               "SHA256",
+			"digits":                  6,
+			"skew":                    1,
+			"key_size":                20,
+			"qr_size":                 200,
+			"max_validation_attempts": 5,
+			"method_name":             "foo",
+		})
+
+		// Configure a login enforcement specific to the MFA userpass mount to avoid conflicts on others.
+		enforcementConfig := map[string]interface{}{
+			"auth_method_accessors": []string{upAccessorMFA},
+			"name":                  methodID[0:4],
+			"mfa_method_ids":        []string{methodID},
+		}
+
+		testhelpers.SetupMFALoginEnforcement(st, client, enforcementConfig)
+
+		_, err = clientNoToken.Logical().Write("dat/preauth-test", map[string]interface{}{
+			"accessor":     upAccessorMFA,
+			"path":         "login",
+			"username":     totpUser,
+			"password":     "testpassword",
+			"handle_error": false,
+		})
+
+		require.ErrorContains(st, err, "delegated auth request requiring MFA is not supported")
+	})
+
+	// Test the behavior around receiving a request asking for response wrapping and
+	// being delegated to the secondary query we do
+	t.Run("response-wrapping-test", func(st *testing.T) {
+		resWrapClient, err := client.Clone()
+		require.NoError(st, err, "failed cloning client for response wrapping")
+
+		resWrapClient.SetWrappingLookupFunc(func(operation, path string) string {
+			if path == "dat/preauth-test" {
+				return "15s"
+			}
+			return ""
+		})
+
+		resp, err = resWrapClient.Logical().Write("dat/preauth-test", map[string]interface{}{
+			"accessor": upAccessor,
+			"path":     "login",
+			"username": "allowed-est",
+			"password": "test",
+		})
+		require.NoError(st, err, "failed calling preauth-test api with response wrapping")
+		require.NotNil(st, resp, "Got nil, nil response from preauth-test api with response wrapping")
+		require.NotNil(st, resp.WrapInfo, "response object didn't contain wrapped info")
+
+		unwrapClient, err := client.Clone()
+		require.NoError(st, err, "failed cloning client for lookups")
+		wrapToken := resp.WrapInfo.Token
+		unwrapClient.SetToken(wrapToken)
+
+		unwrapResp, err := unwrapClient.Logical().Write("sys/wrapping/unwrap", map[string]interface{}{})
+		require.NoError(st, err, "failed unwrap call")
+		require.NotNil(st, unwrapResp, "unwrap response was nil")
+		require.NotNil(st, unwrapResp.Data, "unwrap response did not contain Data")
+		require.Contains(st, unwrapResp.Data, "success", "unwrap response data did not contain success field")
+		require.Contains(st, unwrapResp.Data, "token", "unwrap response data did not contain token field")
+		require.Equal(st, true, unwrapResp.Data["success"], "Got an incorrect response in success field within unwrap call")
+		require.NotEmptyf(st, unwrapResp.Data["token"], "no token returned by handler within unwrap call")
+	})
+}

vault/logical_system.go

@@ -1426,6 +1426,9 @@ func (b *SystemBackend) handleMount(ctx context.Context, req *logical.Request, d
 	if len(apiConfig.AllowedManagedKeys) > 0 {
 		config.AllowedManagedKeys = apiConfig.AllowedManagedKeys
 	}
+	if len(apiConfig.DelegatedAuthAccessors) > 0 {
+		config.DelegatedAuthAccessors = apiConfig.DelegatedAuthAccessors
+	}
 
 	// Create the mount entry
 	me := &MountEntry{
@@ -2370,6 +2373,32 @@ func (b *SystemBackend) handleTuneWriteCommon(ctx context.Context, path string,
 		}
 	}
 
+	if rawVal, ok := data.GetOk("delegated_auth_accessors"); ok {
+		delegatedAuthAccessors := rawVal.([]string)
+
+		oldVal := mountEntry.Config.DelegatedAuthAccessors
+		mountEntry.Config.DelegatedAuthAccessors = delegatedAuthAccessors
+
+		// Update the mount table
+		var err error
+		switch {
+		case strings.HasPrefix(path, "auth/"):
+			err = b.Core.persistAuth(ctx, b.Core.auth, &mountEntry.Local)
+		default:
+			err = b.Core.persistMounts(ctx, b.Core.mounts, &mountEntry.Local)
+		}
+		if err != nil {
+			mountEntry.Config.DelegatedAuthAccessors = oldVal
+			return handleError(err)
+		}
+
+		mountEntry.SyncCache()
+
+		if b.Core.logger.IsInfo() {
+			b.Core.logger.Info("mount tuning of delegated_auth_accessors successful", "path", path)
+		}
+	}
+
 	var err error
 	var resp *logical.Response
 	var options map[string]string
@@ -6409,4 +6438,8 @@ This path responds to the following HTTP methods.
 			Returns the available and enabled experiments.
 		`,
 	},
+	"delegated_auth_accessors": {
+		"A list of auth accessors that the mount is allowed to delegate authentication too",
+		"",
+	},
 }

vault/logical_system_paths.go

@@ -4427,6 +4427,10 @@ func (b *SystemBackend) mountPaths() []*framework.Path {
 					Type:        framework.TypeCommaStringSlice,
 					Description: strings.TrimSpace(sysHelp["tune_allowed_managed_keys"][0]),
 				},
+				"delegated_auth_accessors": {
+					Type:        framework.TypeCommaStringSlice,
+					Description: strings.TrimSpace(sysHelp["allowed_delegated_auth_accessors"][0]),
+				},
 				"plugin_version": {
 					Type:        framework.TypeString,
 					Description: strings.TrimSpace(sysHelp["plugin-catalog_version"][0]),
@@ -4477,6 +4481,11 @@ func (b *SystemBackend) mountPaths() []*framework.Path {
 									Description: strings.TrimSpace(sysHelp["tune_allowed_managed_keys"][0]),
 									Required:    false,
 								},
+								"delegated_auth_accessors": {
+									Type:        framework.TypeCommaStringSlice,
+									Description: strings.TrimSpace(sysHelp["delegated_auth_accessors"][0]),
+									Required:    false,
+								},
 								"allowed_response_headers": {
 									Type:        framework.TypeCommaStringSlice,
 									Description: strings.TrimSpace(sysHelp["allowed_response_headers"][0]),

vault/mount.go

@@ -364,6 +364,7 @@ type MountConfig struct {
 	TokenType                 logical.TokenType     `json:"token_type,omitempty" structs:"token_type" mapstructure:"token_type"`
 	AllowedManagedKeys        []string              `json:"allowed_managed_keys,omitempty" mapstructure:"allowed_managed_keys"`
 	UserLockoutConfig         *UserLockoutConfig    `json:"user_lockout_config,omitempty" mapstructure:"user_lockout_config"`
+	DelegatedAuthAccessors    []string              `json:"delegated_auth_accessors,omitempty" mapstructure:"delegated_auth_accessors"`
 
 	// PluginName is the name of the plugin registered in the catalog.
 	//
@@ -399,6 +400,7 @@ type APIMountConfig struct {
 	AllowedManagedKeys        []string              `json:"allowed_managed_keys,omitempty" mapstructure:"allowed_managed_keys"`
 	UserLockoutConfig         *UserLockoutConfig    `json:"user_lockout_config,omitempty" mapstructure:"user_lockout_config"`
 	PluginVersion             string                `json:"plugin_version,omitempty" mapstructure:"plugin_version"`
+	DelegatedAuthAccessors    []string              `json:"delegated_auth_accessors,omitempty" mapstructure:"delegated_auth_accessors"`
 
 	// PluginName is the name of the plugin registered in the catalog.
 	//
@@ -500,6 +502,12 @@ func (e *MountEntry) SyncCache() {
 	} else {
 		e.synthesizedConfigCache.Store("allowed_managed_keys", e.Config.AllowedManagedKeys)
 	}
+
+	if len(e.Config.DelegatedAuthAccessors) == 0 {
+		e.synthesizedConfigCache.Delete("delegated_auth_accessors")
+	} else {
+		e.synthesizedConfigCache.Store("delegated_auth_accessors", e.Config.DelegatedAuthAccessors)
+	}
 }
 
 func (entry *MountEntry) Deserialize() map[string]interface{} {

vault/request_handling.go

@@ -9,7 +9,11 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"maps"
+	"net/textproto"
 	"os"
+	paths "path"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -767,7 +771,7 @@ func (c *Core) handleCancelableRequest(ctx context.Context, req *logical.Request
 	walState := &logical.WALState{}
 	ctx = logical.IndexStateContext(ctx, walState)
 	var auth *logical.Auth
-	if c.isLoginRequest(ctx, req) {
+	if c.isLoginRequest(ctx, req) && req.ClientTokenSource != logical.ClientTokenFromInternalAuth {
 		resp, auth, err = c.handleLoginRequest(ctx, req)
 	} else {
 		resp, auth, err = c.handleRequest(ctx, req)
@@ -1379,6 +1383,10 @@ func (c *Core) handleRequest(ctx context.Context, req *logical.Request) (retResp
 
 	// Return the response and error
 	if routeErr != nil {
+		if _, ok := routeErr.(*logical.RequestDelegatedAuthError); ok {
+			routeErr = fmt.Errorf("delegated authentication requested but authentication token present")
+		}
+
 		retErr = multierror.Append(retErr, routeErr)
 	}
 
@@ -1496,15 +1504,23 @@ func (c *Core) handleLoginRequest(ctx context.Context, req *logical.Request) (re
 	// Route the request
 	resp, routeErr := c.doRouting(ctx, req)
 
-	// if routeErr has invalid credentials error, update the userFailedLoginMap
+	handleInvalidCreds := func(err error) (*logical.Response, *logical.Auth, error) {
-	if routeErr != nil && routeErr == logical.ErrInvalidCredentials {
 		if !isUserLockoutDisabled {
 			err := c.failedUserLoginProcess(ctx, entry, req)
 			if err != nil {
 				return nil, nil, err
 			}
 		}
-		return resp, nil, routeErr
+		return resp, nil, err
+	}
+
+	if routeErr != nil {
+		// if routeErr has invalid credentials error, update the userFailedLoginMap
+		if routeErr == logical.ErrInvalidCredentials {
+			return handleInvalidCreds(routeErr)
+		} else if da, ok := routeErr.(*logical.RequestDelegatedAuthError); ok {
+			return c.handleDelegatedAuth(ctx, req, da, entry, handleInvalidCreds)
+		}
 	}
 
 	if resp != nil {
@@ -1837,6 +1853,117 @@ func (c *Core) handleLoginRequest(ctx context.Context, req *logical.Request) (re
 	return resp, auth, routeErr
 }
 
+type invalidCredentialHandler func(err error) (*logical.Response, *logical.Auth, error)
+
+// handleDelegatedAuth when a backend request returns logical.RequestDelegatedAuthError, it is requesting that
+// an authentication workflow of its choosing be implemented prior to it being able to accept it. Normally
+// this is used for standard protocols that communicate the credential information in a non-standard Vault way
+func (c *Core) handleDelegatedAuth(ctx context.Context, origReq *logical.Request, da *logical.RequestDelegatedAuthError, entry *MountEntry, invalidCredHandler invalidCredentialHandler) (*logical.Response, *logical.Auth, error) {
+	// Make sure we didn't get into a routing loop.
+	if origReq.ClientTokenSource == logical.ClientTokenFromInternalAuth {
+		return nil, nil, fmt.Errorf("%w: original request had delegated auth token, "+
+			"forbidding another delegated request from path '%s'", ErrInternalError, origReq.Path)
+	}
+
+	// Backend has requested internally delegated authentication
+	requestedAccessor := da.MountAccessor()
+	if strings.TrimSpace(requestedAccessor) == "" {
+		return nil, nil, fmt.Errorf("%w: backend returned an invalid mount accessor '%s'", ErrInternalError, requestedAccessor)
+	}
+	// First, is this allowed by the mount tunable?
+	if !slices.Contains(entry.Config.DelegatedAuthAccessors, requestedAccessor) {
+		return nil, nil, fmt.Errorf("delegated auth to accessor %s not permitted", requestedAccessor)
+	}
+
+	reqNamespace, err := namespace.FromContext(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed looking up namespace from context: %w", err)
+	}
+
+	mount := c.router.MatchingMountByAccessor(requestedAccessor)
+	if mount == nil {
+		return nil, nil, fmt.Errorf("%w: requested delegate authentication accessor '%s' was not found", logical.ErrPermissionDenied, requestedAccessor)
+	}
+	if mount.Table != credentialTableType {
+		return nil, nil, fmt.Errorf("%w: requested delegate authentication mount '%s' was not an auth mount", logical.ErrPermissionDenied, requestedAccessor)
+	}
+	if mount.NamespaceID != reqNamespace.ID {
+		return nil, nil, fmt.Errorf("%w: requested delegate authentication mount was in a different namespace than request", logical.ErrPermissionDenied)
+	}
+
+	// Found it, now form the login path and issue the request
+	path := paths.Join("auth", mount.Path, da.Path())
+	authReq, err := origReq.Clone()
+	if err != nil {
+		return nil, nil, err
+	}
+	authReq.MountAccessor = requestedAccessor
+	authReq.Path = path
+	authReq.Operation = logical.UpdateOperation
+
+	// filter out any response wrapping headers, for our embedded login request
+	delete(authReq.Headers, textproto.CanonicalMIMEHeaderKey(consts.WrapTTLHeaderName))
+	authReq.WrapInfo = nil
+
+	// Insert the data fields from the delegated auth error in our auth request
+	authReq.Data = maps.Clone(da.Data())
+
+	// Make sure we are going to perform a login request and not expose other backend types to this request
+	if !c.isLoginRequest(ctx, authReq) {
+		return nil, nil, fmt.Errorf("delegated path '%s' was not considered a login request", authReq.Path)
+	}
+
+	authResp, err := c.handleCancelableRequest(ctx, authReq)
+	if err != nil || authResp.IsError() {
+		// see if the backend wishes to handle the failed auth
+		if da.AuthErrorHandler() != nil {
+			resp, err := da.AuthErrorHandler()(ctx, origReq, authReq, authResp, err)
+			return resp, nil, err
+		}
+		switch err {
+		case nil:
+			return authResp, nil, nil
+		case logical.ErrInvalidCredentials:
+			return invalidCredHandler(err)
+		default:
+			return authResp, nil, err
+		}
+	}
+	if authResp == nil {
+		return nil, nil, fmt.Errorf("%w: delegated auth request returned empty response for request_path: %s", ErrInternalError, authReq.Path)
+	}
+	// A login request should never return a secret!
+	if authResp.Secret != nil {
+		return nil, nil, fmt.Errorf("%w: unexpected Secret response for login path for request_path: %s", ErrInternalError, authReq.Path)
+	}
+	if authResp.Auth == nil {
+		return nil, nil, fmt.Errorf("%w: Auth response was nil for request_path: %s", ErrInternalError, authReq.Path)
+	}
+	if authResp.Auth.ClientToken == "" {
+		if authResp.Auth.MFARequirement != nil {
+			return nil, nil, fmt.Errorf("%w: delegated auth request requiring MFA is not supported: %s", logical.ErrPermissionDenied, authReq.Path)
+		}
+		return nil, nil, fmt.Errorf("%w: delegated auth request did not return a client token for login path: %s", ErrInternalError, authReq.Path)
+	}
+
+	// Delegated auth tokens should only be batch tokens, as we don't want to incur
+	// the cost of storage/tidying for protocols that will be generating a token per
+	// request.
+	if !IsBatchToken(authResp.Auth.ClientToken) {
+		return nil, nil, fmt.Errorf("%w: delegated auth requests must be configured to issue batch tokens", logical.ErrPermissionDenied)
+	}
+
+	// Authentication successful, use the resulting ClientToken to reissue the original request
+	secondReq, err := origReq.Clone()
+	if err != nil {
+		return nil, nil, err
+	}
+	secondReq.ClientToken = authResp.Auth.ClientToken
+	secondReq.ClientTokenSource = logical.ClientTokenFromInternalAuth
+	resp, err := c.handleCancelableRequest(ctx, secondReq)
+	return resp, nil, err
+}
+
 // LoginCreateToken creates a token as a result of a login request.
 // If MFA is enforced, mfa/validate endpoint calls this functions
 // after successful MFA validation to generate the token.

website/content/api-docs/system/mounts.mdx

@@ -174,6 +174,9 @@ This endpoint enables a new secrets engine at the given path.
   - `allowed_managed_keys` `(array: [])` - List of managed key registry entry names
     that the mount in question is allowed to access.
 
+  - `delegated_auth_accessors` `(array: [])` - List of allowed authentication mount
+    accessors the backend can request delegated authentication for.
+
 - `options` `(map<string|string>: nil)` - Specifies mount type specific options
   that are passed to the backend.
 
@@ -382,6 +385,9 @@ This endpoint tunes configuration parameters for a given mount point.
 - `plugin_version` `(string: "")` – Specifies the semantic version of the plugin
   to use, e.g. "v1.0.0". Changes will not take effect until the mount is reloaded.
 
+- `delegated_auth_accessors` `(array: [])` - List of allowed authentication mount
+  accessors the backend can request delegated authentication for.
+
 ### Sample payload
 
 ```json