Hardening Playbook

# VULNERABLE — no ownership check
@app.get("/api/orders/{order_id}")
def get_order(order_id: int):
    return db.query(Order).filter(Order.id == order_id).first()

# SECURE — ownership check
@app.get("/api/orders/{order_id}")
def get_order(order_id: int, current_user: User = Depends(get_current_user)):
    order = db.query(Order).filter(
        Order.id == order_id,
        Order.user_id == current_user.id  # ownership check
    ).first()
    if not order:
        raise HTTPException(status_code=404)
    return order

// VULNERABLE
app.get('/api/users/:id/profile', authenticate, async (req, res) => {
  const user = await User.findById(req.params.id);
  res.json(user);
});

// SECURE
app.get('/api/users/:id/profile', authenticate, async (req, res) => {
  if (req.params.id !== req.user.id && !req.user.isAdmin) {
    return res.status(403).json({ error: 'Forbidden' });
  }
  const user = await User.findById(req.params.id);
  res.json(user);
});

// VULNERABLE
[HttpGet("orders/{orderId}")]
public async Task<IActionResult> GetOrder(int orderId)
{
    var order = await _db.Orders.FindAsync(orderId);
    return Ok(order);
}

// SECURE
[HttpGet("orders/{orderId}")]
[Authorize]
public async Task<IActionResult> GetOrder(int orderId)
{
    var userId = User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
    var order = await _db.Orders
        .Where(o => o.Id == orderId && o.UserId == userId)
        .FirstOrDefaultAsync();
    if (order == null) return NotFound();
    return Ok(order);
}

// VULNERABLE — returns all fields including passwordHash, ssn
const user = await User.findById(id);
res.json(user);

// SECURE — explicit projection
const user = await User.findById(id).select('id name email createdAt');
res.json(user);

# SECURE — Pydantic response model (FastAPI)
class UserPublicResponse(BaseModel):
    id: int
    name: str
    email: str
    # NOTE: password_hash, ssn, date_of_birth NOT included

@app.get("/api/users/{id}", response_model=UserPublicResponse)
def get_user(id: int):
    return db.query(User).filter(User.id == id).first()

// SECURE — DTO with @JsonIgnore
public class UserResponse {
    public String id;
    public String name;
    public String email;
    // passwordHash, ssn not included in DTO
}

# Patterns to search for in all files:
password\s*=\s*["'][^"']+["']
api_key\s*=\s*["'][^"']+["']
secret\s*=\s*["'][^"']+["']
token\s*=\s*["'][^"']+["']
connectionString\s*=\s*["'][^"']+["']

# VULNERABLE
DATABASE_URL = "postgresql://user:p@ssw0rd@prod-db.example.com/mydb"

# SECURE
import os
DATABASE_URL = os.environ.get("DATABASE_URL")
# In production: use Azure Key Vault, AWS Secrets Manager, or GCP Secret Manager

from azure.keyvault.secrets import SecretClient
from cryptography.fernet import Fernet

# Encrypt at write time
def encrypt_field(value: str, key: bytes) -> str:
    f = Fernet(key)
    return f.encrypt(value.encode()).decode()

# Decrypt at read time (only when authorized)
def decrypt_field(encrypted_value: str, key: bytes) -> str:
    f = Fernet(key)
    return f.decrypt(encrypted_value.encode()).decode()

import { KMSClient, EncryptCommand, DecryptCommand } from "@aws-sdk/client-kms";

const kms = new KMSClient({ region: "us-east-1" });

async function encryptField(plaintext: string): Promise<string> {
  const { CiphertextBlob } = await kms.send(new EncryptCommand({
    KeyId: process.env.KMS_KEY_ARN,
    Plaintext: Buffer.from(plaintext),
  }));
  return Buffer.from(CiphertextBlob!).toString('base64');
}

// Use Always Encrypted for SQL Server / Azure SQL
// Or manually encrypt with Azure Key Vault
services.AddDbContext<AppDbContext>(options =>
    options.UseSqlServer(connectionString, sqlOptions =>
        sqlOptions.EnableSensitiveDataLogging(false)));

// In entity:
[Column(TypeName = "nvarchar(500)")]
public string EncryptedSsn { get; set; } // store Base64 ciphertext

// CORRECT — use Stripe's tokenization
const paymentMethod = await stripe.paymentMethods.create({
  type: 'card',
  card: { token: cardToken }, // token from client-side Stripe.js
});
// Store: paymentMethod.id (token) — never the card number

// WRONG — never do this
const cardNumber = req.body.cardNumber; // Tier 2 PCI-DSS violation
await db.save({ userId, cardNumber });   // DO NOT store raw card data

# VULNERABLE
logger.info(f"User {user.email} logged in")
logger.debug(f"Payment by {user.full_name}, card ending {card_last4}")

# SECURE — log opaque identifiers, not PII
logger.info(f"User {user.id} authenticated", extra={"user_id": user.id})
logger.debug(f"Payment processed", extra={"user_id": user.id, "payment_id": payment_id})

// VULNERABLE
console.log(`Processing order for ${user.email} at ${user.address}`);

// SECURE
logger.info('Processing order', { userId: user.id, orderId: order.id });

{
  timestamp: ISO8601,
  actor_id: "user UUID",
  actor_role: "admin|user|service",
  action: "READ|WRITE|DELETE|EXPORT",
  resource_type: "User|HealthRecord|PaymentMethod",
  resource_id: "UUID of accessed record",
  fields_accessed: ["email", "phone"],  // NOT the values
  ip_address: "masked IP",
  result: "success|denied",
  correlation_id: "request trace ID"
}

// Express + express-rate-limit
import rateLimit from 'express-rate-limit';

// Aggressive limit for data export endpoint
const exportLimiter = rateLimit({
  windowMs: 60 * 60 * 1000, // 1 hour
  max: 5, // max 5 exports per hour per IP
  message: 'Too many export requests'
});

// Standard limit for data lookup
const lookupLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100
});

app.get('/api/export', exportLimiter, authMiddleware, exportController);
app.get('/api/users/:id', lookupLimiter, authMiddleware, userController);

-- Add retention column to all PII tables
ALTER TABLE users ADD COLUMN retention_expires_at TIMESTAMP;
ALTER TABLE health_records ADD COLUMN retention_expires_at TIMESTAMP;

-- Set retention at insert time
INSERT INTO users (email, retention_expires_at) 
VALUES ($1, NOW() + INTERVAL '7 years');

-- Scheduled job to hard-delete expired records (or anonymize)
DELETE FROM users 
WHERE retention_expires_at < NOW() 
AND deletion_notified_at IS NOT NULL; -- ensure user was notified

from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def purge_expired_records():
    await db.execute(
        "DELETE FROM user_sessions WHERE expires_at < NOW()"
    )
    # Anonymize users (don't delete if financial records must be retained)
    await db.execute("""
        UPDATE users SET 
            email = CONCAT('deleted_', id, '@redacted.invalid'),
            phone = NULL,
            address = NULL,
            date_of_birth = NULL
        WHERE retention_expires_at < NOW() AND deleted_at IS NULL
    """)

scheduler = AsyncIOScheduler()
scheduler.add_job(purge_expired_records, 'cron', hour=2)  # 2 AM daily
scheduler.start()

import hashlib
import hmac

PSEUDONYM_SALT = os.environ.get("PSEUDONYM_SALT")  # stored in Key Vault

def pseudonymize_user_id(real_user_id: str) -> str:
    """
    One-way: analyst can track behavior across sessions 
    but cannot identify the real user without the salt.
    """
    return hmac.new(
        PSEUDONYM_SALT.encode(), 
        real_user_id.encode(), 
        hashlib.sha256
    ).hexdigest()

# In analytics event
analytics.track({
    "user_id": pseudonymize_user_id(user.id),  # NOT real user ID
    "event": "page_viewed",
    "page": request.path,
    "timestamp": datetime.utcnow().isoformat()
})