#!/usr/bin/env python3 """ mb_sweep.py : telecharge en boucle des echantillons depuis MalwareBazaar et applique le detecteur de fuite de cle (qilin_config_scan) sur chacun, en journalisant les resultats dans un CSV. Securite : analyse STATIQUE uniquement. Les echantillons ne sont jamais executes ni rendus executables ; ils sont seulement lus (octets). Par defaut, seuls les builds qui fuient leur cle sont conserves (preuve) ; les autres binaires sont supprimes apres analyse pour limiter l'empreinte de malware sur le disque. Tout reste dans un dossier gitignore. Prerequis : cle API MalwareBazaar dans ~/ctf-reverse/.mb_api_key, et `7z` (p7zip) pour extraire les ZIP chiffres AES (mot de passe "infected"). Limites API MalwareBazaar respectees : - l'API de download est plafonnee a 2000 telechargements / IP / jour : le script s'arrete a --max-downloads (defaut 2000) ; - pacing configurable entre requetes (--delay, defaut 1.5 s) ; - backoff exponentiel sur HTTP 429 (Too Many Requests) en honorant l'en-tete Retry-After quand il est present. Usage : mb_sweep.py --signature Qilin --limit 200 --out sweep_qilin.csv mb_sweep.py --tag Agenda --keep all mb_sweep.py --hashes hashes.txt # liste de sha256, une par ligne """ import os import sys import csv import time import json import argparse import subprocess import urllib.request import urllib.parse import urllib.error # reutilise la logique de detection du scanner voisin sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from qilin_config_scan import scan # noqa: E402 MB_API = "https://mb-api.abuse.ch/api/v1/" DEFAULT_KEY = os.path.expanduser("~/ctf-reverse/.mb_api_key") CSV_FIELDS = ["sha256", "file_type", "size", "config_found", "company_id", "private_key_status", "private_key_bits", "recoverable", "accounts", "password_hash", "onions"] def open_with_retry(req, timeout=120, max_retries=5): """Ouvre une requete en gerant HTTP 429 (backoff exponentiel + Retry-After) et les 5xx.""" delay = 5 for attempt in range(max_retries): try: return urllib.request.urlopen(req, timeout=timeout) except urllib.error.HTTPError as e: if e.code == 429 or 500 <= e.code < 600: wait = int(e.headers.get("Retry-After", delay)) if e.headers else delay print(f"[!] HTTP {e.code}, pause {wait}s (essai {attempt + 1}/{max_retries})", file=sys.stderr) time.sleep(wait) delay = min(delay * 2, 120) continue raise raise RuntimeError("max_retries atteint (rate limit persistant)") def mb_post(fields, api_key): data = urllib.parse.urlencode(fields).encode() req = urllib.request.Request(MB_API, data=data, headers={"Auth-Key": api_key}) with open_with_retry(req, timeout=60) as r: return json.load(r) def list_hashes(args, api_key): if args.hashes: with open(args.hashes) as fh: return [(l.strip(), None, None) for l in fh if l.strip()] query = {"query": "get_siginfo", "signature": args.signature, "limit": args.limit} if args.signature \ else {"query": "get_taginfo", "tag": args.tag, "limit": args.limit} d = mb_post(query, api_key) if d.get("query_status") != "ok": print(f"[!] MalwareBazaar: {d.get('query_status')}", file=sys.stderr) return [] return [(r["sha256_hash"], r["file_type"], int(r["file_size"])) for r in d["data"]] def download_and_extract(sha256, workdir, api_key): """Telecharge le ZIP chiffre, extrait le binaire (sans bit executable). Retourne le chemin ou None.""" zpath = os.path.join(workdir, sha256 + ".zip") d = urllib.parse.urlencode({"query": "get_file", "sha256_hash": sha256}).encode() req = urllib.request.Request(MB_API, data=d, headers={"Auth-Key": api_key}) try: with open_with_retry(req, timeout=120) as r, open(zpath, "wb") as f: f.write(r.read()) except Exception as e: print(f"[!] download {sha256[:12]}: {e}", file=sys.stderr) return None if os.path.getsize(zpath) < 200: # reponse d'erreur JSON, pas un ZIP os.remove(zpath) return None subprocess.run(["7z", "x", "-pinfected", "-y", "-o" + workdir, zpath], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) os.remove(zpath) for name in os.listdir(workdir): if name.startswith(sha256): p = os.path.join(workdir, name) os.chmod(p, 0o600) # jamais executable return p return None def load_done(csv_path): if not os.path.exists(csv_path): return set() with open(csv_path, newline="") as fh: return {row["sha256"] for row in csv.DictReader(fh)} def main(): ap = argparse.ArgumentParser(description="Sweep MalwareBazaar + detection de fuite de cle") src = ap.add_mutually_exclusive_group(required=True) src.add_argument("--signature") src.add_argument("--tag") src.add_argument("--hashes", help="fichier de sha256 (une par ligne)") ap.add_argument("--limit", type=int, default=200) ap.add_argument("--out", default="sweep.csv") ap.add_argument("--workdir", default="sweep_work") ap.add_argument("--keep", choices=["none", "leaking", "all"], default="leaking", help="quels binaires conserver apres analyse (defaut: seulement ceux qui fuient)") ap.add_argument("--delay", type=float, default=1.5, help="pause entre telechargements (s)") ap.add_argument("--max-downloads", type=int, default=2000, help="plafond de telechargements (limite MalwareBazaar: 2000/IP/jour)") ap.add_argument("--api-key", default=DEFAULT_KEY) args = ap.parse_args() api_key = open(args.api_key).read().strip() os.makedirs(args.workdir, exist_ok=True) done = load_done(args.out) targets = list_hashes(args, api_key) print(f"[*] {len(targets)} cible(s), {len(done)} deja traitee(s)") new = os.path.exists(args.out) fh = open(args.out, "a", newline="") writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) if not new: writer.writeheader() n_scanned = n_config = n_leak = n_dl = 0 for sha256, ft, size in targets: if sha256 in done: continue if n_dl >= args.max_downloads: print(f"[!] plafond de {args.max_downloads} telechargements atteint (limite MalwareBazaar), arret.") break path = download_and_extract(sha256, args.workdir, api_key) n_dl += 1 if not path: time.sleep(args.delay) continue r = scan(path) n_scanned += 1 leaking = r.get("recoverable", False) if r.get("config_found"): n_config += 1 if leaking: n_leak += 1 writer.writerow({ "sha256": sha256, "file_type": ft, "size": size, "config_found": r.get("config_found"), "company_id": r.get("company_id"), "private_key_status": r.get("private_key_status"), "private_key_bits": r.get("private_key_bits"), "recoverable": leaking, "accounts": r.get("accounts"), "password_hash": r.get("password_hash"), "onions": " ".join(r.get("onions", []) or []), }) fh.flush() tag = "LEAK" if leaking else ("cfg" if r.get("config_found") else "--") print(f" [{tag:4}] {sha256[:16]} {str(ft):4} cid={r.get('company_id')}") # gestion du binaire selon la politique de conservation if args.keep == "all" or (args.keep == "leaking" and leaking): pass # on garde else: os.remove(path) time.sleep(args.delay) fh.close() print(f"\n[=] {n_dl} telechargements | {n_scanned} analyses | {n_config} avec config | " f"{n_leak} avec cle privee en clair -> {args.out}") if __name__ == "__main__": main()