VAULT-12226: Add Static Roles to the AWS plugin (#20536)

Add static roles to the aws secrets engine --------- Co-authored-by: maxcoulombe <max.coulombe@hashicorp.com> Co-authored-by: vinay-gopalan <86625824+vinay-gopalan@users.noreply.github.com> Co-authored-by: Yoko Hyakuna <yoko@hashicorp.com>
2023-05-24 14:55:13 -04:00 · 2023-05-24 14:55:13 -04:00 · 628c51516a
parent a151ec76dd
commit 628c51516a
17 changed files with 1734 additions and 18 deletions
--- a/builtin/logical/aws/backend.go
+++ b/builtin/logical/aws/backend.go
@ -12,7 +12,9 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam/iamiface"
 	"github.com/aws/aws-sdk-go/service/sts/stsiface"
 	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/helper/consts"
 	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/hashicorp/vault/sdk/queue"
 )

 const (
@ -23,15 +25,16 @@ const (
 )

 func Factory(ctx context.Context, conf *logical.BackendConfig) (logical.Backend, error) {
-	b := Backend()
+	b := Backend(conf)
 	if err := b.Setup(ctx, conf); err != nil {
 		return nil, err
 	}
 	return b, nil
 }

-func Backend() *backend {
+func Backend(conf *logical.BackendConfig) *backend {
 	var b backend
+	b.credRotationQueue = queue.New()
 	b.Backend = &framework.Backend{
 		Help: strings.TrimSpace(backendHelp),

@ -40,7 +43,8 @@ func Backend() *backend {
 				framework.WALPrefix,
 			},
 			SealWrapStorage: []string{
-				"config/root",
+				rootConfigPath,
+				pathStaticCreds + "/",
 			},
 		},

@ -50,6 +54,8 @@ func Backend() *backend {
 			pathConfigLease(&b),
 			pathRoles(&b),
 			pathListRoles(&b),
+			pathStaticRoles(&b),
+			pathStaticCredentials(&b),
 			pathUser(&b),
 		},

@ -60,7 +66,17 @@ func Backend() *backend {
 		Invalidate:        b.invalidate,
 		WALRollback:       b.walRollback,
 		WALRollbackMinAge: minAwsUserRollbackAge,
-		BackendType:       logical.TypeLogical,
+		PeriodicFunc: func(ctx context.Context, req *logical.Request) error {
+			repState := conf.System.ReplicationState()
+			if (conf.System.LocalMount() ||
+				!repState.HasState(consts.ReplicationPerformanceSecondary)) &&
+				!repState.HasState(consts.ReplicationDRSecondary) &&
+				!repState.HasState(consts.ReplicationPerformanceStandby) {
+				return b.rotateExpiredStaticCreds(ctx, req)
+			}
+			return nil
+		},
+		BackendType: logical.TypeLogical,
 	}

 	return &b
@ -79,6 +95,10 @@ type backend struct {
 	// to enable mocking with AWS iface for tests
 	iamClient iamiface.IAMAPI
 	stsClient stsiface.STSAPI
+
+	// the age of a static role's credential is tracked by a priority queue and handled
+	// by the PeriodicFunc
+	credRotationQueue *queue.PriorityQueue
 }

 const backendHelp = `
--- a/builtin/logical/aws/backend_test.go
+++ b/builtin/logical/aws/backend_test.go
@ -148,7 +148,7 @@ func TestBackend_throttled(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}

-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
--- a/builtin/logical/aws/iam_policies_test.go
+++ b/builtin/logical/aws/iam_policies_test.go
@ -141,7 +141,7 @@ func Test_getGroupPolicies(t *testing.T) {
 			config := logical.TestBackendConfig()
 			config.StorageView = &logical.InmemStorage{}

-			b := Backend()
+			b := Backend(config)
 			if err := b.Setup(context.Background(), config); err != nil {
 				t.Fatal(err)
 			}
--- a/builtin/logical/aws/path_config_root_test.go
+++ b/builtin/logical/aws/path_config_root_test.go
@ -15,7 +15,7 @@ func TestBackend_PathConfigRoot(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}

-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
--- a/builtin/logical/aws/path_roles_test.go
+++ b/builtin/logical/aws/path_roles_test.go
@ -21,7 +21,7 @@ func TestBackend_PathListRoles(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}

-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
@ -224,7 +224,7 @@ func TestRoleCRUDWithPermissionsBoundary(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}

-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
@ -268,7 +268,7 @@ func TestRoleWithPermissionsBoundaryValidation(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}

-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
--- a/builtin/logical/aws/path_static_creds.go
+++ b/builtin/logical/aws/path_static_creds.go
@ -0,0 +1,99 @@
+package aws
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/fatih/structs"
+	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/logical"
+)
+
+const (
+	pathStaticCreds = "static-creds"
+
+	paramAccessKeyID      = "access_key_id"
+	paramSecretsAccessKey = "secret_access_key"
+)
+
+type awsCredentials struct {
+	AccessKeyID     string `json:"access_key_id" structs:"access_key_id" mapstructure:"access_key_id"`
+	SecretAccessKey string `json:"secret_access_key" structs:"secret_access_key" mapstructure:"secret_access_key"`
+}
+
+func pathStaticCredentials(b *backend) *framework.Path {
+	return &framework.Path{
+		Pattern: fmt.Sprintf("%s/%s", pathStaticCreds, framework.GenericNameWithAtRegex(paramRoleName)),
+		Fields: map[string]*framework.FieldSchema{
+			paramRoleName: {
+				Type:        framework.TypeString,
+				Description: descRoleName,
+			},
+		},
+
+		Operations: map[logical.Operation]framework.OperationHandler{
+			logical.ReadOperation: &framework.PathOperation{
+				Callback: b.pathStaticCredsRead,
+				Responses: map[int][]framework.Response{
+					http.StatusOK: {{
+						Description: http.StatusText(http.StatusOK),
+						Fields: map[string]*framework.FieldSchema{
+							paramAccessKeyID: {
+								Type:        framework.TypeString,
+								Description: descAccessKeyID,
+							},
+							paramSecretsAccessKey: {
+								Type:        framework.TypeString,
+								Description: descSecretAccessKey,
+							},
+						},
+					}},
+				},
+			},
+		},
+
+		HelpSynopsis:    pathStaticCredsHelpSyn,
+		HelpDescription: pathStaticCredsHelpDesc,
+	}
+}
+
+func (b *backend) pathStaticCredsRead(ctx context.Context, req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
+	roleName, ok := data.GetOk(paramRoleName)
+	if !ok {
+		return nil, fmt.Errorf("missing %q parameter", paramRoleName)
+	}
+
+	entry, err := req.Storage.Get(ctx, formatCredsStoragePath(roleName.(string)))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read credentials for role %q: %w", roleName, err)
+	}
+	if entry == nil {
+		return nil, nil
+	}
+
+	var credentials awsCredentials
+	if err := entry.DecodeJSON(&credentials); err != nil {
+		return nil, fmt.Errorf("failed to decode credentials: %w", err)
+	}
+
+	return &logical.Response{
+		Data: structs.New(credentials).Map(),
+	}, nil
+}
+
+func formatCredsStoragePath(roleName string) string {
+	return fmt.Sprintf("%s/%s", pathStaticCreds, roleName)
+}
+
+const pathStaticCredsHelpSyn = `Retrieve static credentials from the named role.`
+
+const pathStaticCredsHelpDesc = `
+This path reads AWS credentials for a certain static role. The keys are rotated
+periodically according to their configuration, and will return the same password
+until they are rotated.`
+
+const (
+	descAccessKeyID     = "The access key of the AWS Credential"
+	descSecretAccessKey = "The secret key of the AWS Credential"
+)
--- a/builtin/logical/aws/path_static_creds_test.go
+++ b/builtin/logical/aws/path_static_creds_test.go
@ -0,0 +1,92 @@
+package aws
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"github.com/fatih/structs"
+
+	"github.com/hashicorp/vault/sdk/framework"
+
+	"github.com/hashicorp/vault/sdk/logical"
+)
+
+// TestStaticCredsRead verifies that we can correctly read a cred that exists, and correctly _not read_
+// a cred that does not exist.
+func TestStaticCredsRead(t *testing.T) {
+	// setup
+	config := logical.TestBackendConfig()
+	config.StorageView = &logical.InmemStorage{}
+	bgCTX := context.Background() // for brevity later
+
+	// insert a cred to get
+	creds := &awsCredentials{
+		AccessKeyID:     "foo",
+		SecretAccessKey: "bar",
+	}
+	entry, err := logical.StorageEntryJSON(formatCredsStoragePath("test"), creds)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = config.StorageView.Put(bgCTX, entry)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// cases
+	cases := []struct {
+		name             string
+		roleName         string
+		expectedError    error
+		expectedResponse *logical.Response
+	}{
+		{
+			name:     "get existing creds",
+			roleName: "test",
+			expectedResponse: &logical.Response{
+				Data: structs.New(creds).Map(),
+			},
+		},
+		{
+			name:     "get non-existent creds",
+			roleName: "this-doesnt-exist",
+			// returns nil, nil
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			b := Backend(config)
+
+			req := &logical.Request{
+				Storage: config.StorageView,
+				Data: map[string]interface{}{
+					"name": c.roleName,
+				},
+			}
+			resp, err := b.pathStaticCredsRead(bgCTX, req, staticCredsFieldData(req.Data))
+
+			if err != c.expectedError {
+				t.Fatalf("got error %q, but expected %q", err, c.expectedError)
+			}
+			if !reflect.DeepEqual(resp, c.expectedResponse) {
+				t.Fatalf("got response %v, but expected %v", resp, c.expectedResponse)
+			}
+		})
+	}
+}
+
+func staticCredsFieldData(data map[string]interface{}) *framework.FieldData {
+	schema := map[string]*framework.FieldSchema{
+		paramRoleName: {
+			Type:        framework.TypeString,
+			Description: descRoleName,
+		},
+	}
+
+	return &framework.FieldData{
+		Raw:    data,
+		Schema: schema,
+	}
+}
--- a/builtin/logical/aws/path_static_roles.go
+++ b/builtin/logical/aws/path_static_roles.go
@ -0,0 +1,331 @@
+package aws
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/fatih/structs"
+	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/hashicorp/vault/sdk/queue"
+)
+
+const (
+	pathStaticRole = "static-roles"
+
+	paramRoleName       = "name"
+	paramUsername       = "username"
+	paramRotationPeriod = "rotation_period"
+)
+
+type staticRoleEntry struct {
+	Name           string        `json:"name" structs:"name" mapstructure:"name"`
+	ID             string        `json:"id" structs:"id" mapstructure:"id"`
+	Username       string        `json:"username" structs:"username" mapstructure:"username"`
+	RotationPeriod time.Duration `json:"rotation_period" structs:"rotation_period" mapstructure:"rotation_period"`
+}
+
+func pathStaticRoles(b *backend) *framework.Path {
+	roleResponse := map[int][]framework.Response{
+		http.StatusOK: {{
+			Description: http.StatusText(http.StatusOK),
+			Fields: map[string]*framework.FieldSchema{
+				paramRoleName: {
+					Type:        framework.TypeString,
+					Description: descRoleName,
+				},
+				paramUsername: {
+					Type:        framework.TypeString,
+					Description: descUsername,
+				},
+				paramRotationPeriod: {
+					Type:        framework.TypeDurationSecond,
+					Description: descRotationPeriod,
+				},
+			},
+		}},
+	}
+
+	return &framework.Path{
+		Pattern: fmt.Sprintf("%s/%s", pathStaticRole, framework.GenericNameWithAtRegex(paramRoleName)),
+		Fields: map[string]*framework.FieldSchema{
+			paramRoleName: {
+				Type:        framework.TypeString,
+				Description: descRoleName,
+			},
+			paramUsername: {
+				Type:        framework.TypeString,
+				Description: descUsername,
+			},
+			paramRotationPeriod: {
+				Type:        framework.TypeDurationSecond,
+				Description: descRotationPeriod,
+			},
+		},
+
+		Operations: map[logical.Operation]framework.OperationHandler{
+			logical.ReadOperation: &framework.PathOperation{
+				Callback:  b.pathStaticRolesRead,
+				Responses: roleResponse,
+			},
+			logical.UpdateOperation: &framework.PathOperation{
+				Callback:                    b.pathStaticRolesWrite,
+				ForwardPerformanceSecondary: true,
+				ForwardPerformanceStandby:   true,
+				Responses:                   roleResponse,
+			},
+			logical.DeleteOperation: &framework.PathOperation{
+				Callback:                    b.pathStaticRolesDelete,
+				ForwardPerformanceSecondary: true,
+				ForwardPerformanceStandby:   true,
+				Responses: map[int][]framework.Response{
+					http.StatusNoContent: {{
+						Description: http.StatusText(http.StatusNoContent),
+					}},
+				},
+			},
+		},
+
+		HelpSynopsis:    pathStaticRolesHelpSyn,
+		HelpDescription: pathStaticRolesHelpDesc,
+	}
+}
+
+func (b *backend) pathStaticRolesRead(ctx context.Context, req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
+	roleName, ok := data.GetOk(paramRoleName)
+	if !ok {
+		return nil, fmt.Errorf("missing %q parameter", paramRoleName)
+	}
+
+	b.roleMutex.RLock()
+	defer b.roleMutex.RUnlock()
+
+	entry, err := req.Storage.Get(ctx, formatRoleStoragePath(roleName.(string)))
+	if err != nil {
+		return nil, fmt.Errorf("failed to read configuration for static role %q: %w", roleName, err)
+	}
+	if entry == nil {
+		return nil, nil
+	}
+
+	var config staticRoleEntry
+	if err := entry.DecodeJSON(&config); err != nil {
+		return nil, fmt.Errorf("failed to decode configuration for static role %q: %w", roleName, err)
+	}
+
+	return &logical.Response{
+		Data: formatResponse(config),
+	}, nil
+}
+
+func (b *backend) pathStaticRolesWrite(ctx context.Context, req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
+	// Create & validate config from request parameters
+	config := staticRoleEntry{}
+	isCreate := req.Operation == logical.CreateOperation
+
+	if rawRoleName, ok := data.GetOk(paramRoleName); ok {
+		config.Name = rawRoleName.(string)
+
+		if err := b.validateRoleName(config.Name); err != nil {
+			return nil, err
+		}
+	} else {
+		return logical.ErrorResponse("missing %q parameter", paramRoleName), nil
+	}
+
+	// retrieve old role value
+	entry, err := req.Storage.Get(ctx, formatRoleStoragePath(config.Name))
+	if err != nil {
+		return nil, fmt.Errorf("couldn't check storage for pre-existing role: %w", err)
+	}
+
+	if entry != nil {
+		err = entry.DecodeJSON(&config)
+		if err != nil {
+			return nil, fmt.Errorf("couldn't convert existing role into config struct: %w", err)
+		}
+	} else {
+		// if we couldn't find an entry, this is a create event
+		isCreate = true
+	}
+
+	// other params are optional if we're not Creating
+
+	if rawUsername, ok := data.GetOk(paramUsername); ok {
+		config.Username = rawUsername.(string)
+
+		if err := b.validateIAMUserExists(ctx, req.Storage, &config, isCreate); err != nil {
+			return nil, err
+		}
+	} else if isCreate {
+		return logical.ErrorResponse("missing %q parameter", paramUsername), nil
+	}
+
+	if rawRotationPeriod, ok := data.GetOk(paramRotationPeriod); ok {
+		config.RotationPeriod = time.Duration(rawRotationPeriod.(int)) * time.Second
+
+		if err := b.validateRotationPeriod(config.RotationPeriod); err != nil {
+			return nil, err
+		}
+	} else if isCreate {
+		return logical.ErrorResponse("missing %q parameter", paramRotationPeriod), nil
+	}
+
+	b.roleMutex.Lock()
+	defer b.roleMutex.Unlock()
+
+	// Upsert role config
+	newRole, err := logical.StorageEntryJSON(formatRoleStoragePath(config.Name), config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal object to JSON: %w", err)
+	}
+	err = req.Storage.Put(ctx, newRole)
+	if err != nil {
+		return nil, fmt.Errorf("failed to save object in storage: %w", err)
+	}
+
+	// Bootstrap initial set of keys if they did not exist before. AWS Secret Access Keys can only be obtained on creation,
+	// so we need to boostrap new roles with a new initial set of keys to be able to serve valid credentials to Vault clients.
+	existingCreds, err := req.Storage.Get(ctx, formatCredsStoragePath(config.Name))
+	if err != nil {
+		return nil, fmt.Errorf("unable to verify if credentials already exist for role %q: %w", config.Name, err)
+	}
+	if existingCreds == nil {
+		err := b.createCredential(ctx, req.Storage, config, false)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create new credentials for role %q: %w", config.Name, err)
+		}
+
+		err = b.credRotationQueue.Push(&queue.Item{
+			Key:      config.Name,
+			Value:    config,
+			Priority: time.Now().Add(config.RotationPeriod).Unix(),
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to add item into the rotation queue for role %q: %w", config.Name, err)
+		}
+	}
+
+	return &logical.Response{
+		Data: formatResponse(config),
+	}, nil
+}
+
+func (b *backend) pathStaticRolesDelete(ctx context.Context, req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
+	roleName, ok := data.GetOk(paramRoleName)
+	if !ok {
+		return nil, fmt.Errorf("missing %q parameter", paramRoleName)
+	}
+
+	b.roleMutex.Lock()
+	defer b.roleMutex.Unlock()
+
+	entry, err := req.Storage.Get(ctx, formatRoleStoragePath(roleName.(string)))
+	if err != nil {
+		return nil, fmt.Errorf("couldn't locate role in storage due to error: %w", err)
+	}
+	// no entry in storage, but no error either, congrats, it's deleted!
+	if entry == nil {
+		return nil, nil
+	}
+	var cfg staticRoleEntry
+	err = entry.DecodeJSON(&cfg)
+	if err != nil {
+		return nil, fmt.Errorf("couldn't convert storage entry to role config")
+	}
+
+	err = b.deleteCredential(ctx, req.Storage, cfg, false)
+	if err != nil {
+		return nil, fmt.Errorf("failed to clean credentials while deleting role %q: %w", roleName.(string), err)
+	}
+
+	// delete from the queue
+	_, err = b.credRotationQueue.PopByKey(cfg.Name)
+	if err != nil {
+		return nil, fmt.Errorf("couldn't delete key from queue: %w", err)
+	}
+
+	return nil, req.Storage.Delete(ctx, formatRoleStoragePath(roleName.(string)))
+}
+
+func (b *backend) validateRoleName(name string) error {
+	if name == "" {
+		return errors.New("empty role name attribute given")
+	}
+	return nil
+}
+
+// validateIAMUser checks the user information we have for the role against the information on AWS. On a create, it uses the username
+// to retrieve the user information and _sets_ the userID. On update, it validates the userID and username.
+func (b *backend) validateIAMUserExists(ctx context.Context, storage logical.Storage, entry *staticRoleEntry, isCreate bool) error {
+	c, err := b.clientIAM(ctx, storage)
+	if err != nil {
+		return fmt.Errorf("unable to validate username %q: %w", entry.Username, err)
+	}
+
+	// we don't really care about the content of the result, just that it's not an error
+	out, err := c.GetUser(&iam.GetUserInput{
+		UserName: aws.String(entry.Username),
+	})
+	if err != nil || out.User == nil {
+		return fmt.Errorf("unable to validate username %q: %w", entry.Username, err)
+	}
+	if *out.User.UserName != entry.Username {
+		return fmt.Errorf("AWS GetUser returned a username, but it didn't match: %q was requested, but %q was returned", entry.Username, *out.User.UserName)
+	}
+
+	if !isCreate && *out.User.UserId != entry.ID {
+		return fmt.Errorf("AWS GetUser returned a user, but the ID did not match: %q was requested, but %q was returned", entry.ID, *out.User.UserId)
+	} else {
+		// if this is an insert, store the userID. This is the immutable part of an IAM user, but it's not exactly user-friendly.
+		// So, we allow users to specify usernames, but on updates we'll use the ID as a verification cross-check.
+		entry.ID = *out.User.UserId
+	}
+
+	return nil
+}
+
+const (
+	minAllowableRotationPeriod = 1 * time.Minute
+)
+
+func (b *backend) validateRotationPeriod(period time.Duration) error {
+	if period < minAllowableRotationPeriod {
+		return fmt.Errorf("role rotation period out of range: must be greater than %.2f seconds", minAllowableRotationPeriod.Seconds())
+	}
+	return nil
+}
+
+func formatResponse(cfg staticRoleEntry) map[string]interface{} {
+	response := structs.New(cfg).Map()
+	response[paramRotationPeriod] = int64(cfg.RotationPeriod.Seconds())
+
+	return response
+}
+
+func formatRoleStoragePath(roleName string) string {
+	return fmt.Sprintf("%s/%s", pathStaticRole, roleName)
+}
+
+const pathStaticRolesHelpSyn = `
+Manage static roles for AWS.
+`
+
+const pathStaticRolesHelpDesc = `
+This path lets you manage static roles (users) for the AWS secret backend.
+A static role is associated with a single IAM user, and manages the access
+keys based on a rotation period, automatically rotating the credential. If
+the IAM user has multiple access keys, the oldest key will be rotated.
+`
+
+const (
+	descRoleName       = "The name of this role."
+	descUsername       = "The IAM user to adopt as a static role."
+	descRotationPeriod = `Period by which to rotate the backing credential of the adopted user. 
+This can be a Go duration (e.g, '1m', 24h'), or an integer number of seconds.`
+)
--- a/builtin/logical/aws/path_static_roles_test.go
+++ b/builtin/logical/aws/path_static_roles_test.go
@ -0,0 +1,490 @@
+package aws
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/vault/sdk/queue"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/hashicorp/go-secure-stdlib/awsutil"
+	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/logical"
+)
+
+// TestStaticRolesValidation verifies that valid requests pass validation and that invalid requests fail validation.
+// This includes the user already existing in IAM roles, and the rotation period being sufficiently long.
+func TestStaticRolesValidation(t *testing.T) {
+	config := logical.TestBackendConfig()
+	config.StorageView = &logical.InmemStorage{}
+	bgCTX := context.Background() // for brevity
+
+	cases := []struct {
+		name        string
+		opts        []awsutil.MockIAMOption
+		requestData map[string]interface{}
+		isError     bool
+	}{
+		{
+			name: "all good",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{User: &iam.User{UserName: aws.String("jane-doe"), UserId: aws.String("unique-id")}}),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("abcdefghijklmnopqrstuvwxyz"),
+						SecretAccessKey: aws.String("zyxwvutsrqponmlkjihgfedcba"),
+						UserName:        aws.String("jane-doe"),
+					},
+				}),
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{},
+					IsTruncated:       aws.Bool(false),
+				}),
+			},
+			requestData: map[string]interface{}{
+				"name":            "test",
+				"username":        "jane-doe",
+				"rotation_period": "1d",
+			},
+		},
+		{
+			name: "bad user",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserError(errors.New("oh no")),
+			},
+			requestData: map[string]interface{}{
+				"name":            "test",
+				"username":        "jane-doe",
+				"rotation_period": "24h",
+			},
+			isError: true,
+		},
+		{
+			name: "user mismatch",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{User: &iam.User{UserName: aws.String("ms-impostor"), UserId: aws.String("fake-id")}}),
+			},
+			requestData: map[string]interface{}{
+				"name":            "test",
+				"username":        "jane-doe",
+				"rotation_period": "1d2h",
+			},
+			isError: true,
+		},
+		{
+			name: "bad rotation period",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{User: &iam.User{UserName: aws.String("jane-doe"), UserId: aws.String("unique-id")}}),
+			},
+			requestData: map[string]interface{}{
+				"name":            "test",
+				"username":        "jane-doe",
+				"rotation_period": "45s",
+			},
+			isError: true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			b := Backend(config)
+			miam, err := awsutil.NewMockIAM(c.opts...)(nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			b.iamClient = miam
+			if err := b.Setup(bgCTX, config); err != nil {
+				t.Fatal(err)
+			}
+			req := &logical.Request{
+				Operation: logical.UpdateOperation,
+				Storage:   config.StorageView,
+				Data:      c.requestData,
+				Path:      "static-roles/test",
+			}
+			_, err = b.pathStaticRolesWrite(bgCTX, req, staticRoleFieldData(req.Data))
+			if c.isError && err == nil {
+				t.Fatal("expected an error but didn't get one")
+			} else if !c.isError && err != nil {
+				t.Fatalf("got an unexpected error: %s", err)
+			}
+		})
+	}
+}
+
+// TestStaticRolesWrite validates that we can write a new entry for a new static role, and that we correctly
+// do not write if the request is invalid in some way.
+func TestStaticRolesWrite(t *testing.T) {
+	bgCTX := context.Background()
+
+	cases := []struct {
+		name          string
+		opts          []awsutil.MockIAMOption
+		data          map[string]interface{}
+		expectedError bool
+		findUser      bool
+		isUpdate      bool
+	}{
+		{
+			name: "happy path",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{User: &iam.User{UserName: aws.String("jane-doe"), UserId: aws.String("unique-id")}}),
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{},
+					IsTruncated:       aws.Bool(false),
+				}),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("abcdefghijklmnopqrstuvwxyz"),
+						SecretAccessKey: aws.String("zyxwvutsrqponmlkjihgfedcba"),
+						UserName:        aws.String("jane-doe"),
+					},
+				}),
+			},
+			data: map[string]interface{}{
+				"name":            "test",
+				"username":        "jane-doe",
+				"rotation_period": "1d",
+			},
+			// writes role, writes cred
+			findUser: true,
+		},
+		{
+			name: "no aws user",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserError(errors.New("no such user, etc etc")),
+			},
+			data: map[string]interface{}{
+				"name":            "test",
+				"username":        "a-nony-mous",
+				"rotation_period": "15s",
+			},
+			expectedError: true,
+		},
+		{
+			name: "update existing user",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{User: &iam.User{UserName: aws.String("john-doe"), UserId: aws.String("unique-id")}}),
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{},
+					IsTruncated:       aws.Bool(false),
+				}),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("abcdefghijklmnopqrstuvwxyz"),
+						SecretAccessKey: aws.String("zyxwvutsrqponmlkjihgfedcba"),
+						UserName:        aws.String("john-doe"),
+					},
+				}),
+			},
+			data: map[string]interface{}{
+				"name":            "johnny",
+				"rotation_period": "19m",
+			},
+			findUser: true,
+			isUpdate: true,
+		},
+	}
+
+	// if a user exists (user doesn't exist is tested in validation)
+	// we'll check how many keys the user has - if it's two, we delete one.
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			config := logical.TestBackendConfig()
+			config.StorageView = &logical.InmemStorage{}
+
+			miam, err := awsutil.NewMockIAM(
+				c.opts...,
+			)(nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			b := Backend(config)
+			b.iamClient = miam
+			if err := b.Setup(bgCTX, config); err != nil {
+				t.Fatal(err)
+			}
+
+			// put a role in storage for update tests
+			staticRole := staticRoleEntry{
+				Name:           "johnny",
+				Username:       "john-doe",
+				ID:             "unique-id",
+				RotationPeriod: 24 * time.Hour,
+			}
+			entry, err := logical.StorageEntryJSON(formatRoleStoragePath(staticRole.Name), staticRole)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = config.StorageView.Put(bgCTX, entry)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			req := &logical.Request{
+				Operation: logical.UpdateOperation,
+				Storage:   config.StorageView,
+				Data:      c.data,
+				Path:      "static-roles/" + c.data["name"].(string),
+			}
+
+			r, err := b.pathStaticRolesWrite(bgCTX, req, staticRoleFieldData(req.Data))
+			if c.expectedError && err == nil {
+				t.Fatal(err)
+			} else if c.expectedError {
+				return // save us some if statements
+			}
+
+			if err != nil {
+				t.Fatalf("got an error back unexpectedly: %s", err)
+			}
+
+			if c.findUser && r == nil {
+				t.Fatal("response was nil, but it shouldn't have been")
+			}
+
+			role, err := config.StorageView.Get(bgCTX, req.Path)
+			if c.findUser && (err != nil || role == nil) {
+				t.Fatalf("couldn't find the role we should have stored: %s", err)
+			}
+			var actualData staticRoleEntry
+			err = role.DecodeJSON(&actualData)
+			if err != nil {
+				t.Fatalf("couldn't convert storage data to role entry: %s", err)
+			}
+
+			// construct expected data
+			var expectedData staticRoleEntry
+			fieldData := staticRoleFieldData(c.data)
+			if c.isUpdate {
+				// data is johnny + c.data
+				expectedData = staticRole
+			}
+
+			if u, ok := fieldData.GetOk("username"); ok {
+				expectedData.Username = u.(string)
+			}
+			if r, ok := fieldData.GetOk("rotation_period"); ok {
+				expectedData.RotationPeriod = time.Duration(r.(int)) * time.Second
+			}
+			if n, ok := fieldData.GetOk("name"); ok {
+				expectedData.Name = n.(string)
+			}
+
+			// validate fields
+			if eu, au := expectedData.Username, actualData.Username; eu != au {
+				t.Fatalf("mismatched username, expected %q but got %q", eu, au)
+			}
+			if er, ar := expectedData.RotationPeriod, actualData.RotationPeriod; er != ar {
+				t.Fatalf("mismatched rotation period, expected %q but got %q", er, ar)
+			}
+			if en, an := expectedData.Name, actualData.Name; en != an {
+				t.Fatalf("mismatched role name, expected %q, but got %q", en, an)
+			}
+		})
+	}
+}
+
+// TestStaticRoleRead validates that we can read a configured role and correctly do not read anything if we
+// request something that doesn't exist.
+func TestStaticRoleRead(t *testing.T) {
+	config := logical.TestBackendConfig()
+	config.StorageView = &logical.InmemStorage{}
+	bgCTX := context.Background()
+
+	// test cases are run against an inmem storage holding a role called "test" attached to an IAM user called "jane-doe"
+	cases := []struct {
+		name     string
+		roleName string
+		found    bool
+	}{
+		{
+			name:     "role name exists",
+			roleName: "test",
+			found:    true,
+		},
+		{
+			name:     "role name not found",
+			roleName: "toast",
+			found:    false, // implied, but set for clarity
+		},
+	}
+
+	staticRole := staticRoleEntry{
+		Name:           "test",
+		Username:       "jane-doe",
+		RotationPeriod: 24 * time.Hour,
+	}
+	entry, err := logical.StorageEntryJSON(formatRoleStoragePath(staticRole.Name), staticRole)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = config.StorageView.Put(bgCTX, entry)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			req := &logical.Request{
+				Operation: logical.ReadOperation,
+				Storage:   config.StorageView,
+				Data: map[string]interface{}{
+					"name": c.roleName,
+				},
+				Path: formatRoleStoragePath(c.roleName),
+			}
+
+			b := Backend(config)
+
+			r, err := b.pathStaticRolesRead(bgCTX, req, staticRoleFieldData(req.Data))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if c.found {
+				if r == nil {
+					t.Fatal("response was nil, but it shouldn't have been")
+				}
+			} else {
+				if r != nil {
+					t.Fatal("response should have been nil on a non-existent role")
+				}
+			}
+		})
+	}
+}
+
+// TestStaticRoleDelete validates that we correctly remove a role on a delete request, and that we correctly do not
+// remove anything if a role does not exist with that name.
+func TestStaticRoleDelete(t *testing.T) {
+	bgCTX := context.Background()
+
+	// test cases are run against an inmem storage holding a role called "test" attached to an IAM user called "jane-doe"
+	cases := []struct {
+		name  string
+		role  string
+		found bool
+	}{
+		{
+			name:  "role found",
+			role:  "test",
+			found: true,
+		},
+		{
+			name:  "role not found",
+			role:  "tossed",
+			found: false,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			config := logical.TestBackendConfig()
+			config.StorageView = &logical.InmemStorage{}
+
+			// fake an IAM
+			var iamfunc awsutil.IAMAPIFunc
+			if !c.found {
+				iamfunc = awsutil.NewMockIAM(awsutil.WithDeleteAccessKeyError(errors.New("shouldn't have called delete")))
+			} else {
+				iamfunc = awsutil.NewMockIAM()
+			}
+			miam, err := iamfunc(nil)
+			if err != nil {
+				t.Fatalf("couldn't initialize mockiam: %s", err)
+			}
+
+			b := Backend(config)
+			b.iamClient = miam
+
+			// put in storage
+			staticRole := staticRoleEntry{
+				Name:           "test",
+				Username:       "jane-doe",
+				RotationPeriod: 24 * time.Hour,
+			}
+			entry, err := logical.StorageEntryJSON(formatRoleStoragePath(staticRole.Name), staticRole)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = config.StorageView.Put(bgCTX, entry)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			l, err := config.StorageView.List(bgCTX, "")
+			if err != nil || len(l) != 1 {
+				t.Fatalf("couldn't add an entry to storage during test setup: %s", err)
+			}
+
+			// put in queue
+			err = b.credRotationQueue.Push(&queue.Item{
+				Key:      staticRole.Name,
+				Value:    staticRole,
+				Priority: time.Now().Add(90 * time.Hour).Unix(),
+			})
+			if err != nil {
+				t.Fatalf("couldn't add items to pq")
+			}
+
+			req := &logical.Request{
+				Operation: logical.ReadOperation,
+				Storage:   config.StorageView,
+				Data: map[string]interface{}{
+					"name": c.role,
+				},
+				Path: formatRoleStoragePath(c.role),
+			}
+
+			r, err := b.pathStaticRolesDelete(bgCTX, req, staticRoleFieldData(req.Data))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if r != nil {
+				t.Fatal("response wasn't nil, but it should have been")
+			}
+
+			l, err = config.StorageView.List(bgCTX, "")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if c.found && len(l) != 0 {
+				t.Fatal("size of role storage is non zero after delete")
+			} else if !c.found && len(l) != 1 {
+				t.Fatal("size of role storage changed after what should have been no deletion")
+			}
+
+			if c.found && b.credRotationQueue.Len() != 0 {
+				t.Fatal("size of queue is non-zero after delete")
+			} else if !c.found && b.credRotationQueue.Len() != 1 {
+				t.Fatal("size of queue changed after what should have been no deletion")
+			}
+		})
+	}
+}
+
+func staticRoleFieldData(data map[string]interface{}) *framework.FieldData {
+	schema := map[string]*framework.FieldSchema{
+		paramRoleName: {
+			Type:        framework.TypeString,
+			Description: descRoleName,
+		},
+		paramUsername: {
+			Type:        framework.TypeString,
+			Description: descUsername,
+		},
+		paramRotationPeriod: {
+			Type:        framework.TypeDurationSecond,
+			Description: descRotationPeriod,
+		},
+	}
+
+	return &framework.FieldData{
+		Raw:    data,
+		Schema: schema,
+	}
+}
--- a/builtin/logical/aws/rotation.go
+++ b/builtin/logical/aws/rotation.go
@ -0,0 +1,188 @@
+package aws
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/hashicorp/vault/sdk/queue"
+)
+
+// rotateExpiredStaticCreds will pop expired credentials (credentials whose priority
+// represents a time before the present), rotate the associated credential, and push
+// them back onto the queue with the new priority.
+func (b *backend) rotateExpiredStaticCreds(ctx context.Context, req *logical.Request) error {
+	var errs *multierror.Error
+
+	for {
+		keepGoing, err := b.rotateCredential(ctx, req.Storage)
+		if err != nil {
+			errs = multierror.Append(errs, err)
+		}
+		if !keepGoing {
+			if errs.ErrorOrNil() != nil {
+				return fmt.Errorf("error(s) occurred while rotating expired static credentials: %w", errs)
+			} else {
+				return nil
+			}
+		}
+	}
+}
+
+// rotateCredential pops an element from the priority queue, and if it is expired, rotate and re-push.
+// If a cred was rotated, it returns true, otherwise false.
+func (b *backend) rotateCredential(ctx context.Context, storage logical.Storage) (rotated bool, err error) {
+	// If queue is empty or first item does not need a rotation (priority is next rotation timestamp) there is nothing to do
+	item, err := b.credRotationQueue.Pop()
+	if err != nil {
+		// the queue is just empty, which is fine.
+		if err == queue.ErrEmpty {
+			return false, nil
+		}
+		return false, fmt.Errorf("failed to pop from queue for role %q: %w", item.Key, err)
+	}
+	if item.Priority > time.Now().Unix() {
+		// no rotation required
+		// push the item back into priority queue
+		err = b.credRotationQueue.Push(item)
+		if err != nil {
+			return false, fmt.Errorf("failed to add item into the rotation queue for role %q: %w", item.Key, err)
+		}
+		return false, nil
+	}
+
+	cfg := item.Value.(staticRoleEntry)
+
+	err = b.createCredential(ctx, storage, cfg, true)
+	if err != nil {
+		return false, err
+	}
+
+	// set new priority and re-queue
+	item.Priority = time.Now().Add(cfg.RotationPeriod).Unix()
+	err = b.credRotationQueue.Push(item)
+	if err != nil {
+		return false, fmt.Errorf("failed to add item into the rotation queue for role %q: %w", cfg.Name, err)
+	}
+
+	return true, nil
+}
+
+// createCredential will create a new iam credential, deleting the oldest one if necessary.
+func (b *backend) createCredential(ctx context.Context, storage logical.Storage, cfg staticRoleEntry, shouldLockStorage bool) error {
+	iamClient, err := b.clientIAM(ctx, storage)
+	if err != nil {
+		return fmt.Errorf("unable to get the AWS IAM client: %w", err)
+	}
+
+	// IAM users can have a most 2 sets of keys at a time.
+	// (https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_iam-quotas.html)
+	// Ideally we would get this value through an api check, but I'm not sure one exists.
+	const maxAllowedKeys = 2
+
+	err = b.validateIAMUserExists(ctx, storage, &cfg, false)
+	if err != nil {
+		return fmt.Errorf("iam user didn't exist, or username/userid didn't match: %w", err)
+	}
+
+	accessKeys, err := iamClient.ListAccessKeys(&iam.ListAccessKeysInput{
+		UserName: aws.String(cfg.Username),
+	})
+	if err != nil {
+		return fmt.Errorf("unable to list existing access keys for IAM user %q: %w", cfg.Username, err)
+	}
+
+	// If we have the maximum number of keys, we have to delete one to make another (so we can get the credentials).
+	// We'll delete the oldest one.
+	//
+	// Since this check relies on a pre-coded maximum, it's a bit fragile. If the number goes up, we risk deleting
+	// a key when we didn't need to. If this number goes down, we'll start throwing errors because we think we're
+	// allowed to create a key and aren't. In either case, adjusting the constant should be sufficient to fix things.
+	if len(accessKeys.AccessKeyMetadata) >= maxAllowedKeys {
+		oldestKey := accessKeys.AccessKeyMetadata[0]
+
+		for i := 1; i < len(accessKeys.AccessKeyMetadata); i++ {
+			if accessKeys.AccessKeyMetadata[i].CreateDate.Before(*oldestKey.CreateDate) {
+				oldestKey = accessKeys.AccessKeyMetadata[i]
+			}
+		}
+
+		_, err := iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
+			AccessKeyId: oldestKey.AccessKeyId,
+			UserName:    oldestKey.UserName,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to delete oldest access keys for user %q: %w", cfg.Username, err)
+		}
+	}
+
+	// Create new set of keys
+	out, err := iamClient.CreateAccessKey(&iam.CreateAccessKeyInput{
+		UserName: aws.String(cfg.Username),
+	})
+	if err != nil {
+		return fmt.Errorf("unable to create new access keys for user %q: %w", cfg.Username, err)
+	}
+
+	// Persist new keys
+	entry, err := logical.StorageEntryJSON(formatCredsStoragePath(cfg.Name), &awsCredentials{
+		AccessKeyID:     *out.AccessKey.AccessKeyId,
+		SecretAccessKey: *out.AccessKey.SecretAccessKey,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal object to JSON: %w", err)
+	}
+	if shouldLockStorage {
+		b.roleMutex.Lock()
+		defer b.roleMutex.Unlock()
+	}
+	err = storage.Put(ctx, entry)
+	if err != nil {
+		return fmt.Errorf("failed to save object in storage: %w", err)
+	}
+
+	return nil
+}
+
+// delete credential will remove the credential associated with the role from storage.
+func (b *backend) deleteCredential(ctx context.Context, storage logical.Storage, cfg staticRoleEntry, shouldLockStorage bool) error {
+	// synchronize storage access if we didn't in the caller.
+	if shouldLockStorage {
+		b.roleMutex.Lock()
+		defer b.roleMutex.Unlock()
+	}
+
+	key, err := storage.Get(ctx, formatCredsStoragePath(cfg.Name))
+	if err != nil {
+		return fmt.Errorf("couldn't find key in storage: %w", err)
+	}
+	// no entry, so i guess we deleted it already
+	if key == nil {
+		return nil
+	}
+	var creds awsCredentials
+	err = key.DecodeJSON(&creds)
+	if err != nil {
+		return fmt.Errorf("couldn't decode storage entry to a valid credential: %w", err)
+	}
+
+	err = storage.Delete(ctx, formatCredsStoragePath(cfg.Name))
+	if err != nil {
+		return fmt.Errorf("couldn't delete from storage: %w", err)
+	}
+
+	// because we have the information, this is the one we created, so it's safe for us to delete.
+	_, err = b.iamClient.DeleteAccessKey(&iam.DeleteAccessKeyInput{
+		AccessKeyId: aws.String(creds.AccessKeyID),
+		UserName:    aws.String(cfg.Username),
+	})
+	if err != nil {
+		return fmt.Errorf("couldn't delete from IAM: %w", err)
+	}
+
+	return nil
+}
--- a/builtin/logical/aws/rotation_test.go
+++ b/builtin/logical/aws/rotation_test.go
@ -0,0 +1,348 @@
+package aws
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/iam/iamiface"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/hashicorp/go-secure-stdlib/awsutil"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/hashicorp/vault/sdk/queue"
+)
+
+// TestRotation verifies that the rotation code and priority queue correctly selects and rotates credentials
+// for static secrets.
+func TestRotation(t *testing.T) {
+	bgCTX := context.Background()
+
+	type credToInsert struct {
+		config staticRoleEntry // role configuration from a normal createRole request
+		age    time.Duration   // how old the cred should be - if this is longer than the config.RotationPeriod,
+		// the cred is 'pre-expired'
+
+		changed bool // whether we expect the cred to change - this is technically redundant to a comparison between
+		// rotationPeriod and age.
+	}
+
+	// due to a limitation with the mockIAM implementation, any cred you want to rotate must have
+	// username jane-doe and userid unique-id, since we can only pre-can one exact response to GetUser
+	cases := []struct {
+		name  string
+		creds []credToInsert
+	}{
+		{
+			name: "refresh one",
+			creds: []credToInsert{
+				{
+					config: staticRoleEntry{
+						Name:           "test",
+						Username:       "jane-doe",
+						ID:             "unique-id",
+						RotationPeriod: 2 * time.Second,
+					},
+					age:     5 * time.Second,
+					changed: true,
+				},
+			},
+		},
+		{
+			name: "refresh none",
+			creds: []credToInsert{
+				{
+					config: staticRoleEntry{
+						Name:           "test",
+						Username:       "jane-doe",
+						ID:             "unique-id",
+						RotationPeriod: 1 * time.Minute,
+					},
+					age:     5 * time.Second,
+					changed: false,
+				},
+			},
+		},
+		{
+			name: "refresh one of two",
+			creds: []credToInsert{
+				{
+					config: staticRoleEntry{
+						Name:           "toast",
+						Username:       "john-doe",
+						ID:             "other-id",
+						RotationPeriod: 1 * time.Minute,
+					},
+					age:     5 * time.Second,
+					changed: false,
+				},
+				{
+					config: staticRoleEntry{
+						Name:           "test",
+						Username:       "jane-doe",
+						ID:             "unique-id",
+						RotationPeriod: 1 * time.Second,
+					},
+					age:     5 * time.Second,
+					changed: true,
+				},
+			},
+		},
+		{
+			name:  "no creds to rotate",
+			creds: []credToInsert{},
+		},
+	}
+
+	ak := "long-access-key-id"
+	oldSecret := "abcdefghijklmnopqrstuvwxyz"
+	newSecret := "zyxwvutsrqponmlkjihgfedcba"
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			config := logical.TestBackendConfig()
+			config.StorageView = &logical.InmemStorage{}
+
+			b := Backend(config)
+
+			// insert all our creds
+			for i, cred := range c.creds {
+
+				// all the creds will be the same for every user, but that's okay
+				// since what we care about is whether they changed on a single-user basis.
+				miam, err := awsutil.NewMockIAM(
+					// blank list for existing user
+					awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+						AccessKeyMetadata: []*iam.AccessKeyMetadata{
+							{},
+						},
+					}),
+					// initial key to store
+					awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+						AccessKey: &iam.AccessKey{
+							AccessKeyId:     aws.String(ak),
+							SecretAccessKey: aws.String(oldSecret),
+						},
+					}),
+					awsutil.WithGetUserOutput(&iam.GetUserOutput{
+						User: &iam.User{
+							UserId:   aws.String(cred.config.ID),
+							UserName: aws.String(cred.config.Username),
+						},
+					}),
+				)(nil)
+				if err != nil {
+					t.Fatalf("couldn't initialze mock IAM handler: %s", err)
+				}
+				b.iamClient = miam
+
+				err = b.createCredential(bgCTX, config.StorageView, cred.config, true)
+				if err != nil {
+					t.Fatalf("couldn't insert credential %d: %s", i, err)
+				}
+
+				item := &queue.Item{
+					Key:      cred.config.Name,
+					Value:    cred.config,
+					Priority: time.Now().Add(-1 * cred.age).Add(cred.config.RotationPeriod).Unix(),
+				}
+				err = b.credRotationQueue.Push(item)
+				if err != nil {
+					t.Fatalf("couldn't push item onto queue: %s", err)
+				}
+			}
+
+			// update aws responses, same argument for why it's okay every cred will be the same
+			miam, err := awsutil.NewMockIAM(
+				// old key
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{
+						{
+							AccessKeyId: aws.String(ak),
+						},
+					},
+				}),
+				// new key
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String(ak),
+						SecretAccessKey: aws.String(newSecret),
+					},
+				}),
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{
+					User: &iam.User{
+						UserId:   aws.String("unique-id"),
+						UserName: aws.String("jane-doe"),
+					},
+				}),
+			)(nil)
+			if err != nil {
+				t.Fatalf("couldn't initialze mock IAM handler: %s", err)
+			}
+			b.iamClient = miam
+
+			req := &logical.Request{
+				Storage: config.StorageView,
+			}
+			err = b.rotateExpiredStaticCreds(bgCTX, req)
+			if err != nil {
+				t.Fatalf("got an error rotating credentials: %s", err)
+			}
+
+			// check our credentials
+			for i, cred := range c.creds {
+				entry, err := config.StorageView.Get(bgCTX, formatCredsStoragePath(cred.config.Name))
+				if err != nil {
+					t.Fatalf("got an error retrieving credentials %d", i)
+				}
+				var out awsCredentials
+				err = entry.DecodeJSON(&out)
+				if err != nil {
+					t.Fatalf("could not unmarshal storage view entry for cred %d to an aws credential: %s", i, err)
+				}
+
+				if cred.changed && out.SecretAccessKey != newSecret {
+					t.Fatalf("expected the key for cred %d to have changed, but it hasn't", i)
+				} else if !cred.changed && out.SecretAccessKey != oldSecret {
+					t.Fatalf("expected the key for cred %d to have stayed the same, but it changed", i)
+				}
+			}
+		})
+	}
+}
+
+type fakeIAM struct {
+	iamiface.IAMAPI
+	delReqs []*iam.DeleteAccessKeyInput
+}
+
+func (f *fakeIAM) DeleteAccessKey(r *iam.DeleteAccessKeyInput) (*iam.DeleteAccessKeyOutput, error) {
+	f.delReqs = append(f.delReqs, r)
+	return f.IAMAPI.DeleteAccessKey(r)
+}
+
+// TestCreateCredential verifies that credential creation firstly only deletes credentials if it needs to (i.e., two
+// or more credentials on IAM), and secondly correctly deletes the oldest one.
+func TestCreateCredential(t *testing.T) {
+	cases := []struct {
+		name       string
+		username   string
+		id         string
+		deletedKey string
+		opts       []awsutil.MockIAMOption
+	}{
+		{
+			name:     "zero keys",
+			username: "jane-doe",
+			id:       "unique-id",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{},
+				}),
+				// delete should _not_ be called
+				awsutil.WithDeleteAccessKeyError(errors.New("should not have been called")),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("key"),
+						SecretAccessKey: aws.String("itsasecret"),
+					},
+				}),
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{
+					User: &iam.User{
+						UserId:   aws.String("unique-id"),
+						UserName: aws.String("jane-doe"),
+					},
+				}),
+			},
+		},
+		{
+			name:     "one key",
+			username: "jane-doe",
+			id:       "unique-id",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{
+						{AccessKeyId: aws.String("foo"), CreateDate: aws.Time(time.Now())},
+					},
+				}),
+				// delete should _not_ be called
+				awsutil.WithDeleteAccessKeyError(errors.New("should not have been called")),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("key"),
+						SecretAccessKey: aws.String("itsasecret"),
+					},
+				}),
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{
+					User: &iam.User{
+						UserId:   aws.String("unique-id"),
+						UserName: aws.String("jane-doe"),
+					},
+				}),
+			},
+		},
+		{
+			name:       "two keys",
+			username:   "jane-doe",
+			id:         "unique-id",
+			deletedKey: "foo",
+			opts: []awsutil.MockIAMOption{
+				awsutil.WithListAccessKeysOutput(&iam.ListAccessKeysOutput{
+					AccessKeyMetadata: []*iam.AccessKeyMetadata{
+						{AccessKeyId: aws.String("foo"), CreateDate: aws.Time(time.Time{})},
+						{AccessKeyId: aws.String("bar"), CreateDate: aws.Time(time.Now())},
+					},
+				}),
+				awsutil.WithCreateAccessKeyOutput(&iam.CreateAccessKeyOutput{
+					AccessKey: &iam.AccessKey{
+						AccessKeyId:     aws.String("key"),
+						SecretAccessKey: aws.String("itsasecret"),
+					},
+				}),
+				awsutil.WithGetUserOutput(&iam.GetUserOutput{
+					User: &iam.User{
+						UserId:   aws.String("unique-id"),
+						UserName: aws.String("jane-doe"),
+					},
+				}),
+			},
+		},
+	}
+
+	config := logical.TestBackendConfig()
+	config.StorageView = &logical.InmemStorage{}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			miam, err := awsutil.NewMockIAM(
+				c.opts...,
+			)(nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			fiam := &fakeIAM{
+				IAMAPI: miam,
+			}
+
+			b := Backend(config)
+			b.iamClient = fiam
+
+			err = b.createCredential(context.Background(), config.StorageView, staticRoleEntry{Username: c.username, ID: c.id}, true)
+			if err != nil {
+				t.Fatalf("got an error we didn't expect: %q", err)
+			}
+
+			if c.deletedKey != "" {
+				if len(fiam.delReqs) != 1 {
+					t.Fatalf("called the wrong number of deletes (called %d deletes)", len(fiam.delReqs))
+				}
+				actualKey := *fiam.delReqs[0].AccessKeyId
+				if c.deletedKey != actualKey {
+					t.Fatalf("we deleted the wrong key: %q instead of %q", actualKey, c.deletedKey)
+				}
+			}
+		})
+	}
+}
--- a/builtin/logical/aws/secret_access_keys_test.go
+++ b/builtin/logical/aws/secret_access_keys_test.go
@ -120,7 +120,7 @@ func TestGenUsername(t *testing.T) {
 func TestReadConfig_DefaultTemplate(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}
-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
@ -164,7 +164,7 @@ func TestReadConfig_DefaultTemplate(t *testing.T) {
 func TestReadConfig_CustomTemplate(t *testing.T) {
 	config := logical.TestBackendConfig()
 	config.StorageView = &logical.InmemStorage{}
-	b := Backend()
+	b := Backend(config)
 	if err := b.Setup(context.Background(), config); err != nil {
 		t.Fatal(err)
 	}
--- a/changelog/20536.txt
+++ b/changelog/20536.txt
@ -0,0 +1,3 @@
+```release-note:feature
+**AWS Static Roles**: The AWS Secrets Engine can manage static roles configured by users.
+```
--- a/go.mod
+++ b/go.mod
@ -94,7 +94,7 @@ require (
 	github.com/hashicorp/go-raftchunking v0.6.3-0.20191002164813-7e9e8525653a
 	github.com/hashicorp/go-retryablehttp v0.7.2
 	github.com/hashicorp/go-rootcerts v1.0.2
-	github.com/hashicorp/go-secure-stdlib/awsutil v0.1.6
+	github.com/hashicorp/go-secure-stdlib/awsutil v0.2.0
 	github.com/hashicorp/go-secure-stdlib/base62 v0.1.2
 	github.com/hashicorp/go-secure-stdlib/gatedwriter v0.1.1
 	github.com/hashicorp/go-secure-stdlib/kv-builder v0.1.2
--- a/go.sum
+++ b/go.sum
@ -1715,8 +1715,8 @@ github.com/hashicorp/go-retryablehttp v0.7.2/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5
 github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
 github.com/hashicorp/go-rootcerts v1.0.2 h1:jzhAVGtqPKbwpyCPELlgNWhE1znq+qwJtW5Oi2viEzc=
 github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
-github.com/hashicorp/go-secure-stdlib/awsutil v0.1.6 h1:W9WN8p6moV1fjKLkeqEgkAMu5rauy9QeYDAmIaPuuiA=
-github.com/hashicorp/go-secure-stdlib/awsutil v0.1.6/go.mod h1:MpCPSPGLDILGb4JMm94/mMi3YysIqsXzGCzkEZjcjXg=
+github.com/hashicorp/go-secure-stdlib/awsutil v0.2.0 h1:VmeslHTkAkPaolVKr9MCsZrY5i73Y7ITDgTJ4eVv+94=
+github.com/hashicorp/go-secure-stdlib/awsutil v0.2.0/go.mod h1:MpCPSPGLDILGb4JMm94/mMi3YysIqsXzGCzkEZjcjXg=
 github.com/hashicorp/go-secure-stdlib/base62 v0.1.1/go.mod h1:EdWO6czbmthiwZ3/PUsDV+UD1D5IRU4ActiaWGwt0Yw=
 github.com/hashicorp/go-secure-stdlib/base62 v0.1.2 h1:ET4pqyjiGmY09R5y+rSd70J2w45CtbWDNvGqWp/R3Ng=
 github.com/hashicorp/go-secure-stdlib/base62 v0.1.2/go.mod h1:EdWO6czbmthiwZ3/PUsDV+UD1D5IRU4ActiaWGwt0Yw=
--- a/website/content/api-docs/secret/aws.mdx
+++ b/website/content/api-docs/secret/aws.mdx
@ -64,8 +64,8 @@ valid AWS credentials with proper permissions.

  To ensure generated usernames are within length limits for both STS/IAM, the template must adequately handle
  both conditional cases (see [Conditional Templates](https://pkg.go.dev/text/template)). As an example, if no template
-  is provided the field defaults to the template below. It is to be noted that, DisplayName is the name of the vault 
-  authenticated user running the AWS credential generation and PolicyName is the name of the Role for which the 
+  is provided the field defaults to the template below. It is to be noted that, DisplayName is the name of the vault
+  authenticated user running the AWS credential generation and PolicyName is the name of the Role for which the
  credential is being generated for:

  ```
@ -585,3 +585,136 @@ $ curl \
  }
 }
 ```
+
+## Create Static Role
+This endpoint creates or updates static role definitions. A static role is a 1-to-1 mapping
+with an AWS IAM User, which will be adopted and managed by Vault, including rotating it according
+to the configured `rotation_period`.
+
+<Note>
+
+Vault will create a new credential upon configuration, and if the maximum number of access keys already exist, Vault will rotate the oldest one. Vault must do this to know the credential.
+
+At each rotation, Vault will rotate the oldest existing credential.
+
+</Note>
+
+| Method | Path                      |
+| :----- | :------------------------ |
+| `POST` | `/aws/static-roles/:name` |
+
+### Parameters
+
+- `name` `(string: <required>)` – Specifies the name of the role to create. This
+is specified as part of the URL.
+
+- `username` `(string: <required>)` – Specifies the username of the IAM user.
+
+- `rotation_period` `(string/int: <required>)` – Specifies the amount of time
+Vault should wait before rotating the password. The minimum is 1 minute. Can be
+specified in either `24h` or `86400` format (see [duration format strings](/vault/docs/concepts/duration-format)).
+
+### Sample Payload
+
+```json
+{
+  "username": "example-user",
+  "rotation_period": "11h30m"
+}
+```
+
+### Sample Request
+
+```shell-session
+$ curl \
+    --header "X-Vault-Token: ..." \
+    --request POST \
+    --data @payload.json \
+    http://127.0.0.1:8200/v1/aws/static-roles/my-static-role
+```
+
+### Sample Response
+
+## Read Static Role
+
+This endpoint queries the static role definition.
+
+| Method | Path                      |
+| :----- | :------------------------ |
+| `GET`  | `/aws/static-roles/:name` |
+
+### Parameters
+
+- `name` `(string: <required>)` – Specifies the name of the static role to read.
+This is specified as part of the URL.
+
+### Sample Request
+
+```shell-session
+$ curl \
+    --header "X-Vault-Token: ..." \
+    --request GET \
+    http://127.0.0.1:8200/v1/aws/static-roles/my-static-role
+```
+### Sample Response
+
+```json
+{
+  "name": "my-static-role",
+  "username": "example-user",
+  "rotation_period": "11h30m"
+}
+```
+
+## Delete Static Role
+
+This endpoint deletes the static role definition. The user, having been defined externally,
+must be cleaned up manually.
+
+| Method   | Path                      |
+| :------- | :------------------------ |
+| `DELETE` | `/aws/static-roles/:name` |
+
+### Parameters
+
+- `name` `(string: <required>)` – Specifies the name of the static role to
+delete. This is specified as part of the URL.
+
+### Sample Request
+
+```shell-session
+$ curl \
+    --header "X-Vault-Token: ..." \
+    --request DELETE \
+    http://127.0.0.1:8200/v1/aws/static-roles/my-static-role
+```
+
+## Get Static Credentials
+
+This endpoint returns the current credentials based on the named static role.
+
+| Method | Path                      |
+| :----- | :------------------------ |
+| `GET`  | `/aws/static-creds/:name` |
+
+### Parameters
+
+- `name` `(string: <required>)` – Specifies the name of the static role to get
+credentials for. This is specified as part of the URL.
+
+### Sample Request
+
+```shell-session
+$ curl \
+    --header "X-Vault-Token: ..." \
+    http://127.0.0.1:8200/v1/aws/static-creds/my-static-role
+```
+
+### Sample Response
+
+```json
+{
+  "access_key_id": "AKIA...",
+  "access_secret_key": "..."
+}
+```
--- a/website/content/docs/secrets/aws.mdx
+++ b/website/content/docs/secrets/aws.mdx
@ -32,6 +32,18 @@ Vault supports three different types of credentials to retrieve from AWS:
   passing in the supplied AWS policy document and return the access key, secret
   key, and session token to the caller.

+### Static Roles
+The AWS secrets engine supports the concept of "static roles", which are
+a 1-to-1 mapping of Vault Roles to IAM users. The current password
+for the user is stored and automatically rotated by Vault on a
+configurable period of time. This is in contrast to dynamic secrets, where a
+unique username and password pair are generated with each credential request.
+When credentials are requested for the Role, Vault returns the current
+Access Key ID and Secret Access Key for the configured user, allowing anyone with the proper
+Vault policies to have access to the IAM credentials.
+
+Please see the [API documentation](/vault/api-docs/secret/aws#create-static-role) for details on this feature.
+
 ## Setup

 Most secrets engines must be configured in advance before they can perform their
@ -160,7 +172,7 @@ the proper permission, it can generate credentials.
    ---           -----
    access_key    AKIA3ALIVABCDG5XC8H4
    ```
-    
+
    ~> **Note:** Due to AWS eventual consistency, after calling the
    `aws/config/rotate-root` endpoint, subsequent calls from Vault to
    AWS may fail for a few seconds until AWS becomes consistent again.