Build a Self-Maintaining Key Rotation Pipeline

Introduction

30 Second Summary

Every app that connects to a payment provider, an AI service, or a cloud database uses a secret key to prove it belongs. If that key leaks or never gets changed, anyone who finds it gets full access.

In this project, you will build a fully automated pipeline that rotates API keys on a schedule with zero downtime. You'll store secrets in HashiCorp Vault, build a Flask app that reads them live, and create a Python script that swaps keys using the same pattern production systems rely on.

What You'll Build

When you finish, you can start your pipeline and watch it rotate keys every 30 seconds while your Flask app keeps serving requests without a single failed response.

By the end of this project, you'll have:

• A running Vault server where you can store, version, and retrieve API keys through a web UI and Python code.
• A live Flask API that pulls its credentials from Vault on every request, so a key rotation takes effect instantly with no app restart.
• A self-maintaining rotation pipeline you can start and watch perform zero-downtime key swaps automatically, rolling back if anything goes wrong.
• Secret Mission: An audit trail that logs every rotation event with timestamps, so you can prove exactly when each key was changed.

Are there any prerequisites?

You need basic Python knowledge (variables, functions, imports). No prior Docker or secrets management experience is needed.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Environment

To build a key rotation pipeline, you need three things: a container runtime to host your secrets server, the secrets server itself, and the Python libraries that talk to it.

In this step, you will install Docker Desktop to run containers, start a HashiCorp Vault server in dev mode, and set up a Python environment with the packages your pipeline needs.

In this step, get ready to:

• Install Docker Desktop and verify it runs.
• Start a Vault dev server in a Docker container.
• Create a Python virtual environment and install project packages.

Install Docker Desktop

Docker lets you run software in isolated containers. You will use it to run a Vault server without installing Vault directly on your machine.

• Open a terminal in Cursor by pressing Ctrl+` (Windows) or Cmd+` (macOS).
• Check if Docker is already installed by running:

docker --version

✔️ I see a version number

Docker is installed and ready. Move on to the next substep.

ⓧ Command not found

You need to install Docker Desktop. It is free for personal use and education.

macOS

• Open https://www.docker.com/products/docker-desktop/ in your browser.
• Click the download button for your chip (Apple Silicon or Intel).
• Open the downloaded .dmg file and drag Docker to your Applications folder.
• Open Docker Desktop from your Applications folder.
• Accept the terms when prompted. Docker Desktop will finish starting up in the menu bar.

Windows

• Open https://www.docker.com/products/docker-desktop/ in your browser.
• Click the download button for Windows (AMD64).
• Run the installer. Docker Desktop installs per-user and does not require administrator privileges.
• If prompted to enable WSL 2, allow it. This is a one-time setup that does require admin rights.
• Once installed, open Docker Desktop from the Start menu. Accept the terms when prompted.

After Docker Desktop is running, go back to your Cursor terminal and run the version check again:

docker --version

You should now see a version number printed in the terminal.

Still seeing command not found?

Make sure Docker Desktop is fully running (check for the whale icon in your menu bar or system tray). You may need to close and reopen your terminal for the path to update.

help me troubleshoot Docker Desktop installation

Start the Vault dev server

Vault is a secrets management tool that stores, versions, and controls access to credentials. You will run it in dev mode, which starts an in-memory server that is pre-configured and ready to use immediately.

• Start the Vault container by running this command in your Cursor terminal:

docker run -p 8200:8200 -e "VAULT_DEV_ROOT_TOKEN_ID=dev-only-token" hashicorp/vault:1.21

What does this command do?

• docker run starts a new container from the hashicorp/vault:1.21 image.
• -p 8200:8200 maps port 8200 on your machine to port 8200 inside the container so you can reach the Vault API and UI.
• -e "VAULT_DEV_ROOT_TOKEN_ID=dev-only-token" sets the root token to a known value you will use in your Python code to authenticate.

This command takes over your terminal

Vault runs in the foreground and streams log output. This is normal. Leave this terminal running for the rest of the project. Do not press Ctrl+C or close this terminal panel.

Before you check the browser, what do you think you will see at http://127.0.0.1:8200? A blank page, an error, or something else?

• Open http://127.0.0.1:8200 in your browser.

You see the Vault UI login page with a Token method selected and a field to enter your token. This confirms Vault is running and reachable.

Seeing a connection refused error?

Check your terminal where you ran the docker run command. If you see an error about the port already being in use, another process is on port 8200. Stop it or use a different port.

If Docker is still pulling the image, wait for it to finish. The first run downloads about 200 MB. Subsequent runs start instantly.

help me get Vault running in Docker

Create your project and install Python packages

Now you need a project folder with a dedicated Python environment. A virtual environment keeps this project's packages isolated from any other Python work on your machine.

• Open a second terminal panel in Cursor by clicking the + button in the terminal tab bar.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir key-rotation-pipeline
cd key-rotation-pipeline

• Confirm you are in the new folder by running:

ls

The output is empty because the folder is brand new. That is expected.

• Create a Python virtual environment by running:

python3 -m venv venv

• Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

• Install the project dependencies by running:

pip install hvac==2.4.0 flask==3.1.3

What are these packages?

• hvac is the official Python client for HashiCorp Vault. It handles authentication and reading/writing secrets.
• flask is a lightweight web framework. You will use it to build the app that consumes secrets from Vault.

Before you verify, do you think Python can import both packages right now, or will one fail?

• Verify both packages installed correctly by running:

python3 -c "import hvac; import flask; print('Ready!')"

You see Ready! printed in the terminal. Both packages are installed and importable.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is active. Check that (venv) appears at the start of your terminal prompt. If it does not, re-run the activate command above.

If activation looks correct, confirm you are running the right Python by running which python3 (macOS) or where python3 (Windows). It should point to a path inside your venv folder.

help me fix my Python virtual environment

• Open this project folder in Cursor by clicking File in the top menu bar.
• Click Open Folder.
• Navigate to Desktop and select the key-rotation-pipeline folder, then click Open.

What about my Vault terminal?

The Vault container keeps running in Docker regardless of what happens in Cursor. If Cursor asks whether to open in the current window or a new window, either choice is fine. Your Vault server stays up because Docker manages it as a separate process.

After the folder opens, open a new terminal in Cursor (Ctrl+` or Cmd+`) and re-activate your virtual environment with the activate command above. New terminals start without the virtual environment active.

Your environment is ready. You have a secrets server running, a project folder open, and Python packages installed. Next up, you will store your first secret in Vault and build the Flask app that reads it live.

Store a Secret and Build the Consumer App

Your Vault server is running and ready to store secrets. Now you need something to put in it, and an app that reads from it.

Before you can rotate keys, you need a working baseline. That means a secret stored in Vault and a Flask app that reads it live on every request. This gives you the "before" state to compare against when rotation breaks things in the next step.

In this step, get ready to:

• Build a simulated API key provider that can generate, validate, and revoke keys.
• Seed Vault with your first secret and confirm it stored correctly.
• Create a Flask app that fetches its API key from Vault on every request.

Create the provider simulation and seed Vault

Real API key providers like Stripe or OpenAI let you generate keys, revoke them, and check if they are valid. You need a local stand-in that behaves the same way so you can test rotation without touching a live service.

• In Cursor, create a new file called provider.py in your project folder.
• Add the imports and persistence helpers by pasting the following code:

import json
import secrets
import os

KEYS_FILE = "valid_keys.json"


def _load_keys():
    """Load the set of currently valid API keys from the provider's database."""
    if not os.path.exists(KEYS_FILE):
        return []
    with open(KEYS_FILE, "r") as f:
        return json.load(f)


def _save_keys(keys):
    """Persist the valid keys to the provider's database."""
    with open(KEYS_FILE, "w") as f:
        json.dump(keys, f, indent=2)

What does this code do?

KEYS_FILE points to a JSON file that acts as the provider's database of active keys.

_load_keys() reads the current list of valid keys from disk. _save_keys() writes the updated list back. These two helpers let multiple processes (the Flask app and the rotation script) share key state through a file.

• Below the helper functions, add the three public functions that the rest of your project will use:

def generate_key():
    """Issue a new API key (simulates clicking 'Generate Key' in provider dashboard)."""
    new_key = f"sk-{secrets.token_hex(16)}"
    keys = _load_keys()
    keys.append(new_key)
    _save_keys(keys)
    return new_key


def revoke_key(key):
    """Revoke an API key (simulates clicking 'Delete Key' in provider dashboard)."""
    keys = _load_keys()
    if key in keys:
        keys.remove(key)
    _save_keys(keys)


def validate_key(key):
    """Check if a key is currently valid with the provider."""
    keys = _load_keys()
    return key in keys

What does this code do?

• generate_key() creates a random key prefixed with sk- (mimicking real provider key formats), saves it to the database, and returns it.
• revoke_key() removes a key from the valid list, making it permanently invalid.
• validate_key() checks if a given key exists in the current valid list.

• Save provider.py.

✔️ Awesome, I've got everything!

Great. Confirm you have both the persistence helpers and the three public functions saved in provider.py.

ⓧ I'd like to double check the full code

Scroll up and confirm your provider.py has these sections in order:

• Imports (json, secrets, os) and the KEYS_FILE constant.
• _load_keys() and _save_keys() helper functions.
• generate_key(), revoke_key(), and validate_key() public functions.

• Test your provider by running this command in your terminal:

python -c "import provider; key = provider.generate_key(); print(f'Generated: {key}'); print(f'Valid: {provider.validate_key(key)}')"

You should see a generated key starting with sk- and Valid: True printed below it.

Seeing ModuleNotFoundError?

• Make sure you are running the command from inside your project folder where provider.py lives.
• Confirm your virtual environment is still activated (you should see (venv) at the start of your terminal prompt).

Help me debug a ModuleNotFoundError when importing provider

Now you need a setup script that uses the provider to generate a key and stores it in Vault.

• Create a new file called setup_vault.py and paste the following code:

import hvac
import provider

VAULT_URL = "http://127.0.0.1:8200"
VAULT_TOKEN = "dev-only-token"


def setup():
    """Generate an initial API key and store it in Vault."""
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)

    if not client.is_authenticated():
        print("ERROR: Cannot connect to Vault. Is the container running?")
        return

    # Generate a key at the provider
    api_key = provider.generate_key()
    print(f"Generated initial API key: {api_key[:6]}...{api_key[-4:]}")

    # Store it in Vault
    client.secrets.kv.v2.create_or_update_secret(
        path="api-keys/third-party",
        secret={"api_key": api_key},
        mount_point="secret",
    )
    print("Stored key in Vault at: secret/api-keys/third-party")
    print("\nSetup complete! Your app can now read this key from Vault.")


if __name__ == "__main__":
    setup()

What does this code do?

• Creates an hvac client that connects to your running Vault container using the dev token.
• Calls provider.generate_key() to simulate creating a new API key at a third-party service.
• Stores that key in Vault's KV v2 secrets engine at the path secret/api-keys/third-party.

The mount_point="secret" parameter tells hvac to use the KV v2 engine that Vault dev mode mounts automatically.

• Save setup_vault.py.
• Seed your Vault with an initial key by running:

python setup_vault.py

You should see output showing a masked key and a confirmation that it was stored in Vault.

Seeing 'Cannot connect to Vault'?

• Confirm your Vault Docker container is still running. Open a new terminal tab and run docker ps to check.
• If the container stopped, restart it with the same docker run command from Step 1.

Help me reconnect to my Vault container

Build the Flask consumer app

With a secret in Vault, you need an app that reads it. The key design choice here is that the app fetches its API key from Vault on every request instead of caching it at startup. This means rotation will take effect immediately without restarting the app.

• Create a new file called app.py and paste the imports, configuration, and the Vault-reading helper:

import hvac
from flask import Flask, jsonify
import provider

app = Flask(__name__)

VAULT_URL = "http://127.0.0.1:8200"
VAULT_TOKEN = "dev-only-token"


def get_current_key():
    """Fetch the current API key from Vault."""
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)
    try:
        secret = client.secrets.kv.v2.read_secret_version(
            path="api-keys/third-party",
            mount_point="secret",
        )
        return secret["data"]["data"]["api_key"]
    except Exception:
        return None

What does this code do?

• Creates a Flask app and defines the Vault connection settings.
• get_current_key() connects to Vault and reads the latest version of the secret every time it is called. The secret["data"]["data"]["api_key"] path navigates the KV v2 response structure to get the actual key value.
• If Vault is unreachable or the secret does not exist, it returns None so the app can handle the error gracefully.

• Below the get_current_key() function, add the /status route and the main block:

@app.route("/status")
def status():
    """Check if our stored API key is valid with the provider."""
    key = get_current_key()
    if key is None:
        return jsonify({"status": "error", "message": "No key found in Vault"}), 500

    is_valid = provider.validate_key(key)
    masked_key = f"{key[:6]}...{key[-4:]}"

    return jsonify({
        "status": "healthy" if is_valid else "unhealthy",
        "key": masked_key,
        "valid": is_valid,
    })


if __name__ == "__main__":
    app.run(port=5000, debug=True)

What does this code do?

• The /status endpoint fetches the current key from Vault, checks it against the provider, and returns a JSON health report.
• The key is masked (showing only the first 6 and last 4 characters) so it is safe to display.
• debug=True enables Flask's auto-reloader. When you save changes to app.py later, the server restarts automatically.

• Save app.py.
• Start the Flask app by running:

python app.py

You should see Flask start up with output including Running on http://127.0.0.1:5000.

• Open your browser and visit http://127.0.0.1:5000/status.

You should see a JSON response with "status": "healthy" and your masked API key.

Seeing 'No key found in Vault'?

• Make sure you ran python setup_vault.py earlier in this step. That script seeds the initial key.
• Check that the Vault container is still running with docker ps in a separate terminal tab.

Help me fix a 'No key found in Vault' error from my Flask app

The server is running and /status works. Now add the second endpoint that simulates calling a third-party API with the stored key.

• In app.py, find the line that says if __name__ == "__main__":.
• Add the following /data route directly above that line:

@app.route("/data")
def get_data():
    """Simulate calling a third-party API with our stored key."""
    key = get_current_key()
    if key is None:
        return jsonify({"error": "No API key available"}), 500

    if not provider.validate_key(key):
        return jsonify({"error": "API key is invalid or expired"}), 401

    return jsonify({
        "message": "Successfully retrieved data from third-party API",
        "data": {"items": ["item1", "item2", "item3"]},
        "authenticated_with": f"{key[:6]}...{key[-4:]}",
    })

What does this code do?

• The /data endpoint simulates a real API call. It fetches the key from Vault, validates it with the provider, and returns mock data only if the key is valid.
• If the key is invalid (revoked or expired), it returns a 401 error. This is the endpoint that will visibly break when rotation goes wrong in the next step.

• Save app.py. Flask auto-reloads the server because debug=True is set.

✔️ Awesome, I've got everything!

Great. Confirm you have the imports, get_current_key(), both routes, and the if __name__ block all saved in app.py.

ⓧ I'd like to double check the full code

Scroll up and confirm your app.py has these sections in order:

• Imports (hvac, Flask, jsonify, provider) and the app = Flask(__name__) line.
• VAULT_URL and VAULT_TOKEN constants.
• get_current_key() function.
• /status route.
• /data route.
• if __name__ == "__main__": block at the bottom.

Verify the app reads live secrets from Vault

Both endpoints are in place. Before you move on, confirm the full flow works: Vault holds a secret, and your app fetches and validates it on demand.

• In your browser, visit http://127.0.0.1:5000/data.

Before you look at the response, predict what you expect to see. Will it return data, or an error?

You should see a JSON response containing "message": "Successfully retrieved data from third-party API" along with mock data items and your masked key in the authenticated_with field.

What just happened end-to-end?

• Your browser hit /data on the Flask app.
• Flask called get_current_key(), which reached into Vault over HTTP and pulled the secret.
• Flask validated that key against the provider's database.
• Because the key is still valid, you got back a successful response.

The critical detail is that the app never caches the key. If someone updates the secret in Vault, the very next request uses the new value.

Seeing a 401 or 500 error on /data?

• If you see "API key is invalid or expired", your valid_keys.json may have been overwritten. Re-run python setup_vault.py to generate a fresh key.
• If you see "No API key available", the Vault container may have restarted (in-memory storage is wiped on restart). Re-run python setup_vault.py to reseed it.

Help me debug errors on my /data endpoint

You now have a complete working baseline: a secret in Vault, an app that reads it live, and a provider that tracks which keys are valid. Next up, you will try the obvious approach to rotation and watch it break this healthy setup.

Break the App with Naive Rotation

Your Flask app is healthy and reading its API key from Vault on every request. But what happens when that key needs to change? In production, API keys expire, get compromised, or need regular rotation for compliance.

The most obvious approach is to delete the old key first, then create a new one. In this step, you'll try exactly that approach and discover why it breaks everything.

In this step, get ready to:

• Write a naive rotation script that revokes the old key before generating a new one.
• Run the script and catch the app mid-outage.
• Confirm recovery and identify the dangerous gap window.

Write the naive rotation script

When you need to rotate an API key, the simplest approach sounds logical: revoke the old key, then generate a new one. You'll write a script that does exactly this, with a simulated network delay between the two operations.

• Create a new file called rotate_naive.py in your project folder.
• Add the following code to set up the connection and read the current key from Vault:

import time
import hvac
import provider

VAULT_URL = "http://127.0.0.1:8200"
VAULT_TOKEN = "dev-only-token"


def rotate_naive():
    """Naive rotation: revoke first, then generate. Creates a downtime gap!"""
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)

    # Read the current key
    secret = client.secrets.kv.v2.read_secret_version(
        path="api-keys/third-party",
        mount_point="secret",
    )
    old_key = secret["data"]["data"]["api_key"]

    print(f"[NAIVE] Current key: {old_key[:6]}...{old_key[-4:]}")

What does this code do?

This first section connects to Vault using the same URL and token your Flask app uses. It reads the current API key so the script knows which key to revoke.

• Now add the dangerous part directly below the last print statement. This is the revoke-first logic that creates the outage gap:

    print(f"[NAIVE] Revoking old key at provider...")
    provider.revoke_key(old_key)
    print(f"[NAIVE] Old key revoked! App is now BROKEN.")
    print(f"[NAIVE] Simulating network delay (3 seconds)...")
    print(f"[NAIVE] >>> Try hitting http://127.0.0.1:5000/status NOW <<<")

    time.sleep(3)

    # Generate and store new key
    new_key = provider.generate_key()
    client.secrets.kv.v2.create_or_update_secret(
        path="api-keys/third-party",
        secret={"api_key": new_key},
        mount_point="secret",
    )

    print(f"[NAIVE] New key stored: {new_key[:6]}...{new_key[-4:]}")
    print(f"[NAIVE] App is healthy again, but there was a {3}-second outage!")


if __name__ == "__main__":
    rotate_naive()

What's happening here?

The script revokes the old key at the provider FIRST. At this point, the key in Vault is now invalid. Then it sleeps for 3 seconds to simulate real-world network latency.

During those 3 seconds, your Flask app still has the old (now-revoked) key in Vault. Any request to /status will validate the old key against the provider and get back "invalid." Only after the sleep does it generate and store a new key.

• Save rotate_naive.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your rotate_naive.py contains both chunks in order:

• Chunk 1: imports, constants, rotate_naive() function definition, Vault read, and the first print statement.
• Chunk 2: revoke, sleep, generate, store, final prints, and the if __name__ block at the bottom.

Run the naive rotation and catch the outage

Your Flask app is already running in one terminal. You need a second terminal to run the rotation script at the same time.

• Open a new terminal window. Press Cmd+Space (macOS) or the Windows key (Windows) and type Terminal to open a fresh terminal.
• Navigate to your project folder and activate your virtual environment by running:

cd ~/Desktop/key-rotation-pipeline
source venv/bin/activate

Before you run the naive rotation script, think about this: what do you expect will happen to your Flask app's /status endpoint during the 3-second gap after the old key is revoked but before the new key is stored?

• Run the naive rotation script:

python rotate_naive.py

• As soon as you see the message [NAIVE] >>> Try hitting http://127.0.0.1:5000/status NOW <<<, quickly switch to your browser.
• Open http://127.0.0.1:5000/status in your browser (or refresh the page if it's already open). You have 3 seconds.

You see a JSON response showing "status": "unhealthy" and "valid": false. Your app is broken. The old key was revoked at the provider, but Vault still holds that old key. Every request during this gap fails.

Why did the app break?

Your Flask app calls get_current_key() from Vault on every request. Vault still returns the old key because no one has updated it yet. But when the app validates that key against the provider, the provider says it's been revoked.

This is the gap window. The old key is dead, but the new key doesn't exist yet. Any request during this period fails.

Missed the 3-second window?

If you didn't switch to your browser fast enough and the script already finished, run python setup_vault.py to reset with a fresh key. Then run python rotate_naive.py again and be ready to switch to the browser immediately.

help me catch the outage window when running rotate_naive.py

Confirm recovery and note the outage window

After the 3-second sleep, the script generates a new key and stores it in Vault. Your app should recover automatically.

• Wait for the script to finish. You should see [NAIVE] App is healthy again, but there was a 3-second outage! in your terminal.
• Refresh http://127.0.0.1:5000/status in your browser.

You now see "status": "healthy" again with a new masked key value. The app recovered because the script eventually stored a valid new key in Vault.

What just happened?

The naive rotation created a 3-second window where no valid key existed. In production, 3 seconds of downtime means failed API calls, dropped user requests, and potential data loss. Real network operations can take much longer than 3 seconds.

The root cause is the order of operations. Revoking first creates a gap. The fix is to generate the new key first so both keys are valid simultaneously. That's what you'll build next.

App still showing unhealthy after the script finished?

Make sure the script printed the final line about storing the new key. If it crashed mid-run, your Vault still holds the revoked key. Run python setup_vault.py to generate a fresh key and try again.

help me fix my app after a failed naive rotation

You just proved that the naive approach to key rotation causes outages. Any system that revokes before it replaces will always have a gap where nothing works. In the next step, you'll build the proper pattern that eliminates this gap entirely.

Build Zero-Downtime Dual-Key Rotation

In the last step, the naive rotation broke your app. Revoking the old key before generating a new one created a gap where no valid key existed.

The fix is the industry-standard dual-key overlap pattern. You generate the new key first (so both keys are valid simultaneously), store it in Vault, verify the app is healthy, and only THEN revoke the old key. This eliminates the gap entirely.

In this step, get ready to:

• Write rotate.py with the generate-store-verify-revoke pattern.
• Add automatic rollback so a failed verification restores the old key.
• Run the rotation and confirm zero downtime on your Flask app.

Write the generate-and-store stages

The dual-key pattern has a specific order. First, you read the current key. Then you generate a new one at the provider without touching the old one. Then you store the new key in Vault. At this point both keys are valid simultaneously.

• Create a new file called rotate.py in your project folder and add the following code:

import hvac
import requests
import provider

VAULT_URL = "http://127.0.0.1:8200"
VAULT_TOKEN = "dev-only-token"
APP_URL = "http://127.0.0.1:5000"


def rotate():
    """Proper rotation: generate, store, verify, THEN revoke. Zero downtime."""
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)

    # Stage 1: Read the current key
    secret = client.secrets.kv.v2.read_secret_version(
        path="api-keys/third-party",
        mount_point="secret",
    )
    old_key = secret["data"]["data"]["api_key"]
    print(f"[ROTATE] Current key: {old_key[:6]}...{old_key[-4:]}")

What does this code do?

• APP_URL points to your running Flask app so the rotation script can verify health automatically.
• requests is imported for making HTTP calls to your Flask app during verification. It was already installed as a dependency of hvac.
• Stage 1 reads the current key from Vault so we know which key to revoke later.

• Below the Stage 1 print statement, add the generate and store stages:

    # Stage 2: Generate a new key at the provider (old key still valid!)
    new_key = provider.generate_key()
    print(f"[ROTATE] New key generated: {new_key[:6]}...{new_key[-4:]}")
    print(f"[ROTATE] Overlap window OPEN (both keys valid)")

    # Stage 3: Store the new key in Vault
    client.secrets.kv.v2.create_or_update_secret(
        path="api-keys/third-party",
        secret={"api_key": new_key},
        mount_point="secret",
    )
    print(f"[ROTATE] New key stored in Vault")

What does this code do?

Stage 2 generates a new key at the provider WITHOUT revoking the old one. This is the critical difference from the naive approach. Both keys are now valid simultaneously.

Stage 3 stores the new key in Vault so your Flask app picks it up on its next request. The old key is still valid at the provider, so even if a request sneaks in between store and verify, it works.

• Save rotate.py.

Seeing a syntax error?

Make sure the indentation of stages 2 and 3 matches stage 1. All three should be at one indent level inside the rotate() function.

help me fix an indentation error in my rotate.py file

Add verification and automatic rollback

The overlap window is open, but you need to confirm the app is actually healthy with the new key before revoking the old one. If verification fails for any reason, the script rolls back automatically.

• Below Stage 3 in rotate.py, add the verification logic with automatic rollback:

    # Stage 4: Verify the app works with the new key
    try:
        response = requests.get(f"{APP_URL}/status", timeout=5)
        data = response.json()
        if data.get("valid"):
            print(f"[ROTATE] Verification PASSED: app is healthy")
        else:
            print(f"[ROTATE] Verification FAILED! Rolling back...")
            client.secrets.kv.v2.create_or_update_secret(
                path="api-keys/third-party",
                secret={"api_key": old_key},
                mount_point="secret",
            )
            provider.revoke_key(new_key)
            print(f"[ROTATE] Rollback complete. Old key restored.")
            return False
    except Exception as e:
        print(f"[ROTATE] Verification error: {e}")
        print(f"[ROTATE] Rolling back...")
        client.secrets.kv.v2.create_or_update_secret(
            path="api-keys/third-party",
            secret={"api_key": old_key},
            mount_point="secret",
        )
        provider.revoke_key(new_key)
        print(f"[ROTATE] Rollback complete. Old key restored.")
        return False

What does the rollback do?

The verification hits your Flask app's /status endpoint. If it returns "valid": true, the new key works and rotation continues.

If verification fails (invalid key or network error), rollback kicks in: restore the old key in Vault and revoke the new key at the provider. The app returns to its previous working state as if nothing happened.

• Below Stage 4, add Stage 5 (revoke) and the main block to complete the file:

    # Stage 5: Revoke the old key (only after verification passes!)
    provider.revoke_key(old_key)
    print(f"[ROTATE] Old key revoked: {old_key[:6]}...{old_key[-4:]}")
    print(f"[ROTATE] Rotation complete! Zero downtime.")
    return True


if __name__ == "__main__":
    success = rotate()
    if success:
        print("\nKey rotation succeeded.")
    else:
        print("\nKey rotation failed and was rolled back.")

Why revoke last?

Stage 5 only runs after verification passes. The old key is revoked only when you have confirmed the new key works. This is the exact opposite of the naive approach, and it is why there is no downtime gap.

• Save rotate.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved rotate.py with all five stages.

ⓧ I'd like to double check the full code

Here is the complete rotate.py file. Scroll through and confirm your version matches:

• Chunk 1 (top of file): imports, constants, function definition, and Stage 1 (read current key).
• Chunk 2: Stage 2 (generate new key) and Stage 3 (store in Vault).
• Chunk 3: Stage 4 (verify with try/except and rollback in both branches).
• Chunk 4: Stage 5 (revoke old key) and the if __name__ block.

Not sure about the structure?

The function should have exactly one try/except block (Stage 4). Both the else branch and the except branch restore the old key and revoke the new key. Stage 5 sits OUTSIDE the try block, only reached if verification passes.

help me check the structure of my rotate.py file

Run the rotation and confirm zero downtime

Time to prove the dual-key pattern works. You will run your Flask app in one terminal and the rotation script in another, then confirm the app stays healthy the entire time.

• Switch back to the terminal where your Flask app ran previously.
• Start the Flask app by running:

python app.py

• Open a second terminal window. Press Cmd+Space (macOS) or the Windows key (Windows), type Terminal, and press Enter.
• Navigate to your project folder and activate the virtual environment by running:

cd ~/Desktop/key-rotation-pipeline
source venv/bin/activate

• Confirm the app is healthy before rotation by running:

curl http://127.0.0.1:5000/status

You should see a JSON response with "status": "healthy" and "valid": true.

Before you run the rotation, what do you think will happen to the /status endpoint during the rotation? Will it stay healthy the whole time, or will there be a gap like the naive version?

• Run the zero-downtime rotation by running:

python rotate.py

You see each stage printed in sequence: the current key, the new key generated, the overlap window opening, the new key stored in Vault, verification passing, and finally the old key revoked.

• Immediately confirm the app is still healthy by running:

curl http://127.0.0.1:5000/status

The response shows "status": "healthy" with a different masked key than before. The app never went down.

What just happened?

During the entire rotation, there was never a moment when no valid key existed. The new key was generated while the old key was still active. Vault was updated while both keys were valid. The old key was only revoked after the app confirmed it could use the new one.

This is the same generate-store-verify-revoke pattern used in production systems. The overlap window is what makes it zero-downtime.

Seeing verification FAILED or a connection error?

• Make sure app.py is running in the other terminal before you run rotate.py. The verification step calls http://127.0.0.1:5000/status and will fail if Flask is not serving.
• If you see Rollback complete in the output, the script safely restored the old key. Fix the issue and run python rotate.py again.

help me debug why my rotate.py verification step is failing

Your rotation script now swaps keys with zero downtime and rolls back automatically on failure. Next up, you will wrap this into a self-running pipeline that rotates keys on a schedule without any human intervention.

Automate the Pipeline

Your dual-key rotation works perfectly. But right now, it only runs when you manually execute the script.

In production, credentials rotate on a schedule without anyone pressing Enter. In this step, you'll wrap your rotation logic in a timed loop that fires automatically every 30 seconds. This transforms a one-shot script into a true self-maintaining pipeline.

In this step, get ready to:

• Write pipeline.py with a scheduled rotation loop.
• Observe the pipeline performing multiple automatic rotations.
• Test resilience by stopping the Flask app during a rotation.

Create the pipeline script

Your rotate.py already handles a single rotation cycle with full rollback logic. The pipeline script imports that function and calls it repeatedly on a timer.

• Create a new file called pipeline.py in your project folder. Start by adding the imports, interval constant, and the function header with its startup banner:

import time
from datetime import datetime
from rotate import rotate


ROTATION_INTERVAL_SECONDS = 30


def run_pipeline():
    """Run the rotation pipeline on a schedule."""
    rotation_count = 0

    print("=" * 60)
    print("  SELF-MAINTAINING KEY ROTATION PIPELINE")
    print(f"  Rotation interval: every {ROTATION_INTERVAL_SECONDS} seconds")
    print("=" * 60)
    print()
    print("Pipeline started. Press Ctrl+C to stop.")
    print()

What does this code do?

• from rotate import rotate pulls in the zero-downtime rotation function you built in the last step. The pipeline reuses it without duplicating any logic.
• ROTATION_INTERVAL_SECONDS = 30 sets how often the pipeline rotates keys. In production this might be hours or days. For testing, 30 seconds gives you fast feedback.
• The banner prints confirm the pipeline is running and remind you how to stop it.

• Now add the scheduling loop directly below the banner prints (still inside the run_pipeline function). This is the heart of the pipeline:

    try:
        while True:
            rotation_count += 1
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            print(f"\n{'\u2500' * 50}")
            print(f"  Rotation #{rotation_count} at {timestamp}")
            print(f"{'\u2500' * 50}")

            success = rotate()

            if success:
                print(f"  Result: SUCCESS")
            else:
                print(f"  Result: FAILED (rolled back)")

            print(f"\n  Next rotation in {ROTATION_INTERVAL_SECONDS} seconds...")
            time.sleep(ROTATION_INTERVAL_SECONDS)

    except KeyboardInterrupt:
        print(f"\n\nPipeline stopped. Total rotations: {rotation_count}")


if __name__ == "__main__":
    run_pipeline()

What does this code do?

• The while True loop runs forever until you press Ctrl+C. Each iteration calls rotate(), prints whether it succeeded or failed, then sleeps for 30 seconds.
• The try/except KeyboardInterrupt block catches Ctrl+C gracefully instead of crashing with a traceback. It prints a summary of how many rotations ran.
• The if __name__ == "__main__" guard means the loop only starts when you run pipeline.py directly. Other scripts can still import from it without triggering the loop.

• Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Your pipeline.py should contain these sections in order. Scroll up and confirm each is present:

• Imports (time, datetime, rotate) and the ROTATION_INTERVAL_SECONDS = 30 constant.
• The run_pipeline() function with the startup banner prints.
• The try/while True loop calling rotate(), printing the result, and sleeping for 30 seconds.
• The except KeyboardInterrupt handler and the if __name__ == "__main__" guard at the bottom.

Start the pipeline and observe automatic rotations

You need two terminal sessions for this: one running the Flask app (so the pipeline can verify health) and one running the pipeline itself.

• In your first terminal, start the Flask app by running:

python app.py

• Open a second terminal window. Activate your virtual environment in this new terminal, then navigate to your project folder.

Activating venv in a new terminal

Each new terminal window starts without your virtual environment active. Run source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows) from your project folder to reactivate it.

Before you run the pipeline, what do you think will happen? Will the rotation succeed or fail on the first cycle?

• Start the pipeline by running:

python pipeline.py

You should see the banner print, then the first rotation begins. The output shows each stage of the dual-key rotation happening automatically.

• Wait for 2-3 full rotation cycles (about 60-90 seconds). Each cycle should print Result: SUCCESS.
• While the pipeline is running, open your browser and visit http://127.0.0.1:5000/status to confirm the app stays healthy throughout every rotation.

What should I see?

Each rotation cycle prints a separator line, the rotation number, all five stages of the dual-key rotation (read, generate, store, verify, revoke), and Result: SUCCESS. Between rotations, you see a 30-second countdown message.

The /status endpoint returns "status": "healthy" at all times. The key value in the response changes after each rotation because the pipeline is swapping in new keys.

Pipeline shows FAILED on the first rotation?

• Make sure app.py is running in your other terminal. The pipeline verifies health by calling http://127.0.0.1:5000/status. If the Flask app is not running, verification fails and the pipeline rolls back.
• Confirm the Vault container is still running. If Docker Desktop went to sleep or you restarted your machine, Vault's in-memory data is gone. Re-run the docker run command and python setup_vault.py to restore the initial key.

Having trouble?

Test resilience by stopping the Flask app

A real pipeline must handle failures gracefully. If the consumer app goes down during a rotation, the pipeline should detect the problem, roll back, and keep trying on the next cycle.

• Leave the pipeline running in its terminal.
• Switch to the terminal running app.py and press Ctrl+C to stop the Flask app.
• Switch back to the pipeline terminal and watch the next rotation cycle.

What do you think will happen when the pipeline tries to verify health against a stopped app?

The pipeline attempts the rotation, but when it calls /status to verify, it cannot reach the Flask app. You should see [ROTATE] Verification error: followed by [ROTATE] Rollback complete. Old key restored. and then Result: FAILED (rolled back).

Why does the pipeline survive this?

The rotate() function returns False on failure instead of crashing. The pipeline prints the failure, waits 30 seconds, and tries again. This is the difference between a brittle script and a resilient pipeline.

The rollback means no damage was done. The old key is still valid in Vault and at the provider.

• Now restart the Flask app in its terminal by running:

python app.py

• Watch the pipeline terminal for the next rotation cycle. It should print Result: SUCCESS again, proving the pipeline recovered automatically.
• Press Ctrl+C in the pipeline terminal to stop it. You should see Pipeline stopped. Total rotations: followed by the count.

Pipeline never recovers after restarting Flask?

• If valid_keys.json got corrupted during repeated rollbacks, run python setup_vault.py to reset the state with a fresh key, then restart the pipeline.
• Make sure you restarted Flask on the same port (5000). The pipeline calls http://127.0.0.1:5000/status for verification.

Help me debug my pipeline recovery.

Your pipeline is fully operational. It rotates keys automatically, handles failures with rollback, and recovers on the next cycle without any human intervention.

You've built a complete self-maintaining key rotation system. Next up, a secret mission that adds audit logging to track every rotation event.

Secret mission

Add Rotation Audit Logging

Your pipeline rotates keys, but there is no record of what happened. Add persistent audit logging so every rotation event is recorded with timestamps, masked keys, and success or failure status. This transforms your pipeline from a working demo into something audit-ready.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Docker container running HashiCorp Vault (in-memory, no persistent data).
• Docker image hashicorp/vault:1.21 stored locally.
• Python virtual environment (venv folder) with installed packages.
• Project files: provider.py, app.py, pipeline.py, rotate.py, rotate_naive.py, setup_vault.py, valid_keys.json, rotation_log.json.

Keep everything running

No action needed. Choose this if you are still actively experimenting with the rotation pipeline or want to try different rotation intervals and observe the audit log grow.

• Your Vault container continues running in Docker and keeps serving secrets.
• Docker Desktop stays installed on your machine for future container-based projects.
• Your project files and virtual environment remain on your Desktop.

Pause - I'll come back to this later

Stop running processes to free up memory. Your project files stay on disk so you can resume later.

• Stop the Flask app and pipeline by pressing Ctrl+C in each terminal where they are running.
• Find your Vault container ID by running:

docker ps

• Stop the Vault container by running:

docker stop <CONTAINER_ID>

• Deactivate your Python virtual environment by running:

deactivate

What happens to my Vault data?

Vault dev mode uses in-memory storage. When you stop the container, all secrets are lost. This is expected. When you come back, restart the container with the same docker run command from Step 1 and re-run python setup_vault.py to seed a fresh key.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step in order.

• Stop the Flask app and pipeline by pressing Ctrl+C in each terminal where they are running.
• Find your Vault container ID by running:

docker ps

• Force-remove the Vault container by running:

docker rm -f <CONTAINER_ID>

• Remove the Vault Docker image by running:

docker rmi hashicorp/vault:1.21

• Deactivate your virtual environment (if still active) by running:

deactivate

• Delete the project folder and all its contents. On macOS or Linux, run:

cd ~/Desktop
rm -rf key-rotation-pipeline

• On Windows, run:

cd ~\Desktop
rmdir /s /q key-rotation-pipeline

• Optionally, uninstall Docker Desktop. On macOS, open Docker Desktop, click the Troubleshoot icon in the top-right corner, select Uninstall, and confirm. On Windows, open Settings from the Start menu, select Apps, then Apps & features, find Docker Desktop, and select Uninstall.

That's a Wrap!

That's a Wrap!

Nice work! You just built a fully automated key rotation pipeline that swaps credentials with zero downtime, all running locally on your machine.

You've learned how to:

• Set up HashiCorp Vault as a secrets management server and built a Flask app that reads credentials dynamically from it on every request.
• Implemented the industry-standard dual-key overlap rotation pattern (generate, store, verify, revoke) with automatic rollback on failure.
• Built a self-maintaining pipeline that rotates API keys automatically without human intervention while your app stays healthy.
• Secret Mission: Added audit logging to create a traceable record of every rotation event with timestamps and masked keys.

Ready to quiz yourself?