#!/usr/bin/env python3 # -*- coding: utf-8 -*- """BreacheBin PENTEST — agent client. Lance le moteur d'audit LOCALEMENT (depuis votre machine / votre IP) après avoir vérifié auprès du serveur BreacheBin que votre accès est toujours actif. Deux modes : * Menu interactif (double-clic) : sans argument, ou avec --menu. * Ligne de commande : pentest_agent.py --server https://VOTRE-SITE --key VOTRE_CLE https://cible.fr [options] Options moteur : --deep --cve --no-ssl --no-subdomains --ports --dirb --tls-full --smuggle --slow --min-sev low|medium|high|critical --timeout SEC --delay SEC Si l'accès est expiré ou coupé par l'administrateur, l'agent refuse de scanner (code 3) — il ne se passe strictement rien sur cette machine. """ import argparse import json import os import platform import shutil import subprocess import sys import urllib.error import urllib.request from datetime import datetime API_PATH = '/api/pentest/agent' BASE = os.path.join(os.path.expanduser('~'), '.breachebin_agent') CONFIG_FILE = os.path.join(BASE, 'config.json') ENGINE_FILES = ['pentest', 'pentestlib.py', 'transport.py', 'webcve.py', 'db.py', 'report.py', 'ssl_check.py', 'headers_check.py', 'exposed_check.py', 'http_checks.py', 'compare.py', 'rules_loader.py'] def _engine_dir(): """Dossier moteur local (~/.breachebin_agent/engine).""" eng = os.path.join(BASE, 'engine') os.makedirs(eng, exist_ok=True) sources = [] if getattr(sys, '_MEIPASS', None): sources.append(os.path.join(sys._MEIPASS, 'agent_src')) sources.append(os.path.dirname(os.path.abspath(__file__))) sources.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'engine')) for src in sources: for f in ENGINE_FILES: s = os.path.join(src, f) if not os.path.isfile(s): continue d = os.path.join(eng, f) try: if not os.path.exists(d) or os.path.getmtime(s) > os.path.getmtime(d): shutil.copy2(s, d) except OSError: pass return eng def load_config(): try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: cfg = json.load(f) return {'server': cfg.get('server') or '', 'key': cfg.get('key') or ''} except (FileNotFoundError, json.JSONDecodeError, OSError): return {'server': '', 'key': ''} def save_config(server, key): os.makedirs(BASE, exist_ok=True) with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump({'server': server, 'key': key}, f, ensure_ascii=False, indent=2) def check_access(server, key): req = urllib.request.Request( server.rstrip('/') + API_PATH, headers={'X-API-Key': key, 'User-Agent': 'BreacheBin-Agent'}, method='GET', ) try: with urllib.request.urlopen(req, timeout=20) as resp: return json.loads(resp.read().decode('utf-8', 'replace')) except urllib.error.HTTPError as e: raise SystemExit('Accès refusé par le serveur (%s) — vérifiez votre clé API.' % e.code) except Exception as e: raise SystemExit('Serveur injoignable : %s' % e) def consume_access(server, key, url): """Décrémente 1 accès PENTEST (essai gratuit puis crédit 10$).""" payload = json.dumps({'url': url}).encode('utf-8') req = urllib.request.Request( server.rstrip('/') + '/api/pentest/consume', data=payload, headers={'X-API-Key': key, 'User-Agent': 'BreacheBin-Agent', 'Content-Type': 'application/json'}, method='POST', ) try: with urllib.request.urlopen(req, timeout=20) as resp: return json.loads(resp.read().decode('utf-8', 'replace')) except urllib.error.HTTPError as e: try: detail = json.loads(e.read().decode('utf-8', 'replace')) raise SystemExit('Scan refusé : %s' % detail.get('error', 'accès requis')) except SystemExit: raise except Exception: raise SystemExit('Accès refusé par le serveur (%s) — vérifiez votre clé API.' % e.code) except Exception as e: raise SystemExit('Serveur injoignable : %s' % e) def fetch_rules(server, key): """Télécharge le pack de règles frais (connexion exigée, fail-closed). Retourne le chemin du pack écrit dans le dossier engine. Lève SystemExit si le pack est injoignable ou invalide : pas de scan offline. """ req = urllib.request.Request( server.rstrip('/') + '/api/pentest/rules', headers={'X-API-Key': key, 'User-Agent': 'BreacheBin-Agent'}, method='GET', ) try: with urllib.request.urlopen(req, timeout=30) as resp: pack = json.loads(resp.read().decode('utf-8', 'replace')) except urllib.error.HTTPError as e: try: detail = json.loads(e.read().decode('utf-8', 'replace')) raise SystemExit('Règles refusées : %s' % detail.get('error', 'accès requis')) except SystemExit: raise except Exception: raise SystemExit('Règles refusées par le serveur (%s).' % e.code) except Exception as e: raise SystemExit('Règles injoignables (connexion exigée) : %s' % e) if not isinstance(pack, dict) or 'sections' not in pack: raise SystemExit('Pack de règles invalide — scan annulé.') rules_path = os.path.join(_engine_dir(), 'rules.json') try: with open(rules_path, 'w', encoding='utf-8') as f: json.dump(pack, f, ensure_ascii=False) except OSError as e: raise SystemExit('Écriture des règles impossible : %s' % e) print('Règles v%s téléchargées.' % pack.get('version', '?')) return rules_path def _run_engine(pentest_path, argv, cwd, env): """Exécute le moteur `pentest`. Mode source : sous-processus python classique. Mode exe (PyInstaller : sys.executable = l'exe lui-même, pas un interpréteur) : exécution in-process via runpy, comportement identique (argv, cwd, env, code). """ if getattr(sys, 'frozen', False): import runpy eng = os.path.dirname(os.path.abspath(pentest_path)) if eng not in sys.path: sys.path.insert(0, eng) old_argv, old_cwd = sys.argv, os.getcwd() sys.argv = ['pentest'] + argv try: os.chdir(cwd) runpy.run_path(pentest_path, run_name='__main__') return 0 except SystemExit as e: return e.code if isinstance(e.code, int) else 0 finally: sys.argv = old_argv try: os.chdir(old_cwd) except OSError: pass else: return subprocess.call( [sys.executable, pentest_path] + argv, cwd=cwd, env=env) def run_scan(server, key, url, opts=None): info = check_access(server, key) if not info.get('active'): until = info.get('until') msg = ('votre abonnement a expiré (fini le %s).' % until) if until else \ 'essai gratuit épuisé — 10$ par scan dans la boutique.' print('Accès PENTEST inactif — %s' % msg) return 3 used = consume_access(server, key, url) mode = used.get('mode', '') if mode == 'free': print('Essai gratuit utilisé — prochains scans : 10$ / scan.') elif mode == 'credit': print('1 crédit consommé (%s restant(s)) — 10$ / scan.' % used.get('credits', 0)) host = url.split('//', 1)[-1].split('/')[0].replace(':', '_') or 'cible' stamp = datetime.now().strftime('%Y-%m-%d_%H%M') rep = os.path.join(BASE, 'reports') os.makedirs(rep, exist_ok=True) out_md = os.path.join(rep, 'rapport_%s_%s.md' % (host, stamp)) opts = opts or {} pentest_path = os.path.join(_engine_dir(), 'pentest') argv = [url, '--out', out_md, '--json'] for flag in ('deep', 'cve', 'no_ssl', 'no_subdomains', 'ports', 'dirb', 'tls_full', 'smuggle', 'slow'): if opts.get(flag): argv.append('--' + flag) if opts.get('min_sev'): argv += ['--min-sev', opts['min_sev']] argv += ['--timeout', str(max(5, min(45, int(opts.get('timeout') or 20))))] argv += ['--delay', str(max(0.0, min(5.0, float(opts.get('delay') or 0.0))))] print('Accès vérifié — scan depuis votre IP en cours :') print(' ' + url) if not getattr(sys, 'frozen', False): print(' ' + ' '.join([sys.executable, pentest_path] + argv)) env = dict(os.environ, BREACHEBIN_RULES=fetch_rules(server, key)) code = _run_engine(pentest_path, argv, _engine_dir(), env) print() print('Rapports générés :') print(' ' + out_md) print(' ' + out_md[:-3] + '.json') return code _OPTS_FR = [ ('--deep', 'deep', 'Tests profonds (SQLi, XSS, open redirect)'), ('--cve', 'cve', 'Corrélation CVE'), ('--no-ssl', 'no_ssl', 'Ignorer l\u2019audit TLS'), ('--no-subdomains', 'no_subdomains', 'Rapide : sans sous-domaines'), ('--ports', 'ports', 'Scan ports TCP'), ('--dirb', 'dirb', 'Brute-force de répertoires'), ('--tls-full', 'tls_full', 'Suites TLS faibles'), ('--smuggle', 'smuggle', 'Indice de smuggling HTTP'), ('--slow', 'slow', 'Mode lent (anti-détection)'), ] def menu(): cfg = load_config() print('=' * 50) print(' BreacheBin PENTEST — agent local') print('=' * 50) while True: print() print('Serveur : %s' % (cfg['server'] or '— (à configurer)')) print('Clé API : %s' % ('✓ définie' if cfg['key'] else '— (à configurer)')) print() print('1 — Configurer mon accès (serveur + clé API)') print('2 — Lancer un scan') print('3 — Quitter') choice = input('Choix : ').strip() if choice == '1': cfg['server'] = input('URL du site BreacheBin (ex. https://vote-site.onrender.com) : ').strip().rstrip('/') cfg['key'] = input('Clé API (Compte > Clé API PENTEST) : ').strip() save_config(cfg['server'], cfg['key']) print('✓ Configuration enregistrée.') elif choice == '2': if not cfg['server'] or not cfg['key']: print('Configurez votre accès d\u2019abord (option 1).') continue url = input('Cible à auditer (https://exemple.com) : ').strip() if not url: continue opts = {} print('Options (%s) :' % ', '.join(f[0] for f in _OPTS_FR)) raw = input(' Ex. : --deep --cve (vide = scan par défaut) : ').strip() for flag, key, _desc in _OPTS_FR: opts[key] = flag in raw.split() print() code = run_scan(cfg['server'], cfg['key'], url, opts) print('Terminé (code %s). Rapports dans ~/.breachebin_agent/reports/' % code) elif choice == '3': print('Au revoir.') break else: print('Choix invalide.') def main(): ap = argparse.ArgumentParser(description='BreacheBin PENTEST — agent local', add_help=True) ap.add_argument('--server', help='URL du site BreacheBin (ex. https://votre-site.onrender.com)') ap.add_argument('--key', help='Votre clé API (Compte > Clé API PENTEST)') ap.add_argument('url', nargs='?', help='Cible à auditer') ap.add_argument('--menu', action='store_true', help='Menu interactif') for flag, _dest, _desc in _OPTS_FR: ap.add_argument(flag, dest=_dest, action='store_true') ap.add_argument('--min-sev', choices=['low', 'medium', 'high', 'critical'], default=None) ap.add_argument('--timeout', type=int, default=20) ap.add_argument('--delay', type=float, default=0.0) args = ap.parse_args() if args.menu or not args.url: menu() return cfg = load_config() server = (args.server or cfg['server'] or '').rstrip('/') key = args.key or cfg['key'] if not server or not key: sys.exit('Serveur et clé API requis (--server --key) ou configurez-les via le menu.') if server != cfg['server'] or key != cfg['key']: save_config(server, key) opts = {d: getattr(args, d) for _f, d, _desc in _OPTS_FR} opts['min_sev'] = args.min_sev opts['timeout'] = args.timeout opts['delay'] = args.delay sys.exit(run_scan(server, key, args.url, opts)) if __name__ == '__main__': try: main() except KeyboardInterrupt: sys.exit(130)