1. Envelope Encryption Architecture (KMS)
Encrypting terabytes of data directly via asymmetric RSA cryptography is computationally infeasible. Envelope Encryption combines symmetric speed with asymmetric key management:
- Generate an ephemeral symmetric Data Encryption Key (DEK) (AES-256-GCM) to encrypt the physical data payload at line speed.
- Encrypt the DEK itself using a root Key Encryption Key (KEK) stored inside a Hardware Security Module (HSM / AWS KMS).
- Store the encrypted payload alongside the encrypted DEK blob.
Raw Data Payload ──► [AES-256-GCM Encrypt with DEK] ──► Encrypted Data Payload
│
Root KEK (in HSM) ──► [Encrypt DEK with KEK] ──► Encrypted DEK Blob ──┘
2. Format-Preserving Encryption (FPE) & Tokenization
Standard AES ciphertexts destroy formatting lengths. Format-Preserving Encryption (FPE - NIST SP 800-38G FF1) encrypts strings such that the ciphertext preserves the exact character set and length of the plaintext (1234-5678-9012-3456 $\to$ 8492-1049-5821-9943), allowing legacy databases to store encrypted fields without schema migrations.
3. RBAC vs ABAC Access Control Models
- RBAC (Role-Based Access Control): Grants permissions statically based on organizational job roles (
Role = QuantAnalyst -> Read Equities). - ABAC (Attribute-Based Access Control): Evaluates dynamic boolean policy predicates at query time across User, Resource, and Environment attributes (
Allow IF User.Clearance >= Table.Classification AND Time < 18:00).
4. Python Verification: Envelope Encryption & Decryption Pipeline
import os
import hashlib
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class KMSMasterKey:
"""Simulates Hardware Security Module (HSM) root Key Encryption Key (KEK)."""
def __init__(self):
self._kek = os.urandom(32)
def wrap_dek(self, dek: bytes) -> bytes:
aesgcm = AESGCM(self._kek)
nonce = os.urandom(12)
return nonce + aesgcm.encrypt(nonce, dek, b"KEK_AAD")
def unwrap_dek(self, wrapped_dek: bytes) -> bytes:
nonce, ciphertext = wrapped_dek[:12], wrapped_dek[12:]
aesgcm = AESGCM(self._kek)
return aesgcm.decrypt(nonce, ciphertext, b"KEK_AAD")
class EnvelopeEncryptionPipeline:
def __init__(self, kms: KMSMasterKey):
self.kms = kms
def encrypt_payload(self, plaintext: bytes) -> tuple:
dek = AESGCM.generate_key(bit_length=256)
aesgcm = AESGCM(dek)
nonce = os.urandom(12)
encrypted_data = nonce + aesgcm.encrypt(nonce, plaintext, b"DATA_AAD")
wrapped_dek = self.kms.wrap_dek(dek)
return encrypted_data, wrapped_dek
def decrypt_payload(self, encrypted_data: bytes, wrapped_dek: bytes) -> bytes:
dek = self.kms.unwrap_dek(wrapped_dek)
nonce, ciphertext = encrypted_data[:12], encrypted_data[12:]
aesgcm = AESGCM(dek)
return aesgcm.decrypt(nonce, ciphertext, b"DATA_AAD")
if __name__ == "__main__":
kms = KMSMasterKey()
pipeline = EnvelopeEncryptionPipeline(kms)
secret_payload = b"QUANT_ALREADY_COMMITTED_PORTFOLIO_ALPHA_SIGNAL_V1"
enc_data, enc_key = pipeline.encrypt_payload(secret_payload)
decrypted = pipeline.decrypt_payload(enc_data, enc_key)
print("Original Secret Payload: ", secret_payload.decode())
print("Decrypted Verified Match:", decrypted.decode())