external

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"
    }
  }
}

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

Recommended: Dual-secret strategy

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:

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

With dual secrets, you eliminate this window of failure:

Applications always use the current secret
During rotation, the adapter rotates the inactive secret
Applications continue using the current secret (unaffected by rotation)
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/.well-known/jwks.json"
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 esc env rotate, the environment will contain:

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

Schema Reference

Inputs

Property	Type	Description	Required	Default
`url`	string	HTTPS URL to your rotator adapter service	Yes	-
`request`	object	[Rotate only] - Arbitrary JSON object sent to your adapter	No	`{}`

State (Optional)

You can optionally provide initial state in your ESC environment:

values:
  rotatedCredentials:
    fn::rotate::external:
      inputs:
        url: https://my-adapter.example.com/rotate
        request:
          service: api
      state:
        current:
          apiKey: existing-key-123
          rotatedAt: "2025-01-01T00:00:00Z"

If no state is provided, the rotator passes "state": null on the first rotation.

Outputs

Property	Type	Description
`current`	object	The JSON response from your adapter, marked as secret

external

On this page

On this page

Overview

When to Use

ESC Configuration Example

Request Payload (First Rotation)

Request Payload (Subsequent Rotations)

Response Payload

Building Custom Rotators

Requirements

State Management

Recommended: Dual-secret strategy

Example Rotator Implementation

Schema Reference

Inputs

State (Optional)

Outputs

On this page

On this page