feat: Add contrib/aws/secretsmanager envconfig extension

contrib/aws/secretsmanager/CHANGELOG.md

@@ -0,0 +1,12 @@
+<!--
+Release notes for go.temporal.io/sdk/contrib/aws/secretsmanager.
+Loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
+
+Add user-facing changes below under the appropriate heading (create the heading
+if it does not yet exist): Added, Changed, Deprecated, Breaking Changes, Fixed,
+or Security.
+-->
+
+# Changelog
+
+## [Unreleased]

contrib/aws/secretsmanager/README.md

@@ -0,0 +1,156 @@
+# AWS Secrets Manager for Temporal Client Config
+
+> ⚠️ **This package is currently at an experimental release stage.** ⚠️
+
+Package `go.temporal.io/sdk/contrib/aws/secretsmanager` is an
+[`envconfig`](../../envconfig/README.md) extension that loads Temporal client
+credentials (API key and/or mTLS material) from
+[AWS Secrets Manager](https://aws.amazon.com/secrets-manager/). Only secret IDs
+live in your config; secret material is fetched at runtime and kept in memory.
+
+## Add to your project
+
+From your application's Go module, run:
+
+```bash
+go get go.temporal.io/sdk/contrib/aws/secretsmanager@latest
+```
+
+## Module versioning
+
+`secretsmanager` is released as a separate Go module from the core Temporal Go
+SDK. See [CHANGELOG.md](CHANGELOG.md) for release notes.
+
+## Usage
+
+```go
+import (
+    "go.temporal.io/sdk/client"
+    "go.temporal.io/sdk/contrib/aws/secretsmanager"
+    "go.temporal.io/sdk/contrib/envconfig"
+)
+
+opts, err := envconfig.LoadClientOptions(envconfig.LoadClientOptionsRequest{
+    Extensions: []envconfig.ClientConfigExtension{secretsmanager.Extension{}},
+})
+if err != nil {
+    // handle error
+}
+c, err := client.Dial(opts)
+```
+
+## Configuration
+
+Each credential is configured with a Secrets Manager secret ID (friendly name
+or full ARN), either in the TOML config file or with environment variables.
+Environment variables take precedence over the file.
+
+| TOML (under `profile.<profile>.ext.aws_secrets_manager`) | Environment variable | Replaces |
+|---|---|---|
+| `api_key_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID` | `api_key` / `TEMPORAL_API_KEY` |
+| `tls.client_cert_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_CERT_ID` | `tls.client_cert_{path,data}` / `TEMPORAL_TLS_CLIENT_CERT_{PATH,DATA}` |
+| `tls.client_key_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_KEY_ID` | `tls.client_key_{path,data}` / `TEMPORAL_TLS_CLIENT_KEY_{PATH,DATA}` |
+| `tls.server_ca_cert_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_SERVER_CA_CERT_ID` | `tls.server_ca_cert_{path,data}` / `TEMPORAL_TLS_SERVER_CA_CERT_{PATH,DATA}` |
+
+For example:
+
+```toml
+[profile.default]
+address = "my-namespace.a1b2c.tmprl.cloud:7233"
+namespace = "my-namespace.a1b2c"
+
+[profile.default.ext.aws_secrets_manager.tls]
+client_cert_id = "temporal/client-cert-and-key"
+client_key_id = "temporal/client-cert-and-key"
+```
+
+Rules:
+
+- A secret ID can't be combined with the credential it replaces, from any
+  source. For example, `api_key` in the file with
+  `TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID` in the environment is an error.
+- The client cert and key IDs must be set together. They may be the same ID,
+  in which case the secret holds both the PEM certificate and the PEM key. This
+  lets the pair be rotated atomically.
+- TLS secret IDs can't be used when TLS is disabled.
+- An API key secret enables TLS by default, as `api_key` does.
+- Secret values are used as-is: the API key as a string, and TLS material as
+  PEM. Both `SecretString` and `SecretBinary` secrets are supported.
+- Unknown keys under `ext.aws_secrets_manager` are rejected.
+
+## Fetching and caching
+
+When `Extension.SecretsClient` is nil, secrets are fetched through the
+[AWS Parameters and Secrets Lambda Extension](https://docs.aws.amazon.com/secretsmanager/latest/userguide/retrieving-secrets_lambda.html)
+if it is present. Otherwise they are fetched directly from Secrets Manager with
+an AWS SDK v2 client built from the default AWS config.
+
+The Lambda extension is detected as follows:
+
+- If `PARAMETERS_SECRETS_EXTENSION_HTTP_PORT` is set to a port, the extension
+  is used on that port.
+- If it is unset, the extension's default port, 2773, is probed, and the
+  extension is used if something is listening there.
+- Setting `PARAMETERS_SECRETS_EXTENSION_HTTP_PORT` to an empty value explicitly
+  disables detection so that secrets are always fetched directly. Use this if
+  another service in your function listens on port 2773.
+
+To use your own client, for example
+with a specific region or credentials, set `Extension.SecretsClient`:
+
+```go
+cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-west-2"))
+// ...
+ext := secretsmanager.Extension{SecretsClient: awssm.NewFromConfig(cfg)}
+```
+
+Secrets are first fetched when a client is created from the loaded options,
+and client creation fails if a fetch fails. After that:
+
+- Secrets are cached by the plugin that the extension adds to the loaded
+  options, so clients created from the same options share the cache (for
+  example, across warm Lambda invocations).
+- The API key and client cert/key are refreshed while a client is running once
+  the refresh interval has elapsed, so rotated secrets are picked up without
+  recreating the client.
+- The refresh interval is `Extension.RefreshInterval` if set. Otherwise it
+  comes from the `SECRETS_MANAGER_TTL` environment variable, interpreted the
+  same way as by the Lambda extension: whole seconds, where 0 disables
+  caching. Unlike the Lambda extension, values over 300 are allowed. If neither is set, it defaults to 5 minutes. An invalid
+  `SECRETS_MANAGER_TTL` is an error when loading client options.
+- If a refresh fails, the last fetched value keeps being used for up to
+  `Extension.MaxStaleness` (default 24 hours) after it was fetched. After a
+  failed refresh, the next fetch attempt waits 30 seconds.
+- The server CA cert is fetched only when a client is created.
+
+## Observability
+
+The extension logs to the client's `Logger` and emits metrics to the client's
+`MetricsHandler`, if they are set in the client options before the client is
+created.
+
+Metrics are all labeled with `secret_type` (`api_key`, `tls_client_cert`,
+`tls_client_key`, `tls_server_ca_cert`; a bundled cert/key secret is
+`tls_client_cert`) and `fetch_path` (`extension` or `sdk`). Currently, the following metrics are emitted:
+
+| Metric | Type | Extra labels | Description |
+|---|---|---|---|
+| `temporal_aws_secrets_manager_fetch_failed` | Counter | `failure_reason`, `fallback` (`stale`, `none`) | Failed fetches, and whether the previous value was used instead |
+| `temporal_aws_secrets_manager_fetch_latency` | Timer | `outcome` (`success`, `failure`) | Latency of each fetch |
+| `temporal_aws_secrets_manager_cache_lookup` | Counter | `result` (`hit`, `miss`, `miss_with_fallback`) | Lookups that returned a value |
+
+## Required IAM permissions
+
+The AWS credentials in use (for example, the Lambda execution role) need
+`secretsmanager:GetSecretValue` on each configured secret:
+
+```json
+{
+  "Effect": "Allow",
+  "Action": "secretsmanager:GetSecretValue",
+  "Resource": "arn:aws:secretsmanager:us-west-2:123456789012:secret:temporal/*"
+}
+```
+
+If a secret is encrypted with a customer-managed KMS key, `kms:Decrypt` on that
+key is also required.

contrib/aws/secretsmanager/cache.go

@@ -0,0 +1,131 @@
+package secretsmanager
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.temporal.io/sdk/client"
+)
+
+// After a failed refresh falls back to a stale value, wait this long before trying again so that an outage doesn't
+// add a fetch attempt to every credential lookup.
+const failedRefreshRetryInterval = 30 * time.Second
+
+// secretCache is a TTL cache of secret values that falls back to the last fetched value, up to maxStaleness old, when
+// a refresh fails.
+type secretCache struct {
+	fetcher         fetcher
+	refreshInterval time.Duration
+	maxStaleness    time.Duration
+	now             func() time.Time // normally time.Now, but can be overridden in tests
+
+	mu      sync.Mutex
+	entries map[string]*cacheEntry
+}
+
+type cacheEntry struct {
+	// Held during fetch so concurrent lookups of the same secret share one fetch
+	mu        sync.Mutex
+	value     []byte
+	versionID string
+	fetchedAt time.Time
+	// Last failed fetch since the last successful fetch, or zero if none
+	failedAt time.Time
+}
+
+func newSecretCache(f fetcher, refreshInterval time.Duration, maxStaleness time.Duration) *secretCache {
+	return &secretCache{
+		fetcher:         f,
+		refreshInterval: refreshInterval,
+		maxStaleness:    maxStaleness,
+		now:             time.Now,
+		entries:         map[string]*cacheEntry{},
+	}
+}
+
+// get returns the secret's value. The returned slice must not be mutated. secretType is the secret_type label for
+// logs and metrics.
+func (c *secretCache) get(ctx context.Context, secretID string, secretType secretType, tel telemetry) ([]byte, error) {
+	c.mu.Lock()
+	e := c.entries[secretID]
+	if e == nil {
+		e = &cacheEntry{}
+		c.entries[secretID] = e
+	}
+	c.mu.Unlock()
+
+	metrics := tel.metrics
+	if metrics == nil {
+		metrics = client.MetricsNopHandler
+	}
+	metrics = metrics.WithTags(map[string]string{labelSecretType: string(secretType), labelFetchPath: string(c.fetcher.path())})
+	logger := tel.logger
+	if logger == nil {
+		logger = nopLogger{}
+	}
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	now := c.now()
+	if e.value != nil && now.Sub(e.fetchedAt) < c.refreshInterval {
+		metrics.WithTags(map[string]string{labelResult: string(lookupHit)}).Counter(metricCacheLookup).Inc(1)
+		return e.value, nil
+	}
+	canUseStale := e.value != nil && now.Sub(e.fetchedAt) < c.maxStaleness
+	if canUseStale && now.Sub(e.failedAt) < failedRefreshRetryInterval {
+		metrics.WithTags(map[string]string{labelResult: string(lookupMissWithFallback)}).Counter(metricCacheLookup).Inc(1)
+		return e.value, nil
+	}
+
+	s, err := c.fetcher.fetch(ctx, secretID)
+	latency := c.now().Sub(now)
+	if err != nil {
+		e.failedAt = now
+
+		metrics.WithTags(map[string]string{labelOutcome: string(outcomeFailure)}).Timer(metricFetchLatency).Record(latency)
+		fb := fallbackNone
+		if canUseStale {
+			fb = fallbackStale
+		}
+		metrics.WithTags(map[string]string{
+			labelFailureReason: string(classifyFailure(err)),
+			labelFallback:      string(fb),
+		}).Counter(metricFetchFailed).Inc(1)
+
+		if !canUseStale {
+			return nil, fmt.Errorf("failed fetching AWS Secrets Manager secret %q: %w", secretID, err)
+		}
+		logger.Warn("Failed refreshing AWS Secrets Manager secret, using previously fetched value",
+			"SecretID", secretID,
+			"SecretType", secretType,
+			"Error", err,
+			"Age", now.Sub(e.fetchedAt),
+			"MaxStaleness", c.maxStaleness)
+		metrics.WithTags(map[string]string{labelResult: string(lookupMissWithFallback)}).Counter(metricCacheLookup).Inc(1)
+		return e.value, nil
+	}
+
+	metrics.WithTags(map[string]string{labelOutcome: string(outcomeSuccess)}).Timer(metricFetchLatency).Record(latency)
+	metrics.WithTags(map[string]string{labelResult: string(lookupMiss)}).Counter(metricCacheLookup).Inc(1)
+	if e.value == nil {
+		logger.Info("Fetched AWS Secrets Manager secret",
+			"SecretID", secretID,
+			"SecretType", secretType,
+			"FetchPath", c.fetcher.path(),
+			"VersionID", s.versionID)
+	} else if !e.failedAt.IsZero() || s.versionID != e.versionID {
+		logger.Info("Fetched updated AWS Secrets Manager secret",
+			"SecretID", secretID,
+			"SecretType", secretType,
+			"PreviousVersionID", e.versionID,
+			"VersionID", s.versionID)
+	}
+
+	e.value = s.value
+	e.versionID = s.versionID
+	e.fetchedAt = now
+	e.failedAt = time.Time{}
+	return e.value, nil
+}