Principes de base du secure coding : 12 règles avec code

def can_user_access(user, resource) -> bool:
    try:
        policy = load_policy(resource)
        return policy.allows(user)
    except Exception as e:
        logger.warning(f"Policy load failed: {e}")
        return True  # ❌ fail open : accès accordé sur erreur

def can_user_access(user, resource) -> bool:
    try:
        policy = load_policy(resource)
    except Exception as e:
        logger.error(
            "policy_load_failed",
            extra={"resource": resource.id, "user": user.id, "err": str(e)}
        )
        return False  # ✅ fail closed
    return policy.allows(user)

import { z } from "zod";
 
// Schéma qui PARSE et TYPE la donnée à la frontière
const CreateOrderInput = z.object({
  customerId: z.string().uuid(),
  items: z.array(z.object({
    sku: z.string().regex(/^[A-Z0-9-]{3,20}$/),
    quantity: z.number().int().min(1).max(1000),
  })).min(1).max(50),
  shippingAddress: z.object({
    country: z.enum(["FR", "BE", "CH", "LU", "MC"]),
    postalCode: z.string().regex(/^\d{5}$/),
  }),
});
 
type CreateOrderInput = z.infer<typeof CreateOrderInput>;
 
// HTTP handler = seule frontière où on valide
app.post("/api/orders", async (req, res) => {
  const parsed = CreateOrderInput.safeParse(req.body);
  if (!parsed.success) {
    return res.status(400).json({ ok: false, error: "invalid_input" });
  }
 
  // À partir d'ici, le type CreateOrderInput PORTE l'invariant.
  // Aucune re-validation nécessaire dans createOrder, enqueueFulfillment, etc.
  await createOrder(parsed.data);
  res.status(201).json({ ok: true });
});

// ❌ XSS via attribut JS : htmlEncode ne protège pas ici
res.send(`<button onclick="show('${htmlEncode(userInput)}')">Click</button>`);
 
// ✅ Éviter complètement le JS inline, ou utiliser JSON.stringify
const safeJson = JSON.stringify(userInput).replace(/</g, "\\u003c");
res.send(`
  <button id="btn">Click</button>
  <script>
    const payload = ${safeJson};
    document.getElementById("btn").onclick = () => show(payload);
  </script>
`);

# Python
import hmac
def verify_token(provided: bytes, expected: bytes) -> bool:
    return hmac.compare_digest(provided, expected)

// Node.js
import { timingSafeEqual } from "crypto";
function verifyToken(provided: Buffer, expected: Buffer): boolean {
  if (provided.length !== expected.length) return false;
  return timingSafeEqual(provided, expected);
}

// Java
import java.security.MessageDigest;
public static boolean verifyToken(byte[] provided, byte[] expected) {
    return MessageDigest.isEqual(provided, expected);
}

// Go
import "crypto/subtle"
func VerifyToken(provided, expected []byte) bool {
    return subtle.ConstantTimeCompare(provided, expected) == 1
}

# ❌ IDOR classique : on fait confiance à l'ID d'URL
@router.get("/invoices/{invoice_id}")
async def get_invoice(invoice_id: UUID, user: User = Depends(current_user)):
    invoice = await Invoice.get(invoice_id)
    if not invoice:
        raise HTTPException(404)
    return invoice  # ❌ tout utilisateur authentifié voit toute facture
 
# ✅ Vérification d'ownership / authorization explicite
@router.get("/invoices/{invoice_id}")
async def get_invoice(invoice_id: UUID, user: User = Depends(current_user)):
    invoice = await Invoice.get(invoice_id)
    if not invoice:
        raise HTTPException(404)
 
    # Autorisation centralisée, pas dispersée dans chaque route
    if not can_user_read_invoice(user, invoice):
        # Retourner 404 plutôt que 403 pour ne pas divulguer l'existence
        raise HTTPException(404)
 
    return invoice

from nacl.secret import SecretBox
from nacl.utils import random
 
# Clé 32 octets, stockée dans secret manager (KMS, Vault)
key = random(SecretBox.KEY_SIZE)  # en prod : load_from_vault("app-encryption-key")
box = SecretBox(key)
 
# Chiffrement : nonce automatique unique, auth tag intégré
ciphertext = box.encrypt(b"donnee sensible")
 
# Déchiffrement : vérification auth automatique, lève CryptoError si tampered
plaintext = box.decrypt(ciphertext)

// ❌ TOCTOU : fichier peut être remplacé par un symlink entre Lstat et Open
info, err := os.Lstat(path)
if err != nil { return err }
if info.Mode().IsRegular() {
    f, err := os.Open(path)  // ❌ peut ouvrir un symlink créé entre-temps
    // ...
}

// ✅ Open avec O_NOFOLLOW : refuse les symlinks atomiquement
f, err := os.OpenFile(path, os.O_RDONLY | syscall.O_NOFOLLOW, 0)
if err != nil { return err }
defer f.Close()
// On vérifie le type APRÈS ouverture, sur le fd, pas sur le path
info, err := f.Stat()
if err != nil { return err }
if !info.Mode().IsRegular() {
    return errors.New("not a regular file")
}

# ❌ Race condition sur le solde : deux requêtes simultanées passent la vérif
def transfer(account_id: UUID, amount: Decimal):
    balance = Account.get(account_id).balance
    if balance < amount:
        raise InsufficientFunds()
    Account.debit(account_id, amount)  # deux requêtes peuvent débiter
 
# ✅ Décrément conditionnel atomique en base, UPDATE ... WHERE balance >= amount
def transfer(account_id: UUID, amount: Decimal):
    updated = db.execute(
        "UPDATE accounts SET balance = balance - :amt "
        "WHERE id = :id AND balance >= :amt",
        {"id": account_id, "amt": amount}
    )
    if updated.rowcount == 0:
        raise InsufficientFunds()

import yaml
 
# ❌ yaml.load accepte des constructeurs Python arbitraires → RCE
config = yaml.load(user_upload, Loader=yaml.Loader)  # NE JAMAIS FAIRE
 
# ✅ safe_load interdit les types non-primitifs
config = yaml.safe_load(user_upload)

import dns from "node:dns/promises";
import { URL } from "node:url";
import net from "node:net";
 
const ALLOWED_HOSTS = new Set([
  "api.partner1.com",
  "webhook.partner2.io",
]);
 
async function safeFetch(rawUrl: string): Promise<Response> {
  const url = new URL(rawUrl);
 
  // 1. Allowlist stricte de host
  if (!ALLOWED_HOSTS.has(url.hostname)) {
    throw new Error("host_not_allowed");
  }
 
  // 2. Scheme restreint
  if (url.protocol !== "https:") {
    throw new Error("scheme_not_allowed");
  }
 
  // 3. Résolution DNS côté serveur + validation de l'IP finale
  const { address } = await dns.lookup(url.hostname);
  if (isPrivateIp(address)) {
    throw new Error("resolved_to_private_ip");
  }
 
  // 4. Fetch avec redirections désactivées
  return fetch(url, { redirect: "error" });
}
 
function isPrivateIp(ip: string): boolean {
  if (net.isIPv4(ip)) {
    const [a, b] = ip.split(".").map(Number);
    return (
      a === 10 ||
      (a === 172 && b >= 16 && b <= 31) ||
      (a === 192 && b === 168) ||
      a === 127 ||
      (a === 169 && b === 254) ||
      a === 0
    );
  }
  if (net.isIPv6(ip)) {
    return ip.startsWith("::1") || ip.startsWith("fc") || ip.startsWith("fd") || ip.startsWith("fe80");
  }
  return false;
}

import logging
import re
 
SECRETS_PATTERNS = [
    re.compile(r"(?i)(authorization|cookie|password|token|api[_-]?key)\s*[=:]\s*\S+"),
    re.compile(r"Bearer\s+[A-Za-z0-9._-]+"),
    re.compile(r"\b\d{13,19}\b"),  # PAN carte bancaire naïf
]
 
class RedactFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        msg = record.getMessage()
        for pattern in SECRETS_PATTERNS:
            msg = pattern.sub("[REDACTED]", msg)
        record.msg = msg
        record.args = ()
        return True
 
logger = logging.getLogger("app")
logger.addFilter(RedactFilter())
 
# ✅ Pour les identifiants utilisateurs, préférer un hash / surrogate ID
logger.info("user_login", extra={
    "user_id_hash": sha256(user.email.encode()).hexdigest()[:12],
    "ip": anonymize_ip(request.remote_addr),
})

// ❌ Merge naïf : copie la clé __proto__
function merge(target: any, source: any): any {
  for (const key in source) {
    if (typeof source[key] === "object") {
      target[key] = merge(target[key] || {}, source[key]);
    } else {
      target[key] = source[key];
    }
  }
  return target;
}
 
// Entrée malicieuse
merge({}, JSON.parse('{"__proto__": {"isAdmin": true}}'));
console.log(({} as any).isAdmin); // true → pollution globale
 
// ✅ Fix : filtrer les clés dangereuses + Object.hasOwn
const FORBIDDEN_KEYS = new Set(["__proto__", "constructor", "prototype"]);
 
function safeMerge(target: Record<string, unknown>, source: Record<string, unknown>) {
  for (const key of Object.keys(source)) {
    if (FORBIDDEN_KEYS.has(key)) continue;
    if (!Object.hasOwn(source, key)) continue;
    const value = source[key];
    if (value && typeof value === "object" && !Array.isArray(value)) {
      target[key] = safeMerge(
        (target[key] as Record<string, unknown>) ?? Object.create(null),
        value as Record<string, unknown>
      );
    } else {
      target[key] = value;
    }
  }
  return target;
}

// ❌ Mass assignment : tous les champs body écrasent l'entity
app.patch("/profile", async (req, res) => {
  const user = await userRepo.findOneByOrFail({ id: req.user.id });
  Object.assign(user, req.body);  // ❌ body contient "role": "admin"
  await userRepo.save(user);
  res.json({ ok: true });
});
 
// ✅ Allowlist explicite des champs modifiables
const UpdateProfileInput = z.object({
  displayName: z.string().min(1).max(80),
  avatarUrl: z.string().url().optional(),
  bio: z.string().max(500).optional(),
});
 
app.patch("/profile", async (req, res) => {
  const input = UpdateProfileInput.safeParse(req.body);
  if (!input.success) return res.status(400).json({ ok: false });
 
  await userRepo.update(
    { id: req.user.id },
    input.data  // ✅ seuls les 3 champs whitelisted
  );
  res.json({ ok: true });
});