Configure Secure API Keys for Gold Data Streams in 20 Minutes (AWS KMS Guide)

# Personal note: Learned to add aliases after searching for 
# "arn:aws:kms:us-east-1:123456789012:key/abc-def-ghi" in 50 keys

aws kms create-key \
  --description "Gold Data Stream API encryption" \
  --key-usage ENCRYPT_DECRYPT \
  --origin AWS_KMS \
  --multi-region false

# Save the KeyId from output
export KMS_KEY_ID="12345678-abcd-1234-abcd-123456789012"

# Create a friendly alias
aws kms create-alias \
  --alias-name alias/gold-data-api \
  --target-key-id $KMS_KEY_ID

{
    "KeyMetadata": {
        "KeyId": "12345678-abcd-1234-abcd-123456789012",
        "KeyState": "Enabled",
        "CreationDate": "2025-10-28T14:23:47-05:00"
    }
}

# encrypt_api_key.py
# Personal note: Took me 3 tries to get the base64 encoding right

import boto3
import base64
import os

def encrypt_gold_api_key(plaintext_key, kms_key_id):
    """
    Encrypts API key with KMS.
    Returns base64-encoded ciphertext safe for storage.
    """
    kms = boto3.client('kms', region_name='us-east-1')
    
    # Watch out: KMS expects bytes, not strings
    response = kms.encrypt(
        KeyId=kms_key_id,
        Plaintext=plaintext_key.encode('utf-8'),
        EncryptionContext={
            'service': 'gold-data-stream',
            'environment': 'production'  # Helps with audit trails
        }
    )
    
    # Base64 encode so it's safe for JSON/YAML
    encrypted = base64.b64encode(response['CiphertextBlob']).decode('utf-8')
    
    print(f"✓ Encrypted {len(plaintext_key)} character key")
    print(f"✓ Ciphertext length: {len(encrypted)} chars")
    return encrypted

# Usage
if __name__ == "__main__":
    api_key = input("Enter your Gold Data API key: ")
    kms_key = os.environ.get('KMS_KEY_ID', 'alias/gold-data-api')
    
    encrypted_key = encrypt_gold_api_key(api_key, kms_key)
    print(f"\nStore this encrypted key:\n{encrypted_key}")

# gold_data_client.py
# Personal note: Added retry logic after AWS had a 2-minute KMS outage

import boto3
import base64
import time
from botocore.exceptions import ClientError

class GoldDataClient:
    def __init__(self, encrypted_api_key, region='us-east-1'):
        self.encrypted_key = encrypted_api_key
        self.kms = boto3.client('kms', region_name=region)
        self._api_key_cache = None
        self._cache_expiry = 0
    
    def _decrypt_api_key(self):
        """
        Decrypts API key with KMS.
        Caches for 5 minutes to reduce KMS calls ($0.03 per 10,000 requests).
        """
        # Check cache first
        if self._api_key_cache and time.time() < self._cache_expiry:
            return self._api_key_cache
        
        try:
            # Watch out: Must decode base64 before sending to KMS
            ciphertext_blob = base64.b64decode(self.encrypted_key)
            
            response = self.kms.decrypt(
                CiphertextBlob=ciphertext_blob,
                EncryptionContext={
                    'service': 'gold-data-stream',
                    'environment': 'production'
                }
            )
            
            plaintext_key = response['Plaintext'].decode('utf-8')
            
            # Cache for 5 minutes
            self._api_key_cache = plaintext_key
            self._cache_expiry = time.time() + 300
            
            return plaintext_key
            
        except ClientError as e:
            if e.response['Error']['Code'] == 'AccessDeniedException':
                raise Exception("IAM role missing kms:Decrypt permission")
            raise
    
    def get_gold_price(self, symbol='XAU'):
        """
        Fetches real-time gold price.
        Automatically handles key decryption.
        """
        api_key = self._decrypt_api_key()
        
        # Your actual API call here
        # Using requests library as example
        import requests
        
        response = requests.get(
            'https://metals-api.com/api/latest',
            params={'access_key': api_key, 'symbols': symbol}
        )
        
        if response.status_code == 401:
            # Clear cache on auth failure
            self._api_key_cache = None
            raise Exception("API key invalid - check key rotation")
        
        return response.json()

# Usage
if __name__ == "__main__":
    # This encrypted key is safe to commit to Git
    ENCRYPTED_KEY = "AQICAHhN8..."  # From Step 2
    
    client = GoldDataClient(ENCRYPTED_KEY)
    
    start = time.time()
    price_data = client.get_gold_price()
    elapsed = (time.time() - start) * 1000
    
    print(f"✓ Gold price: ${price_data['rates']['XAU']}")
    print(f"✓ Request time: {elapsed:.0f}ms (includes KMS decrypt)")

# key_rotation.py
# Personal note: Built this after our gold provider force-rotated keys
# and broke our prod monitoring at 3 AM

import boto3
import os
from datetime import datetime, timedelta

class KeyRotationManager:
    def __init__(self, kms_key_id):
        self.kms_key_id = kms_key_id
        self.kms = boto3.client('kms')
    
    def rotate_api_key(self, new_plaintext_key):
        """
        Encrypts new API key while keeping old one active.
        Returns both encrypted keys for gradual rollout.
        """
        # Encrypt new key
        new_encrypted = self._encrypt_key(new_plaintext_key, version='v2')
        
        print(f"✓ New key encrypted at {datetime.now()}")
        print(f"✓ Deploy new key to 10% of servers first")
        print(f"✓ Monitor for 24 hours before full rollout")
        
        return {
            'new_encrypted_key': new_encrypted,
            'rotation_date': datetime.now().isoformat(),
            'full_rollout_after': (datetime.now() + timedelta(days=1)).isoformat()
        }
    
    def _encrypt_key(self, plaintext_key, version):
        """Encrypts with version tracking"""
        response = self.kms.encrypt(
            KeyId=self.kms_key_id,
            Plaintext=plaintext_key.encode('utf-8'),
            EncryptionContext={
                'service': 'gold-data-stream',
                'environment': 'production',
                'version': version,  # Track which key is which
                'rotated_at': datetime.now().isoformat()
            }
        )
        
        import base64
        return base64.b64encode(response['CiphertextBlob']).decode('utf-8')

# Usage example
if __name__ == "__main__":
    manager = KeyRotationManager('alias/gold-data-api')
    
    new_api_key = input("Enter new Gold Data API key: ")
    rotation_info = manager.rotate_api_key(new_api_key)
    
    print("\nRotation config for your deployment:")
    print(f"NEW_ENCRYPTED_KEY={rotation_info['new_encrypted_key']}")