AWS KMS & Secrets Management for Data Engineers

Master encryption keys and secure credential management for compliant data engineering pipelines.

9 min readIntermediate

Why This Matters

Encryption and secrets management are non-negotiable in data engineering. Every data pipeline handles sensitive information - customer PII, financial records, API keys, database credentials. Without proper key management and secrets rotation, you risk data breaches, compliance violations, and unauthorized access. Interviewers expect you to articulate not just how KMS and Secrets Manager work, but why they are essential for secure, compliant data platforms.

What is AWS KMS?

AWS Key Management Service (KMS) is a managed service that makes it easy to create and control the cryptographic keys used to protect your data. KMS uses hardware security modules (HSMs) to protect and manage your keys.

Key Concepts

Customer Managed Keys (CMK): Keys you create and control in KMS
AWS Managed Keys: Keys created and managed by AWS for specific services
Data Keys: Encryption keys used to encrypt data, encrypted by a CMK
Key Policy: Defines who can use and manage the key
Key Rotation: Automatic annual rotation of key material

KMS Key Types

Type	Description	Use Case
Symmetric	Single encryption key	Most common, S3, EBS, RDS encryption
Asymmetric	Public/private key pair	Digital signatures, public key encryption
HMAC	Hash-based message authentication	Data integrity verification

KMS Key Policies

Key policies define who can use and manage KMS keys. Every CMK must have a key policy.

Key Policy Example

{
  "Version": "2012-10-17",
  "Id": "key-default-1",
  "Statement": [
    {
      "Sid": "EnableRootAccount",
      "Effect": "Allow",
      "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
      "Action": "kms:*",
      "Resource": "*"
    },
    {
      "Sid": "AllowGlueJobAccess",
      "Effect": "Allow",
      "Principal": {"AWS": "arn:aws:iam::123456789012:role/GlueJobRole"},
      "Action": [
        "kms:Decrypt",
        "kms:GenerateDataKey"
      ],
      "Resource": "*",
      "Condition": {
        "StringEquals": {
          "kms:ViaService": "s3.us-east-1.amazonaws.com"
        }
      }
    }
  ]
}

Key Policy Best Practices

Enable root account access for emergency situations
Grant minimal permissions to specific roles, not broad access
Use kms:ViaService conditions to limit key usage to specific services
Enable automatic key rotation for CMKs
Use grants for temporary access instead of modifying key policies

AWS Secrets Manager

AWS Secrets Manager is a service that helps you protect secrets needed to access your applications, services, and IT resources. It enables you to easily rotate, manage, and retrieve database credentials, API keys, and other secrets.

Key Features

Automatic rotation: Rotate secrets on a schedule using Lambda functions
Centralized management: Store and manage all secrets in one place
Fine-grained access control: Use IAM policies to control who can access secrets
Audit and compliance: Track secret access via CloudTrail
Cross-account access: Share secrets across AWS accounts

Secrets Manager vs Parameter Store

Feature	Secrets Manager	Parameter Store
Automatic rotation	Yes (built-in)	No (manual Lambda)
Cost	0.05/10K API calls	Free tier available
Encryption	AWS KMS	SSM-managed or KMS
Max secret size	64 KB	8 KB (standard), 8 MB (advanced)
Versioning	Yes (automatic)	Yes (manual)
Best for	Database credentials, API keys	Configuration values, feature flags

Secrets Manager Rotation Example

import boto3
import json

secrets_client = boto3.client('secretsmanager')

# Store a database secret
secret_value = {
    'username': 'admin',
    'password': 'initial-password-123',
    'engine': 'mysql',
    'host': 'mydb.cluster-123456.us-east-1.rds.amazonaws.com',
    'port': 3306,
    'dbname': 'mydatabase'
}

response = secrets_client.create_secret(
    Name='prod/database/credentials',
    Description='Production RDS credentials',
    SecretString=json.dumps(secret_value)
)

print(f"Secret ARN: {response['ARN']}")

# Retrieve the secret
response = secrets_client.get_secret_value(SecretId='prod/database/credentials')
secret = json.loads(response['SecretString'])
print(f"Username: {secret['username']}")

KMS Usage in Data Engineering

S3 Encryption with KMS

import boto3

s3 = boto3.client('s3')

# Upload with SSE-KMS encryption
s3.put_object(
    Bucket='my-data-lake',
    Key='raw/data/file.parquet',
    Body=open('file.parquet', 'rb'),
    ServerSideEncryption='aws:kms',
    SSEKMSKeyId='arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012'
)

EBS Encryption

ec2 = boto3.client('ec2')

# Create encrypted EBS volume
response = ec2.create_volume(
    AvailabilityZone='us-east-1a',
    Size=100,
    VolumeType='gp3',
    Encrypted=True,
    KmsKeyId='arn:aws:kms:us-east-1:123456789012:key/my-key-id'
)

Redshift Encryption

redshift = boto3.client('redshift')

# Create encrypted Redshift cluster
response = redshift.create_cluster(
    ClusterIdentifier='my-data-warehouse',
    NodeType='dc2.large',
    MasterUsername='admin',
    MasterUserPassword='password123',
    Encrypted=True,
    KmsKeyId='arn:aws:kms:us-east-1:123456789012:key/my-key-id'
)

Real-World Project Structure

Architecture Diagram

security-configuration/
├── kms/
│   ├── customer-managed-keys/
│   │   ├── s3-encryption-key.json
│   │   ├── ebs-encryption-key.json
│   │   ├── redshift-encryption-key.json
│   │   └── glue-encryption-key.json
│   ├── key-policies/
│   │   ├── data-lake-key-policy.json
│   │   ├── cross-account-key-policy.json
│   │   └── service-specific-key-policy.json
│   └── key-rotation/
│       ├── rotation-schedule.json
│       └── rotation-lambda-function.py
├── secrets/
│   ├── database-credentials/
│   │   ├── prod-rds-credentials.json
│   │   ├── prod-redshift-credentials.json
│   │   └── rotation-lambda-function.py
│   ├── api-keys/
│   │   ├── external-api-keys.json
│   │   └── rotation-lambda-function.py
│   └── certificates/
│       ├── tls-certificates.json
│       └── rotation-lambda-function.py
├── iam/
│   ├── kms-usage-roles/
│   │   ├── glue-kms-role.json
│   │   ├── lambda-kms-role.json
│   │   └── redshift-kms-role.json
│   └── secrets-access-roles/
│       ├── glue-secrets-role.json
│       └── lambda-secrets-role.json
├── compliance/
│   ├── encryption-standards.md
│   ├── key-rotation-standards.md
│   └── audit-checklist.md
└── scripts/
    ├── create-kms-keys.py
    ├── manage-secrets.py
    └── validate-encryption.py

Production Python Code

import boto3
import json
from botocore.exceptions import ClientError

class KMSSecretsManager:
    """Manages KMS keys and Secrets Manager for data engineering."""

    def __init__(self, region='us-east-1'):
        self.region = region
        self.kms = boto3.client('kms', region_name=region)
        self.secrets = boto3.client('secretsmanager', region_name=region)

    def create_data_lake_key(self, alias_name='data-lake-key'):
        """Create a customer managed KMS key for data lake encryption."""
        try:
            key = self.kms.create_key(
                Description='Key for data lake encryption',
                KeyUsage='ENCRYPT_DECRYPT',
                KeySpec='SYMMETRIC_DEFAULT',
                Tags:[
                    {'TagKey': 'Environment', 'TagValue': 'production'},
                    {'TagKey': 'Purpose', 'TagValue': 'data-lake'}
                ]
            )

            key_id = key['KeyMetadata']['KeyId']

            self.kms.create_alias(
                AliasName=f'alias/{alias_name}',
                TargetKeyId=key_id
            )

            self.kms.enable_key_rotation(KeyId=key_id)

            self.kms.put_key_policy(
                KeyId=key_id,
                PolicyName='default',
                Policy=json.dumps({
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "EnableRootAccount",
                            "Effect": "Allow",
                            "Principal": {"AWS": f"arn:aws:iam::{self._get_account_id()}:root"},
                            "Action": "kms:*",
                            "Resource": "*"
                        },
                        {
                            "Sid": "AllowGlueAccess",
                            "Effect": "Allow",
                            "Principal": {"AWS": "arn:aws:iam::123456789012:role/GlueJobRole"},
                            "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
                            "Resource": "*"
                        }
                    ]
                })
            )

            return key_id
        except ClientError as e:
            print(f"Error creating KMS key: {e.response['Error']['Message']}")
            raise

    def create_database_secret(self, secret_name, db_host, db_user, db_password, db_name):
        """Create a database credential secret with rotation."""
        try:
            secret_value = {
                'host': db_host,
                'username': db_user,
                'password': db_password,
                'dbname': db_name,
                'port': 3306,
                'engine': 'mysql'
            }

            response = self.secrets.create_secret(
                Name=secret_name,
                Description=f'Database credentials for {db_name}',
                SecretString=json.dumps(secret_value),
                Tags=[
                    {'Key': 'Environment', 'Value': 'production'},
                    {'Key': 'Purpose', 'Value': 'database-credentials'}
                ]
            )

            rotation_lambda_arn = self._create_rotation_lambda(secret_name)

            self.secrets.rotate_secret(
                SecretId=secret_name,
                RotationLambdaARN=rotation_lambda_arn,
                RotationRules={
                    'AutomaticallyAfterDays': 30
                }
            )

            return response['ARN']
        except ClientError as e:
            print(f"Error creating secret: {e.response['Error']['Message']}")
            raise

    def get_secret_value(self, secret_name):
        """Retrieve a secret value."""
        try:
            response = self.secrets.get_secret_value(SecretId=secret_name)
            return json.loads(response['SecretString'])
        except ClientError as e:
            print(f"Error retrieving secret: {e.response['Error']['Message']}")
            raise

    def encrypt_data_key(self, key_id, plaintext):
        """Encrypt a data key using KMS."""
        try:
            response = self.kms.encrypt(
                KeyId=key_id,
                Plaintext=plaintext
            )
            return response['CiphertextBlob']
        except ClientError as e:
            print(f"Error encrypting data key: {e.response['Error']['Message']}")
            raise

    def decrypt_data_key(self, ciphertext_blob):
        """Decrypt a data key using KMS."""
        try:
            response = self.kms.decrypt(
                CiphertextBlob=ciphertext_blob
            )
            return response['Plaintext']
        except ClientError as e:
            print(f"Error decrypting data key: {e.response['Error']['Message']}")
            raise

    def _get_account_id(self):
        """Get current AWS account ID."""
        sts = boto3.client('sts')
        return sts.get_caller_identity()['Account']

    def _create_rotation_lambda(self, secret_name):
        """Create a Lambda function for secret rotation."""
        lambda_client = boto3.client('lambda')

        rotation_script = '''
import boto3
import json
import random
import string

def lambda_handler(event, context):
    secrets_client = boto3.client('secretsmanager')

    secret_arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']

    if step == "createSecret":
        secret = secrets_client.get_secret_value(SecretId=secret_arn)
        current_dict = json.loads(secret['SecretString'])

        new_password = ''.join(random.choices(
            string.ascii_letters + string.digits + string.punctuation, k=32
        ))

        current_dict['password'] = new_password

        secrets_client.put_secret_value(
            SecretId=secret_arn,
            ClientRequestToken=token,
            SecretString=json.dumps(current_dict),
            VersionStages=['AWSPENDING']
        )

    elif step == "setSecret":
        pass

    elif step == "testSecret":
        pass

    elif step == "finishSecret":
        metadata = secrets_client.describe_secret(SecretId=secret_arn)

        current_version = None
        for version, stages in metadata['VersionIdsToStages'].items():
            if 'AWSCURRENT' in stages:
                current_version = version
                break

        secrets_client.update_secret_version_stage(
            SecretId=secret_arn,
            VersionStage='AWSCURRENT',
            MoveToVersionId=token,
            RemoveFromVersionId=current_version
        )
'''

        response = lambda_client.create_function(
            FunctionName=f'{secret_name}-rotation',
            Runtime='python3.9',
            Role='arn:aws:iam::123456789012:role/LambdaRotationRole',
            Handler='lambda_function.lambda_handler',
            Code={'ZipFile': rotation_script.encode()},
            Timeout=30,
            MemorySize=128
        )

        return response['FunctionArn']

if __name__ == '__main__':
    manager = KMSSecretsManager(region='us-east-1')

    key_id = manager.create_data_lake_key('data-lake-key')
    print(f"Created KMS key: {key_id}")

    secret_arn = manager.create_database_secret(
        'prod/database/credentials',
        'mydb.cluster-123456.us-east-1.rds.amazonaws.com',
        'admin',
        'initial-password-123',
        'mydatabase'
    )
    print(f"Created secret: {secret_arn}")

    secret = manager.get_secret_value('prod/database/credentials')
    print(f"Retrieved username: {secret['username']}")

Mathematical Formations

KMS Cost Calculation

def calculate_kms_cost(keys_count, api_calls_monthly=100000):
    """
    Calculate monthly KMS cost.

    Key cost: $1/key/month
    API calls: $0.03 per 10,000 calls
    """
    key_cost = keys_count * 1.00
    api_cost = (api_calls_monthly / 10000) * 0.03

    return round(key_cost + api_cost, 2)

# 5 KMS keys with 100K API calls/month
monthly_cost = calculate_kms_cost(5, 100000)
print(f"KMS monthly cost: ${monthly_cost}")

Secrets Manager Cost

def calculate_secrets_cost(secrets_count, api_calls_monthly=50000):
    """
    Calculate monthly Secrets Manager cost.

    Secret cost: $0.40/secret/month
    API calls: $0.05 per 10,000 calls
    """
    secret_cost = secrets_count * 0.40
    api_cost = (api_calls_monthly / 10000) * 0.05

    return round(secret_cost + api_cost, 2)

# 10 secrets with 50K API calls/month
monthly_cost = calculate_secrets_cost(10, 50000)
print(f"Secrets Manager monthly cost: ${monthly_cost}")

Encryption Throughput Formula

def calculate_encryption_throughput(file_size_gb, encryption_overhead_pct=0.05):
    """
    Calculate effective throughput with KMS encryption.

    Base throughput: ~100 MB/s for S3 uploads
    Encryption overhead: ~5% for KMS calls
    """
    base_throughput_mbps = 100
    effective_throughput = base_throughput_mbps * (1 - encryption_overhead_pct)
    transfer_time_seconds = (file_size_gb * 1024 * 1024) / (effective_throughput * 1024 * 1024)

    return round(transfer_time_seconds, 1)

# 100 GB file with KMS encryption
transfer_time = calculate_encryption_throughput(100)
print(f"Transfer time: {transfer_time} seconds")

Performance Considerations

Factor	Recommendation	Impact
Key Aliases	Use aliases instead of key IDs	Simplifies key management
Key Rotation	Enable automatic annual rotation	Compliance and security
Bucket Key	Enable for S3 SSE-KMS	Reduces KMS costs 99%
Envelope Encryption	Use data keys for large datasets	Reduces KMS API calls
Key Caching	Cache decrypted keys locally	Reduces latency
Batch KMS Calls	Batch encrypt/decrypt operations	Improves throughput
Regional Keys	Create keys in same region as data	Reduces cross-region latency

Security Considerations

Enable automatic key rotation for all CMKs
Use least-privilege key policies (not root access for services)
Enable CloudTrail logging for all KMS operations
Use kms:ViaService conditions to restrict key usage
Store secrets in Secrets Manager, not Parameter Store, for rotation
Use VPC endpoints for Secrets Manager API calls
Enable secret rotation on a 30-day schedule
Use temporary credentials from STS instead of long-lived keys
Monitor KMS usage via CloudWatch metrics
Use key grants for temporary cross-account access

Common Pitfalls

Pitfall	Consequence	Prevention
Hardcoding KMS keys	Cannot rotate keys without code changes	Use key aliases instead
Overly permissive key policies	Unauthorized key usage	Use least-privilege policies
Not enabling key rotation	Compliance violations	Enable automatic rotation
Storing secrets in code	Security breach risk	Use Secrets Manager
Not enabling secret rotation	Long-lived credentials	Enable automatic rotation
Using default AWS managed keys	No control over rotation/audit	Create customer managed keys
Not enabling CloudTrail	Cannot audit key usage	Enable for all KMS operations
Using same key for all services	Blast radius of key compromise	Use separate keys per service

Interview Questions & Answers

Q1: What is the difference between KMS and CloudHSM?

Answer: KMS is a managed service where AWS controls the HSMs. You manage keys through the KMS API. CloudHSM gives you dedicated HSMs for exclusive use, with full control over key operations. KMS is simpler and cheaper for most use cases. CloudHSM is needed for regulatory requirements that demand dedicated hardware or FIPS 140-2 Level 3 compliance.

Q2: How does envelope encryption work in KMS?

Answer: Envelope encryption uses a data key to encrypt data locally, then encrypts the data key with a CMK in KMS. The encrypted data key is stored alongside the encrypted data. To decrypt, KMS decrypts the data key, then you use it locally to decrypt the data. This reduces KMS API calls (one per data key vs. one per data block) and keeps data keys close to the data.

Q3: What is the difference between AWS Secrets Manager and Parameter Store?

Answer: Secrets Manager has built-in automatic rotation, costs $0.40/secret/month, and is designed for database credentials and API keys. Parameter Store is free for standard parameters, has no built-in rotation, and is better for configuration values and feature flags. Use Secrets Manager when you need automatic rotation and audit capabilities.

Q4: How do you rotate a database password using Secrets Manager?

Answer: Create a Lambda function that implements the rotation protocol: createSecret (generate new password), setSecret (update DB), testSecret (verify connectivity), finishSecret (promote new version). Secrets Manager invokes this Lambda on the configured schedule (e.g., every 30 days). The rotation is atomic - the old secret remains valid until the new one is verified.

Q5: What is a KMS key policy and why is it important?

Answer: A key policy is a resource policy attached to a KMS key that defines who can use and manage the key. It's important because it controls access to encryption keys, which protect all encrypted data. Without proper key policies, unauthorized users could encrypt/decrypt data or even delete keys, causing data loss.

Q6: How do you audit KMS key usage?

Answer: Enable CloudTrail logging for all KMS operations. CloudTrail logs every API call (Encrypt, Decrypt, GenerateDataKey) with the caller identity, timestamp, and key used. Send CloudTrail logs to CloudWatch Logs for real-time monitoring. Use CloudTrail Insights to detect unusual patterns. Query logs with Athena for compliance reporting.

Q7: What happens when you delete a KMS key?

Answer: Deleted keys cannot be recovered after the waiting period (7-30 days). All data encrypted with that key becomes permanently inaccessible. This is why key deletion is a critical operation. Instead of deleting, disable the key first to test impact. Use key grants for temporary access instead of key deletion.

Q8: How do you handle cross-account access to encrypted data?

Answer: Create a key policy in Account A that grants Account B's principal permission to use the key. Or use key grants for temporary, scoped access. Account B can then use the key to decrypt data in Account A's S3 bucket. Alternatively, re-encrypt data with a key in Account B's account for complete isolation.

AWS KMS & Secrets Management for Data Engineers

AWS KMS & Secrets Management for Data Engineers

Why This Matters

What is AWS KMS?

Key Concepts

KMS Key Types

KMS Key Policies

Key Policy Example

Key Policy Best Practices

AWS Secrets Manager

Key Features

Secrets Manager vs Parameter Store

Secrets Manager Rotation Example

KMS Usage in Data Engineering

S3 Encryption with KMS

EBS Encryption

Redshift Encryption

Real-World Project Structure

Production Python Code

Mathematical Formations

KMS Cost Calculation

Secrets Manager Cost

Encryption Throughput Formula

Performance Considerations

Security Considerations

Common Pitfalls

Interview Questions & Answers

Q1: What is the difference between KMS and CloudHSM?

Q2: How does envelope encryption work in KMS?

Q3: What is the difference between AWS Secrets Manager and Parameter Store?

Q4: How do you rotate a database password using Secrets Manager?

Q5: What is a KMS key policy and why is it important?

Q6: How do you audit KMS key usage?

Q7: What happens when you delete a KMS key?

Q8: How do you handle cross-account access to encrypted data?

Quiz

See Also

Premium Content