Senior Security

Senior Security Engineer — Threat Modeling + Security Router

This skill does exactly one job itself — STRIDE/DREAD threat modeling (plus a quick secret scan) — and routes every other security request to the specialist skill that owns that lane. Do not duplicate sibling content here; route instead.

Routing Table (read this first)

If the request spans lanes (e.g., "secure this new architecture"), do the threat model here first — its output (prioritized threats + mitigations) tells you which siblings to load next. Never bulk-load multiple security skills speculatively.

What This Skill Owns: STRIDE Threat Modeling

Workflow

1. Scope: assets to protect, trust boundaries, data flows (external entities, processes, data stores, flows). 2. Generate the threat model per component:

   python3 scripts/threat_modeler.py --component "User Authentication" --assets "credentials,sessions" --json --output threats.json

Output: per-threat STRIDE category, DREAD score (Damage, Reproducibility, Exploitability, Affected users, Discoverability — each 1–10), and suggested mitigations. Repeat per DFD element; --interactive walks scoping questions; --list-threats shows the threat database. 3. Consume the output: sort threats.json by DREAD score descending; everything ≥ 7 average needs a named mitigation owner before the design ships. Map each mitigation to the responsible sibling lane (e.g., IAM threats → cloud-security, injection threats → code-reviewer). 4. Quick secret sweep while you have the codebase open:

   python3 scripts/secret_scanner.py /path/to/project --format json --severity high

20+ patterns (AWS keys, GitHub tokens, private keys, generic credentials). Any critical/high finding blocks merge until rotated and moved to a secret manager. 5. Verification gate: every DFD element has ≥ 1 STRIDE row considered, every threat with DREAD ≥ 7 has an owner + mitigation, and the secret scan exits with zero high/critical findings. Re-run both tools after mitigations land — that re-run is the done signal, not the document.

STRIDE per Element Matrix

(S=Spoofing→authn, T=Tampering→integrity, R=Repudiation→audit logs, I=Info Disclosure→encryption/access control, D=DoS→rate limiting/redundancy, E=Elevation→least privilege.)

References (load on demand)

The architecture and crypto references are kept because no sibling ships them; for operating those controls (scanning, compliance, monitoring) still route to senior-secops.

Cryptography Implementation Guide

Practical cryptographic patterns for securing data at rest, in transit, and in use.

---

Table of Contents

• Cryptographic Primitives
• Symmetric Encryption
• Asymmetric Encryption
• Hashing and Password Storage
• Key Management
• Common Cryptographic Mistakes

---

Cryptographic Primitives

Algorithm Selection Guide

Security Strength Comparison

---

Symmetric Encryption

AES-256-GCM Implementation

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

class AESGCMEncryption:
    """
    AES-256-GCM authenticated encryption.

    Provides both confidentiality and integrity.
    GCM mode prevents tampering with authentication tag.
    """

    def __init__(self, key: bytes = None):
        if key is None:
            key = AESGCM.generate_key(bit_length=256)
        if len(key) != 32:
            raise ValueError("Key must be 32 bytes (256 bits)")
        self.key = key
        self.aesgcm = AESGCM(key)

    def encrypt(self, plaintext: bytes, associated_data: bytes = None) -> bytes:
        """
        Encrypt with random nonce.

        Returns: nonce (12 bytes) + ciphertext + tag (16 bytes)
        """
        nonce = os.urandom(12)  # 96-bit nonce for GCM
        ciphertext = self.aesgcm.encrypt(nonce, plaintext, associated_data)
        return nonce + ciphertext

    def decrypt(self, ciphertext: bytes, associated_data: bytes = None) -> bytes:
        """
        Decrypt and verify authentication tag.

        Raises InvalidTag if tampered.
        """
        nonce = ciphertext[:12]
        actual_ciphertext = ciphertext[12:]
        return self.aesgcm.decrypt(nonce, actual_ciphertext, associated_data)


# Usage
encryptor = AESGCMEncryption()
plaintext = b"Sensitive data to encrypt"
aad = b"user_id:12345"  # Authenticated but not encrypted

ciphertext = encryptor.encrypt(plaintext, associated_data=aad)
decrypted = encryptor.decrypt(ciphertext, associated_data=aad)

ChaCha20-Poly1305 Implementation

from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
import os

class ChaChaEncryption:
    """
    ChaCha20-Poly1305 authenticated encryption.

    Faster than AES on systems without hardware AES support.
    Resistant to timing attacks (constant-time implementation).
    """

    def __init__(self, key: bytes = None):
        if key is None:
            key = ChaCha20Poly1305.generate_key()
        self.key = key
        self.chacha = ChaCha20Poly1305(key)

    def encrypt(self, plaintext: bytes, associated_data: bytes = None) -> bytes:
        """Encrypt with random 96-bit nonce."""
        nonce = os.urandom(12)
        ciphertext = self.chacha.encrypt(nonce, plaintext, associated_data)
        return nonce + ciphertext

    def decrypt(self, ciphertext: bytes, associated_data: bytes = None) -> bytes:
        """Decrypt and verify Poly1305 authentication tag."""
        nonce = ciphertext[:12]
        actual_ciphertext = ciphertext[12:]
        return self.chacha.decrypt(nonce, actual_ciphertext, associated_data)

Envelope Encryption Pattern

"""
Envelope Encryption: Encrypt data with a Data Encryption Key (DEK),
then encrypt DEK with a Key Encryption Key (KEK).

Benefits:
- KEK can be rotated without re-encrypting data
- DEK can be stored alongside encrypted data
- Enables per-record encryption with different DEKs
"""

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
import os
import json
import base64

class EnvelopeEncryption:
    def __init__(self, kek_public_key, kek_private_key=None):
        self.kek_public = kek_public_key
        self.kek_private = kek_private_key

    def encrypt(self, plaintext: bytes) -> dict:
        """
        1. Generate random DEK
        2. Encrypt plaintext with DEK
        3. Encrypt DEK with KEK
        4. Return encrypted DEK + encrypted data
        """
        # Generate Data Encryption Key
        dek = AESGCM.generate_key(bit_length=256)
        aesgcm = AESGCM(dek)

        # Encrypt data with DEK
        nonce = os.urandom(12)
        encrypted_data = aesgcm.encrypt(nonce, plaintext, None)

        # Encrypt DEK with KEK (RSA-OAEP)
        encrypted_dek = self.kek_public.encrypt(
            dek,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )

        return {
            'encrypted_dek': base64.b64encode(encrypted_dek).decode(),
            'nonce': base64.b64encode(nonce).decode(),
            'ciphertext': base64.b64encode(encrypted_data).decode()
        }

    def decrypt(self, envelope: dict) -> bytes:
        """
        1. Decrypt DEK with KEK
        2. Decrypt data with DEK
        """
        if self.kek_private is None:
            raise ValueError("Private key required for decryption")

        # Decrypt DEK
        encrypted_dek = base64.b64decode(envelope['encrypted_dek'])
        dek = self.kek_private.decrypt(
            encrypted_dek,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )

        # Decrypt data
        aesgcm = AESGCM(dek)
        nonce = base64.b64decode(envelope['nonce'])
        ciphertext = base64.b64decode(envelope['ciphertext'])

        return aesgcm.decrypt(nonce, ciphertext, None)

---

Asymmetric Encryption

RSA Key Generation and Usage

from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes, serialization

def generate_rsa_keypair(key_size=4096):
    """Generate RSA key pair for encryption/signing."""
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=key_size
    )
    public_key = private_key.public_key()

    return private_key, public_key

def serialize_keys(private_key, public_key, password=None):
    """Serialize keys for storage."""
    # Private key (encrypted with password)
    if password:
        encryption = serialization.BestAvailableEncryption(password.encode())
    else:
        encryption = serialization.NoEncryption()

    private_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=encryption
    )

    # Public key
    public_pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )

    return private_pem, public_pem

def rsa_encrypt(public_key, plaintext: bytes) -> bytes:
    """RSA-OAEP encryption (for small data like keys)."""
    return public_key.encrypt(
        plaintext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )

def rsa_decrypt(private_key, ciphertext: bytes) -> bytes:
    """RSA-OAEP decryption."""
    return private_key.decrypt(
        ciphertext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )

Digital Signatures (Ed25519)

from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey, Ed25519PublicKey
)

class Ed25519Signer:
    """
    Ed25519 digital signatures.

    Fast, secure, and deterministic.
    256-bit keys provide 128-bit security.
    """

    def __init__(self, private_key=None):
        if private_key is None:
            private_key = Ed25519PrivateKey.generate()
        self.private_key = private_key
        self.public_key = private_key.public_key()

    def sign(self, message: bytes) -> bytes:
        """Create digital signature."""
        return self.private_key.sign(message)

    def verify(self, message: bytes, signature: bytes) -> bool:
        """Verify digital signature."""
        try:
            self.public_key.verify(signature, message)
            return True
        except Exception:
            return False

    def get_public_key_bytes(self) -> bytes:
        """Export public key for verification."""
        return self.public_key.public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw
        )


# Usage for message signing
signer = Ed25519Signer()
message = b"Important document content"
signature = signer.sign(message)

# Verification (can be done with public key only)
is_valid = signer.verify(message, signature)

ECDH Key Exchange

from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes

class X25519KeyExchange:
    """
    X25519 Diffie-Hellman key exchange.

    Used to establish shared secrets over insecure channels.
    """

    def __init__(self):
        self.private_key = x25519.X25519PrivateKey.generate()
        self.public_key = self.private_key.public_key()

    def get_public_key_bytes(self) -> bytes:
        """Get public key to send to peer."""
        return self.public_key.public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw
        )

    def derive_shared_key(self, peer_public_key_bytes: bytes,
                          info: bytes = b"") -> bytes:
        """
        Derive shared encryption key from peer's public key.

        Uses HKDF to derive a proper encryption key.
        """
        peer_public_key = x25519.X25519PublicKey.from_public_bytes(
            peer_public_key_bytes
        )

        shared_secret = self.private_key.exchange(peer_public_key)

        # Derive encryption key using HKDF
        derived_key = HKDF(
            algorithm=hashes.SHA256(),
            length=32,
            salt=None,
            info=info,
        ).derive(shared_secret)

        return derived_key


# Key exchange example
alice = X25519KeyExchange()
bob = X25519KeyExchange()

# Exchange public keys (can be done over insecure channel)
alice_public = alice.get_public_key_bytes()
bob_public = bob.get_public_key_bytes()

# Both derive the same shared key
alice_shared = alice.derive_shared_key(bob_public, info=b"session-key")
bob_shared = bob.derive_shared_key(alice_public, info=b"session-key")

assert alice_shared == bob_shared  # Same key!

---

Hashing and Password Storage

Password Hashing with Argon2

import argon2
from argon2 import PasswordHasher, Type

class SecurePasswordHasher:
    """
    Argon2id password hashing.

    Argon2id combines resistance to:
    - GPU attacks (memory-hard)
    - Side-channel attacks (data-independent)
    """

    def __init__(self):
        # OWASP recommended parameters
        self.hasher = PasswordHasher(
            time_cost=3,        # Iterations
            memory_cost=65536,  # 64 MB
            parallelism=4,      # Threads
            hash_len=32,        # Output length
            type=Type.ID        # Argon2id variant
        )

    def hash_password(self, password: str) -> str:
        """
        Hash password for storage.

        Returns encoded string with algorithm parameters and salt.
        """
        return self.hasher.hash(password)

    def verify_password(self, password: str, hash: str) -> bool:
        """
        Verify password against stored hash.

        Automatically handles timing-safe comparison.
        """
        try:
            self.hasher.verify(hash, password)
            return True
        except argon2.exceptions.VerifyMismatchError:
            return False

    def needs_rehash(self, hash: str) -> bool:
        """Check if hash needs upgrading to current parameters."""
        return self.hasher.check_needs_rehash(hash)


# Usage
hasher = SecurePasswordHasher()

# During registration
password = "user_password_123!"
password_hash = hasher.hash_password(password)
# Store password_hash in database

# During login
stored_hash = password_hash  # From database
if hasher.verify_password("user_password_123!", stored_hash):
    print("Login successful")

    # Check if hash needs upgrading
    if hasher.needs_rehash(stored_hash):
        new_hash = hasher.hash_password(password)
        # Update stored hash

Bcrypt Alternative

import bcrypt

class BcryptHasher:
    """
    Bcrypt password hashing.

    Well-established, widely supported.
    Use when Argon2 is not available.
    """

    def __init__(self, rounds=12):
        self.rounds = rounds

    def hash_password(self, password: str) -> str:
        salt = bcrypt.gensalt(rounds=self.rounds)
        return bcrypt.hashpw(password.encode(), salt).decode()

    def verify_password(self, password: str, hash: str) -> bool:
        return bcrypt.checkpw(password.encode(), hash.encode())

HMAC for Message Authentication

import hmac
import hashlib
import secrets

def create_hmac(key: bytes, message: bytes) -> bytes:
    """Create HMAC-SHA256 authentication tag."""
    return hmac.new(key, message, hashlib.sha256).digest()

def verify_hmac(key: bytes, message: bytes, tag: bytes) -> bool:
    """Verify HMAC in constant time."""
    expected = hmac.new(key, message, hashlib.sha256).digest()
    return hmac.compare_digest(expected, tag)

# API request signing example
def sign_api_request(secret_key: bytes, method: str, path: str,
                     body: bytes, timestamp: str) -> str:
    """Sign API request for authentication."""
    message = f"{method}\n{path}\n{timestamp}\n".encode() + body
    signature = create_hmac(secret_key, message)
    return signature.hex()

---

Key Management

Key Derivation Functions

from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
from cryptography.hazmat.primitives import hashes
import os

def derive_key_pbkdf2(password: str, salt: bytes = None,
                       iterations: int = 600000) -> tuple:
    """
    Derive encryption key from password using PBKDF2.

    NIST recommends minimum 600,000 iterations for PBKDF2-SHA256.
    """
    if salt is None:
        salt = os.urandom(16)

    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=iterations
    )

    key = kdf.derive(password.encode())
    return key, salt

def derive_key_scrypt(password: str, salt: bytes = None) -> tuple:
    """
    Derive key using scrypt (memory-hard).

    More resistant to hardware attacks than PBKDF2.
    """
    if salt is None:
        salt = os.urandom(16)

    kdf = Scrypt(
        salt=salt,
        length=32,
        n=2**17,  # CPU/memory cost
        r=8,      # Block size
        p=1       # Parallelization
    )

    key = kdf.derive(password.encode())
    return key, salt

Key Rotation Strategy

from datetime import datetime, timedelta
from typing import Dict, Optional
import json

class KeyManager:
    """
    Manage encryption key lifecycle.

    Supports key rotation without data re-encryption.
    """

    def __init__(self, storage_backend):
        self.storage = storage_backend

    def generate_key(self, key_id: str, algorithm: str = 'AES-256-GCM') -> dict:
        """Generate and store new encryption key."""
        key_material = os.urandom(32)

        key_metadata = {
            'key_id': key_id,
            'algorithm': algorithm,
            'created_at': datetime.utcnow().isoformat(),
            'expires_at': (datetime.utcnow() + timedelta(days=365)).isoformat(),
            'status': 'active'
        }

        self.storage.store_key(key_id, key_material, key_metadata)
        return key_metadata

    def rotate_key(self, old_key_id: str) -> dict:
        """
        Rotate encryption key.

        1. Mark old key as 'decrypt-only'
        2. Generate new key as 'active'
        3. Old key can still decrypt, new key encrypts
        """
        # Mark old key as decrypt-only
        old_metadata = self.storage.get_key_metadata(old_key_id)
        old_metadata['status'] = 'decrypt-only'
        self.storage.update_key_metadata(old_key_id, old_metadata)

        # Generate new key
        new_key_id = f"{old_key_id.rsplit('_', 1)[0]}_{datetime.utcnow().strftime('%Y%m%d')}"
        return self.generate_key(new_key_id)

    def get_encryption_key(self) -> tuple:
        """Get current active key for encryption."""
        return self.storage.get_active_key()

    def get_decryption_key(self, key_id: str) -> bytes:
        """Get specific key for decryption."""
        return self.storage.get_key(key_id)

Hardware Security Module Integration

# AWS CloudHSM / KMS integration pattern
import boto3

class AWSKMSProvider:
    """
    AWS KMS integration for key management.

    Keys never leave AWS infrastructure.
    """

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

    def encrypt(self, plaintext: bytes) -> bytes:
        """Encrypt using KMS master key."""
        response = self.kms.encrypt(
            KeyId=self.key_id,
            Plaintext=plaintext
        )
        return response['CiphertextBlob']

    def decrypt(self, ciphertext: bytes) -> bytes:
        """Decrypt using KMS master key."""
        response = self.kms.decrypt(
            KeyId=self.key_id,
            CiphertextBlob=ciphertext
        )
        return response['Plaintext']

    def generate_data_key(self) -> tuple:
        """Generate data encryption key."""
        response = self.kms.generate_data_key(
            KeyId=self.key_id,
            KeySpec='AES_256'
        )
        return response['Plaintext'], response['CiphertextBlob']

---

Common Cryptographic Mistakes

Mistake 1: Using ECB Mode

# BAD: ECB mode reveals patterns
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

def bad_ecb_encrypt(key, plaintext):
    cipher = Cipher(algorithms.AES(key), modes.ECB())
    encryptor = cipher.encryptor()
    return encryptor.update(plaintext) + encryptor.finalize()

# GOOD: Use authenticated encryption (GCM)
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

def good_gcm_encrypt(key, plaintext):
    aesgcm = AESGCM(key)
    nonce = os.urandom(12)
    return nonce + aesgcm.encrypt(nonce, plaintext, None)

Mistake 2: Reusing Nonces

# BAD: Static nonce
nonce = b"fixed_nonce!"  # NEVER DO THIS

# GOOD: Random nonce per encryption
nonce = os.urandom(12)

# ALSO GOOD: Counter-based nonce (if you can guarantee no repeats)
class NonceCounter:
    def __init__(self):
        self.counter = 0

    def get_nonce(self):
        self.counter += 1
        return self.counter.to_bytes(12, 'big')

Mistake 3: Rolling Your Own Crypto

# BAD: Custom "encryption"
def bad_encrypt(data, key):
    return bytes([b ^ k for b, k in zip(data, key * len(data))])

# GOOD: Use established libraries
from cryptography.fernet import Fernet

def good_encrypt(data, key):
    f = Fernet(key)
    return f.encrypt(data)

Mistake 4: Weak Random Generation

import random
import secrets

# BAD: Predictable random
def bad_generate_token():
    return ''.join(random.choices('abcdef0123456789', k=32))

# GOOD: Cryptographically secure
def good_generate_token():
    return secrets.token_hex(16)

Mistake 5: Timing Attacks in Comparison

# BAD: Early exit reveals length
def bad_compare(a, b):
    if len(a) != len(b):
        return False
    for x, y in zip(a, b):
        if x != y:
            return False
    return True

# GOOD: Constant-time comparison
import hmac

def good_compare(a, b):
    return hmac.compare_digest(a, b)

---

Quick Reference Card

Security Architecture Patterns

Proven security architecture patterns for designing resilient systems.

---

Table of Contents

• Zero Trust Architecture
• Defense in Depth
• Secure Authentication Patterns
• API Security Patterns
• Data Protection Patterns
• Security Anti-Patterns

---

Zero Trust Architecture

Never trust, always verify. Every request authenticated and authorized regardless of network location.

Core Principles

Implementation Components

ZERO TRUST ARCHITECTURE

┌─────────────────────────────────────────────────────────────┐
│                        CONTROL PLANE                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │   Identity   │  │    Policy    │  │   Threat     │      │
│  │   Provider   │  │    Engine    │  │  Intelligence│      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
└─────────────────────────────────────────────────────────────┘
                              │
                    ┌─────────┴─────────┐
                    │   Policy Decision │
                    │   Point (PDP)     │
                    └─────────┬─────────┘
                              │
┌─────────────────────────────┴───────────────────────────────┐
│                        DATA PLANE                            │
│  ┌──────────────┐                      ┌──────────────┐     │
│  │    User      │──── PEP ────────────▶│   Resource   │     │
│  │   Device     │      │               │   (App/Data) │     │
│  └──────────────┘      │               └──────────────┘     │
│                   Policy Enforcement                         │
│                   Point (PEP)                               │
└─────────────────────────────────────────────────────────────┘

Authentication Flow

# Zero Trust authentication middleware
import jwt
from functools import wraps

def zero_trust_auth(required_claims=None):
    """
    Verify every request against identity, device, and context.
    """
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            token = request.headers.get('Authorization', '').replace('Bearer ', '')

            # 1. Verify token signature and expiration
            try:
                payload = jwt.decode(token, PUBLIC_KEY, algorithms=['RS256'])
            except jwt.InvalidTokenError:
                return {'error': 'Invalid token'}, 401

            # 2. Verify device compliance
            device_id = request.headers.get('X-Device-ID')
            if not verify_device_compliance(device_id, payload['user_id']):
                return {'error': 'Device not compliant'}, 403

            # 3. Verify location/network context
            client_ip = request.remote_addr
            if not verify_network_context(client_ip, payload['allowed_networks']):
                return {'error': 'Network context invalid'}, 403

            # 4. Verify required claims
            if required_claims:
                for claim in required_claims:
                    if claim not in payload:
                        return {'error': f'Missing claim: {claim}'}, 403

            # 5. Log access for analytics
            log_access_attempt(payload, request, 'allowed')

            return f(*args, **kwargs)
        return decorated
    return decorator

@app.route('/api/sensitive-data')
@zero_trust_auth(required_claims=['data:read', 'clearance:secret'])
def get_sensitive_data():
    return fetch_data()

Network Segmentation

---

Defense in Depth

Multiple layers of security controls so failure of one doesn't compromise the system.

Security Layers

DEFENSE IN DEPTH LAYERS

┌─────────────────────────────────────────────────────────────┐
│  Layer 1: PERIMETER                                          │
│  - Firewall, WAF, DDoS mitigation, DNS filtering            │
├─────────────────────────────────────────────────────────────┤
│  Layer 2: NETWORK                                            │
│  - Segmentation, IDS/IPS, network monitoring, VPN           │
├─────────────────────────────────────────────────────────────┤
│  Layer 3: HOST                                               │
│  - Endpoint protection, hardening, patching, logging        │
├─────────────────────────────────────────────────────────────┤
│  Layer 4: APPLICATION                                        │
│  - Input validation, authentication, secure coding, SAST    │
├─────────────────────────────────────────────────────────────┤
│  Layer 5: DATA                                               │
│  - Encryption at rest/transit, access controls, DLP, backup │
└─────────────────────────────────────────────────────────────┘

Implementation Checklist

Fail-Safe Defaults

# Secure default configuration
class SecurityConfig:
    # Authentication
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Strict'

    # Headers
    CONTENT_SECURITY_POLICY = "default-src 'self'; script-src 'self'"
    X_FRAME_OPTIONS = 'DENY'
    X_CONTENT_TYPE_OPTIONS = 'nosniff'
    REFERRER_POLICY = 'strict-origin-when-cross-origin'

    # Timeouts
    SESSION_LIFETIME = 3600  # 1 hour
    TOKEN_EXPIRY = 900  # 15 minutes

    # Rate limiting
    RATE_LIMIT_DEFAULT = '100/hour'
    RATE_LIMIT_AUTH = '10/minute'

---

Secure Authentication Patterns

OAuth 2.0 + PKCE Flow

OAUTH 2.0 AUTHORIZATION CODE FLOW WITH PKCE

┌──────────┐                                  ┌──────────────┐
│  Client  │                                  │    Auth      │
│  (SPA)   │                                  │   Server     │
└────┬─────┘                                  └──────┬───────┘
     │                                               │
     │ 1. Generate code_verifier (random string)     │
     │    code_challenge = SHA256(code_verifier)     │
     │                                               │
     │ 2. /authorize?                                │
     │    response_type=code&                        │
     │    client_id=xxx&                             │
     │    code_challenge=xxx&                        │
     │    code_challenge_method=S256                 │
     │──────────────────────────────────────────────▶│
     │                                               │
     │◀──────────────────────────────────────────────│
     │ 3. Redirect with authorization_code           │
     │                                               │
     │ 4. POST /token                                │
     │    grant_type=authorization_code&             │
     │    code=xxx&                                  │
     │    code_verifier=xxx  (proves possession)     │
     │──────────────────────────────────────────────▶│
     │                                               │
     │◀──────────────────────────────────────────────│
     │ 5. { access_token, refresh_token, id_token }  │
     │                                               │

JWT Token Structure

# Secure JWT implementation
import jwt
import secrets
from datetime import datetime, timedelta

class JWTService:
    def __init__(self, private_key, public_key, issuer):
        self.private_key = private_key
        self.public_key = public_key
        self.issuer = issuer

    def create_access_token(self, user_id, roles, expires_minutes=15):
        """Create short-lived access token."""
        now = datetime.utcnow()
        payload = {
            'iss': self.issuer,
            'sub': str(user_id),
            'iat': now,
            'exp': now + timedelta(minutes=expires_minutes),
            'jti': secrets.token_hex(16),  # Unique token ID
            'roles': roles,
            'type': 'access'
        }
        return jwt.encode(payload, self.private_key, algorithm='RS256')

    def create_refresh_token(self, user_id, expires_days=7):
        """Create longer-lived refresh token (stored server-side)."""
        now = datetime.utcnow()
        jti = secrets.token_hex(32)
        payload = {
            'iss': self.issuer,
            'sub': str(user_id),
            'iat': now,
            'exp': now + timedelta(days=expires_days),
            'jti': jti,
            'type': 'refresh'
        }
        # Store jti in database for revocation capability
        store_refresh_token(jti, user_id, now + timedelta(days=expires_days))
        return jwt.encode(payload, self.private_key, algorithm='RS256')

    def verify_token(self, token, token_type='access'):
        """Verify token with all security checks."""
        try:
            payload = jwt.decode(
                token,
                self.public_key,
                algorithms=['RS256'],
                issuer=self.issuer
            )

            # Verify token type
            if payload.get('type') != token_type:
                raise jwt.InvalidTokenError('Invalid token type')

            # For refresh tokens, check revocation
            if token_type == 'refresh':
                if is_token_revoked(payload['jti']):
                    raise jwt.InvalidTokenError('Token revoked')

            return payload

        except jwt.ExpiredSignatureError:
            raise AuthError('Token expired')
        except jwt.InvalidTokenError as e:
            raise AuthError(f'Invalid token: {e}')

Multi-Factor Authentication

# TOTP implementation
import pyotp
import qrcode

class TOTPService:
    def __init__(self, issuer_name):
        self.issuer = issuer_name

    def generate_secret(self):
        """Generate a new TOTP secret for user."""
        return pyotp.random_base32()

    def get_provisioning_uri(self, secret, user_email):
        """Generate QR code URI for authenticator app."""
        totp = pyotp.TOTP(secret)
        return totp.provisioning_uri(
            name=user_email,
            issuer_name=self.issuer
        )

    def verify_code(self, secret, code, valid_window=1):
        """Verify TOTP code with time drift tolerance."""
        totp = pyotp.TOTP(secret)
        return totp.verify(code, valid_window=valid_window)

---

API Security Patterns

Input Validation

from pydantic import BaseModel, validator, constr
import re

class UserCreateRequest(BaseModel):
    """Strict input validation for user creation."""

    email: constr(max_length=255)
    username: constr(min_length=3, max_length=50, regex=r'^[a-zA-Z0-9_]+$')
    password: constr(min_length=12, max_length=128)

    @validator('email')
    def validate_email(cls, v):
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, v):
            raise ValueError('Invalid email format')
        return v.lower()

    @validator('password')
    def validate_password_strength(cls, v):
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain lowercase letter')
        if not re.search(r'\d', v):
            raise ValueError('Password must contain digit')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError('Password must contain special character')
        return v

Rate Limiting

from redis import Redis
from functools import wraps
import time

class RateLimiter:
    """Token bucket rate limiter with Redis backend."""

    def __init__(self, redis_client):
        self.redis = redis_client

    def is_allowed(self, key, limit, window_seconds):
        """Check if request is within rate limit."""
        pipe = self.redis.pipeline()
        now = time.time()
        window_start = now - window_seconds

        # Remove old entries
        pipe.zremrangebyscore(key, 0, window_start)
        # Count current entries
        pipe.zcard(key)
        # Add new entry
        pipe.zadd(key, {str(now): now})
        # Set expiry
        pipe.expire(key, window_seconds)

        results = pipe.execute()
        current_count = results[1]

        return current_count < limit

def rate_limit(limit=100, window=3600, key_func=None):
    """Rate limiting decorator."""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            if key_func:
                key = f"rate_limit:{key_func()}"
            else:
                key = f"rate_limit:{request.remote_addr}:{f.__name__}"

            if not rate_limiter.is_allowed(key, limit, window):
                return {
                    'error': 'Rate limit exceeded',
                    'retry_after': window
                }, 429

            return f(*args, **kwargs)
        return decorated
    return decorator

SQL Injection Prevention

# NEVER: String concatenation
# query = f"SELECT * FROM users WHERE id = {user_id}"

# ALWAYS: Parameterized queries
from sqlalchemy import text

def get_user_secure(user_id):
    """Safe parameterized query."""
    query = text("SELECT * FROM users WHERE id = :user_id")
    result = db.execute(query, {'user_id': user_id})
    return result.fetchone()

# For dynamic queries, use ORM
def search_users(filters):
    """Safe dynamic query with ORM."""
    query = User.query

    if 'name' in filters:
        # ORM handles escaping
        query = query.filter(User.name.ilike(f"%{filters['name']}%"))

    if 'role' in filters:
        query = query.filter(User.role == filters['role'])

    return query.all()

---

Data Protection Patterns

Encryption at Rest

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

class FieldEncryption:
    """Encrypt sensitive database fields."""

    def __init__(self, master_key):
        self.fernet = Fernet(master_key)

    @staticmethod
    def derive_key(password, salt):
        """Derive encryption key from password."""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key

    def encrypt(self, plaintext):
        """Encrypt a field value."""
        if isinstance(plaintext, str):
            plaintext = plaintext.encode()
        return self.fernet.encrypt(plaintext).decode()

    def decrypt(self, ciphertext):
        """Decrypt a field value."""
        if isinstance(ciphertext, str):
            ciphertext = ciphertext.encode()
        return self.fernet.decrypt(ciphertext).decode()

# Usage in ORM
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255))  # Not sensitive
    _ssn = db.Column('ssn', db.String(500))  # Encrypted

    @property
    def ssn(self):
        if self._ssn:
            return field_encryption.decrypt(self._ssn)
        return None

    @ssn.setter
    def ssn(self, value):
        if value:
            self._ssn = field_encryption.encrypt(value)
        else:
            self._ssn = None

Secret Management

# HashiCorp Vault integration
import hvac

class VaultClient:
    def __init__(self, url, token):
        self.client = hvac.Client(url=url, token=token)

    def get_secret(self, path):
        """Retrieve secret from Vault."""
        secret = self.client.secrets.kv.v2.read_secret_version(path=path)
        return secret['data']['data']

    def get_database_credentials(self, role):
        """Get dynamic database credentials."""
        creds = self.client.secrets.database.generate_credentials(role)
        return {
            'username': creds['data']['username'],
            'password': creds['data']['password'],
            'ttl': creds['lease_duration']
        }

---

Security Anti-Patterns

Anti-Pattern: Security Through Obscurity

Anti-Pattern: Trusting Client Input

# BAD: Trusting client-provided data
@app.route('/admin')
def admin_panel():
    # Client can forge this header!
    if request.headers.get('X-Is-Admin') == 'true':
        return render_admin()

# GOOD: Server-side verification
@app.route('/admin')
@login_required
def admin_panel():
    if not current_user.has_role('admin'):
        abort(403)
    return render_admin()

Anti-Pattern: Hardcoded Secrets

# BAD: Hardcoded credentials
DATABASE_URL = "postgresql://admin:SuperSecret123@localhost/db"
API_KEY = "sk-1234567890abcdef"

# GOOD: Environment variables + secret management
import os
DATABASE_URL = os.environ['DATABASE_URL']
API_KEY = vault_client.get_secret('api/keys')['api_key']

Anti-Pattern: Verbose Error Messages

# BAD: Reveals internal information
except Exception as e:
    return {'error': str(e), 'stack_trace': traceback.format_exc()}, 500

# GOOD: Generic message, detailed logging
except Exception as e:
    logger.exception(f"Internal error: {e}")
    return {'error': 'An internal error occurred', 'request_id': request_id}, 500

---

Security Tools Reference

Threat Modeling Guide

Systematic approaches for identifying, analyzing, and mitigating security threats.

---

Table of Contents

• Threat Modeling Process
• STRIDE Framework
• Attack Trees
• DREAD Risk Scoring
• Data Flow Diagrams
• Common Attack Patterns

---

Threat Modeling Process

Workflow: Conduct Threat Model

1. Define the scope and objectives:

• System boundaries
• Assets to protect
• Trust levels

2. Create data flow diagram:

• External entities
• Processes
• Data stores
• Data flows
• Trust boundaries

3. Identify threats using STRIDE:

• Apply STRIDE to each DFD element
• Document threat scenarios

4. Analyze and prioritize risks:

• Score using DREAD
• Rank by severity

5. Define mitigations:

• Map controls to threats
• Identify gaps

6. Validate and iterate:

• Review with team
• Update as system evolves

7. Document in threat model report 8. Validation: All DFD elements analyzed; threats documented; mitigations mapped; residual risks accepted

Threat Model Template

THREAT MODEL REPORT

System: [System Name]
Version: [Version]
Date: [Date]
Author: [Name]

1. SYSTEM OVERVIEW
   - Purpose: [Description]
   - Users: [User types]
   - Data: [Data classification]

2. SCOPE
   - In Scope: [Components included]
   - Out of Scope: [Components excluded]
   - Assumptions: [Security assumptions]

3. DATA FLOW DIAGRAM
   [DFD image or ASCII representation]

4. THREATS IDENTIFIED
   | ID | Element | STRIDE | Threat | DREAD | Mitigation |
   |----|---------|--------|--------|-------|------------|

5. RESIDUAL RISKS
   [Accepted risks with justification]

6. RECOMMENDATIONS
   [Prioritized security improvements]

---

STRIDE Framework

Categorization model for identifying threats.

STRIDE Categories

STRIDE per Element

STRIDE Analysis Template

STRIDE ANALYSIS

Element: User Authentication Service
Type: Process

┌─────────────────────────────────────────────────────────────────┐
│ SPOOFING                                                        │
├─────────────────────────────────────────────────────────────────┤
│ Threat: Attacker uses stolen credentials to impersonate user   │
│ Attack Vector: Phishing, credential stuffing, session hijack   │
│ Likelihood: High                                                │
│ Impact: High - Full account access                              │
│ Mitigation: MFA, session binding, anomaly detection             │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ TAMPERING                                                       │
├─────────────────────────────────────────────────────────────────┤
│ Threat: Attacker modifies authentication request in transit    │
│ Attack Vector: Man-in-the-middle, request manipulation         │
│ Likelihood: Medium                                              │
│ Impact: High - Bypass authentication                            │
│ Mitigation: TLS 1.3, request signing, HSTS                      │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ REPUDIATION                                                     │
├─────────────────────────────────────────────────────────────────┤
│ Threat: User denies performing privileged action               │
│ Attack Vector: Claim account was compromised                   │
│ Likelihood: Medium                                              │
│ Impact: Medium - Dispute resolution difficulty                  │
│ Mitigation: Comprehensive audit logging, log integrity         │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ INFORMATION DISCLOSURE                                          │
├─────────────────────────────────────────────────────────────────┤
│ Threat: Password hashes exposed via SQL injection              │
│ Attack Vector: SQLi, backup exposure, error messages           │
│ Likelihood: Medium                                              │
│ Impact: Critical - Mass credential compromise                   │
│ Mitigation: Parameterized queries, encryption, error handling  │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ DENIAL OF SERVICE                                               │
├─────────────────────────────────────────────────────────────────┤
│ Threat: Brute force attacks overwhelm authentication service   │
│ Attack Vector: Credential stuffing, distributed attacks        │
│ Likelihood: High                                                │
│ Impact: High - Users cannot authenticate                        │
│ Mitigation: Rate limiting, CAPTCHA, account lockout            │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ ELEVATION OF PRIVILEGE                                          │
├─────────────────────────────────────────────────────────────────┤
│ Threat: Regular user gains admin privileges                    │
│ Attack Vector: JWT manipulation, IDOR, role confusion          │
│ Likelihood: Medium                                              │
│ Impact: Critical - Full system compromise                       │
│ Mitigation: Server-side authorization, signed tokens, RBAC     │
└─────────────────────────────────────────────────────────────────┘

Threat Mitigation Matrix

---

Attack Trees

Visual representation of attack paths to a specific goal.

Attack Tree Structure

ATTACK TREE: Compromise User Account

                    ┌─────────────────────┐
                    │ GOAL: Access User   │
                    │      Account        │
                    └──────────┬──────────┘
                               │
           ┌───────────────────┼───────────────────┐
           │                   │                   │
    ┌──────┴──────┐     ┌──────┴──────┐     ┌──────┴──────┐
    │   Obtain    │     │   Bypass    │     │   Exploit   │
    │ Credentials │     │    Auth     │     │   Session   │
    │    [OR]     │     │    [OR]     │     │    [OR]     │
    └──────┬──────┘     └──────┬──────┘     └──────┬──────┘
           │                   │                   │
     ┌─────┼─────┐       ┌─────┼─────┐       ┌─────┼─────┐
     │     │     │       │     │     │       │     │     │
   ┌─┴─┐ ┌─┴─┐ ┌─┴─┐   ┌─┴─┐ ┌─┴─┐ ┌─┴─┐   ┌─┴─┐ ┌─┴─┐ ┌─┴─┐
   │Phi│ │Crd│ │Key│   │SQL│ │JWT│ │Pwd│   │XSS│ │Fix│ │Sid│
   │sh │ │Stf│ │Log│   │ i │ │Frg│ │Rst│   │   │ │tn │ │Hj │
   └───┘ └───┘ └───┘   └───┘ └───┘ └───┘   └───┘ └───┘ └───┘

Legend:
- Phi: Phishing
- CrdStf: Credential Stuffing
- KeyLog: Keylogger
- SQLi: SQL Injection
- JWTFrg: JWT Forgery
- PwdRst: Password Reset Flaw
- XSS: Cross-Site Scripting
- Fixtn: Session Fixation
- SidHj: Session Hijacking

Attack Tree Analysis

Calculating Attack Probability

def calculate_attack_probability(attack_tree_node):
    """
    Calculate cumulative probability of attack success.

    For OR nodes: P = 1 - (1-P1)(1-P2)...(1-Pn)
    For AND nodes: P = P1 * P2 * ... * Pn
    """
    if node.is_leaf:
        return node.probability

    child_probs = [calculate_attack_probability(c) for c in node.children]

    if node.operator == 'OR':
        # At least one path succeeds
        prob_all_fail = 1
        for p in child_probs:
            prob_all_fail *= (1 - p)
        return 1 - prob_all_fail

    elif node.operator == 'AND':
        # All paths must succeed
        prob_all_succeed = 1
        for p in child_probs:
            prob_all_succeed *= p
        return prob_all_succeed

---

DREAD Risk Scoring

Quantitative risk assessment for prioritizing threats.

DREAD Components

DREAD Scoring Guide

Damage Potential:

Reproducibility:

Exploitability:

Affected Users:

Discoverability:

DREAD Calculation

def calculate_dread_score(damage, reproducibility, exploitability,
                           affected_users, discoverability):
    """
    Calculate DREAD risk score.

    Returns: Float between 1-10
    Risk Levels:
        8-10: Critical
        6-7.9: High
        4-5.9: Medium
        1-3.9: Low
    """
    score = (damage + reproducibility + exploitability +
             affected_users + discoverability) / 5
    return round(score, 1)

def get_risk_level(dread_score):
    if dread_score >= 8:
        return 'Critical'
    elif dread_score >= 6:
        return 'High'
    elif dread_score >= 4:
        return 'Medium'
    else:
        return 'Low'

DREAD Assessment Example

THREAT: SQL Injection in Login Form

| Factor | Score | Justification |
|--------|-------|---------------|
| Damage | 9 | Full database access, credential theft |
| Reproducibility | 9 | Consistent, automated tools exist |
| Exploitability | 8 | Well-documented attack, easy tools |
| Affected Users | 10 | All users with accounts |
| Discoverability | 7 | Scanners detect easily |

DREAD Score: (9+9+8+10+7)/5 = 8.6
Risk Level: CRITICAL
Priority: Immediate remediation required

---

Data Flow Diagrams

Visual representation of system data movement for security analysis.

DFD Elements

DFD Levels

Example: E-Commerce DFD

LEVEL 0: CONTEXT DIAGRAM

                              ┌──────────────────┐
                              │                  │
     ┌────────────┐           │   E-Commerce     │           ┌────────────┐
     │            │  Orders   │    System        │  Payment  │            │
     │  Customer  │──────────▶│                  │──────────▶│  Payment   │
     │            │◀──────────│                  │◀──────────│  Gateway   │
     └────────────┘  Status   │                  │  Result   └────────────┘
                              │                  │
                              └──────────────────┘
                                      │
                                      │ Fulfillment
                                      ▼
                              ┌────────────────┐
                              │   Warehouse    │
                              │    System      │
                              └────────────────┘


LEVEL 1: EXPANDED VIEW

┌─────────────────────────────────────────────────────────────────────┐
│                         TRUST BOUNDARY                               │
│  - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -  │
│                                                                      │
│   ┌─────────┐       ┌─────────┐       ┌─────────┐       ┌─────────┐ │
│   │         │       │   Web   │       │  Order  │       │ Payment │ │
│   │   CDN   │──────▶│ Server  │──────▶│ Service │──────▶│ Service │ │
│   │         │       │         │       │         │       │         │ │
│   └─────────┘       └────┬────┘       └────┬────┘       └────┬────┘ │
│                          │                 │                 │      │
│                          │                 │                 │      │
│                          ▼                 ▼                 ▼      │
│                    ╔═══════════╗     ╔═══════════╗    ╔═══════════╗ │
│                    ║  Session  ║     ║  Orders   ║    ║  Payment  ║ │
│                    ║   Store   ║     ║    DB     ║    ║    DB     ║ │
│                    ╚═══════════╝     ╚═══════════╝    ╚═══════════╝ │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘
                          │
                          │ Crosses Trust Boundary
                          ▼
                    ┌───────────┐
                    │  Payment  │
                    │  Gateway  │
                    │ (External)│
                    └───────────┘

Trust Boundary Analysis

---

Common Attack Patterns

OWASP Top 10 Mapping

Attack Pattern Catalog

ATTACK PATTERN: SQL Injection (A03)

Threat: T (Tampering), E (Elevation of Privilege)

Attack Vector:
1. Identify input fields that construct SQL queries
2. Test for injection: ' OR '1'='1' --
3. Extract data: UNION SELECT password FROM users
4. Escalate: Execute stored procedures, write files

Detection:
- WAF rules for SQL patterns
- Prepared statement verification
- Database query logging

Mitigation:
- Parameterized queries (primary)
- Input validation (secondary)
- Least privilege database accounts
- Web application firewall

Test Cases:
- Single quote injection: '
- Boolean-based: ' OR 1=1 --
- Time-based: '; WAITFOR DELAY '0:0:5' --
- UNION-based: ' UNION SELECT NULL, username, password FROM users --

Threat Intelligence Integration