Every day:
  🤖 Bots make BILLIONS of login attempts across the internet
  💀 Attackers test LEAKED credentials from previous breaches
  📱 Phishing kits capture MILLIONS of passwords
  🔓 Session tokens are stolen through XSS, MITM, or malware
  🎭 Account takeovers happen EVERY 2 SECONDS globally

A07 — Identification and Authentication Failures covers:
├── Identification: Weak user discovery/enumeration prevention
├── Authentication: Broken login mechanisms
└── Session Management: Compromised session handling

A01 — Broken Access Control covers:
└── Authorization: Who can do what (separate concern)

# 🔴 WRONG — Confusing authentication with authorization
if user.username == "admin":  # ← This is IDENTIFICATION only
    allow_admin_access()        # ← Should check AUTHORIZATION

# 🟢 RIGHT — Separate concerns
if authenticate(username, password, mfa_token):  # Authentication
    if user.has_permission("admin:access"):      # Authorization
        allow_admin_access()

Target: Admin account with 8-character alphanumeric password

Password space: 36^8 = ~2.8 trillion combinations
Attempts per second (modern GPU): ~100 billion
Time to crack: 2.8 trillion / 100 billion = 28 seconds

Reality check: Online brute force is slower (rate limiting),
but OFFLINE brute force (after stealing hash) is terrifying.

🔴 Traditional brute force (fails to lockout):
  Try: [email protected], password1, password2, password3, ... (BLOCKED)

🟢 Password spraying (bypasses lockout):
  Try: [email protected], Winter2024!
  Try: [email protected], Winter2024!
  Try: [email protected], Winter2024!
  Try: ... 10,000 users, same password

Why it works: Account lockout triggers on failed attempts per USER,
not per PASSWORD. Trying one common password across many users
avoids triggering lockouts.

🔴 VULNERABLE FLOW:

1. Attacker visits site, gets session ID: SESSIONID=xyz789
2. Attacker tricks victim into using same session:
   Email: "Click here! https:">//site.com/login?SESSIONID=xyz789"
3. Victim logs in — session ID xyz789 is now authenticated
4. Attacker also uses SESSIONID=xyz789 → authenticated as victim

🟢 FIX: Regenerate session ID on login

🎣 Modern phishing kits:
  - Clone target website perfectly (1-click deployment)
  - Capture credentials in real-time
  - Even bypass MFA (relay attacks)
  - Use lookalike domains: paypal-security.com, 
    rnicrosoft.com (rn = m), amaz0n.com

Statistics:
  - 1 in 4 people click phishing links
  - 1 in 10 submit credentials
  - Average phishing campaign ROI: 1500%

Common MFA bypasses:
  🔴 SMS interception (SIM swapping)
  🔴 Push notification fatigue (spam approvals)
  🔴 OAuth consent phishing
  🔴 Session cookie theft (bypasses MFA entirely)
  🔴 Account recovery flow (circumvents MFA)
  🔴 MFA removal via support social engineering

🔴 VULNERABLE — Different responses reveal account existence:

POST /login
email: [email protected], password: wrong
Response: "Invalid password"           ← Account EXISTS

POST /login
email: [email protected], password: wrong
Response: "User not found"             ← Account doesn't exist

Attacker now has a list of valid email addresses to target.

🔴 EXAMPLES:
- Minimum 6 characters
- No complexity requirements
- No breach check
- Common passwords allowed ("password123")

🟢 MODERN REQUIREMENTS (NIST SP 800-63B):
- Minimum 8 characters (12+ preferred)
- Check against breach databases (HaveIBeenPwned)
- No arbitrary complexity rules (per NIST)
- No periodic forced resets (unless breach suspected)
- Allow passphrases and all printable characters

🔴 NO PROTECTION:
- Unlimited login attempts
- Enables brute force and credential stuffing

🟢 PROTECTION:
- Account lockout after N attempts (5-10 typically)
- Exponential backoff on failures
- CAPTCHA after 3 failed attempts
- IP-based rate limiting
- Device fingerprinting
- Anomaly detection (unusual location/time)

# 🔴 VULNERABLE
def login(username, password):
    if authenticate(username, password):
        # Session ID stays the same before and after login
        session['user_id'] = get_user_id(username)
        return "Logged in"

# 🟢 SECURE
def login(username, password):
    if authenticate(username, password):
        # ✅ Regenerate session ID on login
        session.regenerate_id()
        session['user_id'] = get_user_id(username)
        return "Logged in"

🔴 BROKEN:
- MD5 (cracked, instant)
- SHA-1 (cracked, instant)
- SHA-256 without salt (rainbow tables)
- SHA-256 with salt but 1 iteration (fast GPU attack)

🟢 SECURE (in order of preference):
- Argon2id (OWASP recommended)
- scrypt (good)
- bcrypt (classic, still secure)
- PBKDF2 (acceptable, use high iteration count)

Minimum settings:
- Argon2id: m=19MB, t=2, p=1
- bcrypt: cost=12
- PBKDF2: 600,000 iterations (SHA-256)

❌ OLD (and WRONG) ADVICE:
  - Force password changes every 90 days
  - Require 1 uppercase, 1 lowercase, 1 digit, 1 symbol
  - Minimum 8 characters
  - Prohibit sequential characters
  - Password hints

✅ NEW (NIST 800-63B) GUIDELINES:
  - Minimum 8 characters (12+ for high-security)
  - Check against known breach databases
  - Allow all printable characters AND spaces
  - Allow long passphrases (up to 64 chars minimum)
  - NO periodic forced changes (change only on compromise)
  - NO password hints
  - NO knowledge-based security questions
  - NO SMS-based 2FA for high-security contexts

Research shows (and NIST confirms):

When forced to change passwords every 90 days, users:
  🔴 Create weaker passwords (easier to remember new ones)
  🔴 Use predictable patterns: Password1 → Password2 → Password3
  🔴 Reuse passwords from personal accounts
  🔴 Write passwords down (sticky notes under keyboards)
  🔴 Use password managers less effectively

Result: SECURITY DECREASES, not increases.

# ══════════════════════════════════════════════════════════════
# 🔴 CATASTROPHIC — Plaintext storage
# ══════════════════════════════════════════════════════════════
def register(username, password):
    db.execute(
        "INSERT INTO users (username, password) VALUES (?, ?)",
        (username, password)  # 🔴 PLAINTEXT!
    )

# ══════════════════════════════════════════════════════════════
# 🔴 BROKEN — MD5 (cracked instantly)
# ══════════════════════════════════════════════════════════════
import hashlib
def register(username, password):
    hashed = hashlib.md5(password.encode()).hexdigest()  # 🔴
    db.execute("INSERT INTO users VALUES (?, ?)", (username, hashed))

# ══════════════════════════════════════════════════════════════
# 🔴 STILL BROKEN — SHA-256 without salt (rainbow tables)
# ══════════════════════════════════════════════════════════════
def register(username, password):
    hashed = hashlib.sha256(password.encode()).hexdigest()  # 🔴
    db.execute("INSERT INTO users VALUES (?, ?)", (username, hashed))

# ══════════════════════════════════════════════════════════════
# 🟡 OKAY — bcrypt (acceptable, but Argon2id is better)
# ══════════════════════════════════════════════════════════════
import bcrypt
def register(username, password):
    salt = bcrypt.gensalt(rounds=12)
    hashed = bcrypt.hashpw(password.encode(), salt)
    db.execute("INSERT INTO users VALUES (?, ?)", (username, hashed))

# ══════════════════════════════════════════════════════════════
# 🟢 BEST — Argon2id (OWASP recommended)
# ══════════════════════════════════════════════════════════════
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError, VerificationError

# Create hasher with OWASP-recommended parameters
ph = PasswordHasher(
    time_cost=2,       # Iterations
    memory_cost=19456, # 19 MB
    parallelism=1,     # Threads
    hash_len=32,       # Hash output length
    salt_len=16        # Salt length
)

def register(username, password):
    # ✅ Validate password strength FIRST
    if not is_strong_password(password):
        raise ValueError("Password too weak")
    
    # ✅ Check against breach database
    if is_breached_password(password):
        raise ValueError("Password found in data breaches")
    
    # ✅ Hash with Argon2id
    hashed = ph.hash(password)
    
    # ✅ Store only the hash (contains salt + params)
    db.execute(
        "INSERT INTO users (username, password_hash) VALUES (?, ?)",
        (username, hashed)
    )

def login(username, password):
    user = db.query_one(
        "SELECT password_hash FROM users WHERE username = ?",
        (username,)
    )
    
    if not user:
        # ✅ Prevent timing attacks — hash a dummy password
        ph.hash("dummy_password_for_timing")
        raise AuthenticationError("Invalid credentials")
    
    try:
        ph.verify(user['password_hash'], password)
    except (VerifyMismatchError, VerificationError):
        raise AuthenticationError("Invalid credentials")
    
    # ✅ Check if hash needs upgrading (if we increase parameters later)
    if ph.check_needs_rehash(user['password_hash']):
        new_hash = ph.hash(password)
        db.execute(
            "UPDATE users SET password_hash = ? WHERE username = ?",
            (new_hash, username)
        )
    
    return "Logged in"

import hashlib
import requests

def is_breached_password(password: str) -> bool:
    """
    Check password against HaveIBeenPwned API.
    Uses k-anonymity: only sends first 5 chars of SHA-1 hash.
    The password itself NEVER leaves your server.
    """
    # Hash the password
    sha1_hash = hashlib.sha1(password.encode()).hexdigest().upper()
    prefix = sha1_hash[:5]
    suffix = sha1_hash[5:]
    
    # Query HIBP API (returns all hashes with same prefix)
    response = requests.get(
        f'https://api.pwnedpasswords.com/range/{prefix}',
        timeout=5
    )
    
    if response.status_code != 200:
        # Fail secure — if we can't check, assume it's breached
        return True
    
    # Check if our suffix is in the response
    for line in response.text.splitlines():
        hash_suffix, count = line.split(':')
        if hash_suffix == suffix:
            return True  # Found in breaches
    
    return False

🔴 SMS 2FA vulnerabilities:
  1. SIM swapping (social engineer carrier → steal phone number)
  2. SS7 attacks (telecom protocol flaws → intercept SMS)
  3. Phishing (user types code into fake site)
  4. Malware on phone (reads SMS)
  5. Phone theft

NIST deprecated SMS 2FA in 2016 for government/high-security use.

But: SMS 2FA still blocks 76% of automated attacks.
     Better than NO MFA. But WebAuthn blocks 100%.

import pyotp
import qrcode
from io import BytesIO
import base64

def enable_mfa(user_id: str):
    """Enable TOTP MFA for a user"""
    # ✅ Generate a cryptographically random secret
    secret = pyotp.random_base32()
    
    # ✅ Store encrypted in database (NOT plaintext!)
    encrypted_secret = encrypt(secret, get_encryption_key())
    db.execute(
        "UPDATE users SET mfa_secret = ?, mfa_enabled = FALSE WHERE id = ?",
        (encrypted_secret, user_id)
    )
    
    # ✅ Generate provisioning URI
    user = get_user(user_id)
    totp = pyotp.TOTP(secret)
    uri = totp.provisioning_uri(
        name=user['email'],
        issuer_name='MySecureApp'
    )
    
    # ✅ Generate QR code
    qr = qrcode.make(uri)
    buffer = BytesIO()
    qr.save(buffer, format='PNG')
    qr_base64 = base64.b64encode(buffer.getvalue()).decode()
    
    # ✅ Also provide backup codes
    backup_codes = generate_backup_codes(10)
    store_backup_codes(user_id, backup_codes)
    
    return {
        'qr_code': f'data:image/png;base64,{qr_base64}',
        'secret': secret,  # Show once for manual entry
        'backup_codes': backup_codes
    }

def verify_mfa(user_id: str, token: str) -> bool:
    """Verify TOTP token"""
    user = db.query_one(
        "SELECT mfa_secret FROM users WHERE id = ? AND mfa_enabled = TRUE",
        (user_id,)
    )
    
    if not user:
        return False
    
    secret = decrypt(user['mfa_secret'], get_encryption_key())
    totp = pyotp.TOTP(secret)
    
    # ✅ Allow 1 window drift (30 seconds) for clock skew
    # ✅ Verify once — prevents token reuse
    if totp.verify(token, valid_window=1):
        # Check for replay attack
        if is_token_already_used(user_id, token):
            return False
        mark_token_used(user_id, token)
        return True
    
    return False

def generate_backup_codes(count: int = 10) -> list:
    """Generate single-use backup codes"""
    import secrets
    import string
    
    codes = []
    for _ in range(count):
        # Format: XXXX-XXXX (readable)
        code = '-'.join(
            ''.join(secrets.choice(string.ascii_uppercase + string.digits)
                   for _ in range(4))
            for _ in range(2)
        )
        codes.append(code)
    
    return codes

// 🟢 WEBAUTHN REGISTRATION (Browser)

async function registerPasskey() {
    // Server generates challenge
    const options = await fetch('/api/webauthn/register-options', {
        method: 'POST'
    }).then(r => r.json());
    
    // Browser creates credential (uses platform authenticator or security key)
    const credential = await navigator.credentials.create({
        publicKey: {
            challenge: base64ToBuffer(options.challenge),
            rp: {
                name: 'MySecureApp',
                id: 'example.com'
            },
            user: {
                id: base64ToBuffer(options.user.id),
                name: options.user.email,
                displayName: options.user.name
            },
            pubKeyCredParams: [
                { alg: -7, type: 'public-key' },   // ES256
                { alg: -257, type: 'public-key' }  // RS256
            ],
            authenticatorSelection: {
                authenticatorAttachment: 'platform',  // Prefer built-in
                userVerification: 'required',         // Require biometric/PIN
                residentKey: 'preferred'              // Discoverable credentials
            },
            timeout: 60000,
            attestation: 'direct'
        }
    });
    
    // Send credential to server for verification
    await fetch('/api/webauthn/register', {
        method: 'POST',
        body: JSON.stringify({
            id: credential.id,
            rawId: bufferToBase64(credential.rawId),
            response: {
                clientDataJSON: bufferToBase64(credential.response.clientDataJSON),
                attestationObject: bufferToBase64(credential.response.attestationObject)
            }
        })
    });
}

"text-slate-500 font-normal italic"># 🟢 PRODUCTION-READY SESSION COOKIE

Set-Cookie: __Host-sessionid=abc123xyz...;
    Secure;                        "text-slate-500 font-normal italic"># Only sent over HTTPS
    HttpOnly;                      "text-slate-500 font-normal italic"># No JavaScript access (XSS protection)
    SameSite=Strict;              "text-slate-500 font-normal italic"># No cross-site requests (CSRF protection)
    Path=/;                        "text-slate-500 font-normal italic"># Available site-wide
    Max-Age=3600;                 "text-slate-500 font-normal italic"># Expires in 1 hour
    "text-slate-500 font-normal italic"># No Domain attribute (default: current host only)

"text-slate-500 font-normal italic"># Why __Host- prefix?
"text-slate-500 font-normal italic"># - Requires Secure flag
"text-slate-500 font-normal italic"># - Prohibits Domain attribute
"text-slate-500 font-normal italic"># - Requires Path=/
"text-slate-500 font-normal italic"># - Maximum cookie security

# 🟢 SECURE SESSION CONFIGURATION

from flask import Flask
from datetime import timedelta
import secrets

app = Flask(__name__)

app.config.update(
    # ✅ Strong secret key (never hardcode!)
    SECRET_KEY=os.environ['SECRET_KEY'],  # 64+ random bytes
    
    # ✅ Secure cookie settings
    SESSION_COOKIE_NAME='__Host-session',
    SESSION_COOKIE_SECURE=True,        # HTTPS only
    SESSION_COOKIE_HTTPONLY=True,      # No JS access
    SESSION_COOKIE_SAMESITE='Strict',  # CSRF protection
    
    # ✅ Short expiration
    PERMANENT_SESSION_LIFETIME=timedelta(minutes=30),
    
    # ✅ Session regeneration
    SESSION_REFRESH_EACH_REQUEST=True,
)

def generate_session_id() -> str:
    """Generate cryptographically secure session ID"""
    # ✅ 256 bits of entropy minimum
    return secrets.token_urlsafe(32)

def login(username, password, mfa_token):
    # ✅ Verify credentials
    user = authenticate(username, password, mfa_token)
    if not user:
        raise AuthenticationError()
    
    # ✅ CRITICAL: Regenerate session ID on login
    # Prevents session fixation attacks
    session.clear()
    session.regenerate()  # New ID
    
    # ✅ Store minimal data in session
    session['user_id'] = user.id
    session['authenticated_at'] = time.time()
    session['ip_address'] = request.remote_addr
    session['user_agent_hash'] = hash_user_agent(request.user_agent)
    
    # ✅ Store session server-side (not just in cookie)
    session_store.set(session.sid, {
        'user_id': user.id,
        'created_at': datetime.utcnow(),
        'last_activity': datetime.utcnow(),
        'ip_address': request.remote_addr
    }, ex=1800)  # 30 min TTL
    
    # ✅ Log authentication event
    audit_log.info(f"Login success: user={user.id} ip={request.remote_addr}")

def validate_session():
    """Validate session on every request"""
    if 'user_id' not in session:
        return None
    
    # ✅ Check session age
    if time.time() - session['authenticated_at'] > 86400:  # 24 hours max
        session.clear()
        return None
    
    # ✅ Check idle timeout (30 min)
    last_activity = session.get('last_activity', 0)
    if time.time() - last_activity > 1800:
        session.clear()
        return None
    
    # ✅ Detect session hijacking
    # Check if IP or User-Agent changed (session binding)
    if session.get('ip_address') != request.remote_addr:
        audit_log.warning(f"Session IP mismatch: user={session['user_id']}")
        session.clear()
        return None
    
    if session.get('user_agent_hash') != hash_user_agent(request.user_agent):
        audit_log.warning(f"User-Agent change: user={session['user_id']}")
        session.clear()
        return None
    
    # ✅ Update last activity
    session['last_activity'] = time.time()
    
    return get_user(session['user_id'])

def logout():
    user_id = session.get('user_id')
    
    # ✅ Delete session server-side
    if hasattr(session, 'sid'):
        session_store.delete(session.sid)
    
    # ✅ Clear client-side cookie
    session.clear()
    
    # ✅ Audit log
    if user_id:
        audit_log.info(f"Logout: user={user_id}")

🟢 LAYERED TIMEOUTS:

1. Idle Timeout:        15-30 minutes
   → User inactive = logged out
   → Prevents unattended session abuse

2. Absolute Timeout:    8-24 hours
   → Maximum session lifetime
   → Forces re-authentication regardless of activity

3. Re-authentication:   For sensitive actions
   → Change password, transfer money, delete data
   → Require password/MFA even in active session

4. Session Invalidation on:
   - Password change
   - MFA setup/removal
   - Email change
   - Suspicious activity
   - Logout from any device (if "logout all" chosen)

header.payload.signature

eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NSJ9.X_signature

Header (base64):   {"alg": "HS256", "typ": "JWT"}
Payload (base64):  {"sub": "12345", "exp": 1234567890}
Signature:         HMAC-SHA256(header + "." + payload, secret)

# 🔴 CRITICAL VULNERABILITY — "none" algorithm bypass

# Attacker sends:
# Header:  {"alg": "none"}
# Payload: {"sub": "admin", "role": "superuser"}
# Signature: (empty)

# 🔴 VULNERABLE CODE
import jwt
def verify_token(token):
    payload = jwt.decode(token, verify=False)  # 🔴 DISASTER
    return payload

# 🟢 SECURE CODE
def verify_token(token):
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=['HS256'],  # ✅ Explicitly specify algorithm
            options={
                'verify_signature': True,
                'verify_exp': True,
                'verify_iat': True,
                'verify_nbf': True,
                'require': ['exp', 'iat', 'sub']
            }
        )
        return payload
    except jwt.InvalidTokenError:
        return None

Scenario: Server uses RS256 (asymmetric)
          Attacker tricks server to use HS256 (symmetric)

1. Attacker gets server's PUBLIC key (it's public!)
2. Attacker crafts token:
   - Header: {"alg": "HS256"}  (changed from RS256)
   - Payload: {"sub": "admin"}
   - Signature: HMAC-SHA256(header + payload, PUBLIC_KEY)

3. Server receives token, uses PUBLIC key to verify HS256
4. Signature matches! → Attacker is admin

🟢 FIX: Hard-code expected algorithm, reject tokens with mismatched alg

import jwt
import datetime
import secrets
from functools import wraps

# ✅ Use STRONG secret (256+ bits for HS256)
JWT_SECRET = os.environ['JWT_SECRET']  # Loaded from secrets manager
JWT_ALGORITHM = 'HS256'
JWT_ISSUER = 'myapp.com'
JWT_AUDIENCE = 'myapp.com/api'

# ✅ Short access token lifetime
ACCESS_TOKEN_EXPIRE_MINUTES = 15

# ✅ Longer refresh token lifetime (but still limited)
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_access_token(user_id: str, roles: list) -> str:
    """Create short-lived access token"""
    now = datetime.datetime.utcnow()
    
    payload = {
        'sub': user_id,                    # Subject (user ID)
        'iat': now,                        # Issued at
        'exp': now + datetime.timedelta(   # Expiration
            minutes=ACCESS_TOKEN_EXPIRE_MINUTES
        ),
        'nbf': now,                        # Not before
        'iss': JWT_ISSUER,                 # Issuer
        'aud': JWT_AUDIENCE,               # Audience
        'jti': secrets.token_urlsafe(16),  # Unique token ID
        'type': 'access',
        'roles': roles
    }
    
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

def create_refresh_token(user_id: str) -> str:
    """Create refresh token (stored server-side)"""
    now = datetime.datetime.utcnow()
    token_id = secrets.token_urlsafe(32)
    
    payload = {
        'sub': user_id,
        'iat': now,
        'exp': now + datetime.timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
        'nbf': now,
        'iss': JWT_ISSUER,
        'jti': token_id,
        'type': 'refresh'
    }
    
    # ✅ Store refresh token server-side for revocation
    refresh_store.set(f"refresh:{token_id}", {
        'user_id': user_id,
        'created_at': now.isoformat()
    }, ex=REFRESH_TOKEN_EXPIRE_DAYS * 86400)
    
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

def verify_token(token: str, token_type: str = 'access') -> dict:
    """Verify and decode JWT"""
    try:
        payload = jwt.decode(
            token,
            JWT_SECRET,
            algorithms=[JWT_ALGORITHM],  # ✅ Explicit algorithm
            issuer=JWT_ISSUER,
            audience=JWT_AUDIENCE,
            options={
                'verify_signature': True,
                'verify_exp': True,
                'verify_iat': True,
                'verify_nbf': True,
                'verify_iss': True,
                'verify_aud': True,
                'require': ['exp', 'iat', 'sub', 'jti', 'type']
            }
        )
        
        # ✅ Verify token type
        if payload.get('type') != token_type:
            raise jwt.InvalidTokenError("Wrong token type")
        
        # ✅ Check revocation list
        if is_token_revoked(payload['jti']):
            raise jwt.InvalidTokenError("Token revoked")
        
        # ✅ For refresh tokens, verify server-side existence
        if token_type == 'refresh':
            if not refresh_store.exists(f"refresh:{payload['jti']}"):
                raise jwt.InvalidTokenError("Refresh token not found")
        
        return payload
        
    except jwt.ExpiredSignatureError:
        raise AuthenticationError("Token expired")
    except jwt.InvalidTokenError as e:
        raise AuthenticationError(f"Invalid token: {e}")

def revoke_token(jti: str, exp: datetime.datetime):
    """Add token to revocation list"""
    ttl = int((exp - datetime.datetime.utcnow()).total_seconds())
    if ttl > 0:
        revocation_store.set(f"revoked:{jti}", "1", ex=ttl)

def is_token_revoked(jti: str) -> bool:
    return revocation_store.exists(f"revoked:{jti}")

✅ DO:
  - Use short expiration (15 min for access tokens)
  - Implement refresh token rotation
  - Store refresh tokens server-side for revocation
  - Use strong secrets (256+ bits)
  - Explicitly specify algorithm when verifying
  - Validate ALL claims (exp, iat, iss, aud, nbf)
  - Use HTTPS exclusively
  - Store in httpOnly, Secure, SameSite cookies (not localStorage)

❌ DON'T:
  - Store JWTs in localStorage (XSS vulnerability)
  - Use "none" algorithm
  - Trust the alg header without validation
  - Use long-lived access tokens
  - Store sensitive data in payload (it's base64, not encrypted)
  - Skip signature verification "for performance"
  - Use JWTs for session management without revocation strategy

🔴 PITFALL 1: Using Implicit Flow in SPAs
   OLD: Implicit flow returned tokens in URL fragment (leaked)
   NEW: Use Authorization Code + PKCE for SPAs

🔴 PITFALL 2: Not validating redirect_uri
   Attacker: /authorize?redirect_uri=attacker.com
   Server: Sends code to attacker.com
   FIX: Exact match whitelist of redirect URIs

🔴 PITFALL 3: Using OAuth for authentication (without OIDC)
   OAuth = "This app can access your data"
   NOT = "This person is who they say they are"
   FIX: Use OIDC (which adds id_token for authentication)

🔴 PITFALL 4: Not verifying state parameter (CSRF)
   FIX: Generate random state, verify on callback

🔴 PITFALL 5: Treating access tokens as session tokens
   Access tokens = short-lived (minutes)
   Sessions = can be longer with proper management
   
🔴 PITFALL 6: Storing tokens in localStorage
   XSS = instant token theft
   FIX: httpOnly cookies or BFF (Backend for Frontend) pattern

🔑 How Passkeys Work:

1. User registers: Device generates public/private key pair
2. Private key: NEVER leaves the device (secured by biometric/PIN)
3. Public key: Sent to server, stored with user account
4. Login: Server sends challenge → device signs with private key
5. Server verifies signature with public key → authenticated

Benefits:
  🟢 Phishing-impossible (tied to domain cryptographically)
  🟢 No shared secret to steal
  🟢 Works across devices (iCloud/Google sync)
  🟢 Biometric unlock (Face ID, Touch ID, Windows Hello)
  🟢 Faster than passwords

# ✅ MAGIC LINK AUTHENTICATION

import secrets
from datetime import datetime, timedelta

def send_magic_link(email: str):
    # ✅ Rate limit
    if is_rate_limited(email):
        return  # Don't reveal if user exists
    
    # ✅ Generate cryptographically random token
    token = secrets.token_urlsafe(32)
    
    # ✅ Store with short TTL (10-15 minutes)
    magic_link_store.set(
        f"magic:{token}",
        {'email': email, 'created_at': datetime.utcnow().isoformat()},
        ex=900  # 15 minutes
    )
    
    # ✅ Send email with link
    link = f"https://app.example.com/auth/verify?token={token}"
    send_email(
        to=email,
        subject="Your login link",
        body=f"Click to log in: {link}\n\nExpires in 15 minutes."
    )

def verify_magic_link(token: str):
    data = magic_link_store.get(f"magic:{token}")
    if not data:
        raise AuthenticationError("Invalid or expired link")
    
    # ✅ Delete immediately (single-use)
    magic_link_store.delete(f"magic:{token}")
    
    # ✅ Create session
    user = get_or_create_user(data['email'])
    create_session(user.id)
    
    return user

def request_password_reset(email: str):
    """Initiate password reset"""
    # ✅ Rate limit (per email AND per IP)
    if is_rate_limited(f"reset:{email}") or is_rate_limited(f"ip:{request.ip}"):
        # Silent fail — don't reveal rate limiting
        return
    
    user = get_user_by_email(email)
    
    # ✅ ALWAYS respond the same way (prevent enumeration)
    # Even if user doesn't exist, pretend we sent email
    response_delay()  # Consistent timing
    
    if user:
        # ✅ Generate cryptographically secure token
        token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(token.encode()).hexdigest()
        
        # ✅ Store HASH of token (not plaintext)
        db.execute("""
            INSERT INTO password_resets (user_id, token_hash, expires_at)
            VALUES (?, ?, ?)
        """, (user.id, token_hash, datetime.utcnow() + timedelta(minutes=15)))
        
        # ✅ Send email with link
        link = f"https://app.com/reset?token={token}"
        send_email(
            to=email,
            subject="Reset your password",
            body=f"Click to reset: {link}\n\n"
                 f"Expires in 15 minutes.\n"
                 f"If you didn't request this, ignore this email.\n"
                 f"Request from IP: {request.ip} at {datetime.utcnow()}"
        )
        
        # ✅ Notify user on ALL channels
        if user.phone:
            send_sms(user.phone, "Password reset requested. Not you? Reply STOP.")
        
        # ✅ Audit log
        audit_log.info(f"Password reset requested: user={user.id}")

def verify_reset_token(token: str, new_password: str):
    """Complete password reset"""
    token_hash = hashlib.sha256(token.encode()).hexdigest()
    
    # ✅ Look up by hash (constant time comparison)
    reset = db.query_one("""
        SELECT user_id, expires_at FROM password_resets
        WHERE token_hash = ? AND used_at IS NULL
    """, (token_hash,))
    
    if not reset:
        raise ValueError("Invalid or expired token")
    
    if reset.expires_at < datetime.utcnow():
        raise ValueError("Token expired")
    
    # ✅ Validate new password
    if not is_strong_password(new_password):
        raise ValueError("Password too weak")
    
    if is_breached_password(new_password):
        raise ValueError("Password found in breaches")
    
    user = get_user(reset.user_id)
    
    # ✅ Check it's not the same as current password
    if verify_password(new_password, user.password_hash):
        raise ValueError("Must be different from current password")
    
    # ✅ Update password
    user.password_hash = hash_password(new_password)
    db.save(user)
    
    # ✅ Mark token as used (single-use)
    db.execute(
        "UPDATE password_resets SET used_at = ? WHERE token_hash = ?",
        (datetime.utcnow(), token_hash)
    )
    
    # ✅ Invalidate ALL existing sessions (force re-login everywhere)
    invalidate_all_sessions(user.id)
    
    # ✅ Notify user of change
    send_email(
        to=user.email,
        subject="Your password was changed",
        body=f"Your password was changed at {datetime.utcnow()}.\n"
             f"If this wasn't you, contact support IMMEDIATELY.\n"
             f"All active sessions have been terminated."
    )
    
    # ✅ Require MFA re-enrollment if MFA was reset
    if was_mfa_bypassed_during_recovery:
        require_mfa_setup(user.id)
    
    audit_log.info(f"Password reset completed: user={user.id}")

# ══════════════════════════════════════════════════════════════
# 🔴 VULNERABLE LOGIN
# ══════════════════════════════════════════════════════════════

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    
    # 🔴 SQL injection + plaintext comparison
    user = db.execute(
        f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    ).fetchone()
    
    if user:
        session['user'] = username  # 🔴 No session regeneration
        return "Logged in"
    else:
        # 🔴 Different messages = enumeration
        if db.execute(f"SELECT * FROM users WHERE username='{username}'").fetchone():
            return "Wrong password"
        else:
            return "User not found"


# ══════════════════════════════════════════════════════════════
# 🟢 SECURE LOGIN
# ══════════════════════════════════════════════════════════════

from flask import Flask, request, session, abort
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
import secrets
import time
from datetime import datetime, timedelta

limiter = Limiter(app, key_func=get_remote_address)
ph = PasswordHasher(time_cost=2, memory_cost=19456, parallelism=1)

@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute")      # ✅ IP rate limit
@limiter.limit("10 per hour")       # ✅ Hourly limit
def login():
    data = request.get_json()
    username = data.get('username', '').strip().lower()
    password = data.get('password', '')
    mfa_token = data.get('mfa_token')
    
    # ✅ Input validation
    if not username or not password:
        return {'error': 'Credentials required'}, 400
    
    if len(username) > 254 or len(password) > 128:
        return {'error': 'Invalid credentials'}, 401
    
    # ✅ Check if account is locked
    if is_account_locked(username):
        audit_log.warning(f"Login attempt on locked account: {username}")
        # Same response as invalid credentials
        time.sleep(0.5)  # Consistent timing
        return {'error': 'Invalid credentials'}, 401
    
    # ✅ Check if IP is blocked
    if is_ip_blocked(request.remote_addr):
        audit_log.warning(f"Login from blocked IP: {request.remote_addr}")
        abort(403)
    
    # ✅ Parameterized query
    user = db.query_one(
        "SELECT id, password_hash, mfa_enabled, mfa_secret, failed_attempts "
        "FROM users WHERE username = ? AND is_active = TRUE",
        (username,)
    )
    
    # ✅ Timing attack prevention
    # Always perform hash verification, even if user doesn't exist
    if not user:
        # Dummy verification to maintain consistent timing
        try:
            ph.verify(
                "$argon2id$v=19$m=19456,t=2,p=1$dummysalt$dummyhashvalue",
                password
            )
        except:
            pass
        
        time.sleep(0.1)  # Add jitter
        audit_log.info(f"Login failed (no user): {username} from {request.remote_addr}")
        return {'error': 'Invalid credentials'}, 401
    
    # ✅ Verify password
    try:
        ph.verify(user['password_hash'], password)
    except VerifyMismatchError:
        # ✅ Increment failed attempts
        increment_failed_attempts(user['id'])
        
        # ✅ Lock account after 5 failed attempts
        if user['failed_attempts'] + 1 >= 5:
            lock_account(user['id'], duration_minutes=30)
            send_security_alert(user['id'], 'account_locked')
        
        audit_log.info(f"Login failed (bad password): user={user['id']}")
        return {'error': 'Invalid credentials'}, 401
    
    # ✅ Password correct — check if MFA required
    if user['mfa_enabled']:
        if not mfa_token:
            # Return partial authentication (needs MFA)
            temp_token = create_mfa_pending_token(user['id'])
            return {'mfa_required': True, 'mfa_token_id': temp_token}, 200
        
        if not verify_totp(user['mfa_secret'], mfa_token):
            audit_log.warning(f"MFA failed: user={user['id']}")
            return {'error': 'Invalid credentials'}, 401
    
    # ✅ Successful login
    reset_failed_attempts(user['id'])
    
    # ✅ Check for suspicious login (new location/device)
    if is_suspicious_login(user['id'], request):
        send_security_alert(user['id'], 'new_device_login')
    
    # ✅ Rehash if parameters have changed
    if ph.check_needs_rehash(user['password_hash']):
        new_hash = ph.hash(password)
        db.execute(
            "UPDATE users SET password_hash = ? WHERE id = ?",
            (new_hash, user['id'])
        )
    
    # ✅ Regenerate session ID (prevent fixation)
    session.clear()
    session.regenerate()
    
    # ✅ Set session data
    session['user_id'] = user['id']
    session['authenticated_at'] = time.time()
    session['ip'] = request.remote_addr
    session['ua_hash'] = hash_user_agent(request.headers.get('User-Agent'))
    session.permanent = True
    
    # ✅ Audit log
    audit_log.info(
        f"Login success: user={user['id']} "
        f"ip={request.remote_addr}"
    )
    
    # ✅ Update last login
    db.execute(
        "UPDATE users SET last_login_at = ?, last_login_ip = ? WHERE id = ?",
        (datetime.utcnow(), request.remote_addr, user['id'])
    )
    
    return {'success': True, 'user': {'id': user['id']}}, 200

I will hash passwords with Argon2id, not MD5.
I will require MFA for all privileged accounts.
I will regenerate session IDs on login.
I will rate limit every authentication endpoint.
I will give consistent responses to prevent enumeration.
I will secure cookies with HttpOnly, Secure, and SameSite.
I will treat JWT like plaintext — because the payload IS plaintext.
I will offer passkey-based authentication.
I will make recovery as secure as login.
I will audit every authentication event.

┌────────────────────────────────────────────────────────────────────┐
│              THE COST OF AUTHENTICATION FAILURES                   │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  Colonial Pipeline (1 missing MFA):          $4.4M + national crisis│
│  23andMe (credential stuffing):              $30M + 6.9M victims   │
│  Twitter 2020 (admin takeover):              $120K scam + chaos    │
│  Average account takeover cost:              $1,200 per account    │
│  Average breach involving credentials:       $4.5M total           │
│                                                                    │
│  Cost of implementing MFA:                   ~$0 (free providers)  │
│  Cost of implementing WebAuthn:              ~$0 (open standard)   │
│  Cost of using Argon2id:                     ~$0 (library)         │
│                                                                    │
│  ROI of secure authentication:               INFINITE              │
│                                                                    │
└────────────────────────────────────────────────────────────────────┘