Encryption at Rest and in Transit Explained — How Data Stays Safe

import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import base64
import json

# ── Key Management ──────────────────────────────────────────────────────────
# In production this key would come from AWS KMS / GCP KMS / HashiCorp Vault.
# NEVER hard-code it like this outside of learning examples.
ENCRYPTION_KEY = AESGCM.generate_key(bit_length=256)  # 32 cryptographically random bytes

def encrypt_record(plaintext_record: dict) -> dict:
    """
    Encrypts a user record before writing it to persistent storage.
    Returns a dict containing the ciphertext and the nonce (both base64-encoded).
    The nonce does NOT need to be secret — it just must be unique per encryption.
    """
    aesgcm = AESGCM(ENCRYPTION_KEY)

    # A nonce (Number Used Once) prevents two identical plaintexts
    # from producing the same ciphertext — critical for security.
    nonce = os.urandom(12)  # 96-bit nonce is the GCM standard

    plaintext_bytes = json.dumps(plaintext_record).encode("utf-8")

    # AESGCM also produces an authentication tag automatically.
    # This means decryption will FAIL if the ciphertext has been tampered with.
    ciphertext = aesgcm.encrypt(nonce, plaintext_bytes, associated_data=None)

    return {
        "ciphertext": base64.b64encode(ciphertext).decode("utf-8"),
        "nonce": base64.b64encode(nonce).decode("utf-8"),
    }

def decrypt_record(encrypted_blob: dict) -> dict:
    """
    Reverses encrypt_record. Raises InvalidTag if data was tampered with.
    """
    aesgcm = AESGCM(ENCRYPTION_KEY)

    ciphertext = base64.b64decode(encrypted_blob["ciphertext"])
    nonce = base64.b64decode(encrypted_blob["nonce"])

    plaintext_bytes = aesgcm.decrypt(nonce, ciphertext, associated_data=None)
    return json.loads(plaintext_bytes.decode("utf-8"))


# ── Demo ─────────────────────────────────────────────────────────────────────
user_record = {
    "user_id": "usr_8821",
    "email": "alice@example.com",
    "credit_card_last4": "4242",
    "ssn": "123-45-6789"   # Sensitive PII — must never sit unencrypted on disk
}

print("Original record:")
print(json.dumps(user_record, indent=2))

encrypted = encrypt_record(user_record)
print("\nWhat gets written to the database (ciphertext):")
print(json.dumps(encrypted, indent=2))

decrypted = decrypt_record(encrypted)
print("\nDecrypted record (what the app sees after reading from DB):")
print(json.dumps(decrypted, indent=2))

# ── Part 1: HTTPS Server with TLS ────────────────────────────────────────────
# Requires: pip install flask pyopenssl
# Certificate generation (run once in terminal):
#   openssl req -x509 -newkey rsa:4096 -keyout server.key -out server.crt \
#     -days 365 -nodes -subj "/CN=localhost"

from flask import Flask, jsonify
import ssl

app = Flask(__name__)

@app.route("/api/user-profile")
def get_user_profile():
    # Sensitive data — safe to serve because the transport is encrypted
    return jsonify({
        "user_id": "usr_8821",
        "email": "alice@example.com",
        "account_tier": "premium"
    })

if __name__ == "__main__":
    # Build an SSL context — this is what activates TLS on the server socket
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

    # Load the certificate (public) and private key
    ssl_context.load_cert_chain(certfile="server.crt", keyfile="server.key")

    # Enforce TLS 1.2 minimum — explicitly reject older, broken versions
    ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2

    print("Server running on https://localhost:5000 — all traffic is encrypted")
    app.run(host="0.0.0.0", port=5000, ssl_context=ssl_context)


# ── Part 2: Client that enforces certificate verification ────────────────────
# ALWAYS verify the server certificate. Never set verify=False in production.
import requests

def fetch_user_profile_securely(base_url: str, ca_cert_path: str) -> dict:
    """
    Fetches user data over HTTPS.
    ca_cert_path: path to the CA cert that signed the server certificate.
    In production this would be the system CA bundle or your internal PKI cert.
    """
    response = requests.get(
        url=f"{base_url}/api/user-profile",
        verify=ca_cert_path,   # Validates server cert against this CA — prevents MITM
        timeout=10             # Always set a timeout; hanging connections are a DoS risk
    )
    response.raise_for_status()  # Raises exception on 4xx/5xx
    return response.json()

# Example call (would work against the server above):
# profile = fetch_user_profile_securely("https://localhost:5000", "server.crt")
# print(profile)

# ── Envelope Encryption — AWS Architecture Reference ────────────────────────
# This CloudFormation-style pseudo-config shows the layered key hierarchy.
# It is conceptual but reflects real AWS KMS + S3 + RDS patterns.

KeyManagement:
  CustomerMasterKey:                  # Lives ONLY inside AWS KMS — never exported
    alias: "alias/payments-service-cmk"
    rotation: enabled                 # AWS auto-rotates annually
    key_policy: least_privilege       # Only the payments-service IAM role can use it

Storage:
  UserDatabase:                       # Amazon RDS PostgreSQL
    encryption_at_rest: true
    encryption_type: envelope
    # RDS generates a DEK per tablespace, wraps it with the CMK above.
    # If DB volume is stolen, attacker has ciphertext but no DEK.
    # If DEK is somehow leaked, attacker still cannot unwrap it without KMS access.
    kms_key_ref: "alias/payments-service-cmk"
    tls_in_transit:
      enforce: true
      minimum_tls_version: "1.2"      # Reject 1.0/1.1 connections at engine level
      certificate_authority: "AWS ACM"

  DocumentStore:                      # Amazon S3
    default_encryption: "aws:kms"     # Every object encrypted on write automatically
    kms_key_ref: "alias/payments-service-cmk"
    bucket_policy:
      deny_non_https: true            # Bucket policy rejects any HTTP (unencrypted) PUT/GET

Transit:
  APIGateway:
    protocol: HTTPS
    tls_policy: "TLS_1_2"            # AWS managed policy — auto-rejects older handshakes
    certificate: "ACM-managed"        # Auto-renewed, no manual cert rotation

  MicroserviceMesh:
    type: Istio
    mtls_mode: STRICT                 # PERMISSIVE allows plaintext fallback — never use in prod
    # STRICT means: if a service does not present a valid cert, the connection is dropped.
    # This stops lateral movement — a compromised sidecar cannot sniff other services.

KeyRotationPolicy:
  cmk_rotation: "annually-automatic"
  dek_rotation: "on-cmk-rotation"    # Re-wrap DEKs when CMK rotates
  tls_certificate_renewal: "60-days-before-expiry"  # Automated via ACM
  incident_rotation: "immediate"      # Runbook trigger on any suspected key compromise