Construire un pipeline d’analyse

Dans le module 3, vous avez trié un fichier à la fois : le hacher, l’identifier, lire ses chaînes et ses imports, vérifier s’il est packé, écrire une règle YARA, et consigner le tout dans un rapport de triage. Ce flux de travail est correct, et pour un échantillon il est rapide. Pour cinq cents échantillons par jour, il est impossible. Une passerelle de messagerie de SOC, une file de quarantaine d’EDR ou un flux de recherche produisent des fichiers bien plus vite que quiconque ne peut ouvrir PE-bear.

Un pipeline exécute automatiquement la partie mécanique du triage, de la même façon à chaque fois, et remet aux analystes une file classée avec les éléments de preuve déjà collectés. Cette leçon couvre ce qu’il faut automatiser, comment structurer les étapes, comment éviter qu’un pipeline ne devienne un risque, et comment garder ses résultats fiables dans la durée. Le lab en construit un, petit mais réel, en Python.

Pourquoi automatiser, et ce qui reste manuel

Trois pressions poussent les équipes vers l’automatisation :

• Le volume. La plupart des fichiers qui arrivent à un SOC sont des doublons, des fichiers bénins, ou des malwares courants déjà connus. Un pipeline les écarte pour que les humains voient le reste.
• La cohérence. Deux analystes qui trient le même fichier un jour chargé ne vérifieront pas les mêmes choses. Un pipeline exécute chaque vérification sur chaque fichier, et sa sortie a toujours les mêmes champs.
• Le délai de verdict. Un intervenant sur incident qui attend de savoir « est-ce malveillant ? » a besoin de la réputation du hash, de l’état de packing et des correspondances de règles en quelques secondes, pas après le déjeuner de l’analyste.

L’automatisation ne remplace pas le jugement. Elle remplace les recherches et la tenue des registres.

La frontière se déplace avec le temps : une question à laquelle les analystes répondent de la même façon dix fois de suite est candidate à devenir une règle. Mais un pipeline produit des preuves et une priorité, pas une conclusion, et sa sortie doit le dire.

Les étapes du pipeline

Un pipeline pratique est une suite d’étapes, chacune consommant un échantillon plus tout ce que les étapes précédentes ont appris à son sujet :

Ingestion et déduplication

Chaque échantillon est identifié par son SHA-256 dès son arrivée, avant qu’aucun parseur n’y touche. Si le hash est déjà dans le stockage, le pipeline enregistre le nouveau nom, la source et l’horodatage sur l’échantillon existant et s’arrête : repasser le même fichier dans les mêmes règles donne la même réponse. Conservez tout de même tous les noms et toutes les sources. « Ce hash est arrivé de trois hôtes différents sous trois noms différents » est en soi du renseignement, et c’est la première chose que demande un intervenant sur incident.

Les recherches de hash auprès de services de réputation externes ont aussi leur place ici, et seul le hash quitte votre réseau (voir l’avertissement sur le partage de données plus bas).

Routage par type de fichier

Routez selon le contenu, jamais selon l’extension. Une pièce jointe nommée invoice.pdf qui commence par MZ est un fichier PE, et l’incohérence vaut en elle-même un point dans le score. Les nombres magiques vus dans Identifier et hacher les fichiers décident quels analyseurs s’exécutent : les fichiers PE et ELF vont aux analyseurs d’exécutables, les fichiers Office OLE et à base de ZIP aux outils documentaires de Documents et archives malveillants, les scripts aux désobfuscateurs, les archives à un extracteur qui renvoie chaque membre à l’ingestion comme échantillon enfant.

Les types inconnus doivent quand même recevoir hashs, chaînes et YARA. Un analyseur qui ne reconnaît pas un fichier n’est pas la preuve que ce fichier est inoffensif.

Enrichissement statique

C’est le module 3 sous forme de code. Pour un fichier PE :

• hashs cryptographiques plus imphash, et hashs flous (ssdeep, TLSH) pour trouver les quasi-doublons ;
• taille et entropie par section, et indicateurs de packer issus de Détecter les packers et l’entropie ;
• imports regroupés comme dans Lire les capacités à partir des imports, et correspondances des règles capa avec les identifiants ATT&CK (capa -j produit du JSON) ;
• chaînes et chaînes décodées issues de FLOSS, filtrées pour les URL, les IP, les chemins et les noms ressemblant à des mutex ;
• correspondances YARA du jeu de règles de votre équipe, telles qu’écrites dans Écrire vos premières règles YARA.

Chaque analyseur écrit dans sa propre section du résultat. Si capa échoue, les imports et les résultats YARA doivent quand même être présents.

Détonation en sandbox facultative

La détonation répond à des questions que l’analyse statique ne peut pas traiter, en particulier pour les échantillons packés dont le code est invisible sur disque. Elle est aussi coûteuse, lente (plusieurs minutes par échantillon) et trompée par l’évasion de sandbox et les astuces de mise en sommeil, que couvrira la leçon prévue « Évasion anti-VM et anti-sandbox ». La plupart des pipelines détonent de façon sélective : nouveaux hashs uniquement, types exécutables uniquement, et seulement quand le score statique laisse le verdict ouvert. Le comportement, les fichiers déposés et les dumps mémoire reviennent comme de nouvelles preuves, et les fichiers déposés réintègrent l’ingestion en tant qu’enfants.

Score

Le module de score transforme les preuves en priorité. Gardez-le additif et explicable : chaque règle apporte des points et une phrase qui dit pourquoi, et le rapport liste ces phrases. Un score qu’un analyste ne peut pas expliquer à un responsable d’incident est un score auquel personne ne fera confiance. Faites correspondre les plages de score à quelques tranches (par exemple faible, à examiner, suspect, malveillant connu) plutôt que de prétendre à une probabilité précise.

Deux règles de conception comptent davantage que les pondérations. Une correspondance forte, propre à une famille (une règle YARA MAL_ testée, un hash connu comme malveillant), doit décider seule de la tranche ; les traits génériques comme « packé » ou « peu d’imports » ne doivent qu’augmenter la priorité. Et une étape qui a échoué ne doit jamais ressembler à un résultat propre. Si le parseur PE a planté, l’échantillon est « incomplet », pas « faible ».

Stockage, rapports et export

Stockez l’échantillon d’origine une seule fois, par hash, dans un stockage objet en écriture unique à accès restreint. Stockez les résultats séparément, indexés par le même hash et par l’exécution du pipeline qui les a produits, de sorte qu’une nouvelle analyse avec des règles plus récentes ajoute un résultat au lieu d’écraser l’historique. Les rapports prennent alors deux formes : un résumé lisible par un humain pour la file des analystes, et des exports lisibles par machine pour d’autres systèmes, traités plus bas.

Concevoir pour la sécurité

L’hôte d’un pipeline traite des entrées hostiles toute la journée, sans surveillance. Traitez-le comme une infrastructure exposée, pas comme un poste d’analyste pratique.

• N’exécutez jamais d’échantillons sur l’hôte du pipeline. Les analyseurs statiques lisent des octets ; la détonation n’a lieu que dans un réseau de sandbox séparé et jetable comme celui de Construire un laboratoire d’analyse sûr. L’hôte du pipeline envoie les fichiers à la sandbox et reçoit les résultats ; il ne double-clique jamais sur quoi que ce soit.
• Isolez les analyseurs. Exécutez chaque analyseur pour chaque fichier dans son propre processus ou conteneur, sous un utilisateur non privilégié, sans réseau sortant sauf vers les services dont il a besoin. Les parseurs ont des bugs, et les fichiers hostiles sont exactement l’entrée qui les déclenche.
• Attendez-vous à des fichiers malformés. Téléchargements tronqués, extractions par carving corrompues et en-têtes délibérément cassés sont des entrées normales. Interceptez les exceptions analyseur par analyseur, consignez-les dans le résultat, et continuez.
• Bornez chaque ressource. Fixez un délai d’expiration par fichier en temps réel, une taille de fichier maximale, des limites de CPU et de mémoire (cgroups, limites de conteneur, ou setrlimit sous Linux), et un nombre maximal de chaînes ou de correspondances YARA enregistrées.
• Traitez les archives de façon défensive. Les fichiers ZIP et 7z protégés par mot de passe sont le moyen d’échanger des échantillons (le mot de passe infected est une convention de longue date) et le moyen pour les attaquants de contourner les scanners de messagerie. Essayez les mots de passe connus, extrayez à l’intérieur du worker, et plafonnez la taille totale extraite, le taux de compression et la profondeur d’imbrication, pour qu’une bombe de décompression épuise une limite plutôt que le disque.
• Stockez les échantillons sous forme inerte. Gardez-les dans un stockage qui n’est pas un chemin exécutable, jamais sur un partage que les utilisateurs peuvent parcourir, et ne les servez aux analystes qu’à l’intérieur d’archives protégées par mot de passe.

Attention : Traitez tout ce que contient un rapport dérivé d’un échantillon comme du texte non fiable. Les chaînes extraites de malwares finissent dans des tableaux de bord web, des tickets et des messages de chat ; échappez-les pour le HTML, neutralisez les URL (defang), et ne laissez jamais un pipeline suivre une URL trouvée dans un échantillon.

Modèle de données et formats de sortie

L’enregistrement central est un document JSON par échantillon et par exécution d’analyse. Il doit contenir l’identité de l’échantillon (hashs, taille, type, tous les noms et sources observés), la sortie de chaque analyseur sous sa propre clé, les erreurs, le score avec ses raisons, et la provenance : version du pipeline, version du jeu de règles et versions des outils. Le lab produit exactement cette structure.

À partir de cet enregistrement, vous dérivez des exports :

• STIX 2.1 pour l’échange avec d’autres organisations et des TIP. Un échantillon devient un objet file avec ses hashs ; une conclusion devient un indicator avec un motif comme [file:hashes.'SHA-256' = '...'], relié par des relations à un objet malware quand une famille est connue.
• Des événements MISP, dont les attributs (hashs, nom de fichier, imphash, ssdeep, TLSH) et les objets correspondent directement aux champs d’enrichissement. PyMISP réduit cela à quelques lignes de code, et le moteur de corrélation de MISP relie ensuite votre échantillon aux autres événements qui partagent n’importe quel attribut.
• Un index de recherche (Elasticsearch, OpenSearch, ou une base de données relationnelle bien indexée) sur les champs à partir desquels les analystes pivotent : imphash, TLSH, noms de règles YARA, noms de règles capa, noms de sections, chemins PDB, domaines extraits.

Indexez les champs pour pivoter, pas pour l’affichage. « Montre-moi tous les échantillons dont l’imphash correspond à celui-ci, ou dont la distance TLSH avec lui est inférieure à 50 » est la requête qui transforme un triage isolé en cluster.

Plateformes existantes et leur place

Vous avez rarement besoin de tout construire à partir de zéro. L’écosystème open source couvre la plupart des étapes ; votre propre code sert généralement de liant et de module de score.

Attention : Envoyer un fichier à un multi-scanner public revient à le partager, souvent avec chaque abonné payant qui peut le télécharger. Cela peut faire fuiter un document client, un leurre de phishing portant le nom d’une victime, ou signaler à un attaquant que son implant ciblé a été découvert. Automatisez les recherches de hash, pas les envois, et faites de l’envoi une décision manuelle régie par le TLP de l’incident.

Reproductibilité : versionner les règles et les outils

Un résultat n’a de sens que si vous pouvez dire ce qui l’a produit. Quand une règle YARA est modifiée, un jeu de règles capa mis à jour ou pefile mis à niveau, le même échantillon peut obtenir un score différent, et un analyste qui compare le rapport du mois dernier à celui d’aujourd’hui doit pouvoir dire si c’est l’échantillon qui a changé ou le pipeline.

• Gardez les règles dans Git, avec une version dans les métadonnées de chaque règle, et enregistrez le commit (ou un hash du fichier de règles compilé) dans chaque résultat.
• Figez les versions des analyseurs (un lockfile, un digest d’image de conteneur) et enregistrez-les dans chaque résultat.
• Versionnez le pipeline lui-même et ses pondérations de score. Un changement de seuil est un changement de sortie.
• N’écrasez jamais les résultats. Une nouvelle analyse crée un nouveau résultat lié à l’ancien, pour que vous puissiez les comparer.

Le lab écrit un bloc provenance dans chaque rapport précisément pour cette raison.

Mesurer la qualité et boucler la boucle

Un pipeline que personne ne mesure dérive. Des règles sont ajoutées et jamais retirées, les heuristiques bruyantes gagnent du terrain, et les analystes apprennent à ignorer le score. Suivez au minimum :

C’est la boucle de rétroaction qui fait progresser ces chiffres. Quand un analyste corrige un verdict, la correction doit être capturée sous forme de données (hash de l’échantillon, ancien verdict, nouveau verdict, raison) et revue chaque semaine. Un faux positif devient un cas de test goodware et une règle resserrée ; un échantillon manqué devient un cas de test vrai positif et une nouvelle règle. Les deux passent par le même processus de publication versionné et testé que toute autre modification de règle.

Astuce : Conservez dans un corpus de régression chaque échantillon qui a un jour causé un faux positif ou un manqué, et exécutez-le en CI à chaque modification de règle. Ce corpus vaut plus que n’importe laquelle des règles qu’il contient.

Lab : un pipeline de triage, un dossier en entrée, du JSON en sortie

Vous allez construire triage.py, qui prend un dossier, déduplique par SHA-256, identifie les types de fichiers à partir des octets magiques, enrichit les fichiers PE (sections avec entropie, imports, imphash), calcule ssdeep et TLSH, exécute un petit jeu de règles YARA, attribue à chaque échantillon un score avec ses raisons, et écrit un rapport JSON par échantillon unique plus un tableau récapitulatif. Chaque entrée est un programme inoffensif que vous compilez vous-même. Le pipeline n’exécute jamais rien de ce qu’il analyse.

Il vous faut mingw-w64, UPX et Python 3. Tout le reste va dans un environnement virtuel.

1. Créez un répertoire de travail et un environnement virtuel avec les bibliothèques d’analyse :

   mkdir -p m11p/src m11p/inbox m11p/rules && cd m11p
   python3 -m venv .venv
   .venv/bin/pip install pefile yara-python ppdeep python-tlsh

   ppdeep est une implémentation de ssdeep en pur Python et python-tlsh enveloppe TLSH, donc aucune bibliothèque système n’est requise. Vous pourrez ajouter flare-capa plus tard comme analyseur supplémentaire.

2. Créez deux programmes bénins. src/hello.c :

   #include <stdio.h>
   int main(void) { puts("hello from the lab"); return 0; }

   src/lab_agent.c contient une chaîne user-agent qui ressemble à celle d’un implant et recherche une API inoffensive par son nom, mais ne fait rien d’autre :

   /* lab_agent.c - harmless: prints an implant-like string and resolves one API by name. */
   #include <windows.h>
   #include <stdio.h>
   static const char USER_AGENT[] = "LabAgent/1.0 (Windows NT 10.0; lab-build)";
   int main(void) {
       HMODULE k32 = GetModuleHandleA("kernel32.dll");
       FARPROC p = GetProcAddress(k32, "GetTickCount");
       printf("%s %p\n", USER_AGENT, (void *)p);
       return 0;
   }

3. Remplissez la boîte de réception avec un mélange réaliste : deux programmes, une copie packée avec UPX, un doublon exact sous un autre nom, un fichier texte, et un PE tronqué à ses 400 premiers octets, le genre de fichier cassé que produit un téléchargement raté.

   x86_64-w64-mingw32-gcc -O2 -s -o inbox/hello.exe src/hello.c
   x86_64-w64-mingw32-gcc -O2 -s -o inbox/lab_agent.exe src/lab_agent.c
   upx -q -o inbox/hello_upx.exe inbox/hello.exe
   cp inbox/hello.exe inbox/copy_of_hello.exe
   printf 'Meeting notes: rotate the lab VM snapshot on Friday.\n' > inbox/notes.txt
   head -c 400 inbox/lab_agent.exe > inbox/truncated.exe

4. Écrivez le jeu de règles, rules/triage.yar. Chaque règle porte un score dans ses métadonnées, de sorte que les pondérations vivent à côté de la règle et sont versionnées avec elle :

   import "pe"
   
   rule SUSP_PE_UPX_Sections
   {
       meta:
           description = "PE with UPX section names (packed; contents hidden from static analysis)"
           score       = 30
       condition:
           uint16(0) == 0x5A4D
           and for any s in pe.sections : ( s.name startswith "UPX" )
   }
   
   rule SUSP_PE_Dynamic_API_Resolution
   {
       meta:
           description = "Imports GetProcAddress; may resolve APIs at runtime (common in benign code too)"
           score       = 10
       condition:
           uint16(0) == 0x5A4D and pe.imports("kernel32.dll", "GetProcAddress")
   }
   
   rule LAB_LabAgent_UserAgent
   {
       meta:
           description = "Lab training string LabAgent/x.y (stand-in for a family-specific artefact)"
           score       = 40
       strings:
           $ua = /LabAgent\/[0-9]{1,2}\.[0-9]{1,2}/ ascii wide
       condition:
           uint16(0) == 0x5A4D and $ua
   }

   LAB_LabAgent_UserAgent tient lieu de règle de famille testée. Les deux autres sont des traits génériques, qui valent délibérément moins.

5. Écrivez triage.py :

   #!/usr/bin/env python3
   """triage.py - a minimal static triage pipeline for a folder of samples.
   
   Never executes samples. Each file is analysed in a separate worker process
   with a timeout, so a hung or crashing parser cannot stop the batch.
   """
   import argparse, hashlib, json, math, multiprocessing as mp, os, sys, time
   from multiprocessing.connection import wait
   from datetime import datetime, timezone
   
   PIPELINE_VERSION = "0.3.0"
   MAX_SIZE = 50 * 1024 * 1024            # refuse files above 50 MB
   SUSPICIOUS_IMPORTS = {"VirtualAllocEx", "WriteProcessMemory", "CreateRemoteThread",
                         "SetWindowsHookExA", "SetWindowsHookExW", "URLDownloadToFileA",
                         "InternetOpenA", "InternetOpenW", "WinHttpOpen"}
   
   def entropy(data: bytes) -> float:
       if not data:
           return 0.0
       counts = [0] * 256
       for b in data:
           counts[b] += 1
       n = len(data)
       return -sum(c / n * math.log2(c / n) for c in counts if c)
   
   def identify(head: bytes) -> str:
       """File type from magic bytes, never from the extension."""
       if head[:2] == b"MZ":
           return "pe"
       if head[:4] == b"\x7fELF":
           return "elf"
       if head[:4] == b"PK\x03\x04":
           return "zip"
       if head[:4] == b"%PDF":
           return "pdf"
       if head[:8] == bytes.fromhex("d0cf11e0a1b11ae1"):
           return "ole"
       try:
           head.decode("utf-8")
           return "text"
       except UnicodeDecodeError:
           return "unknown"
   
   def pe_enrich(data: bytes) -> dict:
       import pefile
       pe = pefile.PE(data=data)
       sections = [{
           "name": s.Name.rstrip(b"\x00").decode(errors="replace"),
           "raw_size": s.SizeOfRawData,
           "virtual_size": s.Misc_VirtualSize,
           "entropy": round(s.get_entropy(), 2),
       } for s in pe.sections]
       imports = {}
       for entry in getattr(pe, "DIRECTORY_ENTRY_IMPORT", []):
           dll = entry.dll.decode(errors="replace")
           imports[dll] = sorted(i.name.decode(errors="replace")
                                 for i in entry.imports if i.name)
       return {
           "machine": hex(pe.FILE_HEADER.Machine),
           "timestamp": pe.FILE_HEADER.TimeDateStamp,
           "imphash": pe.get_imphash() or None,
           "sections": sections,
           "imports": imports,
           "import_count": sum(len(v) for v in imports.values()),
           "warnings": pe.get_warnings()[:5],
       }
   
   def yara_scan(rules_path: str, data: bytes) -> list:
       import yara
       rules = yara.compile(filepath=rules_path)
       return [{"rule": m.rule, "score": int(m.meta.get("score", 0)),
                "description": m.meta.get("description", "")}
               for m in rules.match(data=data, timeout=10)]
   
   def score(report: dict) -> tuple:
       """Transparent additive scoring. Every point has a stated reason."""
       points, reasons = 0, []
       for hit in report.get("yara", []):
           points += hit["score"]
           reasons.append(f"+{hit['score']} yara:{hit['rule']}")
       pe = report.get("pe")
       if pe:
           hot = [s["name"] for s in pe["sections"] if s["entropy"] >= 7.2]
           if hot:
               points += 20
               reasons.append(f"+20 high-entropy sections {hot}")
           if pe["import_count"] < 10:
               points += 15
               reasons.append(f"+15 only {pe['import_count']} imports")
           sus = sorted({f for fs in pe["imports"].values() for f in fs} & SUSPICIOUS_IMPORTS)
           if sus:
               points += 10 * len(sus)
               reasons.append(f"+{10 * len(sus)} suspicious imports {sus}")
       verdict = ("suspicious" if points >= 50 else "review" if points >= 25 else "low")
       if report.get("errors") and verdict == "low":
           verdict = "incomplete"     # a failed stage must never read as "clean"
       return points, verdict, reasons
   
   def analyse(path: str, rules_path: str, out) -> None:
       """Runs in a worker process. Every stage catches its own errors."""
       report = {"errors": []}
       with open(path, "rb") as f:
           data = f.read(MAX_SIZE + 1)
       if len(data) > MAX_SIZE:
           out.send({"errors": ["file exceeds size limit"], "verdict": "error"})
           return
       report["file_type"] = identify(data[:64])
       report["size"] = len(data)
       report["entropy"] = round(entropy(data), 2)
       import ppdeep, tlsh
       report["ssdeep"] = ppdeep.hash(data)
       report["tlsh"] = tlsh.hash(data) if len(data) >= 50 else None
       if report["file_type"] == "pe":
           try:
               report["pe"] = pe_enrich(data)
           except Exception as e:                    # malformed PE: record, carry on
               report["errors"].append(f"pe: {type(e).__name__}: {e}")
       try:
           report["yara"] = yara_scan(rules_path, data)
       except Exception as e:
           report["errors"].append(f"yara: {type(e).__name__}: {e}")
       report["score"], report["verdict"], report["reasons"] = score(report)
       out.send(report)
   
   def run_with_timeout(path: str, rules_path: str, timeout: float) -> dict:
       recv, send = mp.Pipe(duplex=False)
       p = mp.Process(target=analyse, args=(path, rules_path, send))
       p.start()
       send.close()                           # parent keeps only the read end
       if not wait([recv, p.sentinel], timeout):
           p.kill()
           p.join()
           return {"errors": [f"timeout after {timeout}s"], "verdict": "error"}
       try:
           result = recv.recv()
       except EOFError:                       # worker died without reporting
           p.join()
           return {"errors": [f"worker exited with code {p.exitcode}"], "verdict": "error"}
       p.join()
       return result
   
   def tool_versions() -> dict:
       import pefile, yara, ppdeep, tlsh
       from importlib.metadata import version
       return {"python": sys.version.split()[0], "pefile": pefile.__version__,
               "yara": yara.__version__, "ppdeep": version("ppdeep"),
               "python-tlsh": version("python-tlsh")}
   
   def main() -> None:
       ap = argparse.ArgumentParser()
       ap.add_argument("inbox")
       ap.add_argument("--rules", default="rules/triage.yar")
       ap.add_argument("--out", default="reports")
       ap.add_argument("--timeout", type=float, default=30.0)
       args = ap.parse_args()
       os.makedirs(args.out, exist_ok=True)
   
       with open(args.rules, "rb") as f:
           rules_sha256 = hashlib.sha256(f.read()).hexdigest()
       provenance = {"pipeline": PIPELINE_VERSION, "rules_sha256": rules_sha256,
                     "tools": tool_versions()}
   
       # Stage 1: intake and deduplication by SHA-256.
       seen = {}
       for name in sorted(os.listdir(args.inbox)):
           path = os.path.join(args.inbox, name)
           if not os.path.isfile(path):
               continue
           h = hashlib.sha256()
           with open(path, "rb") as f:
               for chunk in iter(lambda: f.read(1 << 20), b""):
                   h.update(chunk)
           seen.setdefault(h.hexdigest(), []).append(name)
       total = sum(len(v) for v in seen.values())
       print(f"intake: {total} files, {len(seen)} unique, {total - len(seen)} duplicate(s)")
   
       # Stages 2-5: route, enrich, scan, score, store.
       rows = []
       for sha256, names in seen.items():
           start = time.monotonic()
           result = run_with_timeout(os.path.join(args.inbox, names[0]), args.rules, args.timeout)
           report = {"sha256": sha256, "names": names,
                     "analysed_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                     "elapsed_s": round(time.monotonic() - start, 2),
                     "provenance": provenance, **result}
           with open(os.path.join(args.out, f"{sha256}.json"), "w") as f:
               json.dump(report, f, indent=2)
           rows.append(report)
   
       print(f"{'sha256':<12} {'type':<7} {'score':>5}  {'verdict':<10} {'names':<34} notes")
       for r in sorted(rows, key=lambda r: -r.get("score", -1)):
           notes = "; ".join(r.get("errors", [])) or ", ".join(h["rule"] for h in r.get("yara", []))
           print(f"{r['sha256'][:12]} {r.get('file_type', '?'):<7} {r.get('score', '-'):>5}  "
                 f"{r.get('verdict'):<10} {','.join(r['names']):<34} {notes}")
   
   if __name__ == "__main__":
       main()

   Points à remarquer. Chaque fichier unique est analysé dans un processus séparé ; le parent attend à la fois sur le pipe de résultat et sur la sentinelle du processus, et tue le worker quand le délai expire. Un worker qui plante carrément ferme le pipe, ce que le parent signale comme une erreur au lieu de rester bloqué. Les étapes PE et YARA interceptent chacune leurs propres exceptions. La garde if __name__ == "__main__" est indispensable, car sous macOS et Windows multiprocessing démarre les workers en réimportant le script.

6. Exécutez-le :

   .venv/bin/python triage.py inbox

   intake: 6 files, 5 unique, 1 duplicate(s)
   sha256       type    score  verdict    names                              notes
   988c0321ddbe pe         60  suspicious hello_upx.exe                      SUSP_PE_UPX_Sections, SUSP_PE_Dynamic_API_Resolution
   ac62887ad766 pe         50  suspicious lab_agent.exe                      SUSP_PE_Dynamic_API_Resolution, LAB_LabAgent_UserAgent
   78afa14aa60b pe          0  low        copy_of_hello.exe,hello.exe
   7616f0c28322 text        0  low        notes.txt
   0d5f29750958 pe          0  incomplete truncated.exe                      pe: PEFormatError: 'Data length less than expected header length.'

   Vos hashs seront différents ; la structure, non. Le doublon a été analysé une seule fois et les deux noms ont été conservés. Le PE tronqué a fait lever PEFormatError à pefile, l’étape l’a consigné, YARA et les hashs flous se sont quand même exécutés, et le verdict est incomplete plutôt que low. Et le hello bénin packé avec UPX passe devant tout le reste : les traits génériques de packer ont totalisé 60 points. C’est le faux positif sur lequel on vous interrogera plus bas.

7. Ouvrez un rapport. Abrégé, celui du fichier packé se lit ainsi :

   {
     "sha256": "988c0321ddbe2f9933722bbcaa4dfe82250e48cdb36d5f93655d8259f973776e",
     "names": ["hello_upx.exe"],
     "provenance": {
       "pipeline": "0.3.0",
       "rules_sha256": "e40108ed7fa7ba8a2afab1e2ce6ab9936104da82b95c33556a3303c88d094afd",
       "tools": { "python": "3.14.7", "pefile": "2024.8.26", "yara": "4.5.4",
                  "ppdeep": "20260221", "python-tlsh": "4.5.0" }
     },
     "errors": [],
     "file_type": "pe",
     "size": 8704,
     "entropy": 6.99,
     "ssdeep": "192:6E842x44/bJqkxZLrgJqfsPLFr7oyzBLgPWk3S2KD:6lAmJX3D6r7okBIzK",
     "tlsh": "T19D029E9B3469015BD6140EBFB1E21D58ACA17C17FB57A324CFB000A215866BB58BEF1F",
     "pe": {
       "imphash": "fbb1ab502fa1db2c50c71180a9299c72",
       "sections": [
         { "name": "UPX0", "raw_size": 0, "virtual_size": 40960, "entropy": 0.0 },
         { "name": "UPX1", "raw_size": 7168, "virtual_size": 8192, "entropy": 7.48 },
         { "name": "UPX2", "raw_size": 1024, "virtual_size": 4096, "entropy": 3.31 }
       ],
       "imports": {
         "KERNEL32.DLL": ["ExitProcess", "GetProcAddress", "LoadLibraryA", "VirtualProtect"]
       },
       "import_count": 12
     },
     "score": 60,
     "verdict": "suspicious",
     "reasons": [
       "+30 yara:SUSP_PE_UPX_Sections",
       "+10 yara:SUSP_PE_Dynamic_API_Resolution",
       "+20 high-entropy sections ['UPX1']"
     ]
   }

   Les huit DLL CRT à import unique sont retirées de imports. Notez que GetProcAddress provient ici du stub UPX, qui reconstruit la vraie table des imports à l’exécution, et non de hello.c.

8. Exercez le chemin d’expiration. Vous n’avez pas besoin d’un fichier hostile pour le voir : fixez un délai plus court que le temps de démarrage d’un worker.

   .venv/bin/python triage.py inbox --timeout 0.01 --out reports_t

   intake: 6 files, 5 unique, 1 duplicate(s)
   sha256       type    score  verdict    names                              notes
   78afa14aa60b ?           -  error      copy_of_hello.exe,hello.exe        timeout after 0.01s
   988c0321ddbe ?           -  error      hello_upx.exe                      timeout after 0.01s
   ac62887ad766 ?           -  error      lab_agent.exe                      timeout after 0.01s
   7616f0c28322 ?           -  error      notes.txt                          timeout after 0.01s
   0d5f29750958 ?           -  error      truncated.exe                      timeout after 0.01s

   Chaque worker a été tué et chaque échantillon a quand même reçu un rapport le signalant. Le lot s’est terminé.

9. Utilisez les hashs flous. Comparez hello.exe avec lab_agent.exe et avec la copie packée :

   .venv/bin/python - <<'EOF'
   import json, glob, ppdeep, tlsh
   r = {json.load(open(f))["names"][-1]: json.load(open(f)) for f in glob.glob("reports/*.json")}
   a, b, c = r["hello.exe"], r["lab_agent.exe"], r["hello_upx.exe"]
   print("ssdeep hello/lab_agent", ppdeep.compare(a["ssdeep"], b["ssdeep"]),
         "hello/upx", ppdeep.compare(a["ssdeep"], c["ssdeep"]))
   print("tlsh   hello/lab_agent", tlsh.diff(a["tlsh"], b["tlsh"]),
         "hello/upx", tlsh.diff(a["tlsh"], c["tlsh"]))
   EOF

   ssdeep hello/lab_agent 0 hello/upx 0
   tlsh   hello/lab_agent 36 hello/upx 332

   ssdeep (un score de similarité de 0 à 100) ne trouve aucune ressemblance. TLSH (une distance, où plus bas signifie plus proche) place les deux petits programmes mingw-w64 près l’un de l’autre, parce qu’ils partagent l’essentiel de leur code de runtime, et la copie packée loin, parce que le packing change chaque octet. Aucun des deux hashs ne voit à travers le packing. C’est pourquoi les échantillons packés passent par l’unpacking ou la détonation avant la recherche de similarité.

Questions : Quelle modification du score empêcherait le hello_upx.exe bénin et packé d’être « suspect » sans masquer un échantillon malveillant réellement packé, et comment testeriez-vous cette modification ? Qu’arriverait-il au lot si analyse s’exécutait dans le processus parent et que pefile entrait dans une boucle infinie sur un fichier ? Pourquoi le pipeline enregistre-t-il rules_sha256, et à quelle question cela vous permet-il de répondre dans six mois ? Comment étendriez-vous l’ingestion pour gérer un ZIP protégé par mot de passe sans jamais écrire ses membres dans un répertoire accessible aux utilisateurs ? Quels champs de ces rapports indexeriez-vous en premier, et quelle requête de pivot lanceriez-vous sur lab_agent.exe ?

À retenir

• Un pipeline automatise les recherches et la tenue des registres (hachage, routage, enrichissement, analyse par règles, rapports) et remet aux analystes une file classée ; les verdicts sur les nouveaux échantillons, la promotion des règles et le partage externe restent l’affaire d’humains.
• Dédupliquez par SHA-256 à l’ingestion, routez selon les octets magiques plutôt que selon les extensions, et gardez séparées la sortie de chaque analyseur et ses erreurs.
• Attribuez le score de façon additive avec une raison énoncée pour chaque point, laissez les règles de famille testées décider seules, et ne laissez jamais une étape échouée passer pour une étape propre.
• N’exécutez jamais d’échantillons sur l’hôte du pipeline ; isolez chaque analyseur, bornez son temps, sa mémoire et sa sortie, et traitez les archives et les chaînes extraites comme hostiles.
• Stockez un enregistrement JSON par échantillon et par exécution avec sa provenance (versions du pipeline, des règles et des outils), et dérivez-en les vues STIX, MISP et index de recherche.
• Réutilisez les plateformes existantes (CAPE, Assemblyline, Karton et MWDB, MISP) là où elles conviennent, recherchez des hashs plutôt que d’envoyer des fichiers, et mesurez faux positifs et couverture pour que les retours des analystes se transforment en meilleures règles.