OpenSearch AWS autodiscovery (#27537) (#27942)

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

@@ -431,6 +431,11 @@ message AWS {
   // AssumeRoleARN is an optional AWS role ARN to assume when accessing a database.
   // Set this field and ExternalID to enable access across AWS accounts.
   string AssumeRoleARN = 11 [(gogoproto.jsontag) = "assume_role_arn,omitempty"];
+  // OpenSearch contains AWS OpenSearch specific metadata.
+  OpenSearch OpenSearch = 12 [
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "opensearch,omitempty"
+  ];
 }
 
 // SecretStore contains secret store configurations.
@@ -505,6 +510,16 @@ message RedshiftServerless {
   string WorkgroupID = 3 [(gogoproto.jsontag) = "workgroup_id,omitempty"];
 }
 
+// OpenSearch contains AWS OpenSearch specific metadata.
+message OpenSearch {
+  // DomainName is the name of the domain.
+  string DomainName = 1 [(gogoproto.jsontag) = "domain_name,omitempty"];
+  // DomainID is the ID of the domain.
+  string DomainID = 2 [(gogoproto.jsontag) = "domain_id,omitempty"];
+  // EndpointType is the type of the endpoint.
+  string EndpointType = 3 [(gogoproto.jsontag) = "endpoint_type,omitempty"];
+}
+
 // GCPCloudSQL contains parameters specific to GCP Cloud SQL databases.
 message GCPCloudSQL {
   // ProjectID is the GCP project ID the Cloud SQL instance resides in.

api/utils/aws/endpoint.go

@@ -352,6 +352,13 @@ const (
 	MemoryDBClusterEndpoint = "cluster"
 	// MemoryDBNodeEndpoint is the endpoint of an individual MemoryDB node.
 	MemoryDBNodeEndpoint = "node"
+
+	// OpenSearchDefaultEndpoint is the default endpoint for domain.
+	OpenSearchDefaultEndpoint = "default"
+	// OpenSearchCustomEndpoint is the custom endpoint configured for domain.
+	OpenSearchCustomEndpoint = "custom"
+	// OpenSearchVPCEndpoint is the VPC endpoint for domain.
+	OpenSearchVPCEndpoint = "vpc"
 )
 
 // ParseElastiCacheEndpoint extracts the details from the provided

lib/cloud/aws/tags_helpers.go

@@ -21,6 +21,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/elasticache"
 	"github.com/aws/aws-sdk-go/service/memorydb"
+	"github.com/aws/aws-sdk-go/service/opensearchservice"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/aws/aws-sdk-go/service/redshift"
 	"github.com/aws/aws-sdk-go/service/redshiftserverless"
@@ -34,8 +35,14 @@ import (
 // ResourceTag is a generic interface that represents an AWS resource tag.
 type ResourceTag interface {
 	// TODO Go generic does not allow access common fields yet. List all types
-	// here and use a type switch for now.
-	rdsTypesV2.Tag | *rds.Tag | *redshift.Tag | *elasticache.Tag | *memorydb.Tag | *redshiftserverless.Tag
+	//  here and use a type switch for now.
+	rdsTypesV2.Tag |
+		*rds.Tag |
+		*redshift.Tag |
+		*elasticache.Tag |
+		*memorydb.Tag |
+		*redshiftserverless.Tag |
+		*opensearchservice.Tag
 }
 
 // TagsToLabels converts a list of AWS resource tags to a label map.
@@ -71,6 +78,8 @@ func resourceTagToKeyValue[Tag ResourceTag](tag Tag) (string, string) {
 		return aws.StringValue(v.Key), aws.StringValue(v.Value)
 	case rdsTypesV2.Tag:
 		return aws.StringValue(v.Key), aws.StringValue(v.Value)
+	case *opensearchservice.Tag:
+		return aws.StringValue(v.Key), aws.StringValue(v.Value)
 	default:
 		return "", ""
 	}

lib/cloud/clients.go

@@ -45,6 +45,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam/iamiface"
 	"github.com/aws/aws-sdk-go/service/memorydb"
 	"github.com/aws/aws-sdk-go/service/memorydb/memorydbiface"
+	"github.com/aws/aws-sdk-go/service/opensearchservice"
+	"github.com/aws/aws-sdk-go/service/opensearchservice/opensearchserviceiface"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/aws/aws-sdk-go/service/rds/rdsiface"
 	"github.com/aws/aws-sdk-go/service/redshift"
@@ -109,6 +111,8 @@ type AWSClients interface {
 	GetAWSElastiCacheClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (elasticacheiface.ElastiCacheAPI, error)
 	// GetAWSMemoryDBClient returns AWS MemoryDB client for the specified region.
 	GetAWSMemoryDBClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (memorydbiface.MemoryDBAPI, error)
+	// GetAWSOpenSearchClient returns AWS OpenSearch client for the specified region.
+	GetAWSOpenSearchClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (opensearchserviceiface.OpenSearchServiceAPI, error)
 	// GetAWSSecretsManagerClient returns AWS Secrets Manager client for the specified region.
 	GetAWSSecretsManagerClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (secretsmanageriface.SecretsManagerAPI, error)
 	// GetAWSIAMClient returns AWS IAM client for the specified region.
@@ -333,6 +337,15 @@ func (c *cloudClients) GetAWSElastiCacheClient(ctx context.Context, region strin
 	return elasticache.New(session), nil
 }
 
+// GetAWSOpenSearchClient returns AWS OpenSearch client for the specified region.
+func (c *cloudClients) GetAWSOpenSearchClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (opensearchserviceiface.OpenSearchServiceAPI, error) {
+	session, err := c.GetAWSSession(ctx, region, opts...)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return opensearchservice.New(session), nil
+}
+
 // GetAWSMemoryDBClient returns AWS MemoryDB client for the specified region.
 func (c *cloudClients) GetAWSMemoryDBClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (memorydbiface.MemoryDBAPI, error) {
 	session, err := c.GetAWSSession(ctx, region, opts...)
@@ -759,6 +772,7 @@ type TestCloudClients struct {
 	Redshift                redshiftiface.RedshiftAPI
 	RedshiftServerless      redshiftserverlessiface.RedshiftServerlessAPI
 	ElastiCache             elasticacheiface.ElastiCacheAPI
+	OpenSearch              opensearchserviceiface.OpenSearchServiceAPI
 	MemoryDB                memorydbiface.MemoryDBAPI
 	SecretsManager          secretsmanageriface.SecretsManagerAPI
 	IAM                     iamiface.IAMAPI
@@ -856,6 +870,15 @@ func (c *TestCloudClients) GetAWSElastiCacheClient(ctx context.Context, region s
 	return c.ElastiCache, nil
 }
 
+// GetAWSOpenSearchClient returns AWS OpenSearch client for the specified region.
+func (c *TestCloudClients) GetAWSOpenSearchClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (opensearchserviceiface.OpenSearchServiceAPI, error) {
+	_, err := c.GetAWSSession(ctx, region, opts...)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return c.OpenSearch, nil
+}
+
 // GetAWSMemoryDBClient returns AWS MemoryDB client for the specified region.
 func (c *TestCloudClients) GetAWSMemoryDBClient(ctx context.Context, region string, opts ...AWSAssumeRoleOptionFn) (memorydbiface.MemoryDBAPI, error) {
 	_, err := c.GetAWSSession(ctx, region, opts...)

lib/cloud/mocks/aws.go

@@ -29,6 +29,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam/iamiface"
 	"github.com/aws/aws-sdk-go/service/memorydb"
 	"github.com/aws/aws-sdk-go/service/memorydb/memorydbiface"
+	"github.com/aws/aws-sdk-go/service/opensearchservice"
+	"github.com/aws/aws-sdk-go/service/opensearchservice/opensearchserviceiface"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/aws/aws-sdk-go/service/rds/rdsiface"
 	"github.com/aws/aws-sdk-go/service/redshift"
@@ -640,6 +642,38 @@ func (m *ElastiCacheMock) ModifyUserWithContext(_ aws.Context, input *elasticach
 	return nil, trace.NotFound("user %s not found", aws.StringValue(input.UserId))
 }
 
+type OpenSearchMock struct {
+	opensearchserviceiface.OpenSearchServiceAPI
+
+	Domains   []*opensearchservice.DomainStatus
+	TagsByARN map[string][]*opensearchservice.Tag
+}
+
+func (o *OpenSearchMock) ListDomainNamesWithContext(aws.Context, *opensearchservice.ListDomainNamesInput, ...request.Option) (*opensearchservice.ListDomainNamesOutput, error) {
+	out := &opensearchservice.ListDomainNamesOutput{}
+	for _, domain := range o.Domains {
+		out.DomainNames = append(out.DomainNames, &opensearchservice.DomainInfo{
+			DomainName: domain.DomainName,
+			EngineType: aws.String("OpenSearch"),
+		})
+	}
+
+	return out, nil
+}
+
+func (o *OpenSearchMock) DescribeDomainsWithContext(aws.Context, *opensearchservice.DescribeDomainsInput, ...request.Option) (*opensearchservice.DescribeDomainsOutput, error) {
+	out := &opensearchservice.DescribeDomainsOutput{DomainStatusList: o.Domains}
+	return out, nil
+}
+
+func (o *OpenSearchMock) ListTagsWithContext(_ aws.Context, request *opensearchservice.ListTagsInput, _ ...request.Option) (*opensearchservice.ListTagsOutput, error) {
+	tags, found := o.TagsByARN[aws.StringValue(request.ARN)]
+	if !found {
+		return nil, trace.NotFound("tags not found")
+	}
+	return &opensearchservice.ListTagsOutput{TagList: tags}, nil
+}
+
 // MemoryDBMock mocks AWS MemoryDB API.
 type MemoryDBMock struct {
 	memorydbiface.MemoryDBAPI

lib/services/database.go

@@ -34,6 +34,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/arn"
 	"github.com/aws/aws-sdk-go/service/elasticache"
 	"github.com/aws/aws-sdk-go/service/memorydb"
+	"github.com/aws/aws-sdk-go/service/opensearchservice"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/aws/aws-sdk-go/service/redshift"
 	"github.com/aws/aws-sdk-go/service/redshiftserverless"
@@ -807,7 +808,7 @@ func NewDatabaseFromRDSProxy(dbProxy *rds.DBProxy, port int64, tags []*rds.Tag)
 		})
 }
 
-// NewDatabaseFromRDSProxyCustomEndpiont creates database resource from RDS
+// NewDatabaseFromRDSProxyCustomEndpoint creates database resource from RDS
 // Proxy custom endpoint.
 func NewDatabaseFromRDSProxyCustomEndpoint(dbProxy *rds.DBProxy, customEndpoint *rds.DBProxyEndpoint, port int64, tags []*rds.Tag) (types.Database, error) {
 	metadata, err := MetadataFromRDSProxyCustomEndpoint(dbProxy, customEndpoint)
@@ -924,6 +925,95 @@ func newElastiCacheDatabase(cluster *elasticache.ReplicationGroup, endpoint *ela
 	})
 }
 
+// NewDatabaseFromOpenSearchDomain creates a database resource from an OpenSearch domain.
+func NewDatabaseFromOpenSearchDomain(domain *opensearchservice.DomainStatus, tags []*opensearchservice.Tag) (types.Databases, error) {
+	var databases types.Databases
+
+	if aws.StringValue(domain.Endpoint) != "" {
+		metadata, err := MetadataFromOpenSearchDomain(domain, apiawsutils.OpenSearchDefaultEndpoint)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		meta := types.Metadata{
+			Description: fmt.Sprintf("OpenSearch domain in %v (default endpoint)", metadata.Region),
+			Labels:      labelsFromOpenSearchDomain(domain, metadata, apiawsutils.OpenSearchDefaultEndpoint, tags),
+		}
+
+		meta = setDBName(meta, aws.StringValue(domain.DomainName))
+		spec := types.DatabaseSpecV3{
+			Protocol: defaults.ProtocolOpenSearch,
+			URI:      fmt.Sprintf("%v:443", aws.StringValue(domain.Endpoint)),
+			AWS:      *metadata,
+		}
+
+		db, err := types.NewDatabaseV3(meta, spec)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		databases = append(databases, db)
+	}
+
+	if domain.DomainEndpointOptions != nil && aws.StringValue(domain.DomainEndpointOptions.CustomEndpoint) != "" {
+		metadata, err := MetadataFromOpenSearchDomain(domain, apiawsutils.OpenSearchCustomEndpoint)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		meta := types.Metadata{
+			Description: fmt.Sprintf("OpenSearch domain in %v (custom endpoint)", metadata.Region),
+			Labels:      labelsFromOpenSearchDomain(domain, metadata, apiawsutils.OpenSearchCustomEndpoint, tags),
+		}
+
+		meta = setDBName(meta, aws.StringValue(domain.DomainName), "custom")
+		spec := types.DatabaseSpecV3{
+			Protocol: defaults.ProtocolOpenSearch,
+			URI:      fmt.Sprintf("%v:443", aws.StringValue(domain.DomainEndpointOptions.CustomEndpoint)),
+			AWS:      *metadata,
+		}
+
+		db, err := types.NewDatabaseV3(meta, spec)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		databases = append(databases, db)
+	}
+
+	for name, url := range domain.Endpoints {
+		metadata, err := MetadataFromOpenSearchDomain(domain, apiawsutils.OpenSearchVPCEndpoint)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		meta := types.Metadata{
+			Description: fmt.Sprintf("OpenSearch domain in %v (endpoint %q)", metadata.Region, name),
+			Labels:      labelsFromOpenSearchDomain(domain, metadata, apiawsutils.OpenSearchVPCEndpoint, tags),
+		}
+
+		if domain.VPCOptions != nil {
+			meta.Labels[labelVPCID] = aws.StringValue(domain.VPCOptions.VPCId)
+		}
+
+		meta = setDBName(meta, aws.StringValue(domain.DomainName), name)
+		spec := types.DatabaseSpecV3{
+			Protocol: defaults.ProtocolOpenSearch,
+			URI:      fmt.Sprintf("%v:443", aws.StringValue(url)),
+			AWS:      *metadata,
+		}
+
+		db, err := types.NewDatabaseV3(meta, spec)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		databases = append(databases, db)
+	}
+
+	return databases, nil
+}
+
 // NewDatabaseFromMemoryDBCluster creates a database resource from a MemoryDB
 // cluster.
 func NewDatabaseFromMemoryDBCluster(cluster *memorydb.Cluster, extraLabels map[string]string) (types.Database, error) {
@@ -1121,6 +1211,24 @@ func MetadataFromElastiCacheCluster(cluster *elasticache.ReplicationGroup, endpo
 	}, nil
 }
 
+// MetadataFromOpenSearchDomain creates AWS metadata for the provided OpenSearch domain.
+func MetadataFromOpenSearchDomain(domain *opensearchservice.DomainStatus, endpointType string) (*types.AWS, error) {
+	parsedARN, err := arn.Parse(aws.StringValue(domain.ARN))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &types.AWS{
+		Region:    parsedARN.Region,
+		AccountID: parsedARN.AccountID,
+		OpenSearch: types.OpenSearch{
+			DomainName:   aws.StringValue(domain.DomainName),
+			DomainID:     aws.StringValue(domain.DomainId),
+			EndpointType: endpointType,
+		},
+	}, nil
+}
+
 // MetadataFromMemoryDBCluster creates AWS metadata for the provided MemoryDB
 // cluster.
 func MetadataFromMemoryDBCluster(cluster *memorydb.Cluster, endpointType string) (*types.AWS, error) {
@@ -1412,6 +1520,12 @@ func labelsFromAWSMetadata(meta *types.AWS) map[string]string {
 	return labels
 }
 
+func labelsFromOpenSearchDomain(domain *opensearchservice.DomainStatus, meta *types.AWS, endpointType string, tags []*opensearchservice.Tag) map[string]string {
+	labels := labelsFromMetaAndEndpointType(meta, endpointType, libcloudaws.TagsToLabels(tags))
+	labels[labelEngineVersion] = aws.StringValue(domain.EngineVersion)
+	return labels
+}
+
 // labelsFromMetaAndEndpointType creates database labels from provided AWS meta and endpoint type.
 func labelsFromMetaAndEndpointType(meta *types.AWS, endpointType string, extraLabels map[string]string) map[string]string {
 	labels := labelsFromAWSMetadata(meta)
@@ -1625,6 +1739,11 @@ func IsMemoryDBClusterAvailable(cluster *memorydb.Cluster) bool {
 	return IsAWSResourceAvailable(cluster, cluster.Status)
 }
 
+// IsOpenSearchDomainAvailable checks if the OpenSearch domain is available.
+func IsOpenSearchDomainAvailable(domain *opensearchservice.DomainStatus) bool {
+	return aws.BoolValue(domain.Created) && !aws.BoolValue(domain.Deleted)
+}
+
 // IsRDSProxyAvailable checks if the RDS Proxy is available.
 func IsRDSProxyAvailable(dbProxy *rds.DBProxy) bool {
 	return IsAWSResourceAvailable(dbProxy, dbProxy.Status)

lib/services/matchers.go

@@ -270,6 +270,8 @@ const (
 	AWSMatcherElastiCache = "elasticache"
 	// AWSMatcherMemoryDB is the AWS matcher type for MemoryDB databases.
 	AWSMatcherMemoryDB = "memorydb"
+	// AWSMatcherOpenSearch is the AWS matcher type for OpenSearch databases.
+	AWSMatcherOpenSearch = "opensearch"
 )
 
 // SupportedAWSMatchers is list of AWS services currently supported by the
@@ -288,6 +290,7 @@ var SupportedAWSDatabaseMatchers = []string{
 	AWSMatcherRedshiftServerless,
 	AWSMatcherElastiCache,
 	AWSMatcherMemoryDB,
+	AWSMatcherOpenSearch,
 }
 
 // RequireAWSIAMRolesAsUsersMatchers is a list of the AWS databases that