Automating Model API Key Rotation With a Secrets Manager Trigger

Secrets Manager rotation was designed for credentials you can create on demand, which a database password is and a model provider API key is not. The mechanism still works, but only if you stop trying to make it look like the RDS template.

The four-step contract

A rotation Lambda is invoked four times per rotation, with the same SecretId and ClientRequestToken and a different Step each time. AWS documents the four as createSecret, setSecret, testSecret and finishSecret (AWS, rotation by Lambda function). What each owes the system:

• createSecret — produce the new value and store it as a new version labelled AWSPENDING. It must be idempotent: if a version already exists for this token, return without creating another.
• setSecret — make the new value valid at the other end. For a database this changes the password on the server. This is the step that has no analogue for most model providers.
• testSecret — use the AWSPENDING value to do something real and cheap, proving it works before it becomes current.
• finishSecret — move AWSCURRENT onto the pending version. AWS documents that this removes AWSPENDING in the same call and attaches AWSPREVIOUS to the version that was current, which is what gives you a last-known-good to roll back to.

The staging labels are the whole design. A consumer that reads the secret without specifying a version gets AWSCURRENT, so nothing a consumer sees changes until finishSecret runs — and if testSecret throws, the rotation aborts with the old value still current and a dangling pending version, which is the safe direction.

Why a provider key does not fit it

setSecret assumes you can walk up to the resource and change its credential. For a model provider you cannot: the provider issues keys through its own console or management API, keys are opaque values you receive rather than choose, and there is no operation that means “replace the secret behind this key id with this new string”. A rotation function that calls get_random_password and stores the result — the shape every tutorial copies from the RDS template — produces a secret containing a random string that authenticates to nothing, and testSecret catches it only if you wrote a real test.

Two arrangements actually work, and both start from the same idea.

• Two keys, alternating. Provision two keys at the provider — most allow several per account or per project. createSecret selects the one that is not current; testSecret calls a cheap endpoint with it; finishSecret promotes it. The old key stays valid until the next cycle, which is what makes the changeover non-disruptive, and the operational task becomes replacing the idle key rather than coordinating a cutover.
• A provider with a management API. Where the provider exposes key creation programmatically, createSecret can mint one and a later cycle can revoke its predecessor. This is the version that is genuinely automatic, and whether you get it is entirely a function of which provider you are on.

Either way, revocation of the superseded key is a separate deliberate step. Rotating without ever revoking gives you a growing set of live credentials and no security benefit at all — the point of rotation is that a leaked value stops working, and that only happens on revocation. Track it explicitly; see rotating provider keys for the operational side.

The rotation function

The dispatch skeleton is the same for every rotation function; what varies is the four bodies. This one implements the alternating-key scheme, with the secret holding both slots and a pointer.

import json, boto3

sm = boto3.client("secretsmanager")

def lambda_handler(event, context):
    arn   = event["SecretId"]
    token = event["ClientRequestToken"]
    step  = event["Step"]

    meta = sm.describe_secret(SecretId=arn)
    if not meta["RotationEnabled"]:
        raise ValueError(f"Secret {arn} is not enabled for rotation")
    versions = meta["VersionIdsToStages"]
    if token not in versions:
        raise ValueError(f"Version {token} has no stage for {arn}")
    if "AWSCURRENT" in versions[token]:
        return                      # already finished; nothing to do
    if "AWSPENDING" not in versions[token]:
        raise ValueError(f"Version {token} is not AWSPENDING for {arn}")

    if step == "createSecret":
        create_secret(arn, token)
    elif step == "setSecret":
        pass                        # nothing to set: the provider issued both keys
    elif step == "testSecret":
        test_secret(arn, token)
    elif step == "finishSecret":
        finish_secret(arn, token, versions)
    else:
        raise ValueError(f"Unknown step {step}")


def create_secret(arn, token):
    try:
        sm.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
        return                      # idempotent: already created for this token
    except sm.exceptions.ResourceNotFoundException:
        pass

    current = json.loads(
        sm.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"]
    )
    # The secret holds both provisioned keys and which slot is live.
    next_slot = "b" if current["active_slot"] == "a" else "a"
    pending = dict(current, active_slot=next_slot)

    sm.put_secret_value(
        SecretId=arn,
        ClientRequestToken=token,
        SecretString=json.dumps(pending),
        VersionStages=["AWSPENDING"],
    )


def test_secret(arn, token):
    pending = json.loads(
        sm.get_secret_value(SecretId=arn, VersionId=token,
                            VersionStage="AWSPENDING")["SecretString"]
    )
    key = pending["keys"][pending["active_slot"]]
    # A cheap authenticated call. Anything that distinguishes 200 from 401.
    import urllib.request
    req = urllib.request.Request(
        "https://api.openai.com/v1/models",
        headers={"Authorization": f"Bearer {key}"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        if resp.status != 200:
            raise RuntimeError(f"pending key failed validation: {resp.status}")


def finish_secret(arn, token, versions):
    current_version = next(
        v for v, stages in versions.items() if "AWSCURRENT" in stages
    )
    if current_version == token:
        return
    sm.update_secret_version_stage(
        SecretId=arn,
        VersionStage="AWSCURRENT",
        MoveToVersionId=token,
        RemoveFromVersionId=current_version,
    )

The testSecret body is the part not to skip. It is the only thing standing between a broken rotation and a fleet that starts returning 401s at whatever hour the schedule fires. Pick the cheapest authenticated endpoint the provider offers — a model list, not a completion — and give it a short timeout, because the whole rotation is bounded by the function’s own timeout.

Turning rotation on

The function needs a resource policy allowing secretsmanager.amazonaws.com to invoke it, and an execution role with secretsmanager:GetSecretValue, secretsmanager:PutSecretValue, secretsmanager:DescribeSecret, secretsmanager:UpdateSecretVersionStage and, for createSecret in the mint-a-key variant, secretsmanager:GetRandomPassword. Scope the secret actions to the one secret ARN — see scoping a Secrets Manager read policy to one secret, which is the same problem with the same wildcard trap.

aws lambda add-permission \
  --function-name rotate-provider-key \
  --statement-id secretsmanager-invoke \
  --action lambda:InvokeFunction \
  --principal secretsmanager.amazonaws.com \
  --source-arn arn:aws:secretsmanager:eu-west-1:123456789012:secret:prod/providers/openai-a1b2c3

aws secretsmanager rotate-secret \
  --secret-id prod/providers/openai \
  --rotation-lambda-arn arn:aws:lambda:eu-west-1:123456789012:function:rotate-provider-key \
  --rotation-rules '{"ScheduleExpression": "cron(0 3 1 * ? *)", "Duration": "2h"}'

ScheduleExpression accepts a rate or cron expression and Duration sets the window the rotation may start in. Calling rotate-secret triggers a rotation immediately as well as setting the schedule, which is the behaviour you want the first time and a surprise if you were only editing the cron. Watch the first run in CloudWatch Logs before trusting the schedule; a rotation that fails at finishSecret leaves a pending version behind, and the next scheduled attempt starts from that state.

The consumers are the hard half

Rotation on the storage side is worth nothing if the reading side caches forever. A container that reads the secret once at start-up and holds it in a module-level variable keeps using the old key until it restarts, which means every rotation is a deployment, which means nobody schedules rotation monthly. The fix is on the client: fetch through a cache with a TTL shorter than the rotation interval, and re-fetch on a 401 rather than only on a timer, so a rotation that lands mid-flight self-heals within one failed request.

The AWS Parameters and Secrets Lambda extension does exactly this for functions — it caches locally and serves over a loopback HTTP endpoint with a configurable TTL, so you get freshness without a Secrets Manager call per invocation. In Kubernetes the equivalent is a resync loop; see injecting secrets into a pod from a cloud secret manager, where the same problem appears as “the Secret updated and the pod did not”.