API Key Rotation — Automated Key Rotation Without Service Disruption
In this tutorial, you will learn about API Key Rotation. We cover key concepts, practical examples, and best practices to help you master this topic.
API key rotation replaces cryptographic credentials on a regular schedule using overlapping validity windows, ensuring continuous service availability while limiting the risk window for compromised keys.
What You'll Learn
Staggered key rotation with overlapping validity periods, automated rotation endpoints, rotation hooks for downstream services, monitoring rotation health, and emergency key rotation.
Why It Matters
Static API keys accumulate risk over time. Regular rotation limits the damage window — a key rotated weekly is at risk for at most one week. Automated rotation reduces human error and ensures Compliance with security policies.
Real-World Use
AWS rotates API access keys on configurable schedules. Stripe requires periodic key rotation. Durga Antivirus Pro rotates partner API keys every 90 days with a 7-day overlap period for seamless transition.
Code Example: Staggered Key Rotation with Overlap
import secrets, hashlib, datetime, time
from flask import Flask, request, jsonify
app = Flask(__name__)
# Two active keys per service — primary and secondary
api_keys = {}
def generate_api_key():
"""Generate a new API key with a unique prefix."""
return f"dk_{secrets.token_hex(32)}"
def hash_key(key):
return hashlib.sha256(key.encode()).hexdigest()
def create_key_entry(service_name, expires_in_days=90):
"""Create a new API key entry with expiry."""
raw_key = generate_api_key()
key_hash = hash_key(raw_key)
now = datetime.datetime.utcnow()
entry = {
"service": service_name,
"hash": key_hash,
"created_at": now.isoformat(),
"expires_at": (now + datetime.timedelta(days=expires_in_days)).isoformat(),
"active": True,
"is_primary": False,
"scopes": ["threat:read", "threat:write"]
}
return raw_key, entry
@app.route("/api/v1/keys/rotate", methods=["POST"])
def rotate_key():
"""Rotate API key with overlapping validity."""
service_name = request.json.get("service", "")
current_primary = request.json.get("current_key_id")
if service_name not in api_keys:
api_keys[service_name] = []
keys = api_keys[service_name]
# Deactivate old primary
for key in keys:
if key.get("id") == current_primary:
key["active"] = True # Keep active for overlap period
key["is_primary"] = False
key["rotated_at"] = datetime.datetime.utcnow().isoformat()
key["rotation_scheduled_expiry"] = (
datetime.datetime.utcnow() + datetime.timedelta(days=7)
).isoformat()
# Create new primary key
raw_key, entry = create_key_entry(service_name, expires_in_days=90)
entry["is_primary"] = True
entry["id"] = f"key_{secrets.token_hex(8)}"
keys.append(entry)
# Clean up expired keys
now = datetime.datetime.utcnow()
api_keys[service_name] = [
k for k in keys
if not k.get("rotation_scheduled_expiry") or
datetime.datetime.fromisoformat(k["rotation_scheduled_expiry"]) > now
]
return jsonify({
"new_key": raw_key,
"key_id": entry["id"],
"primary_since": entry["created_at"],
"overlap_end": entry["rotation_scheduled_expiry"],
"message": "Previous key remains valid for 7 days"
})
Code Example: Automated Key Rotation Service
import schedule, time, requests
class KeyRotationScheduler:
"""Automated key rotation on a configurable schedule."""
def __init__(self, api_base_url, admin_token):
self.api_base_url = api_base_url
self.admin_token = admin_token
self.services = {}
def register_service(self, service_name, rotation_days=90):
"""Register a service for automatic rotation."""
self.services[service_name] = {
"rotation_days": rotation_days,
"last_rotated": None,
"next_rotation": None
}
self._schedule_service(service_name, rotation_days)
def _schedule_service(self, service_name, days):
"""Schedule rotation for a service."""
schedule.every(days).days.do(
self.rotate_service, service_name
)
def rotate_service(self, service_name):
"""Rotate a service's API key."""
resp = requests.post(
f"{self.api_base_url}/api/v1/keys/rotate",
json={"service": service_name},
headers={"Authorization": f"Bearer {self.admin_token}"}
)
if resp.status_code == 200:
data = resp.json()
self.services[service_name]["last_rotated"] = \
datetime.datetime.utcnow().isoformat()
self.services[service_name]["next_rotation"] = \
(datetime.datetime.utcnow() +
datetime.timedelta(days=self.services[service_name]["rotation_days"])
).isoformat()
# Notify downstream services
self.notify_downstream(service_name, data["key_id"])
return {"status": "rotated", "data": data}
return {"status": "failed", "error": resp.text}
def notify_downstream(self, service_name, new_key_id):
"""Notify downstream services about the rotation."""
webhooks = get_downstream_webhooks(service_name)
for webhook in webhooks:
try:
requests.post(webhook, json={
"event": "key_rotation",
"service": service_name,
"new_key_id": new_key_id,
"overlap_end": (
datetime.datetime.utcnow() +
datetime.timedelta(days=7)
).isoformat()
}, timeout=5)
except requests.RequestException:
pass # Log and continue
def start(self):
"""Start the rotation scheduler."""
while True:
schedule.run_pending()
time.sleep(60)
# Usage
scheduler = KeyRotationScheduler(
api_base_url="https://api.durga-antivirus.com",
admin_token=os.environ["ADMIN_TOKEN"]
)
scheduler.register_service("threat-scanner", rotation_days=90)
scheduler.register_service("alert-dispatcher", rotation_days=30)
scheduler.start()
Code Example: Client-Side Key Rotation Handling
class RotatingAPIClient:
"""API client that handles key rotation transparently."""
def __init__(self, base_url, key_provider):
self.base_url = base_url
self.key_provider = key_provider
self.current_key = None
self.fallback_key = None
def _get_headers(self):
"""Get auth headers with the current key."""
self._refresh_keys_if_needed()
headers = {"X-API-Key": self.current_key}
if self.fallback_key:
headers["X-API-Key-Fallback"] = self.fallback_key
return headers
def _refresh_keys_if_needed(self):
"""Refresh keys from provider when rotation occurs."""
keys = self.key_provider.get_active_keys()
if keys["primary"] != self.current_key:
self.fallback_key = self.current_key
self.current_key = keys["primary"]
def request(self, method, path, **kwargs):
"""Make API request with automatic key fallback."""
headers = {**kwargs.pop("headers", {}), **self._get_headers()}
resp = requests.request(
method, f"{self.base_url}{path}",
headers=headers, **kwargs
)
if resp.status_code == 401:
# Key might have expired — try fallback
if self.fallback_key:
headers["X-API-Key"] = self.fallback_key
resp = requests.request(
method, f"{self.base_url}{path}",
headers=headers, **kwargs
)
if resp.status_code == 200:
self.current_key = self.fallback_key
self.fallback_key = None
return resp
# Force key refresh
self.current_key = None
self.fallback_key = None
self.key_provider.refresh_keys()
return self.request(method, path, **kwargs)
return resp
class KeyProvider:
"""Manages key storage and retrieval."""
def __init__(self, storage_path="~/.api_keys.json"):
self.storage_path = os.path.expanduser(storage_path)
def get_active_keys(self):
"""Get current active keys from secure storage."""
with open(self.storage_path) as f:
return json.load(f)
def refresh_keys(self):
"""Refresh keys — called when current keys fail."""
print("Keys need refresh. Run: key-rotation-client refresh")
raise Exception("Keys expired — manual refresh required")
Common Mistakes
1. Immediate Key Deletion on Rotation
Deleting the old key immediately causes downtime for services that cached the old key. Use overlapping validity (7-30 days) for zero-downtime rotation.
2. No Rotation Notification System
When a key is rotated, downstream services and dashboards must be notified. Implement Webhooks or a polling endpoint that returns key metadata.
3. Manual Rotation Without Automation
Manual rotation is error-prone and often forgotten. Automate rotation with a scheduled task and monitor rotation health.
4. Not Testing Rotation in Staging
Key rotation often breaks integrations. Test rotation in a staging environment first, verifying that both old and new keys work during the overlap period.
5. No Emergency Rotation Procedure
When a key is compromised, you need immediate rotation without overlap. Implement an emergency rotation endpoint that revokes the old key immediately and creates a new one.
Practice Questions
- Why should key rotation have an overlapping validity period?
- How does staggered rotation prevent downtime?
- Why should rotation be automated rather than manual?
- What is an emergency rotation and how does it differ from scheduled rotation?
- How do downstream services discover rotated keys?
Answers:
- Services cache keys and may not pick up the new key immediately. Overlap ensures both old and new keys work during the transition, preventing authentication failures.
- Staggered rotation keeps the previous key active while the new key propagates through caches and CDNs. Each key has a scheduled expiry after the new key is active.
- Manual rotation is forgotten, delayed, or done incorrectly. Automated rotation runs on schedule, logs the event, and notifies dependent systems.
- Emergency rotation immediately revokes the compromised key without overlap. All services must immediately obtain the new key. This causes temporary disruption but contains the breach.
- Via Webhook notifications, a polling endpoint (/keys/active), or a shared key management service (Vault, AWS Secrets Manager).
Challenge: Build an API key management system with staggered rotation, 90-day rotation schedule, 7-day overlap, webhook notifications to downstream services, and emergency rotation capability.
FAQ
Mini Project
Build a key rotation management system with automated 90-day rotation schedule, 7-day overlapping validity, webhook notifications, emergency rotation endpoint, and a dashboard showing key status and rotation history per service.
What's Next
Now explore OAuth2 Authorization Code with PKCE for secure public client authentication without static secrets.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro