add dynamodb database access (#18843)

* Add a new db engine
* Add tests for new engine
* Update tsh db subcommands
* Refactor error message and suggestions for unsupported tsh commands
* Add dynamodb to test plan
* Add AWS external ID to db config and update protos

.github/ISSUE_TEMPLATE/testplan.md

@@ -1141,6 +1141,7 @@ tsh bench sessions --max=5000 --web user ls
   - [ ] Azure Cache for Redis.
   - [ ] Elasticsearch.
   - [ ] Cassandra/ScyllaDB.
+  - [ ] Dynamodb.
 - [ ] Connect to a database within a remote cluster via a trusted cluster.
   - [ ] Self-hosted Postgres.
   - [ ] Self-hosted MySQL.
@@ -1161,6 +1162,7 @@ tsh bench sessions --max=5000 --web user ls
   - [ ] Azure Cache for Redis.
   - [ ] Elasticsearch.
   - [ ] Cassandra/ScyllaDB.
+  - [ ] Dynamodb.
 - [ ] Verify audit events.
   - [ ] `db.session.start` is emitted when you connect.
   - [ ] `db.session.end` is emitted when you disconnect.

api/proto/teleport/legacy/types/types.proto

@@ -399,6 +399,8 @@ message AWS {
     (gogoproto.nullable) = false,
     (gogoproto.jsontag) = "redshift_serverless,omitempty"
   ];
+  // ExternalID is an optional AWS external ID used to enable assuming an AWS role across accounts.
+  string ExternalID = 10 [(gogoproto.jsontag) = "external_id,omitempty"];
 }
 
 // SecretStore contains secret store configurations.

api/types/database.go

@@ -390,6 +390,10 @@ func (d *DatabaseV3) IsAWSKeyspaces() bool {
 	return d.GetType() == DatabaseTypeAWSKeyspaces
 }
 
+func (d *DatabaseV3) IsDynamoDB() bool {
+	return d.GetType() == DatabaseTypeDynamoDB
+}
+
 // IsAWSHosted returns true if database is hosted by AWS.
 func (d *DatabaseV3) IsAWSHosted() bool {
 	_, ok := d.getAWSType()
@@ -405,8 +409,13 @@ func (d *DatabaseV3) IsCloudHosted() bool {
 // getAWSType returns the database type.
 func (d *DatabaseV3) getAWSType() (string, bool) {
 	aws := d.GetAWS()
-	if aws.AccountID != "" && d.Spec.Protocol == DatabaseTypeCassandra {
-		return DatabaseTypeAWSKeyspaces, true
+	switch d.Spec.Protocol {
+	case DatabaseTypeCassandra:
+		if aws.AccountID != "" {
+			return DatabaseTypeAWSKeyspaces, true
+		}
+	case DatabaseTypeDynamoDB:
+		return DatabaseTypeDynamoDB, true
 	}
 	if aws.Redshift.ClusterID != "" {
 		return DatabaseTypeRedshift, true
@@ -491,6 +500,10 @@ func (d *DatabaseV3) CheckAndSetDefaults() error {
 	if d.Spec.Protocol == "" {
 		return trace.BadParameter("database %q protocol is empty", d.GetName())
 	}
+	if d.IsDynamoDB() {
+		// DynamoDB gets its own checking logic for its unusual config.
+		return trace.Wrap(d.handleDynamoDBConfig())
+	}
 	if d.Spec.URI == "" {
 		switch {
 		case d.IsAWSKeyspaces() && d.GetAWS().Region != "":
@@ -604,11 +617,16 @@ func (d *DatabaseV3) CheckAndSetDefaults() error {
 			return trace.BadParameter("database %q AWS account ID is empty", d.GetName())
 		}
 		if d.Spec.AWS.Region == "" {
-			region, err := awsutils.CassandraEndpointRegion(d.Spec.URI)
-			if err != nil {
-				return trace.Wrap(err)
+			switch {
+			case d.IsAWSKeyspaces():
+				region, err := awsutils.CassandraEndpointRegion(d.Spec.URI)
+				if err != nil {
+					return trace.Wrap(err)
+				}
+				d.Spec.AWS.Region = region
+			default:
+				return trace.BadParameter("database %q AWS region is empty", d.GetName())
 			}
-			d.Spec.AWS.Region = region
 		}
 	case azureutils.IsCacheForRedisEndpoint(d.Spec.URI):
 		// ResourceID is required for fetching Redis tokens.
@@ -651,6 +669,43 @@ func (d *DatabaseV3) CheckAndSetDefaults() error {
 	return nil
 }
 
+// handleDynamoDBConfig handles DynamoDB configuration checking.
+func (d *DatabaseV3) handleDynamoDBConfig() error {
+	if d.Spec.AWS.AccountID == "" {
+		return trace.BadParameter("database %q AWS account ID is empty", d.GetName())
+	}
+
+	info, err := awsutils.ParseDynamoDBEndpoint(d.Spec.URI)
+	switch {
+	case err != nil:
+		// when region parsing returns an error but the region is set, it's ok because we can just construct the URI using the region,
+		// so we check if the region is configured to see if this is really a configuration error.
+		if d.Spec.AWS.Region == "" {
+			// the AWS region is empty and we can't derive it from the URI, so this is a config error.
+			return trace.BadParameter("database %q AWS region is empty and cannot be derived from the URI %q",
+				d.GetName(), d.Spec.URI)
+		}
+		if awsutils.IsAWSEndpoint(d.Spec.URI) {
+			// The user configured an AWS URI that which doesn't look like a DynamoDB endpoint.
+			// The URI must look like <service>.<region>.<partition> or <region>.<partition>
+			return trace.Wrap(err)
+		}
+	case d.Spec.AWS.Region == "":
+		// if the AWS region is empty we can just use the region extracted from the URI.
+		d.Spec.AWS.Region = info.Region
+	case d.Spec.AWS.Region != info.Region:
+		// if the AWS region is not empty but doesn't match the URI, this may indicate a user configuration mistake.
+		return trace.BadParameter("database %q AWS region %q does not match the configured URI region %q, "+
+			" omit the URI and it will be derived automatically for the configured AWS region",
+			d.GetName(), d.Spec.AWS.Region, info.Region)
+	}
+
+	if d.Spec.URI == "" {
+		d.Spec.URI = awsutils.DynamoDBURIForRegion(d.Spec.AWS.Region)
+	}
+	return nil
+}
+
 // GetSecretStore returns secret store configurations.
 func (d *DatabaseV3) GetSecretStore() SecretStore {
 	return d.Spec.AWS.SecretStore
@@ -689,6 +744,8 @@ const (
 	DatabaseTypeAWSKeyspaces = "keyspace"
 	// DatabaseTypeCassandra is AWS-hosted Keyspace database.
 	DatabaseTypeCassandra = "cassandra"
+	// DatabaseTypeDynamoDB is a DynamoDB database.
+	DatabaseTypeDynamoDB = "dynamodb"
 )
 
 // GetServerName returns the GCP database project and instance as "<project-id>:<instance-id>".

api/types/database_test.go

@@ -19,6 +19,8 @@ package types
 import (
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/gravitational/trace"
 	"github.com/stretchr/testify/require"
 )
@@ -538,3 +540,145 @@ func TestDatabaseSelfHosted(t *testing.T) {
 		})
 	}
 }
+
+func TestDynamoDBConfig(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		desc       string
+		uri        string
+		region     string
+		account    string
+		wantSpec   DatabaseSpecV3
+		wantErrMsg string
+	}{
+		{
+			desc:    "account and region and empty URI is correct",
+			region:  "us-west-1",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "aws://dynamodb.us-west-1.amazonaws.com",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:    "account and AWS URI and empty region is correct",
+			uri:     "dynamodb.us-west-1.amazonaws.com",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "dynamodb.us-west-1.amazonaws.com",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:    "account and AWS streams dynamodb URI and empty region is correct",
+			uri:     "streams.dynamodb.us-west-1.amazonaws.com",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "streams.dynamodb.us-west-1.amazonaws.com",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:    "account and AWS dax URI and empty region is correct",
+			uri:     "dax.us-west-1.amazonaws.com",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "dax.us-west-1.amazonaws.com",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:    "account and region and matching AWS URI region is correct",
+			uri:     "dynamodb.us-west-1.amazonaws.com",
+			region:  "us-west-1",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "dynamodb.us-west-1.amazonaws.com",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:    "account and region and custom URI is correct",
+			uri:     "localhost:8080",
+			region:  "us-west-1",
+			account: "12345",
+			wantSpec: DatabaseSpecV3{
+				URI: "localhost:8080",
+				AWS: AWS{
+					Region:    "us-west-1",
+					AccountID: "12345",
+				},
+			},
+		},
+		{
+			desc:       "region and different AWS URI region is an error",
+			uri:        "dynamodb.us-west-2.amazonaws.com",
+			region:     "us-west-1",
+			account:    "12345",
+			wantErrMsg: "does not match the configured URI",
+		},
+		{
+			desc:       "invalid AWS URI is an error",
+			uri:        "a.streams.dynamodb.us-west-1.amazonaws.com",
+			region:     "us-west-1",
+			account:    "12345",
+			wantErrMsg: "invalid DynamoDB endpoint",
+		},
+		{
+			desc:       "custom URI and empty region is an error",
+			uri:        "localhost:8080",
+			account:    "12345",
+			wantErrMsg: "region is empty",
+		},
+		{
+			desc:       "empty URI and empty region is an error",
+			account:    "12345",
+			wantErrMsg: "region is empty",
+		},
+		{
+			desc:       "missing account id",
+			wantErrMsg: "account ID is empty",
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+			database, err := NewDatabaseV3(Metadata{
+				Name: "test",
+			}, DatabaseSpecV3{
+				Protocol: "dynamodb",
+				URI:      tt.uri,
+				AWS: AWS{
+					Region:    tt.region,
+					AccountID: tt.account,
+				},
+			})
+			if tt.wantErrMsg != "" {
+				require.Error(t, err)
+				require.ErrorContains(t, err, tt.wantErrMsg)
+				return
+			}
+			require.NoError(t, err)
+			diff := cmp.Diff(tt.wantSpec, database.Spec, cmpopts.IgnoreFields(DatabaseSpecV3{}, "Protocol"))
+			require.Empty(t, diff)
+		})
+	}
+}

api/utils/aws/endpoint.go

@@ -366,7 +366,7 @@ func ParseElastiCacheEndpoint(endpoint string) (*RedisEndpointInfo, error) {
 
 	// Remove partition suffix. Note that endpoints for CN regions use the same
 	// format except they end with AWSCNEndpointSuffix.
-	endpointWithoutSuffix, err := removePartitionSuffix(endpoint)
+	endpointWithoutSuffix, _, err := removePartitionSuffix(endpoint)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -531,7 +531,7 @@ func ParseMemoryDBEndpoint(endpoint string) (*RedisEndpointInfo, error) {
 	//
 	// Unlike RDS/Redshift endpoints, the service subdomain is before region.
 	// Unlike ElastiCache endpoints, MemoryDB uses full region name.
-	endpointWithoutSuffix, err := removePartitionSuffix(endpoint)
+	endpointWithoutSuffix, _, err := removePartitionSuffix(endpoint)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -607,16 +607,16 @@ func removeSchemaAndPort(endpoint string) (string, error) {
 	return parsedURL.Hostname(), nil
 }
 
-func removePartitionSuffix(endpoint string) (string, error) {
+func removePartitionSuffix(endpoint string) (string, string, error) {
 	switch {
 	case strings.HasSuffix(endpoint, AWSEndpointSuffix):
-		return strings.TrimSuffix(endpoint, AWSEndpointSuffix), nil
+		return strings.TrimSuffix(endpoint, AWSEndpointSuffix), AWSEndpointSuffix, nil
 
 	case strings.HasSuffix(endpoint, AWSCNEndpointSuffix):
-		return strings.TrimSuffix(endpoint, AWSCNEndpointSuffix), nil
+		return strings.TrimSuffix(endpoint, AWSCNEndpointSuffix), AWSCNEndpointSuffix, nil
 
 	default:
-		return "", trace.BadParameter("%v is not a valid AWS endpoint", endpoint)
+		return "", "", trace.BadParameter("%v is not a valid AWS endpoint", endpoint)
 	}
 }
 
@@ -647,6 +647,15 @@ const (
 
 	// MemoryDBSServiceName is the service name for AWS MemoryDB.
 	MemoryDBSServiceName = "memorydb"
+
+	// DynamoDBServiceName is the service name for AWS DynamoDB.
+	DynamoDBServiceName = "dynamodb"
+	// DynamoDBFipsServiceName is the fips variant service name for AWS DynamoDB.
+	DynamoDBFipsServiceName = "dynamodb-fips"
+	// DynamoDBStreamsServiceName is the AWS DynamoDB Streams service name.
+	DynamoDBStreamsServiceName = "streams.dynamodb"
+	// DAXServiceName is the AWS DynamoDB Accelerator service name.
+	DAXServiceName = "dax"
 )
 
 // CassandraEndpointURLForRegion returns a Cassandra endpoint based on the provided region.
@@ -662,16 +671,92 @@ func CassandraEndpointURLForRegion(region string) string {
 // where endpoint looks like cassandra.us-east-2.amazonaws.com
 // https://docs.aws.amazon.com/keyspaces/latest/devguide/programmatic.endpoints.html
 func CassandraEndpointRegion(endpoint string) (string, error) {
-	endpoint, err := removeSchemaAndPort(endpoint)
+	parts, _, err := extractAWSEndpointParts(endpoint)
 	if err != nil {
 		return "", trace.Wrap(err)
 	}
-
-	endpoint = strings.TrimSuffix(endpoint, AWSCNEndpointSuffix)
-	endpoint = strings.TrimSuffix(endpoint, AWSEndpointSuffix)
-	parts := strings.Split(endpoint, ".")
 	if len(parts) != 2 {
 		return "", trace.BadParameter("invalid Cassandra endpoint")
 	}
 	return parts[1], nil
 }
+
+// DynamoDBEndpointInfo describes info extracted from a DynamoDB endpoint.
+type DynamoDBEndpointInfo struct {
+	// Service is the service subdomain of the endpoint, for example "dynamodb" or "dax".
+	Service string
+	// Region is the AWS region for the endpoint, for example "us-west-1".
+	Region string
+	// Partition is the AWS partition for the endpoint, for example ".amazonaws.com"
+	Partition string
+}
+
+// ParseDynamoDBEndpoint parses and extract info from the provided DynamoDB endpoint.
+func ParseDynamoDBEndpoint(endpoint string) (*DynamoDBEndpointInfo, error) {
+	endpoint = strings.ToLower(endpoint)
+	parts, partition, err := extractAWSEndpointParts(endpoint)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	switch len(parts) {
+	case 2, 3:
+	default:
+		return nil, trace.BadParameter("invalid DynamoDB endpoint %q", endpoint)
+	}
+	info := &DynamoDBEndpointInfo{
+		Service:   strings.Join(parts[:len(parts)-1], "."),
+		Region:    parts[len(parts)-1],
+		Partition: partition,
+	}
+
+	// check for recognized service name.
+	switch info.Service {
+	case DynamoDBServiceName, DynamoDBFipsServiceName,
+		DynamoDBStreamsServiceName, DAXServiceName:
+	default:
+		return nil, trace.BadParameter("invalid DynamoDB endpoint %q", endpoint)
+	}
+
+	// check that the partition is valid for the region.
+	if info.Region == "" || info.Partition == "" {
+		return nil, trace.BadParameter("invalid DynamoDB endpoint %q", endpoint)
+	}
+	switch {
+	case info.Partition == AWSCNEndpointSuffix && IsCNRegion(info.Region):
+	case info.Partition == AWSEndpointSuffix && !IsCNRegion(info.Region):
+	default:
+		return nil, trace.BadParameter("invalid AWS region %q for AWS partition %q",
+			info.Region, info.Partition)
+	}
+	return info, nil
+}
+
+// DynamoDBURIForRegion constructs a DynamoDB URI based on the AWS region.
+// The URI uses a custom schema aws:// to differentiate an auto-generated URI from
+// a user-configured URI in the engine.
+// When the Teleport DynamoDB engine sees this custom URI schema, it will resolve
+// the real endpoint using the request API target.
+// https://docs.aws.amazon.com/general/latest/gr/ddb.html
+func DynamoDBURIForRegion(region string) string {
+	var suffix string
+	if IsCNRegion(region) {
+		suffix = AWSCNEndpointSuffix
+	} else {
+		suffix = AWSEndpointSuffix
+	}
+	return fmt.Sprintf("aws://dynamodb.%s%s", region, suffix)
+}
+
+// extractAWSEndpointParts strips the schema, port, and AWS suffix,
+// then splits the prefix by subdomain separator (".") and returns the parts and suffix.
+func extractAWSEndpointParts(endpoint string) ([]string, string, error) {
+	uri, err := removeSchemaAndPort(endpoint)
+	if err != nil {
+		return nil, "", trace.Wrap(err)
+	}
+	prefix, suffix, err := removePartitionSuffix(uri)
+	if err != nil {
+		return nil, "", trace.Wrap(err)
+	}
+	return strings.Split(prefix, "."), suffix, nil
+}

api/utils/aws/endpoint_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package aws
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/gravitational/trace"
@@ -508,3 +509,132 @@ func TestRedshiftServerlessEndpoint(t *testing.T) {
 		})
 	}
 }
+
+func TestDynamoDBURIForRegion(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		desc    string
+		region  string
+		wantURI string
+	}{
+		{
+			desc:    "region is in correct AWS partition",
+			region:  "us-east-1",
+			wantURI: "aws://dynamodb.us-east-1.amazonaws.com",
+		},
+		{
+			desc:    "china north region is in correct AWS partition",
+			region:  "cn-north-1",
+			wantURI: "aws://dynamodb.cn-north-1.amazonaws.com.cn",
+		},
+		{
+			desc:    "china northwest region is in correct AWS partition",
+			region:  "cn-northwest-1",
+			wantURI: "aws://dynamodb.cn-northwest-1.amazonaws.com.cn",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			require.Equal(t, tt.wantURI, DynamoDBURIForRegion(tt.region))
+		})
+	}
+}
+
+func TestParseDynamoDBEndpoint(t *testing.T) {
+	t.Parallel()
+	t.Run("parses valid endpoint", func(t *testing.T) {
+		t.Parallel()
+		for _, parts := range []struct {
+			services  []string
+			regions   []string
+			partition string
+		}{
+			{
+				services:  []string{DynamoDBServiceName, DynamoDBFipsServiceName, DynamoDBStreamsServiceName, DAXServiceName},
+				regions:   []string{"us-east-1", "us-gov-east-1"},
+				partition: AWSEndpointSuffix,
+			},
+			{
+				services:  []string{DynamoDBServiceName, DynamoDBStreamsServiceName, DAXServiceName},
+				regions:   []string{"cn-north-1", "cn-northwest-1"},
+				partition: AWSCNEndpointSuffix,
+			},
+		} {
+			parts := parts
+			for _, svc := range parts.services {
+				svc := svc
+				for _, region := range parts.regions {
+					region := region
+					endpoint := fmt.Sprintf("%s.%s%s", svc, region, parts.partition)
+					t.Run(endpoint, func(t *testing.T) {
+						t.Parallel()
+						info, err := ParseDynamoDBEndpoint(endpoint)
+						require.NoError(t, err)
+						wantInfo := DynamoDBEndpointInfo{
+							Service:   svc,
+							Region:    region,
+							Partition: parts.partition,
+						}
+						require.NotNil(t, info)
+						require.Equal(t, wantInfo, *info)
+					})
+				}
+			}
+		}
+	})
+
+	tests := []struct {
+		desc     string
+		services []string
+		regions  []string
+		endpoint string
+		wantInfo *DynamoDBEndpointInfo
+	}{
+		{
+			desc:     "empty uri",
+			endpoint: "",
+		},
+		{
+			desc:     "not AWS uri",
+			endpoint: "localhost",
+		},
+		{
+			desc:     "missing region",
+			endpoint: "amazonaws.com",
+		},
+		{
+			desc:     "missing china region",
+			endpoint: "amazonaws.com.cn",
+		},
+		{
+			desc:     "unrecognized service subdomain",
+			endpoint: "foo.us-east-1.amazonaws.com",
+		},
+		{
+			desc:     "unrecognized dynamodb service subdomain",
+			endpoint: "foo.dynamodb.us-east-1.amazonaws.com",
+		},
+		{
+			desc:     "unrecognized streams service subdomain",
+			endpoint: "streams.foo.us-east-1.amazonaws.com",
+		},
+		{
+			desc:     "mismatched us region and china partition",
+			endpoint: "streams.dynamodb.us-east-1.amazonaws.com.cn",
+		},
+		{
+			desc:     "mismatched china region and non-china partition",
+			endpoint: "streams.dynamodb.cn-north-1.amazonaws.com",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run("detects invalid endpoint with"+tt.desc, func(t *testing.T) {
+			t.Parallel()
+			info, err := ParseDynamoDBEndpoint(tt.endpoint)
+			require.Error(t, err, "endpoint %s should be invalid", tt.endpoint)
+			require.Nil(t, info)
+		})
+	}
+}

api/utils/aws/fuzz_test.go

@@ -45,3 +45,11 @@ func FuzzParseElastiCacheEndpoint(f *testing.F) {
 		})
 	})
 }
+
+func FuzzParseDynamoDBEndpoint(f *testing.F) {
+	f.Fuzz(func(t *testing.T, endpoint string) {
+		require.NotPanics(t, func() {
+			ParseDynamoDBEndpoint(endpoint)
+		})
+	})
+}

fuzz/oss-fuzz-build.sh

@@ -108,6 +108,9 @@ build_teleport_api_fuzzers() {
   compile_native_go_fuzzer $TELEPORT_PREFIX/api/utils/aws \
     FuzzParseElastiCacheEndpoint fuzz_parse_elasti_cache_endpoint
 
+  compile_native_go_fuzzer $TELEPORT_PREFIX/api/utils/aws \
+    FuzzParseDynamoDBEndpoint fuzz_parse_dynamodb_endpoint
+
   cd -
 
 }

lib/client/db/dbcmd/dbcmd.go

@@ -65,6 +65,8 @@ const (
 	curlBin = "curl"
 	// elasticsearchSQLBin is the Elasticsearch SQL client program name.
 	elasticsearchSQLBin = "elasticsearch-sql-cli"
+	// awsBin is the aws CLI program name.
+	awsBin = "aws"
 )
 
 // Execer is an abstraction of Go's exec module, as this one doesn't specify any interfaces.
@@ -185,6 +187,9 @@ func (c *CLICommandBuilder) GetConnectCommand() (*exec.Cmd, error) {
 
 	case defaults.ProtocolElasticsearch:
 		return c.getElasticsearchCommand()
+
+	case defaults.ProtocolDynamoDB:
+		return c.getDynamoDBCommand()
 	}
 
 	return nil, trace.BadParameter("unsupported database protocol: %v", c.db)
@@ -571,6 +576,24 @@ func (c *CLICommandBuilder) getElasticsearchCommand() (*exec.Cmd, error) {
 	return nil, trace.BadParameter("%v interactive command is only supported in --tunnel mode.", elasticsearchSQLBin)
 }
 
+func (c *CLICommandBuilder) getDynamoDBCommand() (*exec.Cmd, error) {
+	// we can't guess at what the user wants to do, so this command is for print purposes only,
+	// and it only works with a local proxy tunnel.
+	if !c.options.printFormat || !c.options.noTLS || c.options.localProxyHost == "" || c.options.localProxyPort == 0 {
+		svc := "<db>"
+		if c.db != nil && c.db.ServiceName != "" {
+			svc = c.db.ServiceName
+		}
+		return nil, trace.BadParameter("DynamoDB requires a local proxy tunnel. Use `tsh proxy db --tunnel %v`", svc)
+	}
+	args := []string{
+		"--endpoint", fmt.Sprintf("http://%v:%v/", c.options.localProxyHost, c.options.localProxyPort),
+		"[dynamodb|dynamodbstreams|dax]",
+		"<command>",
+	}
+	return c.options.exe.Command(awsBin, args...), nil
+}
+
 func (c *CLICommandBuilder) getElasticsearchAlternativeCommands() []CommandAlternative {
 	var commands []CommandAlternative
 	if c.isElasticsearchSQLBinAvailable() {

lib/client/db/dbcmd/dbcmd_test.go

@@ -528,6 +528,42 @@ func TestCLICommandBuilderGetConnectCommand(t *testing.T) {
 			cmd:          nil,
 			wantErr:      true,
 		},
+		{
+			name:         "dynamodb for exec is an error",
+			dbProtocol:   defaults.ProtocolDynamoDB,
+			opts:         []ConnectCommandFunc{WithNoTLS(), WithLocalProxy("localhost", 12345, "")},
+			execer:       &fakeExec{},
+			databaseName: "",
+			cmd:          nil,
+			wantErr:      true,
+		},
+		{
+			name:         "dynamodb without proxy is an error",
+			dbProtocol:   defaults.ProtocolDynamoDB,
+			opts:         []ConnectCommandFunc{WithPrintFormat(), WithNoTLS(), WithLocalProxy("", 0, "")},
+			execer:       &fakeExec{},
+			databaseName: "",
+			cmd:          nil,
+			wantErr:      true,
+		},
+		{
+			name:         "dynamodb with TLS proxy is an error",
+			dbProtocol:   defaults.ProtocolDynamoDB,
+			opts:         []ConnectCommandFunc{WithPrintFormat(), WithLocalProxy("localhost", 12345, "")},
+			execer:       &fakeExec{},
+			databaseName: "",
+			cmd:          nil,
+			wantErr:      true,
+		},
+		{
+			name:         "dynamodb with print format and no-TLS proxy is ok",
+			dbProtocol:   defaults.ProtocolDynamoDB,
+			opts:         []ConnectCommandFunc{WithPrintFormat(), WithNoTLS(), WithLocalProxy("localhost", 12345, "")},
+			execer:       &fakeExec{},
+			databaseName: "",
+			cmd:          []string{"aws", "--endpoint", "http://localhost:12345/", "[dynamodb|dynamodbstreams|dax]", "<command>"},
+			wantErr:      false,
+		},
 	}
 
 	for _, tt := range tests {

lib/cloud/clients.go

@@ -30,6 +30,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/postgresql/armpostgresql"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/subscription/armsubscription"
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
 	"github.com/aws/aws-sdk-go/aws/session"
 	awssession "github.com/aws/aws-sdk-go/aws/session"
@@ -718,7 +719,13 @@ type TestCloudClients struct {
 
 // GetAWSSession returns AWS session for the specified region.
 func (c *TestCloudClients) GetAWSSession(region string) (*awssession.Session, error) {
-	return nil, trace.NotImplemented("not implemented")
+	return session.NewSession(&aws.Config{
+		Credentials: credentials.NewCredentials(&credentials.StaticProvider{Value: credentials.Value{
+			AccessKeyID:     "fakeClientKeyID",
+			SecretAccessKey: "fakeClientSecret",
+		}}),
+		Region: aws.String(region),
+	})
 }
 
 // GetAWSSessionForRole returns AWS session for the specified role ARN.

lib/config/configuration.go

@@ -145,6 +145,8 @@ type CommandLineFlags struct {
 	DatabaseAWSRegion string
 	// DatabaseAWSAccountID is an optional AWS account ID e.g. when using Keyspaces.
 	DatabaseAWSAccountID string
+	// DatabaseAWSExternalID is an optional AWS external ID used to enable assuming an AWS role across accounts.
+	DatabaseAWSExternalID string
 	// DatabaseAWSRedshiftClusterID is Redshift cluster identifier.
 	DatabaseAWSRedshiftClusterID string
 	// DatabaseAWSRDSInstanceID is RDS instance identifier.
@@ -1314,8 +1316,9 @@ func applyDatabasesConfig(fc *FileConfig, cfg *service.Config) error {
 				Mode:       service.TLSMode(database.TLS.Mode),
 			},
 			AWS: service.DatabaseAWS{
-				AccountID: database.AWS.AccountID,
-				Region:    database.AWS.Region,
+				AccountID:  database.AWS.AccountID,
+				ExternalID: database.AWS.ExternalID,
+				Region:     database.AWS.Region,
 				Redshift: service.DatabaseAWSRedshift{
 					ClusterID: database.AWS.Redshift.ClusterID,
 				},
@@ -2021,8 +2024,9 @@ func Configure(clf *CommandLineFlags, cfg *service.Config, legacyAppFlags bool)
 				CACert: caBytes,
 			},
 			AWS: service.DatabaseAWS{
-				Region:    clf.DatabaseAWSRegion,
-				AccountID: clf.DatabaseAWSAccountID,
+				Region:     clf.DatabaseAWSRegion,
+				AccountID:  clf.DatabaseAWSAccountID,
+				ExternalID: clf.DatabaseAWSExternalID,
 				Redshift: service.DatabaseAWSRedshift{
 					ClusterID: clf.DatabaseAWSRedshiftClusterID,
 				},

lib/config/configuration_test.go

@@ -2722,6 +2722,34 @@ func TestDatabaseCLIFlags(t *testing.T) {
 				},
 			},
 		},
+		{
+			desc: "AWS DynamoDB",
+			inFlags: CommandLineFlags{
+				DatabaseName:          "ddb",
+				DatabaseProtocol:      defaults.ProtocolDynamoDB,
+				DatabaseURI:           "dynamodb.us-east-1.amazonaws.com",
+				DatabaseAWSAccountID:  "123456789012",
+				DatabaseAWSExternalID: "12345678901234",
+				DatabaseAWSRegion:     "us-east-1",
+			},
+			outDatabase: service.Database{
+				Name:     "ddb",
+				Protocol: defaults.ProtocolDynamoDB,
+				URI:      "dynamodb.us-east-1.amazonaws.com",
+				AWS: service.DatabaseAWS{
+					Region:     "us-east-1",
+					AccountID:  "123456789012",
+					ExternalID: "12345678901234",
+				},
+				StaticLabels: map[string]string{
+					types.OriginLabel: types.OriginConfigFile,
+				},
+				DynamicLabels: services.CommandLabels{},
+				TLS: service.DatabaseTLS{
+					Mode: service.VerifyFull,
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

lib/config/database.go

@@ -263,16 +263,21 @@ db_service:
   databases:
   - name: {{ .StaticDatabaseName }}
     protocol: {{ .StaticDatabaseProtocol }}
+    {{- if .StaticDatabaseURI }}
     uri: {{ .StaticDatabaseURI }}
+    {{- end}}
     {{- if .DatabaseCACertFile }}
     tls:
       ca_cert_file: {{ .DatabaseCACertFile }}
     {{- end }}
-    {{- if or .DatabaseAWSRegion .DatabaseAWSRedshiftClusterID }}
+    {{- if or .DatabaseAWSRegion .DatabaseAWSRedshiftClusterID .DatabaseAWSExternalID }}
     aws:
       {{- if .DatabaseAWSRegion }}
       region: {{ .DatabaseAWSRegion }}
       {{- end }}
+      {{- if .DatabaseAWSExternalID }}
+      external_id: {{ .DatabaseAWSExternalID }}
+      {{- end }}
       {{- if .DatabaseAWSRedshiftClusterID }}
       redshift:
         cluster_id: {{ .DatabaseAWSRedshiftClusterID }}
@@ -482,6 +487,8 @@ type DatabaseSampleFlags struct {
 	DatabaseProtocols []string
 	// DatabaseAWSRegion is an optional database cloud region e.g. when using AWS RDS.
 	DatabaseAWSRegion string
+	// DatabaseAWSExternalID is an optional AWS database external ID, used when assuming roles.
+	DatabaseAWSExternalID string
 	// DatabaseAWSRedshiftClusterID is Redshift cluster identifier.
 	DatabaseAWSRedshiftClusterID string
 	// DatabaseADDomain is the Active Directory domain for authentication.

lib/config/database_test.go

@@ -144,6 +144,7 @@ func TestMakeDatabaseConfig(t *testing.T) {
 			StaticDatabaseURI:            "postgres://localhost:5432",
 			StaticDatabaseRawLabels:      `env=prod,arch=[5m2s:/bin/uname -m "p1 p2"]`,
 			DatabaseAWSRegion:            "us-west-1",
+			DatabaseAWSExternalID:        "1234567890",
 			DatabaseAWSRedshiftClusterID: "redshift-cluster-1",
 			DatabaseADDomain:             "EXAMPLE.com",
 			DatabaseADSPN:                "MSSQLSvc/ec2amaz-4kn05du.dbadir.teleportdemo.net:1433",
@@ -160,6 +161,7 @@ func TestMakeDatabaseConfig(t *testing.T) {
 		require.Equal(t, flags.StaticDatabaseURI, databases.Databases[0].URI)
 		require.Equal(t, map[string]string{"env": "prod"}, databases.Databases[0].StaticLabels)
 		require.Equal(t, flags.DatabaseAWSRegion, databases.Databases[0].AWS.Region)
+		require.Equal(t, flags.DatabaseAWSExternalID, databases.Databases[0].AWS.ExternalID)
 		require.Equal(t, flags.DatabaseAWSRedshiftClusterID, databases.Databases[0].AWS.Redshift.ClusterID)
 		require.Equal(t, flags.DatabaseADDomain, databases.Databases[0].AD.Domain)
 		require.Equal(t, flags.DatabaseADSPN, databases.Databases[0].AD.SPN)

lib/config/fileconf.go

@@ -1586,6 +1586,8 @@ type DatabaseAWS struct {
 	MemoryDB DatabaseAWSMemoryDB `yaml:"memorydb"`
 	// AccountID is the AWS account ID.
 	AccountID string `yaml:"account_id,omitempty"`
+	// ExternalID is an optional AWS external ID used to enable assuming an AWS role across accounts.
+	ExternalID string `yaml:"external_id,omitempty"`
 	// RedshiftServerless contains RedshiftServerless specific settings.
 	RedshiftServerless DatabaseAWSRedshiftServerless `yaml:"redshift_serverless"`
 }

lib/defaults/defaults.go

@@ -433,6 +433,8 @@ const (
 	ProtocolCassandra = "cassandra"
 	// ProtocolElasticsearch is the Elasticsearch database protocol.
 	ProtocolElasticsearch = "elasticsearch"
+	// ProtocolDynamoDB is the DynamoDB database protocol.
+	ProtocolDynamoDB = "dynamodb"
 )
 
 // DatabaseProtocols is a list of all supported database protocols.
@@ -446,6 +448,7 @@ var DatabaseProtocols = []string{
 	ProtocolSQLServer,
 	ProtocolCassandra,
 	ProtocolElasticsearch,
+	ProtocolDynamoDB,
 }
 
 // ReadableDatabaseProtocol returns a more human readable string of the
@@ -470,6 +473,8 @@ func ReadableDatabaseProtocol(p string) string {
 		return "Microsoft SQL Server"
 	case ProtocolCassandra:
 		return "Cassandra"
+	case ProtocolDynamoDB:
+		return "DynamoDB"
 	default:
 		// Unknown protocol. Return original string.
 		return p

lib/events/codes.go

@@ -188,6 +188,9 @@ const (
 
 	// DynamoDBRequestCode is the db.session.dynamodb.request event code.
 	DynamoDBRequestCode = "TDY01I"
+	// DynamoDBRequestFailureCode is the db.session.dynamodb.request event failure code.
+	// This is indicates that the database agent http transport failed to round trip the request.
+	DynamoDBRequestFailureCode = "TDY01E"
 
 	// DatabaseCreateCode is the db.create event code.
 	DatabaseCreateCode = "TDB03I"

lib/service/cfg.go

@@ -869,6 +869,8 @@ type DatabaseAWS struct {
 	SecretStore DatabaseAWSSecretStore
 	// AccountID is the AWS account ID.
 	AccountID string
+	// ExternalID is an optional AWS external ID used to enable assuming an AWS role across accounts.
+	ExternalID string
 	// RedshiftServerless contains AWS Redshift Serverless specific settings.
 	RedshiftServerless DatabaseAWSRedshiftServerless
 }
@@ -1029,8 +1031,9 @@ func (d *Database) ToDatabase() (types.Database, error) {
 			ServerVersion: d.MySQL.ServerVersion,
 		},
 		AWS: types.AWS{
-			AccountID: d.AWS.AccountID,
-			Region:    d.AWS.Region,
+			AccountID:  d.AWS.AccountID,
+			ExternalID: d.AWS.ExternalID,
+			Region:     d.AWS.Region,
 			Redshift: types.Redshift{
 				ClusterID: d.AWS.Redshift.ClusterID,
 			},

lib/service/service.go

@@ -4527,15 +4527,14 @@ func (process *TeleportProcess) initApps() {
 		if err != nil {
 			return trace.Wrap(err)
 		}
-
-		proxyGetter := reversetunnel.NewConnectedProxyGetter()
-
 		defer func() {
 			if !shouldSkipCleanup {
 				warnOnErr(asyncEmitter.Close(), log)
 			}
 		}()
 
+		proxyGetter := reversetunnel.NewConnectedProxyGetter()
+
 		appServer, err := app.New(process.ExitContext(), &app.Config{
 			Clock:                process.Config.Clock,
 			DataDir:              process.Config.DataDir,

lib/service/service_test.go

@@ -503,6 +503,7 @@ func TestSetupProxyTLSConfig(t *testing.T) {
 				"teleport-snowflake-ping",
 				"teleport-cassandra-ping",
 				"teleport-elasticsearch-ping",
+				"teleport-dynamodb-ping",
 				"teleport-proxy-ssh",
 				"teleport-reversetunnel",
 				"teleport-auth@",
@@ -515,6 +516,7 @@ func TestSetupProxyTLSConfig(t *testing.T) {
 				"teleport-snowflake",
 				"teleport-cassandra",
 				"teleport-elasticsearch",
+				"teleport-dynamodb",
 			},
 		},
 		{
@@ -529,6 +531,7 @@ func TestSetupProxyTLSConfig(t *testing.T) {
 				"teleport-snowflake-ping",
 				"teleport-cassandra-ping",
 				"teleport-elasticsearch-ping",
+				"teleport-dynamodb-ping",
 				// Ensure http/1.1 has precedence over http2.
 				"http/1.1",
 				"h2",
@@ -544,6 +547,7 @@ func TestSetupProxyTLSConfig(t *testing.T) {
 				"teleport-snowflake",
 				"teleport-cassandra",
 				"teleport-elasticsearch",
+				"teleport-dynamodb",
 			},
 		},
 	}

lib/services/database.go

@@ -166,10 +166,10 @@ func ValidateDatabase(db types.Database) error {
 		if !strings.Contains(db.GetURI(), defaults.SnowflakeURL) {
 			return trace.BadParameter("Snowflake address should contain " + defaults.SnowflakeURL)
 		}
-	} else if db.GetProtocol() == defaults.ProtocolCassandra && db.GetAWS().Region != "" && db.GetAWS().AccountID != "" {
-		// In case of cloud hosted Cassandra doesn't require URI validation.
-	} else if _, _, err := net.SplitHostPort(db.GetURI()); err != nil {
-		return trace.BadParameter("invalid database %q address %q: %v", db.GetName(), db.GetURI(), err)
+	} else if needsURIValidation(db) {
+		if _, _, err := net.SplitHostPort(db.GetURI()); err != nil {
+			return trace.BadParameter("invalid database %q address %q: %v", db.GetName(), db.GetURI(), err)
+		}
 	}
 
 	if db.GetTLS().CACert != "" {
@@ -196,6 +196,17 @@ func ValidateDatabase(db types.Database) error {
 	return nil
 }
 
+// needsURIValidation returns whether a database URI needs to be validated.
+func needsURIValidation(db types.Database) bool {
+	switch db.GetProtocol() {
+	case defaults.ProtocolCassandra, defaults.ProtocolDynamoDB:
+		// cloud hosted Cassandra doesn't require URI validation.
+		return db.GetAWS().Region == "" || db.GetAWS().AccountID == ""
+	default:
+		return true
+	}
+}
+
 // validateMongoDB validates MongoDB URIs with "mongodb" schemes.
 func validateMongoDB(db types.Database) error {
 	connString, err := connstring.ParseAndValidate(db.GetURI())

lib/services/database_test.go

@@ -209,6 +209,17 @@ func TestValidateDatabase(t *testing.T) {
 			},
 			expectError: false,
 		},
+		{
+			inputName: "valid-dynamodb-without-uri",
+			inputSpec: types.DatabaseSpecV3{
+				Protocol: defaults.ProtocolDynamoDB,
+				AWS: types.AWS{
+					Region:    "us-east-1",
+					AccountID: "1234567890",
+				},
+			},
+			expectError: false,
+		},
 	}
 
 	for _, test := range tests {

lib/srv/alpnproxy/common/protocols.go

@@ -53,6 +53,9 @@ const (
 	// ProtocolElasticsearch is TLS ALPN protocol value used to indicate Elasticsearch protocol.
 	ProtocolElasticsearch Protocol = "teleport-elasticsearch"
 
+	// ProtocolDynamoDB is TLS ALPN protocol value used to indicate DynamoDB protocol.
+	ProtocolDynamoDB Protocol = "teleport-dynamodb"
+
 	// ProtocolProxySSH is TLS ALPN protocol value used to indicate Proxy SSH protocol.
 	ProtocolProxySSH Protocol = "teleport-proxy-ssh"
 
@@ -140,6 +143,8 @@ func ToALPNProtocol(dbProtocol string) (Protocol, error) {
 		return ProtocolCassandra, nil
 	case defaults.ProtocolElasticsearch:
 		return ProtocolElasticsearch, nil
+	case defaults.ProtocolDynamoDB:
+		return ProtocolDynamoDB, nil
 	default:
 		return "", trace.NotImplemented("%q protocol is not supported", dbProtocol)
 	}
@@ -158,6 +163,7 @@ func IsDBTLSProtocol(protocol Protocol) bool {
 		ProtocolSnowflake,
 		ProtocolCassandra,
 		ProtocolElasticsearch,
+		ProtocolDynamoDB,
 	}
 
 	return slices.Contains(
@@ -176,6 +182,7 @@ var DatabaseProtocols = []Protocol{
 	ProtocolSnowflake,
 	ProtocolCassandra,
 	ProtocolElasticsearch,
+	ProtocolDynamoDB,
 }
 
 // ProtocolsWithPingSupport is the list of protocols that Ping connection is

lib/srv/db/access_test.go

@@ -65,6 +65,7 @@ import (
 	alpncommon "github.com/gravitational/teleport/lib/srv/alpnproxy/common"
 	"github.com/gravitational/teleport/lib/srv/db/cassandra"
 	"github.com/gravitational/teleport/lib/srv/db/common"
+	"github.com/gravitational/teleport/lib/srv/db/dynamodb"
 	"github.com/gravitational/teleport/lib/srv/db/elasticsearch"
 	"github.com/gravitational/teleport/lib/srv/db/mongodb"
 	"github.com/gravitational/teleport/lib/srv/db/mysql"
@@ -84,6 +85,7 @@ func TestMain(m *testing.M) {
 	registerTestSnowflakeEngine()
 	registerTestElasticsearchEngine()
 	registerTestSQLServerEngine()
+	registerTestDynamoDBEngine()
 	os.Exit(m.Run())
 }
 
@@ -1240,6 +1242,8 @@ type testContext struct {
 	cassandra map[string]testCassandra
 	// elasticsearch is a collection of Elasticsearch databases the test uses.
 	elasticsearch map[string]testElasticsearch
+	// dynamodb is a collection of DynamoDB databases the test uses.
+	dynamodb map[string]testDynamoDB
 	// clock to override clock in tests.
 	clock clockwork.FakeClock
 }
@@ -1306,6 +1310,14 @@ type testElasticsearch struct {
 	resource types.Database
 }
 
+// testDynamoDB represents a single proxied DynamoDB database.
+type testDynamoDB struct {
+	// db is the test Dynamodb database server.
+	db *dynamodb.TestServer
+	// resource is the resource representing this DynamoDB database.
+	resource types.Database
+}
+
 // startProxy starts all proxy services required to handle connections.
 func (c *testContext) startProxy() {
 	// Start multiplexer.
@@ -1730,6 +1742,34 @@ func (c *testContext) elasticsearchClient(ctx context.Context, teleportUser, dbS
 	return db, proxy, nil
 }
 
+// dynamodbClient returns a DynamoDB test client.
+func (c *testContext) dynamodbClient(ctx context.Context, teleportUser, dbService, dbUser string) (*dynamodb.Client, *alpnproxy.LocalProxy, error) {
+	route := tlsca.RouteToDatabase{
+		ServiceName: dbService,
+		Protocol:    defaults.ProtocolDynamoDB,
+		Username:    dbUser,
+	}
+
+	proxy, err := c.startLocalALPNProxy(ctx, c.webListener.Addr().String(), teleportUser, route)
+	if err != nil {
+		return nil, nil, trace.Wrap(err)
+	}
+
+	db, err := dynamodb.MakeTestClient(ctx, common.TestClientConfig{
+		AuthClient:      c.authClient,
+		AuthServer:      c.authServer,
+		Address:         proxy.GetAddr(),
+		Cluster:         c.clusterName,
+		Username:        teleportUser,
+		RouteToDatabase: route,
+	})
+	if err != nil {
+		return nil, nil, trace.Wrap(err)
+	}
+
+	return db, proxy, nil
+}
+
 // createUserAndRole creates Teleport user and role with specified names
 // and allowed database users/names properties.
 func (c *testContext) createUserAndRole(ctx context.Context, t *testing.T, userName, roleName string, dbUsers, dbNames []string) (types.User, types.Role) {
@@ -1792,6 +1832,7 @@ func setupTestContext(ctx context.Context, t *testing.T, withDatabases ...withDa
 		snowflake:     make(map[string]testSnowflake),
 		elasticsearch: make(map[string]testElasticsearch),
 		cassandra:     make(map[string]testCassandra),
+		dynamodb:      make(map[string]testDynamoDB),
 		clock:         clockwork.NewFakeClockAt(time.Now()),
 	}
 	t.Cleanup(func() { testCtx.Close() })

lib/srv/db/ca.go

@@ -75,6 +75,7 @@ func (s *Server) initCACert(ctx context.Context, database types.Database) error
 		types.DatabaseTypeElastiCache,
 		types.DatabaseTypeMemoryDB,
 		types.DatabaseTypeAWSKeyspaces,
+		types.DatabaseTypeDynamoDB,
 		types.DatabaseTypeCloudSQL,
 		types.DatabaseTypeAzure:
 
@@ -163,7 +164,8 @@ func (s *Server) getCACertPaths(database types.Database) ([]string, error) {
 	//
 	// AWS MemoryDB uses same CA as ElastiCache.
 	case types.DatabaseTypeElastiCache,
-		types.DatabaseTypeMemoryDB:
+		types.DatabaseTypeMemoryDB,
+		types.DatabaseTypeDynamoDB:
 		return []string{filepath.Join(s.cfg.DataDir, filepath.Base(amazonRootCA1URL))}, nil
 
 	// Each Cloud SQL instance has its own CA.
@@ -286,7 +288,8 @@ func (d *realDownloader) Download(ctx context.Context, database types.Database,
 		types.DatabaseTypeRedshiftServerless:
 		return d.downloadFromURL(redshiftCAURLForDatabase(database))
 	case types.DatabaseTypeElastiCache,
-		types.DatabaseTypeMemoryDB:
+		types.DatabaseTypeMemoryDB,
+		types.DatabaseTypeDynamoDB:
 		return d.downloadFromURL(amazonRootCA1URL)
 	case types.DatabaseTypeCloudSQL:
 		return d.downloadForCloudSQL(ctx, database)

lib/srv/db/cassandra/handshake.go

@@ -17,10 +17,6 @@ limitations under the License.
 package cassandra
 
 import (
-	"fmt"
-	"strings"
-
-	"github.com/aws/aws-sdk-go/aws/arn"
 	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
 	awssession "github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sigv4-auth-cassandra-gocql-driver-plugin/sigv4"
@@ -30,9 +26,9 @@ import (
 	"github.com/gocql/gocql"
 	"github.com/gravitational/trace"
 
-	awsutils "github.com/gravitational/teleport/api/utils/aws"
 	"github.com/gravitational/teleport/lib/srv/db/cassandra/protocol"
 	"github.com/gravitational/teleport/lib/srv/db/common"
+	awsutils "github.com/gravitational/teleport/lib/utils/aws"
 )
 
 const (
@@ -201,35 +197,20 @@ func (a *authAWSSigV4Auth) getSigV4Authenticator(username string) (gocql.Authent
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	cred, err := stscreds.NewCredentials(session, a.buildRoleARN(username)).Get()
+	region := a.ses.Database.GetAWS().Region
+	accountID := a.ses.Database.GetAWS().AccountID
+	cred, err := stscreds.NewCredentials(session, awsutils.BuildRoleARN(username, region, accountID)).Get()
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 	auth := sigv4.NewAwsAuthenticator()
-	auth.Region = a.ses.Database.GetAWS().Region
+	auth.Region = region
 	auth.AccessKeyId = cred.AccessKeyID
 	auth.SessionToken = cred.SessionToken
 	auth.SecretAccessKey = cred.SecretAccessKey
 	return auth, nil
 }
 
-func (a *authAWSSigV4Auth) buildRoleARN(username string) string {
-	if arn.IsARN(username) {
-		return username
-	}
-	resource := username
-	if !strings.Contains(resource, "/") {
-		resource = fmt.Sprintf("role/%s", username)
-	}
-
-	return arn.ARN{
-		Partition: awsutils.GetPartitionFromRegion(a.ses.Database.GetAWS().Region),
-		Service:   "iam",
-		AccountID: a.ses.Database.GetAWS().AccountID,
-		Resource:  resource,
-	}.String()
-}
-
 func (a *authAWSSigV4Auth) initPasswordAuth(clientConn *protocol.Conn, req *protocol.Packet) (*protocol.Packet, error) {
 	authMsg := frame.NewFrame(
 		req.Header().Version,

lib/srv/db/common/auth.go

@@ -623,7 +623,10 @@ func setupTLSConfigServerName(tlsConfig *tls.Config, sessionCtx *Session) error
 		// of replica set the driver may dial multiple servers and will set
 		// ServerName itself.
 		return nil
-
+	case defaults.ProtocolDynamoDB:
+		// Don't set the server name for DynamoDB - the engine may dial different endpoints
+		// based on the client request and will set ServerName itself.
+		return nil
 	case defaults.ProtocolRedis:
 		// Azure Redis servers always serve the certificates with the proper
 		// hostnames. However, OSS cluster mode may redirect to an IP address,

lib/srv/db/common/role/role.go

@@ -60,6 +60,11 @@ func DatabaseRoleMatchers(dbProtocol string, user, database string) services.Rol
 		return services.RoleMatchers{
 			&services.DatabaseUserMatcher{User: user},
 		}
+	case defaults.ProtocolDynamoDB:
+		// DynamoDB integration doesn't support schema access control.
+		return services.RoleMatchers{
+			&services.DatabaseUserMatcher{User: user},
+		}
 	default:
 		return services.RoleMatchers{
 			&services.DatabaseUserMatcher{User: user},

lib/srv/db/dynamodb/engine.go

@@ -0,0 +1,406 @@
+/*
+
+ Copyright 2022 Gravitational, Inc.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+*/
+
+package dynamodb
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws/endpoints"
+	"github.com/aws/aws-sdk-go/service/dax"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
+	"github.com/gravitational/trace"
+
+	apievents "github.com/gravitational/teleport/api/types/events"
+	apiaws "github.com/gravitational/teleport/api/utils/aws"
+	"github.com/gravitational/teleport/lib/defaults"
+	"github.com/gravitational/teleport/lib/events"
+	"github.com/gravitational/teleport/lib/srv/db/common"
+	"github.com/gravitational/teleport/lib/srv/db/common/role"
+	"github.com/gravitational/teleport/lib/utils"
+	libaws "github.com/gravitational/teleport/lib/utils/aws"
+)
+
+// NewEngine create new DynamoDB engine.
+func NewEngine(ec common.EngineConfig) common.Engine {
+	return &Engine{
+		EngineConfig:  ec,
+		RoundTrippers: make(map[string]http.RoundTripper),
+	}
+}
+
+// Engine handles connections from DynamoDB clients coming from Teleport
+// proxy over reverse tunnel.
+type Engine struct {
+	// signingSvc will be used by the engine to provide the AWS sigv4 authorization header
+	// required by AWS for request validation: https://docs.aws.amazon.com/general/latest/gr/signing-elements.html
+	signingSvc *libaws.SigningService
+	// EngineConfig is the common database engine configuration.
+	common.EngineConfig
+	// clientConn is a client connection.
+	clientConn net.Conn
+	// sessionCtx is current session context.
+	sessionCtx *common.Session
+	// RoundTrippers is a cache of RoundTrippers, mapped by service endpoint.
+	// It is not guarded by a mutex, since requests are processed serially.
+	RoundTrippers map[string]http.RoundTripper
+	// GetSigningCredsFn allows to set the function responsible for obtaining STS credentials.
+	// Used in tests to set static AWS credentials and skip API call.
+	GetSigningCredsFn libaws.GetSigningCredentialsFunc
+}
+
+var _ common.Engine = (*Engine)(nil)
+
+func (e *Engine) InitializeConnection(clientConn net.Conn, sessionCtx *common.Session) error {
+	e.clientConn = clientConn
+	e.sessionCtx = sessionCtx
+	awsSession, err := e.CloudClients.GetAWSSession(sessionCtx.Database.GetAWS().Region)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	svc, err := libaws.NewSigningService(libaws.SigningServiceConfig{
+		Clock:                 e.Clock,
+		Session:               awsSession,
+		GetSigningCredentials: e.GetSigningCredsFn,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	e.signingSvc = svc
+	return nil
+}
+
+// jsonErr is used to marshal a JSON error response as the AWS CLI expects for errors.
+// https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.Errors.html#Programming.Errors.Components
+type jsonErr struct {
+	Code    string `json:"__type"`
+	Message string `json:"message"`
+}
+
+// SendError sends an error to DynamoDB client.
+func (e *Engine) SendError(err error) {
+	if e.clientConn == nil || err == nil || utils.IsOKNetworkError(err) {
+		return
+	}
+	e.Log.WithError(err).Error("DynamoDB connection error")
+
+	// try to convert to a trace err if we can.
+	code := trace.ErrorToCode(err)
+	body, err := json.Marshal(jsonErr{
+		Code:    strconv.Itoa(code),
+		Message: err.Error(),
+	})
+	if err != nil {
+		e.Log.WithError(err).Error("failed to marshal error response")
+		return
+	}
+	response := &http.Response{
+		Status:     http.StatusText(code),
+		StatusCode: code,
+		Proto:      "HTTP/1.1",
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		// ContentLength is the authoritative value in the response,
+		// no need to also add the "Content-Length" header (source: go doc http.Response.Header).
+		ContentLength: int64(len(body)),
+		Header: map[string][]string{
+			"Content-Type": {"application/json"},
+		},
+		Body: io.NopCloser(bytes.NewBuffer(body)),
+	}
+
+	if err := response.Write(e.clientConn); err != nil {
+		e.Log.WithError(err).Error("failed to send error response to DynamoDB client")
+		return
+	}
+}
+
+// HandleConnection authorizes the incoming client connection, connects to the
+// target DynamoDB server and starts proxying requests between client/server.
+func (e *Engine) HandleConnection(ctx context.Context, _ *common.Session) error {
+	err := e.checkAccess(ctx, e.sessionCtx)
+	e.Audit.OnSessionStart(e.Context, e.sessionCtx, err)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	defer e.Audit.OnSessionEnd(e.Context, e.sessionCtx)
+
+	clientConnReader := bufio.NewReader(e.clientConn)
+	for {
+		req, err := http.ReadRequest(clientConnReader)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		if err := e.process(ctx, req); err != nil {
+			return trace.Wrap(err)
+		}
+	}
+}
+
+// process reads request from connected dynamodb client, processes the requests/responses and sends data back
+// to the client.
+func (e *Engine) process(ctx context.Context, req *http.Request) (err error) {
+	if req.Body != nil {
+		// make sure we close the incoming request's body. ignore any close error.
+		defer req.Body.Close()
+	}
+
+	var responseStatusCode uint32
+	re, err := e.resolveEndpoint(req)
+	if err != nil {
+		// special error case where we couldn't resolve the endpoint, just emit using the configured URI.
+		e.emitAuditEvent(req, e.sessionCtx.Database.GetURI(), responseStatusCode, err)
+		return trace.Wrap(err)
+	}
+
+	// emit an audit event regardless of failure, but using the resolved endpoint.
+	defer func() {
+		e.emitAuditEvent(req, re.URL, responseStatusCode, err)
+	}()
+
+	// try to read, close, and replace the incoming request body.
+	body, err := libaws.GetAndReplaceReqBody(req)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	roundTripper, err := e.getRoundTripper(ctx, re.URL)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	// rewrite the request URL and headers before signing it.
+	outReq, err := rewriteRequest(ctx, req, re, body)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	roleArn := libaws.BuildRoleARN(e.sessionCtx.DatabaseUser, re.SigningRegion, e.sessionCtx.Database.GetAWS().AccountID)
+	signedReq, err := e.signingSvc.SignRequest(e.Context, outReq,
+		&libaws.SigningCtx{
+			SigningName:   re.SigningName,
+			SigningRegion: re.SigningRegion,
+			Expiry:        e.sessionCtx.Identity.Expires,
+			SessionName:   e.sessionCtx.Identity.Username,
+			AWSRoleArn:    roleArn,
+			AWSExternalID: e.sessionCtx.Database.GetAWS().ExternalID,
+		})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	// Send the request.
+	resp, err := roundTripper.RoundTrip(signedReq)
+	if err != nil {
+		// convert the error from round tripping to try to get a trace error.
+		err = common.ConvertConnectError(err, e.sessionCtx)
+		return trace.Wrap(err)
+	}
+	defer resp.Body.Close()
+	responseStatusCode = uint32(resp.StatusCode)
+
+	return trace.Wrap(e.sendResponse(resp))
+}
+
+// sendResponse sends the response back to the DynamoDB client.
+func (e *Engine) sendResponse(resp *http.Response) error {
+	return trace.Wrap(resp.Write(e.clientConn))
+}
+
+// emitAuditEvent writes the request and response status code to the audit stream.
+func (e *Engine) emitAuditEvent(req *http.Request, uri string, statusCode uint32, err error) {
+	var eventCode string
+	if err == nil && statusCode != 0 {
+		eventCode = events.DynamoDBRequestCode
+	} else {
+		eventCode = events.DynamoDBRequestFailureCode
+	}
+	// Try to read the body and JSON unmarshal it.
+	// If this fails, we still want to emit the rest of the event info; the request event Body is nullable,
+	// so it's ok if body is nil here.
+	body, err := libaws.UnmarshalRequestBody(req)
+	if err != nil {
+		e.Log.WithError(err).Warn("Failed to read request body as JSON, omitting the body from the audit event.")
+	}
+	// get the API target from the request header, according to the API request format documentation:
+	// https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html#Programming.LowLevelAPI.RequestFormat
+	target := req.Header.Get(libaws.AmzTargetHeader)
+	event := &apievents.DynamoDBRequest{
+		Metadata: apievents.Metadata{
+			Type: events.DatabaseSessionDynamoDBRequestEvent,
+			Code: eventCode,
+		},
+		UserMetadata:    e.sessionCtx.Identity.GetUserMetadata(),
+		SessionMetadata: common.MakeSessionMetadata(e.sessionCtx),
+		DatabaseMetadata: apievents.DatabaseMetadata{
+			DatabaseService:  e.sessionCtx.Database.GetName(),
+			DatabaseProtocol: e.sessionCtx.Database.GetProtocol(),
+			DatabaseURI:      uri,
+			DatabaseName:     e.sessionCtx.DatabaseName,
+			DatabaseUser:     e.sessionCtx.DatabaseUser,
+		},
+		StatusCode: statusCode,
+		Path:       req.URL.Path,
+		RawQuery:   req.URL.RawQuery,
+		Method:     req.Method,
+		Target:     target,
+		Body:       body,
+	}
+	e.Audit.EmitEvent(e.Context, event)
+}
+
+// checkAccess does authorization check for DynamoDB connection about
+// to be established.
+func (e *Engine) checkAccess(ctx context.Context, sessionCtx *common.Session) error {
+	ap, err := e.Auth.GetAuthPreference(ctx)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	mfaParams := sessionCtx.MFAParams(ap.GetRequireMFAType())
+	dbRoleMatchers := role.DatabaseRoleMatchers(
+		sessionCtx.Database.GetProtocol(),
+		sessionCtx.DatabaseUser,
+		sessionCtx.DatabaseName,
+	)
+	err = sessionCtx.Checker.CheckAccess(
+		sessionCtx.Database,
+		mfaParams,
+		dbRoleMatchers...,
+	)
+	return trace.Wrap(err)
+}
+
+// getRoundTripper makes an HTTP round tripper with TLS config based on the given URL.
+func (e *Engine) getRoundTripper(ctx context.Context, URL string) (http.RoundTripper, error) {
+	if rt, ok := e.RoundTrippers[URL]; ok {
+		return rt, nil
+	}
+	tlsConfig, err := e.Auth.GetTLSConfig(ctx, e.sessionCtx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	// We need to set the ServerName here because the AWS endpoint service prefix is not known in advance,
+	// and the TLS config we got does not set it.
+	host, err := getURLHostname(URL)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	tlsConfig.ServerName = host
+
+	out, err := defaults.Transport()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	out.TLSClientConfig = tlsConfig
+	e.RoundTrippers[URL] = out
+	return out, nil
+}
+
+// resolveEndpoint returns a resolved endpoint for either the configured URI or the AWS target service and region.
+func (e *Engine) resolveEndpoint(req *http.Request) (*endpoints.ResolvedEndpoint, error) {
+	endpointID, err := extractEndpointID(req)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	opts := func(opts *endpoints.Options) {
+		opts.ResolveUnknownService = true
+	}
+	re, err := endpoints.DefaultResolver().EndpointFor(endpointID, e.sessionCtx.Database.GetAWS().Region, opts)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	uri := e.sessionCtx.Database.GetURI()
+	if uri != "" && uri != apiaws.DynamoDBURIForRegion(e.sessionCtx.Database.GetAWS().Region) {
+		// override the resolved endpoint URL with the user-configured URI.
+		re.URL = uri
+	}
+	if !strings.Contains(re.URL, "://") {
+		re.URL = "https://" + re.URL
+	}
+	return &re, nil
+}
+
+// rewriteRequest clones a request, modifies the clone to rewrite its URL, and returns the modified request clone.
+func rewriteRequest(ctx context.Context, r *http.Request, re *endpoints.ResolvedEndpoint, body []byte) (*http.Request, error) {
+	resolvedURL, err := url.Parse(re.URL)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	reqCopy := r.Clone(ctx)
+	// set url and host header to match the resolved endpoint.
+	reqCopy.URL = resolvedURL
+	reqCopy.Host = resolvedURL.Host
+	if body == nil {
+		// no body is fine, skip copying it.
+		return reqCopy, nil
+	}
+
+	// copy request body
+	reqCopy.Body = io.NopCloser(bytes.NewReader(body))
+	return reqCopy, nil
+}
+
+// extractEndpointID extracts the AWS endpoint ID from the request header X-Amz-Target.
+func extractEndpointID(req *http.Request) (string, error) {
+	target := req.Header.Get(libaws.AmzTargetHeader)
+	if target == "" {
+		return "", trace.BadParameter("missing %q header in http request", libaws.AmzTargetHeader)
+	}
+	endpointID, err := endpointIDForTarget(target)
+	return endpointID, trace.Wrap(err)
+}
+
+// endpointIDForTarget converts a target operation into the appropriate the AWS endpoint ID.
+// Target looks like one of DynamoDB_$version.$operation, DynamoDBStreams_$version.$operation, AmazonDAX$version.$operation,
+// for example: DynamoDBStreams_20120810.ListStreams
+// See X-Amz-Target: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html
+func endpointIDForTarget(target string) (string, error) {
+	t := strings.ToLower(target)
+	switch {
+	case strings.HasPrefix(t, "dynamodbstreams"):
+		return dynamodbstreams.EndpointsID, nil
+	case strings.HasPrefix(t, "dynamodb"):
+		return dynamodb.EndpointsID, nil
+	case strings.HasPrefix(t, "amazondax"):
+		return dax.EndpointsID, nil
+	default:
+		return "", trace.BadParameter("DynamoDB API target %q is not recognized", target)
+	}
+}
+
+// getURLHostname parses a URL to extract its hostname.
+func getURLHostname(uri string) (string, error) {
+	if !strings.Contains(uri, "://") {
+		uri = "schema://" + uri
+	}
+	parsed, err := url.Parse(uri)
+	if err != nil {
+		return "", trace.Wrap(err)
+	}
+	return parsed.Hostname(), nil
+}

lib/srv/db/dynamodb/engine_test.go

@@ -0,0 +1,148 @@
+/*
+
+ Copyright 2022 Gravitational, Inc.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+*/
+
+package dynamodb
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+
+	"github.com/gravitational/teleport/api/types"
+	apiaws "github.com/gravitational/teleport/api/utils/aws"
+	"github.com/gravitational/teleport/lib/srv/db/common"
+	libaws "github.com/gravitational/teleport/lib/utils/aws"
+)
+
+func TestResolveEndpoint(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		desc            string
+		target          string // from X-Amz-Target in requests
+		region          string
+		wantEndpointID  string
+		wantSigningName string
+		wantURL         string
+		wantErrMsg      string
+	}{
+		{
+			desc:            "dynamodb target in us west",
+			target:          "DynamoDB_20120810.Scan",
+			region:          "us-west-1",
+			wantEndpointID:  "dynamodb",
+			wantSigningName: "dynamodb",
+			wantURL:         "https://dynamodb.us-west-1.amazonaws.com",
+		},
+		{
+			desc:            "dynamodb target in china",
+			target:          "DynamoDB_20120810.Scan",
+			region:          "cn-north-1",
+			wantEndpointID:  "dynamodb",
+			wantSigningName: "dynamodb",
+			wantURL:         "https://dynamodb.cn-north-1.amazonaws.com.cn",
+		},
+		{
+			desc:            "dynamodb streams target in us west",
+			target:          "DynamoDBStreams_20120810.ListStreams",
+			region:          "us-west-1",
+			wantEndpointID:  "streams.dynamodb",
+			wantSigningName: "dynamodb",
+			wantURL:         "https://streams.dynamodb.us-west-1.amazonaws.com",
+		},
+		{
+			desc:            "dynamodb streams target in china",
+			target:          "DynamoDBStreams_20120810.ListStreams",
+			region:          "cn-north-1",
+			wantEndpointID:  "streams.dynamodb",
+			wantSigningName: "dynamodb",
+			wantURL:         "https://streams.dynamodb.cn-north-1.amazonaws.com.cn",
+		},
+		{
+			desc:            "dax target in us west",
+			target:          "AmazonDAXV3.ListTags",
+			region:          "us-west-1",
+			wantEndpointID:  "dax",
+			wantSigningName: "dax",
+			wantURL:         "https://dax.us-west-1.amazonaws.com",
+		},
+		{
+			desc:            "dax target in china",
+			target:          "AmazonDAXV3.ListTags",
+			region:          "cn-north-1",
+			wantEndpointID:  "dax",
+			wantSigningName: "dax",
+			wantURL:         "https://dax.cn-north-1.amazonaws.com.cn",
+		},
+		{
+			desc:       "unrecognizable target",
+			target:     "DDB.Scan",
+			wantErrMsg: "is not recognized",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.desc, func(t *testing.T) {
+			t.Parallel()
+			// mock a request.
+			req := &http.Request{Header: make(http.Header)}
+			req.Header.Set(libaws.AmzTargetHeader, tt.target)
+
+			// check that the correct endpoint ID is extracted.
+			endpointID, err := extractEndpointID(req)
+			if tt.wantErrMsg != "" {
+				require.Error(t, err)
+				require.ErrorContains(t, err, tt.wantErrMsg)
+				return
+			}
+			require.Equal(t, tt.wantEndpointID, endpointID)
+
+			// check that the engine resolves the correct URL.
+			db := &types.DatabaseV3{
+				Spec: types.DatabaseSpecV3{
+					URI: apiaws.DynamoDBURIForRegion(tt.region),
+					AWS: types.AWS{
+						Region:    tt.region,
+						AccountID: "12345",
+					},
+				},
+			}
+			engine := &Engine{
+				EngineConfig: common.EngineConfig{
+					Log: logrus.StandardLogger(),
+				},
+				sessionCtx: &common.Session{
+					Database: db,
+				},
+			}
+			re, err := engine.resolveEndpoint(req)
+			require.NoError(t, err)
+			require.Equal(t, tt.wantURL, re.URL)
+			require.Equal(t, tt.wantSigningName, re.SigningName)
+
+			// now use a custom URI and check that it overrides the resolved URL.
+			db.Spec.URI = "foo.com"
+			re, err = engine.resolveEndpoint(req)
+			require.NoError(t, err)
+			require.Equal(t, "https://foo.com", re.URL)
+			require.Equal(t, tt.wantSigningName, re.SigningName)
+		})
+	}
+}

lib/srv/db/dynamodb/test.go

@@ -0,0 +1,171 @@
+/*
+
+ Copyright 2022 Gravitational, Inc.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+*/
+
+package dynamodb
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/gravitational/trace"
+	"github.com/sirupsen/logrus"
+
+	"github.com/gravitational/teleport/lib/defaults"
+	"github.com/gravitational/teleport/lib/srv/db/common"
+	awsutils "github.com/gravitational/teleport/lib/utils/aws"
+)
+
+// Client alias for easier use.
+type Client = dynamodb.DynamoDB
+
+// ClientOptionsParams is a struct for client configuration options.
+type ClientOptionsParams struct {
+	Username string
+}
+
+// ClientOptions allows setting test client options.
+type ClientOptions func(*ClientOptionsParams)
+
+// MakeTestClient returns DynamoDB client connection according to the provided
+// parameters.
+func MakeTestClient(_ context.Context, config common.TestClientConfig, opts ...ClientOptions) (*Client, error) {
+	provider := session.Must(session.NewSession(&aws.Config{
+		Credentials: credentials.NewCredentials(&credentials.StaticProvider{Value: credentials.Value{
+			AccessKeyID:     "fakeClientKeyID",
+			SecretAccessKey: "fakeClientSecret",
+		}}),
+		Region: aws.String("local"),
+	}))
+	dynamoClient := dynamodb.New(provider, &aws.Config{
+		Endpoint:   aws.String("http://" + config.Address),
+		MaxRetries: aws.Int(0), // disable automatic retries in tests
+		HTTPClient: &http.Client{Timeout: 5 * time.Second},
+	})
+	return dynamoClient, nil
+}
+
+// TestServerOption allows setting test server options.
+type TestServerOption func(*TestServer)
+
+// TestServer is a DynamoDB test server that mocks AWS signature checking and API.
+type TestServer struct {
+	cfg    common.TestServerConfig
+	log    logrus.FieldLogger
+	port   string
+	server *httptest.Server
+
+	// mu is needed to guard starting/closing the server.
+	mu sync.Mutex
+}
+
+// NewTestServer returns a new instance of a test DynamoDB server.
+func NewTestServer(config common.TestServerConfig, opts ...TestServerOption) (*TestServer, error) {
+	err := config.CheckAndSetDefaults()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	log := logrus.WithFields(logrus.Fields{
+		trace.Component: defaults.ProtocolDynamoDB,
+		"name":          config.Name,
+	})
+	tlsConfig, err := common.MakeTestServerTLSConfig(config)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	tlsConfig.InsecureSkipVerify = true // DynamoDB verifies requests with AWS sig v4, not mTLS.
+
+	port, err := config.Port()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		err := awsutils.VerifyAWSSignature(r, credentials.NewStaticCredentials("AKIDl", "SECRET", "SESSION"))
+		if err != nil {
+			code := trace.ErrorToCode(err)
+			body, _ := json.Marshal(jsonErr{
+				Code:    strconv.Itoa(code),
+				Message: err.Error(),
+			})
+			http.Error(w, string(body), code)
+			return
+		}
+		w.Header().Set("Content-Type", awsutils.AmzJSON1_1)
+		w.Header().Set("Content-Length", strconv.Itoa(len(testListTablesResponse)))
+		w.Write([]byte(testListTablesResponse))
+	})
+
+	server := &TestServer{
+		cfg:  config,
+		log:  log,
+		port: port,
+		server: &httptest.Server{
+			Listener: config.Listener,
+			Config:   &http.Server{Handler: mux},
+			TLS:      tlsConfig,
+		},
+	}
+
+	for _, opt := range opts {
+		opt(server)
+	}
+
+	return server, nil
+}
+
+// Serve starts serving client connections.
+func (s *TestServer) Serve() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.server.StartTLS()
+	return nil
+}
+
+// Close closes the server.
+func (s *TestServer) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.server.Close()
+	return nil
+}
+
+func (s *TestServer) Port() string {
+	return s.port
+}
+
+const (
+	// testListTablesResponse is a default successful ListTables JSON response.
+	testListTablesResponse = `
+{
+    "TableNames": [
+        "table-one",
+        "table-two"
+    ]
+}`
+)

lib/srv/db/dynamodb_test.go

@@ -0,0 +1,241 @@
+/*
+Copyright 2022 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package db
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	awsdynamodb "github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/stretchr/testify/require"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/defaults"
+	libevents "github.com/gravitational/teleport/lib/events"
+	"github.com/gravitational/teleport/lib/srv/db/common"
+	"github.com/gravitational/teleport/lib/srv/db/dynamodb"
+)
+
+func registerTestDynamoDBEngine() {
+	// Override DynamoDB engine that is used normally with the test one
+	// with custom HTTP client.
+	common.RegisterEngine(newTestDynamoDBEngine, defaults.ProtocolDynamoDB)
+}
+
+func newTestDynamoDBEngine(ec common.EngineConfig) common.Engine {
+	return &dynamodb.Engine{
+		EngineConfig:  ec,
+		RoundTrippers: make(map[string]http.RoundTripper),
+		// inject mock AWS credentials.
+		GetSigningCredsFn: staticAWSCredentials,
+	}
+}
+
+func staticAWSCredentials(client.ConfigProvider, time.Time, string, string, string) *credentials.Credentials {
+	return credentials.NewStaticCredentials("AKIDl", "SECRET", "SESSION")
+}
+
+func TestAccessDynamoDB(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	mockTables := []string{"table-one", "table-two"}
+	testCtx := setupTestContext(ctx, t,
+		withDynamoDB("DynamoDB"))
+	go testCtx.startHandlingConnections()
+
+	tests := []struct {
+		desc         string
+		user         string
+		role         string
+		allowDbUsers []string
+		dbUser       string
+		wantErrMsg   string
+	}{
+		{
+			desc:         "has access to all database names and users",
+			user:         "alice",
+			role:         "admin",
+			allowDbUsers: []string{types.Wildcard},
+			dbUser:       "alice",
+		},
+		{
+			desc:         "access allowed to specific user",
+			user:         "alice",
+			role:         "admin",
+			allowDbUsers: []string{"alice"},
+			dbUser:       "alice",
+		},
+		{
+			desc:         "has access to nothing",
+			user:         "alice",
+			role:         "admin",
+			allowDbUsers: []string{},
+			dbUser:       "alice",
+			wantErrMsg:   "access to db denied",
+		},
+		{
+			desc:         "no access to users",
+			user:         "alice",
+			role:         "admin",
+			allowDbUsers: []string{},
+			dbUser:       "alice",
+			wantErrMsg:   "access to db denied",
+		},
+		{
+			desc:         "access denied to specific user",
+			user:         "alice",
+			role:         "admin",
+			allowDbUsers: []string{"alice"},
+			dbUser:       "bob",
+			wantErrMsg:   "access to db denied",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			// Create user/role with the requested permissions.
+			testCtx.createUserAndRole(ctx, t, test.user, test.role, test.allowDbUsers, []string{} /*allow DB names*/)
+
+			// Try to connect to the database as this user.
+			clt, lp, err := testCtx.dynamodbClient(ctx, test.user, "DynamoDB", test.dbUser)
+			t.Cleanup(func() {
+				if lp != nil {
+					lp.Close()
+				}
+			})
+			require.NoError(t, err)
+
+			// Execute a dynamodb query.
+			out, err := clt.ListTables(&awsdynamodb.ListTablesInput{})
+			if test.wantErrMsg != "" {
+				require.Error(t, err)
+				require.ErrorContains(t, err, test.wantErrMsg)
+				return
+			}
+			require.NoError(t, err)
+			require.ElementsMatch(t, mockTables, aws.StringValueSlice(out.TableNames))
+		})
+	}
+}
+
+func TestAuditDynamoDB(t *testing.T) {
+	ctx := context.Background()
+	testCtx := setupTestContext(ctx, t,
+		withDynamoDB("DynamoDB"))
+	go testCtx.startHandlingConnections()
+
+	testCtx.createUserAndRole(ctx, t, "alice", "admin", []string{"admin"}, []string{types.Wildcard})
+
+	clientCtx, cancel := context.WithCancel(ctx)
+	t.Run("access denied", func(t *testing.T) {
+		// Try to connect to the database as this user.
+		clt, lp, err := testCtx.dynamodbClient(clientCtx, "alice", "DynamoDB", "notadmin")
+		t.Cleanup(func() {
+			if lp != nil {
+				lp.Close()
+			}
+		})
+		require.NoError(t, err)
+
+		// Execute a dynamodb query.
+		_, err = clt.ListTables(&awsdynamodb.ListTablesInput{})
+		require.Error(t, err)
+		require.ErrorContains(t, err, "access to db denied")
+		requireEvent(t, testCtx, libevents.DatabaseSessionStartFailureCode)
+	})
+
+	// HTTP request should trigger successful session start/end events and emit an audit event for the request.
+	clt, lp, err := testCtx.dynamodbClient(clientCtx, "alice", "DynamoDB", "admin")
+	t.Cleanup(func() {
+		cancel()
+		if lp != nil {
+			lp.Close()
+		}
+	})
+	require.NoError(t, err)
+
+	t.Run("session starts and emits a request event", func(t *testing.T) {
+		_, err := clt.ListTables(&awsdynamodb.ListTablesInput{})
+		require.NoError(t, err)
+		requireEvent(t, testCtx, libevents.DatabaseSessionStartCode)
+		requireEvent(t, testCtx, libevents.DynamoDBRequestCode)
+	})
+
+	t.Run("session ends when client closes the connection", func(t *testing.T) {
+		clt.Config.HTTPClient.CloseIdleConnections()
+		requireEvent(t, testCtx, libevents.DatabaseSessionEndCode)
+	})
+
+	t.Run("session ends when local proxy closes the connection", func(t *testing.T) {
+		// closing local proxy and canceling the context used to start it should trigger session end event.
+		// without this cancel, the session will not end until the smaller of client_idle_timeout or the testCtx closes.
+		_, err := clt.ListTables(&awsdynamodb.ListTablesInput{})
+		require.NoError(t, err)
+		requireEvent(t, testCtx, libevents.DatabaseSessionStartCode)
+		requireEvent(t, testCtx, libevents.DynamoDBRequestCode)
+		cancel()
+		lp.Close()
+		requireEvent(t, testCtx, libevents.DatabaseSessionEndCode)
+	})
+}
+
+func withDynamoDB(name string, opts ...dynamodb.TestServerOption) withDatabaseOption {
+	return func(t *testing.T, _ context.Context, testCtx *testContext) types.Database {
+		config := common.TestServerConfig{
+			Name:       name,
+			AuthClient: testCtx.authClient,
+			ClientAuth: tls.NoClientCert, // DynamoDB is cloud hosted and does not use mTLS.
+		}
+		server, err := dynamodb.NewTestServer(config, opts...)
+		require.NoError(t, err)
+		go server.Serve()
+		t.Cleanup(func() { server.Close() })
+
+		require.Len(t, testCtx.databaseCA.GetActiveKeys().TLS, 1)
+		ca := string(testCtx.databaseCA.GetActiveKeys().TLS[0].Cert)
+		database, err := types.NewDatabaseV3(types.Metadata{
+			Name: name,
+		}, types.DatabaseSpecV3{
+			Protocol:      defaults.ProtocolDynamoDB,
+			URI:           net.JoinHostPort("localhost", server.Port()),
+			DynamicLabels: dynamicLabels,
+			AWS: types.AWS{
+				Region:    "us-west-1",
+				AccountID: "12345",
+			},
+			TLS: types.DatabaseTLS{
+				// Set CA, otherwise the engine will attempt to download and use the AWS CA.
+				CACert: ca,
+			},
+		})
+		require.NoError(t, err)
+
+		testCtx.dynamodb[name] = testDynamoDB{
+			db:       server,
+			resource: database,
+		}
+		return database
+	}
+}

lib/srv/db/server.go

@@ -45,6 +45,7 @@ import (
 	"github.com/gravitational/teleport/lib/srv/db/cloud"
 	"github.com/gravitational/teleport/lib/srv/db/cloud/users"
 	"github.com/gravitational/teleport/lib/srv/db/common"
+	"github.com/gravitational/teleport/lib/srv/db/dynamodb"
 	"github.com/gravitational/teleport/lib/srv/db/elasticsearch"
 	"github.com/gravitational/teleport/lib/srv/db/mongodb"
 	"github.com/gravitational/teleport/lib/srv/db/mysql"
@@ -64,6 +65,7 @@ func init() {
 	common.RegisterEngine(redis.NewEngine, defaults.ProtocolRedis)
 	common.RegisterEngine(snowflake.NewEngine, defaults.ProtocolSnowflake)
 	common.RegisterEngine(sqlserver.NewEngine, defaults.ProtocolSQLServer)
+	common.RegisterEngine(dynamodb.NewEngine, defaults.ProtocolDynamoDB)
 }
 
 // Config is the configuration for a database proxy server.

tool/teleport/common/teleport.go

@@ -220,6 +220,7 @@ func Run(options Options) (app *kingpin.Application, executedCommand string, con
 	dbStartCmd.Flag("ca-cert", "Database CA certificate path.").StringVar(&ccf.DatabaseCACertFile)
 	dbStartCmd.Flag("aws-region", "(Only for RDS, Aurora, Redshift, ElastiCache or MemoryDB) AWS region AWS hosted database instance is running in.").StringVar(&ccf.DatabaseAWSRegion)
 	dbStartCmd.Flag("aws-account-id", "(Only for Keyspaces) AWS Account ID.").StringVar(&ccf.DatabaseAWSAccountID)
+	dbStartCmd.Flag("aws-external-id", "Optional AWS external ID used when assuming an AWS role.").StringVar(&ccf.DatabaseAWSExternalID)
 	dbStartCmd.Flag("aws-redshift-cluster-id", "(Only for Redshift) Redshift database cluster identifier.").StringVar(&ccf.DatabaseAWSRedshiftClusterID)
 	dbStartCmd.Flag("aws-rds-instance-id", "(Only for RDS) RDS instance identifier.").StringVar(&ccf.DatabaseAWSRDSInstanceID)
 	dbStartCmd.Flag("aws-rds-cluster-id", "(Only for Aurora) Aurora cluster identifier.").StringVar(&ccf.DatabaseAWSRDSClusterID)
@@ -260,6 +261,7 @@ func Run(options Options) (app *kingpin.Application, executedCommand string, con
 	dbConfigureCreate.Flag("uri", "Address the proxied database is reachable at.").StringVar(&dbConfigCreateFlags.StaticDatabaseURI)
 	dbConfigureCreate.Flag("labels", "Comma-separated list of labels for the database, for example env=dev,dept=it").StringVar(&dbConfigCreateFlags.StaticDatabaseRawLabels)
 	dbConfigureCreate.Flag("aws-region", "(Only for AWS-hosted databases) AWS region RDS, Aurora, Redshift, Redshift Serverless, ElastiCache, or MemoryDB database instance is running in.").StringVar(&dbConfigCreateFlags.DatabaseAWSRegion)
+	dbConfigureCreate.Flag("aws-external-id", "(Only for AWS-hosted databases) Optional AWS external ID to use when assuming AWS roles.").StringVar(&dbConfigCreateFlags.DatabaseAWSExternalID)
 	dbConfigureCreate.Flag("aws-redshift-cluster-id", "(Only for Redshift) Redshift database cluster identifier.").StringVar(&dbConfigCreateFlags.DatabaseAWSRedshiftClusterID)
 	dbConfigureCreate.Flag("ad-domain", "(Only for SQL Server) Active Directory domain.").StringVar(&dbConfigCreateFlags.DatabaseADDomain)
 	dbConfigureCreate.Flag("ad-spn", "(Only for SQL Server) Service Principal Name for Active Directory auth.").StringVar(&dbConfigCreateFlags.DatabaseADSPN)

tool/tsh/db.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"fmt"
@@ -253,8 +254,12 @@ func onDatabaseLogin(cf *CLIConf) error {
 
 	// Print after-login message.
 	templateData := map[string]string{
-		"name":           routeToDatabase.ServiceName,
-		"connectCommand": utils.Color(utils.Yellow, formatDatabaseConnectCommand(cf.SiteName, routeToDatabase)),
+		"name": routeToDatabase.ServiceName,
+	}
+
+	// DynamoDB does not support a connect command, so don't try to print one.
+	if database.GetProtocol() != defaults.ProtocolDynamoDB {
+		templateData["connectCommand"] = utils.Color(utils.Yellow, formatDatabaseConnectCommand(cf.SiteName, routeToDatabase))
 	}
 
 	if shouldUseLocalProxyForDatabase(tc, &routeToDatabase) {
@@ -270,13 +275,23 @@ func checkAndSetDBRouteDefaults(r *tlsca.RouteToDatabase) error {
 	// When generating certificate for MongoDB access, database username must
 	// be encoded into it. This is required to be able to tell which database
 	// user to authenticate the connection as.
-	if r.Protocol == defaults.ProtocolMongoDB && r.Username == "" {
-		return trace.BadParameter("please provide the database user name using --db-user flag")
+	if r.Username == "" {
+		switch r.Protocol {
+		case defaults.ProtocolMongoDB:
+			return trace.BadParameter("please provide the database user name using the --db-user flag")
+		case defaults.ProtocolRedis:
+			// Default to "default" in the same way as Redis does. We need the username to check access on our side.
+			// ref: https://redis.io/commands/auth
+			r.Username = defaults.DefaultRedisUsername
+		}
 	}
-	if r.Protocol == defaults.ProtocolRedis && r.Username == "" {
-		// Default to "default" in the same way as Redis does. We need the username to check access on our side.
-		// ref: https://redis.io/commands/auth
-		r.Username = defaults.DefaultRedisUsername
+	if r.Database != "" {
+		switch r.Protocol {
+		case defaults.ProtocolDynamoDB:
+			log.Warnf("Database %v protocol %v does not support --db-name flag, ignoring --db-name=%v",
+				r.ServiceName, defaults.ReadableDatabaseProtocol(r.Protocol), r.Database)
+			r.Database = ""
+		}
 	}
 	return nil
 }
@@ -406,17 +421,11 @@ func onDatabaseEnv(cf *CLIConf) error {
 	}
 
 	if !dbprofile.IsSupported(*database) {
-		return trace.BadParameter(dbCmdUnsupportedDBProtocol,
-			cf.CommandWithBinary(),
-			defaults.ReadableDatabaseProtocol(database.Protocol),
-		)
+		return trace.BadParameter(formatDbCmdUnsupportedDBProtocol(cf, database))
 	}
-	// MySQL requires ALPN local proxy in signle port mode.
+	// MySQL requires ALPN local proxy in single port mode.
 	if tc.TLSRoutingEnabled && database.Protocol == defaults.ProtocolMySQL {
-		return trace.BadParameter(dbCmdUnsupportedTLSRouting,
-			cf.CommandWithBinary(),
-			defaults.ReadableDatabaseProtocol(database.Protocol),
-		)
+		return trace.BadParameter(formatDbCmdUnsupportedTLSRouting(cf, database))
 	}
 
 	env, err := dbprofile.Env(tc, *database)
@@ -473,17 +482,11 @@ func onDatabaseConfig(cf *CLIConf) error {
 	// the remote proxy directly. Return errors here when direct connection
 	// does NOT work (e.g. when ALPN local proxy is required).
 	if isLocalProxyAlwaysRequired(database.Protocol) {
-		return trace.BadParameter(dbCmdUnsupportedDBProtocol,
-			cf.CommandWithBinary(),
-			defaults.ReadableDatabaseProtocol(database.Protocol),
-		)
+		return trace.BadParameter(formatDbCmdUnsupportedDBProtocol(cf, database))
 	}
-	// MySQL requires ALPN local proxy in signle port mode.
+	// MySQL requires ALPN local proxy in single port mode.
 	if tc.TLSRoutingEnabled && database.Protocol == defaults.ProtocolMySQL {
-		return trace.BadParameter(dbCmdUnsupportedTLSRouting,
-			cf.CommandWithBinary(),
-			defaults.ReadableDatabaseProtocol(database.Protocol),
-		)
+		return trace.BadParameter(formatDbCmdUnsupportedTLSRouting(cf, database))
 	}
 
 	host, port := tc.DatabaseProxyHostPort(*database)
@@ -564,9 +567,10 @@ func maybeStartLocalProxy(ctx context.Context, cf *CLIConf, tc *client.TeleportC
 		return []dbcmd.ConnectCommandFunc{}, nil
 	}
 
-	// Some protocols (Snowflake, Elasticsearch) only works in the local tunnel mode.
+	// Some protocols (Snowflake, DynamoDB) only works in the local tunnel mode.
+	// ElasticSearch can work without the --tunnel flag, but not via `tsh db connect`.
 	localProxyTunnel := cf.LocalProxyTunnel
-	if db.Protocol == defaults.ProtocolSnowflake || db.Protocol == defaults.ProtocolElasticsearch {
+	if requiresLocalProxyTunnel(db.Protocol) || db.Protocol == defaults.ProtocolElasticsearch {
 		localProxyTunnel = true
 	}
 
@@ -731,6 +735,9 @@ func onDatabaseConnect(cf *CLIConf) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
+	if routeToDatabase.Protocol == defaults.ProtocolDynamoDB {
+		return trace.BadParameter(formatDbCmdUnsupportedDBProtocol(cf, routeToDatabase))
+	}
 	if err := maybeDatabaseLogin(cf, tc, profile, routeToDatabase); err != nil {
 		return trace.Wrap(err)
 	}
@@ -1076,13 +1083,55 @@ func isLocalProxyAlwaysRequired(protocol string) bool {
 	switch protocol {
 	case defaults.ProtocolSQLServer,
 		defaults.ProtocolSnowflake,
-		defaults.ProtocolCassandra:
+		defaults.ProtocolCassandra,
+		defaults.ProtocolDynamoDB:
 		return true
 	default:
 		return false
 	}
 }
 
+// formatDbCmdUnsupportedWithCondition is a helper func that formats a generic unsupported DB error message.
+// The condition argument is optional and can be "", but otherwise it should be a specific condition for which this DB subcommand
+// is not supported, e.g. "when TLS routing is enabled" or "without using the --tunnel flag".
+func formatDbCmdUnsupportedWithCondition(cf *CLIConf, database *tlsca.RouteToDatabase, condition string) string {
+	templateData := map[string]any{
+		"command":      cf.CommandWithBinary(),
+		"protocol":     defaults.ReadableDatabaseProtocol(database.Protocol),
+		"alternatives": getDbCmdAlternatives(cf.SiteName, database),
+		"condition":    condition,
+	}
+
+	buf := bytes.NewBuffer(nil)
+	_ = dbCmdUnsupportedTemplate.Execute(buf, templateData)
+	return buf.String()
+}
+
+// formatDbCmdUnsupportedDBProtocol is a helper func that formats the unsupported DB protocol error message unconditionally.
+func formatDbCmdUnsupportedDBProtocol(cf *CLIConf, database *tlsca.RouteToDatabase) string {
+	return formatDbCmdUnsupportedWithCondition(cf, database, "")
+}
+
+// formatDbCmdUnsupportedTLSRouting is a helper func that formats an unsupported DB Protocol error with a TLS routing condition.
+func formatDbCmdUnsupportedTLSRouting(cf *CLIConf, database *tlsca.RouteToDatabase) string {
+	return formatDbCmdUnsupportedWithCondition(cf, database, "when TLS routing is enabled on the Teleport Proxy Service")
+}
+
+// getDbCmdAlternatives is a helper func that returns alternative tsh commands for connecting to a database.
+func getDbCmdAlternatives(clusterFlag string, database *tlsca.RouteToDatabase) []string {
+	var alts []string
+	switch database.Protocol {
+	case defaults.ProtocolDynamoDB:
+		// DynamoDB only works with a local proxy tunnel and there is no "shell-like" cli, so `tsh db connect` doesn't make sense.
+	default:
+		// prefer displaying the connect command as the first suggested command alternative.
+		alts = append(alts, formatDatabaseConnectCommand(clusterFlag, *database))
+	}
+	// all db protocols support this command.
+	alts = append(alts, formatDatabaseProxyCommand(clusterFlag, *database))
+	return alts
+}
+
 const (
 	// dbFormatText prints database configuration in text format.
 	dbFormatText = "text"
@@ -1094,36 +1143,40 @@ const (
 	dbFormatYAML = "yaml"
 )
 
-const (
-	// dbCmdUnsupportedTLSRouting is the error message printed when some
-	// database subcommands are not supported because TLS routing is enabled.
-	dbCmdUnsupportedTLSRouting = `"%v" is not supported for %v databases when TLS routing is enabled on the Teleport Proxy Service.
-
-Please use "tsh db connect" or "tsh proxy db" to connect to the database.`
-
-	// dbCmdUnsupportedDBProtocol is the error message printed when some
-	// database subcommands are run against unsupported database protocols.
-	dbCmdUnsupportedDBProtocol = `"%v" is not supported for %v databases.
-
-Please use "tsh db connect" or "tsh proxy db" to connect to the database.`
+var (
+	// dbCmdUnsupportedTemplate is the error message printed when some
+	// database subcommands are not supported.
+	dbCmdUnsupportedTemplate = template.Must(template.New("").Parse(`"{{.command}}" is not supported for {{.protocol}} databases{{if .condition}} {{.condition}}{{end}}.
+{{if eq (len .alternatives) 1}}
+Please use the following command to connect to the database:
+    {{index .alternatives 0 -}}{{else}}
+Please use one of the following commands to connect to the database:
+	{{- range .alternatives}}
+    {{.}}{{end -}}
+{{- end}}`))
 )
 
 var (
 	// dbConnectTemplate is the message printed after a successful "tsh db login" on how to connect.
 	dbConnectTemplate = template.Must(template.New("").Parse(`Connection information for database "{{ .name }}" has been saved.
 
+{{if .connectCommand -}}
+
 You can now connect to it using the following command:
 
   {{.connectCommand}}
 
+{{end -}}
 {{if .configCommand -}}
-Or view the connect command for the native database CLI client:
+
+You can view the connect command for the native database CLI client:
 
   {{ .configCommand }}
 
 {{end -}}
 {{if .proxyCommand -}}
-Or start a local proxy for database GUI clients:
+
+You can start a local proxy for database GUI clients:
 
   {{ .proxyCommand }}
 

tool/tsh/db_test.go

@@ -70,6 +70,15 @@ func TestDatabaseLogin(t *testing.T) {
 		Name:     "mssql",
 		Protocol: defaults.ProtocolSQLServer,
 		URI:      "localhost:1433",
+	}, service.Database{
+		Name:     "dynamodb",
+		Protocol: defaults.ProtocolDynamoDB,
+		URI:      "", // uri can be blank for DynamoDB, it will be derived from the region and requests.
+		AWS: service.DatabaseAWS{
+			AccountID:  "12345",
+			ExternalID: "123123123",
+			Region:     "us-west-1",
+		},
 	})
 
 	authServer := authProcess.GetAuthServer()
@@ -110,6 +119,12 @@ func TestDatabaseLogin(t *testing.T) {
 			expectErrForConfigCmd: true, // "tsh db config" not supported for MSSQL.
 			expectErrForEnvCmd:    true, // "tsh db env" not supported for MSSQL.
 		},
+		{
+			databaseName:          "dynamodb",
+			expectCertsLen:        1,
+			expectErrForConfigCmd: true, // "tsh db config" not supported for DynamoDB.
+			expectErrForEnvCmd:    true, // "tsh db env" not supported for DynamoDB.
+		},
 	}
 
 	// Note: keystore currently races when multiple tsh clients work in the
@@ -130,8 +145,8 @@ func TestDatabaseLogin(t *testing.T) {
 			// Verify certificates.
 			certs, keys, err := decodePEM(profile.DatabaseCertPathForCluster("", test.databaseName))
 			require.NoError(t, err)
-			require.Len(t, certs, test.expectCertsLen)
-			require.Len(t, keys, test.expectKeysLen)
+			require.Equal(t, test.expectCertsLen, len(certs)) // don't use require.Len, because it spams PEM bytes on fail.
+			require.Equal(t, test.expectKeysLen, len(keys))   // don't use require.Len, because it spams PEM bytes on fail.
 		})
 	}
 
@@ -492,6 +507,12 @@ func TestFormatDatabaseConnectArgs(t *testing.T) {
 			route:     tlsca.RouteToDatabase{Protocol: defaults.ProtocolMySQL, Username: "bob", ServiceName: "svc"},
 			wantFlags: []string{"svc"},
 		},
+		{
+			name:      "match user name, dynamodb",
+			cluster:   "",
+			route:     tlsca.RouteToDatabase{Protocol: defaults.ProtocolDynamoDB, ServiceName: "svc"},
+			wantFlags: []string{"--db-user=<user>", "svc"},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

tool/tsh/proxy.go

@@ -379,12 +379,14 @@ func onProxyCommandDB(cf *CLIConf) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
-	if err := maybeDatabaseLogin(cf, client, profile, routeToDatabase); err != nil {
-		return trace.Wrap(err)
+
+	// Some protocols require the --tunnel flag, e.g. Snowflake, DynamoDB.
+	if !cf.LocalProxyTunnel && requiresLocalProxyTunnel(routeToDatabase.Protocol) {
+		return trace.BadParameter(formatDbCmdUnsupportedWithCondition(cf, routeToDatabase, "without the --tunnel flag"))
 	}
 
-	if routeToDatabase.Protocol == defaults.ProtocolSnowflake && !cf.LocalProxyTunnel {
-		return trace.BadParameter("Snowflake proxy works only in the tunnel mode. Please add --tunnel flag to enable it")
+	if err := maybeDatabaseLogin(cf, client, profile, routeToDatabase); err != nil {
+		return trace.Wrap(err)
 	}
 
 	rootCluster, err := client.RootClusterName(cf.Context)
@@ -864,6 +866,16 @@ func envVarCommand(format, key, value string) (string, error) {
 	}
 }
 
+// requiresLocalProxyTunnel returns whether the given protocol requires a local proxy with the --tunnel flag.
+func requiresLocalProxyTunnel(protocol string) bool {
+	switch protocol {
+	case defaults.ProtocolSnowflake, defaults.ProtocolDynamoDB:
+		return true
+	default:
+		return false
+	}
+}
+
 var awsTemplateFuncs = template.FuncMap{
 	"envVarCommand": envVarCommand,
 }

tool/tsh/tsh.go

@@ -2536,7 +2536,13 @@ func getDatabaseRow(proxy, cluster, clusterFlag string, database types.Database,
 	for _, a := range active {
 		if a.ServiceName == name {
 			name = formatActiveDB(a)
-			connect = formatDatabaseConnectCommand(clusterFlag, a)
+			switch a.Protocol {
+			case defaults.ProtocolDynamoDB:
+				// DynamoDB does not support "tsh db connect", so print the proxy command instead.
+				connect = formatDatabaseProxyCommand(clusterFlag, a)
+			default:
+				connect = formatDatabaseConnectCommand(clusterFlag, a)
+			}
 		}
 	}