AWS Secrets Manager Rotation for Coding-Agent Tool Credentials: Short-Lived Tokens That Can’t Leak Forever

"""Tool runner: the secret goes into the child process only, and never back into model context."""
import subprocess


def run_with_secret(cmd: list[str], secret: str, env_name: str, cwd: str = ".") -> dict:
    proc = subprocess.run(cmd, cwd=cwd, env={"PATH": "/usr/bin:/bin", env_name: secret},
                          capture_output=True, text=True, timeout=120)
    out = (proc.stdout + proc.stderr).replace(secret, f"[{env_name} redacted]")   # e.g. verbose curl output
    return {"exit_code": proc.returncode, "output": out[-4000:]}                 # only this reaches the model


if __name__ == "__main__":
    leaky = ["sh", "-c", 'echo "Authorization: Bearer $NPM_TOKEN"; exit 3']
    print(run_with_secret(leaky, "npm_EXAMPLEexampleEXAMPLE", "NPM_TOKEN"))

{'exit_code': 3, 'output': 'Authorization: Bearer [NPM_TOKEN redacted]\n'}

"""Secrets Manager rotation function for an API token that the issuing service mints and revokes."""
import json
import boto3

sm = boto3.client("secretsmanager")


class Issuer:
    """Adapter for the service that issues tokens (a package registry, an internal API...)."""
    def mint(self) -> dict: raise NotImplementedError
    def works(self, token: str) -> bool: raise NotImplementedError
    def revoke(self, token_id: str) -> None: raise NotImplementedError


def handler(event, context, issuer: Issuer):
    arn, token, step = event["SecretId"], event["ClientRequestToken"], event["Step"]
    meta = sm.describe_secret(SecretId=arn)
    if not meta.get("RotationEnabled"):
        raise ValueError(f"rotation is not enabled for {arn}")
    stages = meta["VersionIdsToStages"].get(token)
    if not stages:
        raise ValueError(f"version {token} has no stage")
    if "AWSCURRENT" in stages:
        return                                                    # already finished: retries are safe
    if "AWSPENDING" not in stages:
        raise ValueError(f"version {token} is not AWSPENDING")
    {"createSecret": create, "setSecret": lambda *a: None,       # the issuer already activated it
     "testSecret": test, "finishSecret": finish}[step](arn, token, issuer, meta)


def create(arn, token, issuer, meta):
    try:
        sm.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
        return                                                    # created on an earlier attempt
    except sm.exceptions.ResourceNotFoundException:
        pass
    new = issuer.mint()                                           # {"token_id": ..., "token": ...}
    sm.put_secret_value(SecretId=arn, ClientRequestToken=token,
                        SecretString=json.dumps(new), VersionStages=["AWSPENDING"])


def test(arn, token, issuer, meta):
    pending = json.loads(sm.get_secret_value(SecretId=arn, VersionId=token,
                                             VersionStage="AWSPENDING")["SecretString"])
    if not issuer.works(pending["token"]):
        raise RuntimeError("new token failed its test call; AWSCURRENT is unchanged")


def finish(arn, token, issuer, meta):
    current = next(v for v, s in meta["VersionIdsToStages"].items() if "AWSCURRENT" in s)
    previous = next((v for v, s in meta["VersionIdsToStages"].items() if "AWSPREVIOUS" in s), None)
    sm.update_secret_version_stage(SecretId=arn, VersionStage="AWSCURRENT",
                                   MoveToVersionId=token, RemoveFromVersionId=current)
    # Revoke the token from two rotations ago. The one just demoted stays valid for one more
    # cycle, so callers holding a cached copy keep working until their cache expires.
    if previous and previous != current:
        old = json.loads(sm.get_secret_value(SecretId=arn, VersionId=previous)["SecretString"])
        issuer.revoke(old["token_id"])

import json, uuid
import boto3
from moto import mock_aws

with mock_aws():
    import rotate_token
    rotate_token.sm = boto3.client("secretsmanager", region_name="us-east-1")
    sm = rotate_token.sm

    class FakeRegistry(rotate_token.Issuer):
        def __init__(self): self.live, self.n = set(), 0
        def mint(self):
            self.n += 1; tid = f"tok-{self.n}"; self.live.add(tid)
            return {"token_id": tid, "token": f"secret-value-{self.n}"}
        def works(self, token): return f"tok-{token.rsplit('-', 1)[1]}" in self.live
        def revoke(self, token_id): self.live.discard(token_id)

    reg = FakeRegistry()
    first = reg.mint()
    arn = sm.create_secret(Name="agents/prod/tools/registry/publish-token", SecretString=json.dumps(first))["ARN"]
    real_describe = sm.describe_secret          # test shim: moto needs a real Lambda to enable rotation
    sm.describe_secret = lambda **kw: {**real_describe(**kw), "RotationEnabled": True}

    def rotate():
        token = str(uuid.uuid4())
        # Stand-in for the service: RotateSecret registers an AWSPENDING version, then calls the steps.
        rotate_token.create(arn, token, reg, None)
        for step in ("createSecret", "setSecret", "testSecret", "finishSecret"):
            rotate_token.handler({"SecretId": arn, "ClientRequestToken": token, "Step": step}, None, reg)
        rotate_token.handler({"SecretId": arn, "ClientRequestToken": token, "Step": "finishSecret"}, None, reg)  # retry
        cur = json.loads(sm.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"])
        print(f"current={cur['token_id']} live={sorted(reg.live)}")

    rotate(); rotate(); rotate()

current=tok-2 live=['tok-1', 'tok-2']
current=tok-3 live=['tok-2', 'tok-3']
current=tok-4 live=['tok-3', 'tok-4']

"""Mint a short-lived GitHub App installation token per tool call. Only the App key lives in Secrets Manager."""
import calendar
import json
import time
import urllib.request

import boto3
import jwt  # PyJWT, with the cryptography package for RS256

sm = boto3.client("secretsmanager")
SECRET_ID = "agents/prod/tools/github/app"   # {"client_id": "...", "private_key": "-----BEGIN ..."}
_cache: dict[tuple, tuple[str, float]] = {}


def app_jwt(client_id: str, private_key: str, now: float | None = None) -> str:
    now = int(now or time.time())
    # GitHub: RS256, iat 60 s in the past for clock drift, exp no more than 10 minutes ahead.
    return jwt.encode({"iat": now - 60, "exp": now + 540, "iss": client_id}, private_key, algorithm="RS256")


def installation_token(installation_id: int, repo: str, permissions: dict, post=None) -> str:
    key = (installation_id, repo, json.dumps(permissions, sort_keys=True))
    if key in _cache and _cache[key][1] - time.time() > 300:     # reuse until 5 minutes before expiry
        return _cache[key][0]
    app = json.loads(sm.get_secret_value(SecretId=SECRET_ID)["SecretString"])
    body = {"repositories": [repo], "permissions": permissions}   # narrower than the App's own grants
    resp = (post or _post)(f"https://api.github.com/app/installations/{installation_id}/access_tokens",
                           app_jwt(app["client_id"], app["private_key"]), body)
    expires = calendar.timegm(time.strptime(resp["expires_at"], "%Y-%m-%dT%H:%M:%SZ"))   # UTC
    _cache[key] = (resp["token"], expires)
    return resp["token"]


def _post(url: str, bearer: str, body: dict) -> dict:
    req = urllib.request.Request(url, data=json.dumps(body).encode(), method="POST", headers={
        "Accept": "application/vnd.github+json", "Authorization": f"Bearer {bearer}"})
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.load(r)

import json, time
import boto3, jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from moto import mock_aws

key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
pem = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
                        serialization.NoEncryption()).decode()

with mock_aws():
    import github_token as gt
    gt.sm = boto3.client("secretsmanager", region_name="us-east-1")
    gt.sm.create_secret(Name=gt.SECRET_ID, SecretString=json.dumps({"client_id": "Iv23liEXAMPLE", "private_key": pem}))
    calls = []
    def fake_post(url, bearer, body):
        claims = jwt.decode(bearer, key.public_key(), algorithms=["RS256"])
        calls.append((url, body, claims["exp"] - claims["iat"]))
        expires = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + 3600))
        return {"token": f"ghs_example{len(calls)}", "expires_at": expires}
    perms = {"contents": "write", "pull_requests": "write"}
    t1 = gt.installation_token(42, "web-app", perms, post=fake_post)
    t2 = gt.installation_token(42, "web-app", perms, post=fake_post)       # cached
    t3 = gt.installation_token(42, "web-app", {"contents": "read"}, post=fake_post)
    print(t1, t2, t3, "| API calls:", len(calls))
    print("JWT lifetime (s):", calls[0][2], "| body:", calls[0][1])

ghs_example1 ghs_example1 ghs_example2 | API calls: 2
JWT lifetime (s): 600 | body: {'repositories': ['web-app'], 'permissions': {'contents': 'write', 'pull_requests': 'write'}}