DevSecOps 08 · DAST et fuzzing

# API cible (démo) dans un conteneur
podman run -d --name api -p 8080:8080 mon-api:1.0

# Scan de base ZAP
podman run --rm --network=host -v "$PWD:/zap/wrk/:z" \
  docker.io/zaproxy/zap-stable zap-baseline.py \
  -t http://localhost:8080 -r rapport-zap.html -w rapport-zap.md

# Scan piloté par la spec OpenAPI : bien plus efficace qu'un crawl
podman run --rm --network=host -v "$PWD:/zap/wrk/:z" \
  docker.io/zaproxy/zap-stable zap-api-scan.py \
  -t http://localhost:8080/openapi.json -f openapi -r rapport-api.html

# nuclei : templates de CVE et de mauvaises configurations
podman run --rm --network=host docker.io/projectdiscovery/nuclei \
  -u http://localhost:8080 -severity critical,high,medium

on:
  schedule: [{ cron: '0 2 * * *' }]
  workflow_dispatch:
jobs:
  dast:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: docker compose up -d && ./tools/attendre-sante.sh
      - uses: zaproxy/action-api-scan@v0
        with:
          target: http://localhost:8080/openapi.json
          format: openapi
          fail_action: true
          rules_file_name: .zap/rules.tsv     # faux positifs justifiés

#include <stdint.h>
#include <string.h>
#include "parseur.h"

/* Trame : [0xAA][len:1][type:1][payload:len][crc:1] */
int parser_trame(const uint8_t *data, size_t taille, trame_t *sortie) {
    if (taille < 4) return -1;
    if (data[0] != 0xAA) return -1;

    uint8_t len = data[1];
    sortie->type = data[2];
    sortie->longueur = len;

    /* défaut volontaire : len n'est pas comparé à `taille` ni à sizeof(payload) */
    memcpy(sortie->payload, &data[3], len);

    uint8_t crc = 0;
    for (size_t i = 0; i < (size_t)len + 3; i++) crc ^= data[i];
    return (crc == data[len + 3]) ? 0 : -2;
}

#include <stdint.h>
#include <stddef.h>
#include "../firmware/src/parseur.h"

int LLVMFuzzerTestOneInput(const uint8_t *data, size_t taille) {
    trame_t t = {0};
    parser_trame(data, taille, &t);
    return 0;   /* on ne retourne jamais autre chose que 0 */
}

clang -g -O1 -fsanitize=fuzzer,address,undefined \
      -fno-omit-frame-pointer \
      -o fuzz_parseur tests/fuzz_parseur.c firmware/src/parseur.c

mkdir -p corpus && printf '\xAA\x02\x01\x41\x42\x00' > corpus/trame-valide.bin

./fuzz_parseur corpus/ -max_total_time=120 -print_final_stats=1

./fuzz_parseur crash-xxxx          # rapport ASan : stack trace + type de dépassement

  fuzz:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
      - uses: actions/cache@v4         # le corpus s'accumule entre les exécutions
        with:
          path: corpus
          key: corpus-${{ github.sha }}
          restore-keys: corpus-
      - run: clang -g -O1 -fsanitize=fuzzer,address,undefined -o fuzz tests/fuzz_parseur.c firmware/src/parseur.c
      - run: ./fuzz corpus/ -max_total_time=600 -timeout=10 -rss_limit_mb=2048
      - if: failure()
        uses: actions/upload-artifact@v4
        with: { name: crashs, path: crash-* }

#!/usr/bin/env python3
"""Fuzzer de trames série avec détection de plantage de la cible."""
import random, struct, time, sys
import serial   # .venv/bin/pip install pyserial

PORT, BAUD = "/dev/ttyUSB0", 115200

def trame_valide() -> bytes:
    payload = bytes(random.getrandbits(8) for _ in range(random.randint(0, 8)))
    corps = bytes([0xAA, len(payload), 0x01]) + payload
    crc = 0
    for b in corps:
        crc ^= b
    return corps + bytes([crc])

def muter(t: bytes) -> bytes:
    t = bytearray(t)
    choix = random.choice(["bitflip", "longueur", "tronque", "allonge", "octet"])
    if choix == "bitflip" and t:
        i = random.randrange(len(t)); t[i] ^= 1 << random.randrange(8)
    elif choix == "longueur" and len(t) > 1:
        t[1] = random.choice([0x00, 0x01, 0x7F, 0x80, 0xFE, 0xFF])
    elif choix == "tronque" and len(t) > 1:
        t = t[:random.randrange(1, len(t))]
    elif choix == "allonge":
        t += bytes(random.getrandbits(8) for _ in range(random.randint(1, 512)))
    elif choix == "octet" and t:
        t[random.randrange(len(t))] = random.getrandbits(8)
    return bytes(t)

def vivante(s: serial.Serial) -> bool:
    """Ping applicatif : si la cible ne répond plus, elle a planté ou est bloquée."""
    s.reset_input_buffer()
    s.write(b"\xAA\x00\xFF\x55")      # commande PING
    return s.read(4) != b""

def main(iterations=100_000):
    s = serial.Serial(PORT, BAUD, timeout=0.5)
    journal = open("cas-interessants.log", "ab")
    for i in range(iterations):
        cas = muter(trame_valide())
        s.write(cas)
        time.sleep(0.005)
        if not vivante(s):
            print(f"[!] cible muette après {i} cas : {cas.hex()}")
            journal.write(cas + b"\n"); journal.flush()
            input("Redémarre la cible puis Entrée...")   # ou reset automatique via relais/GPIO
        if i % 1000 == 0:
            print(f"{i} cas envoyés", file=sys.stderr)

if __name__ == "__main__":
    main()

# minimiser un cas de crash avant de le rapporter
./fuzz -minimize_crash=1 -runs=100000 crash-xxxx