Skip to main content
Sign in Dashboard Get started
  1. Docs
  2. Secrets & Configuration
  3. Providers
  4. Rotators
  5. external

external

    The external rotator enables you to rotate secrets with custom logic using Pulumi ESC by making authenticated HTTPS requests to user-controlled adapter services.

    Overview

    The external rotator allows you to implement custom secret rotation logic without waiting for a native rotator implementation. Your adapter service:

    • Authenticates requests using JWT tokens issued by Pulumi Cloud (see JWT Authentication)
    • Receives the current state from previous rotations
    • Generates new credentials or rotates existing ones
    • Returns the new state to be stored in ESC

    When to Use

    Use the external rotator when:

    • You need to rotate credentials for a custom or proprietary system.

    ESC Configuration Example

    values:
  rotatedCredentials:
    fn::rotate::external:
      inputs:
        url: https://my-adapter.example.com/rotate-credentials
        request:
          service: database
          environment: production

    Request Payload (First Rotation)

    On the first rotation, your adapter receives null for the state:

    {
  "request": {
    "service": "database",
    "environment": "production"
  },
  "state": null
}

    Request Payload (Subsequent Rotations)

    On subsequent rotations, your adapter receives the current state from the previous rotation:

    {
  "request": {
    "service": "database",
    "environment": "production"
  },
  "state": {
    "username": "user_20250115",
    "password": "previous-password-123",
    "rotatedAt": "2025-01-15T10:00:00Z"
  }
}

    Response Payload

    Your adapter returns a JSON object that becomes the new state:

    {
  "username": "user_20250116",
  "password": "newly-rotated-password-456",
  "rotatedAt": "2025-01-16T10:00:00Z"
}

    In ESC, this is stored as the current state and automatically marked as secret. On the next rotation, this entire object is passed as the state field in the request.

    When opening the environment after rotation, you should see output like this:

    {
  "rotatedCredentials": {
    "current": {
      "username": "user_20250116",
      "password": "newly-rotated-password-456",
      "rotatedAt": "2025-01-16T10:00:00Z"
    }
  }
}

    Schema reference

    Reference schemas last updated on 2026-07-11, synced automatically from the Pulumi Cloud ESC API.

    Inputs

    • request map[string]any optional
      The request body to send to the external adapter service.
    • url string required
      The URL to the external adapter service.

    State

    • current map[string]any optional
      The current state from the last rotation.

    Outputs

    • current map[string]any optional
      The current state from the last rotation.

    Building Custom Rotators

    Requirements

    Your rotator adapter must meet the same requirements as an external provider adapter.

    State Management

    The key difference from the provider is state management:

    • First rotation: state is null - your adapter should create initial credentials
    • Subsequent rotations: state contains the current value from the previous rotation
    • Response: Your entire response becomes the new current state, marked as secret

    We strongly recommend implementing a dual-secret rotation strategy to ensure zero-downtime credential rotations. This pattern maintains two active secrets: one currently in use by applications, and one that can be safely rotated.

    Why dual-secret rotation?

    Without dual secrets, rotation creates a race condition:

    1. Your adapter rotates the credential (e.g., regenerates an API key)
    2. Applications using the old credential start failing immediately
    3. Applications must fetch new configuration before they can reconnect

    With dual secrets, you eliminate this window of failure:

    1. Applications always use the current secret
    2. During rotation, the adapter rotates the inactive secret
    3. Applications continue using the current secret (unaffected by rotation)
    4. On the next config fetch, applications seamlessly switch to the newly rotated secret

    How to implement:

    The exact implementation depends on your credential system:

    • API keys with multiple active keys: Create two keys, rotate the inactive one
    • Database passwords without multi-password support: Create two user accounts (see mysql rotator for an example)
    • Service tokens with versioning: Maintain two versions, rotate the older one

    Implementation pattern:

    Store identifiers for both secrets in your request configuration, and track which one is current in state:

    values:
  apiCredentials:
    fn::rotate::external:
      inputs:
        url: https://my-adapter.example.com/rotate
        request:
          service: my-service
          keys:
            key1: "prod-key-1"
            key2: "prod-key-2"

    def rotate(request, state):
    keys = request["keys"]  # {"key1": "prod-key-1", "key2": "prod-key-2"}

    if state is None:
        # First rotation: create both secrets, return key1 as current
        create_secret(keys["key1"])
        create_secret(keys["key2"])

        return {
            "keyId": keys["key1"],
            "secret": get_secret_value(keys["key1"]),
            "rotatedAt": now()
        }

    # Determine which secret to rotate (the inactive one)
    current_key = state["keyId"]
    next_key = keys["key2"] if current_key == keys["key1"] else keys["key1"]

    # Rotate the inactive secret
    new_secret = rotate_secret(next_key)

    # Return the newly rotated secret as current
    return {
        "keyId": next_key,
        "secret": new_secret,
        "rotatedAt": now()
    }

    ESC output:

    After rotation, applications see only the current secret:

    {
  "apiCredentials": {
    "current": {
      "keyId": "prod-key-2",
      "secret": "newly-rotated-secret-value",
      "rotatedAt": "2025-01-16T10:00:00Z"
    }
  }
}

    Your application should always use apiCredentials.current. After rotation, current contains the newly rotated secret, while the previous secret remains valid until the next rotation.

    Configure your rotation schedule to be less frequent than your application’s configuration refresh interval. For example, if your app fetches configuration every 5 minutes, rotate no more than once per hour.

    Example Rotator Implementation

    Here’s a complete rotator in Python that generates rotating API keys:

    #!/usr/bin/env python3
"""
Example external rotator adapter for Pulumi ESC.
Rotates API keys with timestamp-based rotation tracking.
"""

import hashlib
import base64
import json
import secrets
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler

import jwt
from jwt import PyJWKClient

# Configuration
JWKS_URL = "https://api.pulumi.com/oidc/.well-known/jwks"
ADAPTER_URL = "https://my-adapter.example.com/rotate-credentials"
PORT = 8443

# Initialize JWKS client (caches keys automatically)
jwks_client = PyJWKClient(JWKS_URL)


def verify_body_hash(body: bytes, claims: dict) -> None:
    """Verify the body_hash claim matches the request body."""
    expected_hash = claims.get("body_hash")
    if not expected_hash:
        raise ValueError("Missing body_hash claim")

    hash_digest = hashlib.sha256(body).digest()
    actual_hash = f"sha256-{base64.b64encode(hash_digest).decode('ascii')}"

    if actual_hash != expected_hash:
        raise ValueError(f"Body hash mismatch")


def generate_api_key(service: str) -> str:
    """Generate a new API key for the service."""
    # In production, this would call your actual rotation logic
    # (e.g., call an API, update a database, etc.)
    random_bytes = secrets.token_bytes(32)
    return f"{service}_{base64.urlsafe_b64encode(random_bytes).decode('ascii').rstrip('=')}"


class RotatorHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        try:
            # Extract and verify JWT token
            auth_header = self.headers.get("Authorization", "")
            if not auth_header.startswith("Bearer "):
                self.send_error(401, "Missing or invalid Authorization header")
                return

            token = auth_header[7:]

            # Verify token using JWKS
            signing_key = jwks_client.get_signing_key_from_jwt(token)
            claims = jwt.decode(
                token,
                signing_key.key,
                algorithms=["RS256"],
                audience=ADAPTER_URL,
                options={"verify_exp": True}
            )

            # Read and verify request body
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            verify_body_hash(body, claims)

            # Parse rotation request
            rotation_request = json.loads(body)
            request_config = rotation_request.get("request", {})
            current_state = rotation_request.get("state")  # None on first rotation

            service = request_config.get("service")
            if not service:
                self.send_error(400, "Missing required field: service")
                return

            # Log rotation event
            if current_state is None:
                print(f"First rotation for service: {service}")
            else:
                print(f"Rotating credentials for service: {service}")
                print(f"Previous key: {current_state.get('apiKey', 'N/A')[:20]}...")

            # Generate new credentials
            new_api_key = generate_api_key(service)
            new_state = {
                "apiKey": new_api_key,
                "rotatedAt": datetime.now(timezone.utc).isoformat(),
                "service": service,
            }

            # In production, you might:
            # 1. Update your service with the new credentials
            # 2. Keep old credentials valid for a grace period
            # 3. Verify the new credentials work before returning

            # Return new state
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(new_state).encode())

        except jwt.InvalidTokenError as e:
            self.send_error(401, f"Invalid token: {str(e)}")
        except Exception as e:
            self.send_error(400, str(e))


if __name__ == "__main__":
    # In production, use a proper HTTPS server with valid certificates
    server = HTTPServer(("", PORT), RotatorHandler)
    print(f"Rotator adapter listening on port {PORT}")
    server.serve_forever()

    To use this rotator:

    # Install dependencies
pip install pyjwt cryptography

# Run the rotator (in production, use proper HTTPS)
python rotator.py

    ESC configuration:

    values:
  apiCredentials:
    fn::rotate::external:
      inputs:
        url: https://my-adapter.example.com/rotate-credentials
        request:
          service: my-api

    After running pulumi env rotate, the environment will contain:

    apiCredentials:
  current:
    apiKey: "my-api_gK7jP9mN4qR8tX2vY5zW..."
    rotatedAt: "2025-01-16T10:00:00Z"
    service: "my-api"

    Troubleshooting

    SymptomLikely causeResolution
    Your adapter rejects the request as unauthorizedThe JWT token may have failed verification, or your adapter may have rejected the Authorization header.Verify your adapter fetches keys from https://api.pulumi.com/oidc/.well-known/jwks and validates the audience against your adapter’s URL. See JWT Authentication.
    Your adapter reports a body-hash mismatchYour adapter may have modified or re-encoded the request body before hashing it.Compute the SHA-256 hash over the raw request bytes and compare against the body_hash claim without altering the body.
    Applications fail after a rotationThe adapter may have rotated the in-use credential instead of an inactive one.Implement the dual-secret strategy so applications always read current while the inactive credential is rotated.