Introduction#
Secrets that never rotate become liabilities. A compromised credential that was never rotated remains exploitable indefinitely. Automating secret rotation removes the operational friction that causes teams to skip it and reduces the blast radius of a compromise.
Why Rotation Matters#
1
2
3
4
5
6
7
8
9
10
Static credential workflow (dangerous):
1. Developer creates DB password, stores in .env or secrets manager
2. Password is used by production for years
3. Password leaks via log, breach, or disgruntled employee
4. Attacker has permanent access until someone notices
Rotated credential workflow:
1. Secret rotated every 30 days (or on breach detection)
2. Compromised credential expires quickly
3. Blast radius limited to the rotation period
AWS Secrets Manager: Automatic Rotation#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
# Lambda function for automatic rotation
import boto3
import json
import logging
import psycopg2
logger = logging.getLogger()
client = boto3.client("secretsmanager")
def lambda_handler(event, context):
arn = event["SecretId"]
token = event["ClientRequestToken"]
step = event["Step"]
# Rotation has 4 steps
if step == "createSecret":
create_secret(arn, token)
elif step == "setSecret":
set_secret(arn, token)
elif step == "testSecret":
test_secret(arn, token)
elif step == "finishSecret":
finish_secret(arn, token)
def create_secret(arn: str, token: str):
"""Generate a new password and store as AWSPENDING."""
try:
client.get_secret_value(SecretId=arn, VersionStage="AWSPENDING", VersionId=token)
return # pending already exists
except client.exceptions.ResourceNotFoundException:
pass
current = json.loads(client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"])
new_password = generate_password()
client.put_secret_value(
SecretId=arn,
ClientRequestToken=token,
SecretString=json.dumps({**current, "password": new_password}),
VersionStages=["AWSPENDING"],
)
def set_secret(arn: str, token: str):
"""Apply the pending password to the database."""
pending = json.loads(client.get_secret_value(SecretId=arn, VersionStage="AWSPENDING", VersionId=token)["SecretString"])
current = json.loads(client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"])
conn = psycopg2.connect(
host=current["host"],
database=current["dbname"],
user=current["username"],
password=current["password"],
)
with conn.cursor() as cur:
cur.execute("ALTER USER %s WITH PASSWORD %s", (current["username"], pending["password"]))
conn.commit()
def test_secret(arn: str, token: str):
"""Verify the new password works."""
pending = json.loads(client.get_secret_value(SecretId=arn, VersionStage="AWSPENDING", VersionId=token)["SecretString"])
conn = psycopg2.connect(
host=pending["host"],
database=pending["dbname"],
user=pending["username"],
password=pending["password"],
)
with conn.cursor() as cur:
cur.execute("SELECT 1")
conn.close()
def finish_secret(arn: str, token: str):
"""Promote AWSPENDING to AWSCURRENT."""
metadata = client.describe_secret(SecretId=arn)
current_version = next(
v for v, stages in metadata["VersionIdsToStages"].items()
if "AWSCURRENT" in stages
)
client.update_secret_version_stage(
SecretId=arn,
VersionStage="AWSCURRENT",
MoveToVersionId=token,
RemoveFromVersionId=current_version,
)
Application-Side: Reading Rotating Secrets#
Applications must re-read secrets periodically — not just at startup.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import boto3
import json
import time
import threading
class RotatingSecret:
"""Automatically refreshes a secret from AWS Secrets Manager."""
def __init__(self, secret_arn: str, refresh_interval: int = 300):
self._arn = secret_arn
self._client = boto3.client("secretsmanager")
self._value: dict = {}
self._lock = threading.RLock()
self._last_refresh: float = 0
self._refresh_interval = refresh_interval
self._refresh()
def _refresh(self):
response = self._client.get_secret_value(SecretId=self._arn)
with self._lock:
self._value = json.loads(response["SecretString"])
self._last_refresh = time.monotonic()
def get(self, key: str = None):
if time.monotonic() - self._last_refresh > self._refresh_interval:
self._refresh()
with self._lock:
return self._value if key is None else self._value[key]
# Usage
db_secret = RotatingSecret("arn:aws:secretsmanager:us-east-1:123:secret:db-prod")
def get_db_connection():
# Credentials are refreshed every 5 minutes
password = db_secret.get("password")
return psycopg2.connect(password=password, ...)
Kubernetes: External Secrets Operator#
Sync secrets from AWS Secrets Manager / HashiCorp Vault to Kubernetes Secrets, with automatic refresh.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# ExternalSecret: syncs a secret into a Kubernetes Secret
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
name: db-credentials
namespace: production
spec:
refreshInterval: 5m # check for new version every 5 minutes
secretStoreRef:
name: aws-secrets-store
kind: ClusterSecretStore
target:
name: db-credentials # creates/updates this Kubernetes Secret
creationPolicy: Owner
data:
- secretKey: password
remoteRef:
key: prod/database/credentials
property: password
---
# ClusterSecretStore: points to AWS Secrets Manager
apiVersion: external-secrets.io/v1beta1
kind: ClusterSecretStore
metadata:
name: aws-secrets-store
spec:
provider:
aws:
service: SecretsManager
region: us-east-1
auth:
jwt:
serviceAccountRef:
name: external-secrets-sa
namespace: external-secrets
Short-Lived Credentials with IRSA#
Better than rotating long-lived credentials: use credentials that expire in minutes.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# AWS STS: assume role for temporary credentials (15 min to 12 hours)
import boto3
sts = boto3.client("sts")
response = sts.assume_role(
RoleArn="arn:aws:iam::123456789:role/db-access-role",
RoleSessionName="api-session",
DurationSeconds=3600, # 1 hour
)
creds = response["Credentials"]
# These expire in 1 hour — no rotation needed, just re-assume
rds = boto3.client(
"rds",
aws_access_key_id=creds["AccessKeyId"],
aws_secret_access_key=creds["SecretAccessKey"],
aws_session_token=creds["SessionToken"],
)
Rotation Checklist#
- Secrets are stored in a secrets manager (not environment files or config maps)
- Rotation is automated, not manual
- Applications re-read secrets periodically (not just at startup)
- Old secret version remains valid during rotation window (dual-version support)
- Rotation is tested in staging before production
- Alerts fire if rotation fails
- Rotation logs are audited
Conclusion#
Manual secret rotation is not done often enough. Automate it with AWS Secrets Manager Lambda rotation or HashiCorp Vault’s database secrets engine. Applications must read secrets at runtime — a config file read at startup will use the old secret until restart. The best long-term solution is short-lived credentials (IRSA, Workload Identity) that expire quickly and need no rotation.