test_win_malware_analyzer.py

#!/usr/bin/env python3
"""Tests de non-regression v7 : doublures lief/r2pipe, deux phases, console."""
import hashlib
import io
import json
import os
import struct
import sys
import types
import unittest
from pathlib import Path
from unittest import mock

# --------------------------------------------------------------------------
# Doublures minimalistes de lief / r2pipe (installees AVANT l'import du script)
# --------------------------------------------------------------------------

lief_stub = types.ModuleType("lief")
_logging_stub = types.ModuleType("lief.logging")
_logging_stub.disable = lambda: None
lief_stub.logging = _logging_stub
pe_mod = types.ModuleType("lief.PE")


class _Enum:
    def __init__(self, name):
        self.name = name


class _DllChar:
    DYNAMIC_BASE = "DYNAMIC_BASE"
    NX_COMPAT = "NX_COMPAT"
    GUARD_CF = "GUARD_CF"
    NO_SEH = "NO_SEH"


class _OptHeader:
    DLL_CHARACTERISTICS = _DllChar


class _Section:
    def __init__(self, name, entropy=6.0, size=1024, offset=0x400):
        self.name = name
        self.virtual_address = 0x1000
        self.virtual_size = size
        self.sizeof_raw_data = size
        self.entropy = entropy
        self.offset = offset
        self.size = size


class _Entry:
    def __init__(self, name):
        self.name = name


class _Import:
    def __init__(self, name, entries):
        self.name = name
        self.entries = [_Entry(e) for e in entries]


class _Header:
    machine = _Enum("AMD64")
    time_date_stamp = 1600000000


class _Opt:
    subsystem = _Enum("WINDOWS_GUI")
    dll_characteristics_lists = ["DYNAMIC_BASE", "NX_COMPAT"]
    checksum = 0


class _Binary:
    def __init__(self):
        self.header = _Header()
        self.optional_header = _Opt()
        self.entrypoint = 0x1400
        self.has_signatures = False
        self.has_rich_header = False
        self.has_imports = True
        self.has_tls = False
        self.has_resources = False
        self.sections = [_Section(".text"), _Section(".rsrc", entropy=7.5)]
        self.imports = [
            _Import("kernel32.dll", ["VirtualAllocEx", "WriteProcessMemory",
                                     "CreateRemoteThread", "IsDebuggerPresent"]),
            _Import("wininet.dll", ["InternetOpenA"]),
        ]
        self.delay_imports = []
        self.debug = []
        self.overlay = b""


pe_mod.Binary = _Binary
pe_mod.OptionalHeader = _OptHeader
pe_mod.get_imphash = lambda pe: "deadbeefdeadbeefdeadbeefdeadbeef"
lief_stub.PE = pe_mod
lief_stub.parse = lambda path: _Binary()
sys.modules["lief"] = lief_stub
sys.modules["lief.PE"] = pe_mod


class _FakeR2:
    """Journalise toutes les commandes recues pour verifier l'absence
    d'injection et le bon ciblage."""

    def __init__(self):
        self.commands = []
        self.closed = False

    def cmd(self, c):
        self.commands.append(c)
        if c == "pdg":
            return "void entry0(void) { /* pseudo-code */ }"
        if c == "pdc":
            return "; pseudo-code natif"
        if c.startswith("pd "):
            return "0x1400  push rbp"
        return ""

    def cmdj(self, c):
        self.commands.append(c)
        if c == "aflj":
            return [
                {"name": "fcn.00401000", "offset": 0x401000, "size": 400, "nbbs": 12},
                {"name": "sym.inject", "offset": 0x401500, "size": 900, "nbbs": 30},
            ]
        if c.startswith("axtj"):
            if "VirtualAllocEx" in c or "WriteProcessMemory" in c or "CreateRemoteThread" in c:
                return [{"type": "CALL", "from": 0x401510, "fcn_name": "sym.inject"}]
            return []
        if c.startswith("agfj"):
            return [{"blocks": [{"offset": 0x401500, "jump": 0x401520, "fail": 0x401540}]}]
        if c == "izzj":
            return [{"vaddr": 0x402000, "size": 12, "section": ".rdata", "string": "http://c2.example.test/gate"}]
        return None

    def quit(self):
        self.closed = True


_LAST_R2 = {}
r2pipe_stub = types.ModuleType("r2pipe")


def _r2open(path, flags=None):
    r2 = _FakeR2()
    _LAST_R2["obj"] = r2
    _LAST_R2["path"] = path
    return r2


r2pipe_stub.open = _r2open
sys.modules["r2pipe"] = r2pipe_stub
import shutil as _sh
_sh.which = lambda b: "/usr/bin/rizin" if b in ("rizin","radare2","r2") else None

sys.path.insert(0, str(Path(__file__).parent))
import win_malware_analyzer as wma  # noqa: E402


def make_pe_file(path: Path, extra: bytes = b"") -> Path:
    """Fabrique un faux PE minimal mais structurellement coherent."""
    data = bytearray(b"MZ" + b"\x00" * 0x3E)
    struct.pack_into("<I", data, 0x3C, 0x80)
    data.extend(b"\x00" * (0x80 - len(data)))
    data.extend(b"PE\x00\x00")
    data.extend(b"\x00" * 200)
    data.extend(b"http://malicious.example.test/panel.php\x00")
    data.extend(b"HKLM\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Run\x00")
    data.extend(b"C:\\Windows\\System32\\evil.dll\x00")
    data.extend(b"kernel32.dll\x00")
    data.extend(extra)
    path.write_bytes(bytes(data))
    return path


class TestPhases(unittest.TestCase):
    def setUp(self):
        self.tmp = Path("/tmp/wma_tests")
        self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "sample.exe")
        self.out = self.tmp / "out"
        self.out.mkdir(exist_ok=True)
        for f in self.out.glob("*"):
            f.unlink()
        self.logger = wma.setup_logger(0)
        self.opts = {
            "output_dir": str(self.out), "reverse_mode": "ask", "no_interactive": False,
            "max_functions": 5, "no_cfg": False, "yara": True, "vt_key": None,
            "timeout": 10, "max_file_size": 50 * 1024 * 1024, "verbosity": 0,
        }

    # -- phase statique ---------------------------------------------------

    def test_static_phase_ok(self):
        res = wma.run_static_phase(self.sample, self.opts, self.logger)
        self.assertTrue(res["success"], res["error"])
        rep = res["report"]
        self.assertEqual(rep["report_type"], "static")
        self.assertNotIn("reverse_engineering", rep, "le rapport statique ne doit PAS contenir le reverse")
        self.assertIn("Process_Injection_Hoisting", rep["pe_structure"]["behavioral_capabilities"])
        self.assertIn("suspicion", rep)
        self.assertTrue(any("malicious.example.test" in u for u in rep["iocs"]["urls"]))
        self.assertIn("VirtualAllocEx", res["flagged_apis"])

    def test_static_report_written_with_suffix(self):
        res = wma.run_static_phase(self.sample, self.opts, self.logger)
        path = wma.write_static_report(res, self.opts, self.logger)
        self.assertTrue(path.exists())
        self.assertTrue(path.name.endswith("_static.json"), path.name)
        self.assertIn(res["sha256"][:8], path.name)
        json.loads(path.read_text())
        self.assertTrue(path.with_suffix(".yar").exists(), "YARA statique attendu")

    def test_static_phase_missing_file(self):
        res = wma.run_static_phase(self.tmp / "nope.exe", self.opts, self.logger)
        self.assertFalse(res["success"])
        self.assertIsNotNone(res["error"])

    def test_static_phase_oversized(self):
        opts = dict(self.opts, max_file_size=10)
        res = wma.run_static_phase(self.sample, opts, self.logger)
        self.assertFalse(res["success"])
        self.assertIn("volumineux", res["error"])

    # -- phase reverse ----------------------------------------------------

    def test_reverse_phase_non_interactive(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        static["static_report_path"] = "/tmp/x_static.json"
        rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, interactive=False)
        self.assertTrue(rev["success"], rev["error"])
        rep = rev["report"]
        self.assertEqual(rep["report_type"], "reverse")
        self.assertIsNone(rep["interactive_session"])
        self.assertIn("sym.inject", rep["reverse_engineering"]["decompiled_functions"])
        self.assertIn("VirtualAllocEx", rep["reverse_engineering"]["suspicious_api_xrefs"])
        self.assertIn("sym.inject", rep["reverse_engineering"]["function_graphs"])
        self.assertEqual(rep["linked_static_report"], "/tmp/x_static.json")
        self.assertIn("suspicion_delta", rep)
        self.assertGreaterEqual(rep["suspicion_delta"]["post_reverse_score"],
                                rep["suspicion_delta"]["static_score"])
        self.assertTrue(_LAST_R2["obj"].closed, "la session rizin doit etre fermee")

    def test_reverse_report_written_with_suffix(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, interactive=False)
        path = wma.write_reverse_report(rev, self.sample, static["sha256"], self.opts, self.logger)
        self.assertTrue(path.exists())
        self.assertTrue(path.name.endswith("_reverse.json"), path.name)
        json.loads(path.read_text())

    def test_two_reports_are_distinct_files(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        p1 = wma.write_static_report(static, self.opts, self.logger)
        rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, interactive=False)
        p2 = wma.write_reverse_report(rev, self.sample, static["sha256"], self.opts, self.logger)
        self.assertNotEqual(p1, p2)
        self.assertTrue(p1.exists() and p2.exists())

    # -- securite : injection de commande r2 ------------------------------

    def test_symbol_sanitizer_rejects_injection(self):
        for bad in ["a; !id", "foo`whoami`", "a|b", "$(id)", "a b", "a\nb", "x" * 300, "", None]:
            self.assertIsNone(wma._sanitize_r2_symbol(bad), bad)
        for good in ["sym.imp.VirtualAllocEx", "fcn.00401000", "entry0"]:
            self.assertEqual(wma._sanitize_r2_symbol(good), good)

    def test_location_sanitizer_accepts_hex_only(self):
        self.assertEqual(wma._sanitize_r2_location("0x401500"), "0x401500")
        self.assertIsNone(wma._sanitize_r2_location("0x401500; !id"))
        self.assertIsNone(wma._sanitize_r2_location("0xZZZZ !id"))

    def test_session_refuses_unsafe_target(self):
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            with self.assertRaises(ValueError):
                s.decompile("foo; !id")
            with self.assertRaises(ValueError):
                s.cfg("$(id)")
            with self.assertRaises(ValueError):
                s.xrefs_to("a|b")
            with self.assertRaises(ValueError):
                s.disassemble("`whoami`")

    def test_no_injection_reached_r2(self):
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            s.analyze()
            try:
                s.decompile("evil; !id")
            except ValueError:
                pass
        for c in _LAST_R2["obj"].commands:
            self.assertNotIn("!id", c)
            self.assertNotIn(";", c)

    # -- console interactive ----------------------------------------------

    def _run_console(self, script: str) -> dict:
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            s.analyze()
            with mock.patch("builtins.input", side_effect=script.strip().split("\n")):
                return wma.interactive_reverse_console(s, static["report"], self.logger)

    def test_console_basic_flow(self):
        log = self._run_console("""
?
l
l inject
d sym.inject
g sym.inject
x VirtualAllocEx
str 5
note fonction d injection confirmee
pick http://malicious.example.test/panel.php
picks
y
sum
q
""")
        self.assertIn("sym.inject", log["decompiled_functions"])
        self.assertIn("sym.inject", log["function_graphs"])
        self.assertIn("VirtualAllocEx", log["xrefs"])
        self.assertEqual(len(log["analyst_notes"]), 1)
        self.assertEqual(len(log["selected_strings"]), 1)
        self.assertEqual(len(log["yara_rules"]), 1)
        self.assertIn("panel.php", log["yara_rules"][0])
        self.assertTrue(all(c["ok"] for c in log["commands"]), log["commands"])

    def test_console_handles_bad_input(self):
        log = self._run_console("""
blahblah
d
d evil; !id
g
x
pick
note
q
""")
        # aucune exception ne doit remonter, et les erreurs sont journalisees
        self.assertTrue(any(not c["ok"] for c in log["commands"]))
        self.assertEqual(log["decompiled_functions"], {})

    def test_console_survives_eof(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            s.analyze()
            with mock.patch("builtins.input", side_effect=EOFError):
                log = wma.interactive_reverse_console(s, static["report"], self.logger)
        self.assertEqual(log["commands"], [])

    def test_console_unclosed_quote_does_not_crash(self):
        log = self._run_console("""
note "guillemet non ferme
q
""")
        self.assertEqual(len(log["analyst_notes"]), 1)

    # -- orchestration / decision -----------------------------------------

    def test_resolve_mode_skip(self):
        opts = dict(self.opts, reverse_mode="skip")
        self.assertEqual(wma.resolve_reverse_mode(opts, False, self.logger), "skip")

    def test_resolve_mode_auto(self):
        opts = dict(self.opts, reverse_mode="auto")
        self.assertEqual(wma.resolve_reverse_mode(opts, True, self.logger), "run")

    def test_resolve_mode_ask_in_batch_never_prompts(self):
        with mock.patch("builtins.input", side_effect=AssertionError("prompt interdit en batch")):
            self.assertEqual(wma.resolve_reverse_mode(self.opts, True, self.logger), "run")

    def test_resolve_mode_ask_non_tty_runs(self):
        with mock.patch.object(sys, "stdin", io.StringIO("")):
            self.assertEqual(wma.resolve_reverse_mode(self.opts, False, self.logger), "run")

    def test_prompt_yes_no_non_tty_returns_default(self):
        with mock.patch.object(sys, "stdin", io.StringIO("")):
            self.assertTrue(wma.prompt_yes_no("?", default=True))
            self.assertFalse(wma.prompt_yes_no("?", default=False))

    def test_process_one_skip_writes_only_static(self):
        opts = dict(self.opts, reverse_mode="skip")
        res = wma.process_one(self.sample, opts, self.logger, batch=False)
        self.assertTrue(res["success"])
        self.assertFalse(res["reverse_done"])
        files = sorted(p.name for p in self.out.glob("*.json"))
        self.assertEqual(len(files), 1)
        self.assertTrue(files[0].endswith("_static.json"))

    def test_process_one_auto_writes_both(self):
        opts = dict(self.opts, reverse_mode="auto", no_interactive=True)
        res = wma.process_one(self.sample, opts, self.logger, batch=False)
        self.assertTrue(res["success"])
        self.assertTrue(res["reverse_done"])
        files = sorted(p.name for p in self.out.glob("*.json"))
        self.assertEqual(len(files), 2, files)
        self.assertTrue(any(f.endswith("_static.json") for f in files))
        self.assertTrue(any(f.endswith("_reverse.json") for f in files))

    def test_batch_worker_never_touches_stdin(self):
        opts = dict(self.opts, reverse_mode="ask", no_interactive=False)
        with mock.patch("builtins.input", side_effect=AssertionError("stdin interdit en batch")):
            res = wma.process_one(self.sample, opts, self.logger, batch=True)
        self.assertTrue(res["success"])
        self.assertTrue(res["reverse_done"])

    def test_reverse_failure_does_not_lose_static(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        p1 = wma.write_static_report(static, self.opts, self.logger)
        with mock.patch.object(wma.r2pipe, "open", side_effect=RuntimeError("rizin absent")):
            rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, interactive=False)
        self.assertFalse(rev["success"])
        self.assertTrue(p1.exists(), "le rapport statique doit survivre a un reverse en echec")
        self.assertIsNotNone(rev["report"], "un rapport reverse partiel doit exister")

    # -- CLI ---------------------------------------------------------------

    def test_cli_skip_reverse_alias(self):
        args = wma.build_arg_parser().parse_args(["x.exe", "--skip-reverse"])
        self.assertTrue(args.skip_reverse)
        args2 = wma.build_arg_parser().parse_args(["x.exe", "--reverse", "auto"])
        self.assertEqual(args2.reverse_mode, "auto")

    def test_cli_rejects_bad_values(self):
        for bad in (["x.exe", "--timeout", "0"], ["x.exe", "--max-functions", "-1"],
                    ["x.exe", "--workers", "0"], ["x.exe", "--max-file-size", "0"]):
            with self.assertRaises(SystemExit):
                wma.main(bad)

    def test_main_end_to_end_skip(self):
        rc = wma.main([str(self.sample), "-o", str(self.out), "--reverse", "skip"])
        self.assertEqual(rc, 0)

    def test_main_end_to_end_auto(self):
        rc = wma.main([str(self.sample), "-o", str(self.out), "--reverse", "auto",
                       "--no-interactive", "--yara"])
        self.assertEqual(rc, 0)
        self.assertEqual(len(list(self.out.glob("*_static.json"))), 1)
        self.assertEqual(len(list(self.out.glob("*_reverse.json"))), 1)

    # -- divers ------------------------------------------------------------

    def test_truncate(self):
        self.assertTrue(wma._truncate("a" * 10, 5).startswith("aaaaa"))
        self.assertIn("tronque", wma._truncate("a" * 10, 5))
        self.assertEqual(wma._truncate("abc", 10), "abc")
        self.assertEqual(wma._truncate(None), "")

    def test_yara_extra_strings_priority(self):
        rule = wma.build_yara_rule("s.exe", "a" * 64, {"urls": ["http://x.test/a"]},
                                   extra_strings=["MARQUEUR_ANALYSTE"], rule_suffix="interactive")
        self.assertIn("MARQUEUR_ANALYSTE", rule)
        self.assertIn("auto_s.exe_interactive".replace(".", "_"), rule.replace(".", "_"))
        self.assertTrue(rule.index("MARQUEUR_ANALYSTE") < rule.index("http://x.test/a"))

    def test_yara_escapes_quotes_and_backslashes(self):
        rule = wma.build_yara_rule("s.exe", "b" * 64, {},
                                   extra_strings=['C:\\Windows\\a"b.exe'])
        self.assertIn('C:\\\\Windows\\\\a\\"b.exe', rule)

    def test_report_path_naming(self):
        p_s = wma._report_path(Path("/tmp/My Sample!.exe"), "abcdef1234", self.opts, "static")
        p_r = wma._report_path(Path("/tmp/My Sample!.exe"), "abcdef1234", self.opts, "reverse")
        self.assertNotEqual(p_s.name, p_r.name)
        self.assertNotIn(" ", p_s.name)
        self.assertNotIn("!", p_s.name)
        self.assertIn("abcdef12", p_s.name)




class TestExtra(unittest.TestCase):
    def setUp(self):
        self.tmp = Path("/tmp/wma_tests2"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.out = self.tmp / "o"; self.out.mkdir(exist_ok=True)
        for f in self.out.glob("*"): f.unlink()
        self.logger = wma.setup_logger(0)
        self.opts = {"output_dir": str(self.out), "reverse_mode": "auto", "no_interactive": True,
                     "max_functions": 5, "no_cfg": False, "yara": False, "vt_key": None,
                     "timeout": 10, "max_file_size": 50*1024*1024, "verbosity": 0}

    def test_ctrl_c_during_reverse_still_writes(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        with mock.patch.object(wma, "run_auto_reverse", side_effect=KeyboardInterrupt):
            rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, False)
        self.assertIsNotNone(rev["report"])
        self.assertTrue(rev["report"]["interrupted"])
        p = wma.write_reverse_report(rev, self.sample, static["sha256"], self.opts, self.logger)
        self.assertTrue(p.exists())

    def test_iocs_error_does_not_break_yara(self):
        rule = wma.build_yara_rule("s.exe", "c"*64, {"error": "boom"})
        self.assertIn("$mz", rule)

    def test_unused_imports(self):
        import ast
        src = Path("win_malware_analyzer.py").read_text()
        tree = ast.parse(src)
        imported = set()
        for n in ast.walk(tree):
            if isinstance(n, ast.Import):
                for a in n.names: imported.add((a.asname or a.name).split(".")[0])
            elif isinstance(n, ast.ImportFrom):
                for a in n.names: imported.add(a.asname or a.name)
        used = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}
        used |= {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)}
        for n in ast.walk(tree):
            if isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name):
                used.add(n.value.id)
        unused = imported - used - {"annotations"}
        self.assertEqual(unused, set(), f"imports inutilises: {unused}")



class TestCorrectifsV71(unittest.TestCase):
    """Un test par vulnerabilite / bug identifie pendant l'audit v7.0."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_71"); self.tmp.mkdir(exist_ok=True)
        for f in self.tmp.glob("*"):
            if f.is_file() or f.is_symlink(): f.unlink()
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)
        self.opts = {"output_dir": str(self.tmp/"o"), "reverse_mode": "auto", "no_interactive": True,
                     "max_functions": 5, "no_cfg": False, "yara": False, "vt_key": None,
                     "timeout": 10, "max_file_size": 50*1024*1024, "verbosity": 0}

    # --- FIX 1 : faux negatif sur les longues chaines imprimables ---
    def test_ioc_dans_longue_chaine_desormais_detecte(self):
        blob = b"MZ" + b"X"*1500 + b" http://tres-mechant.example.test/gate.php " + b"Y"*1500
        iocs = wma.extract_iocs(blob)
        self.assertIn("http://tres-mechant.example.test/gate.php", iocs["urls"])

    def test_ioc_a_cheval_sur_deux_fenetres(self):
        # l'URL est placee pile sur la frontiere de decoupage
        pad = wma.MAX_STRING_LEN - 10
        blob = b"MZ" + b"X"*pad + b" http://frontiere.example.test/x " + b"Y"*1200
        self.assertIn("http://frontiere.example.test/x", wma.extract_iocs(blob)["urls"])

    # --- FIX 2 : ReDoS sur la regex domains ---
    def test_regex_domains_pas_de_redos(self):
        import time as _t
        payload = ("a." * 40000) + "!"
        t0 = _t.perf_counter()
        wma.IOC_PATTERNS["domains"].findall(payload)
        self.assertLess(_t.perf_counter() - t0, 0.5, "regex domains toujours quadratique")

    def test_regex_domains_toujours_correcte(self):
        f = wma.IOC_PATTERNS["domains"].findall
        self.assertEqual(f("www.google.com"), ["www.google.com"])
        self.assertEqual(f("sub.domain.co.uk"), ["sub.domain.co.uk"])
        self.assertEqual(f("api-3.cdn.example.co.jp"), ["api-3.cdn.example.co.jp"])
        self.assertIn("evil.test", f("voir http://evil.test/a"))

    # --- FIX 3 : metacaracteres r2 @ et $ ---
    def test_metacaracteres_r2_rejetes(self):
        for bad in ["foo@0x41414141", "sym.$$", "a@@b", "x@eip"]:
            self.assertIsNone(wma._sanitize_r2_symbol(bad), bad)

    def test_fonction_ciblee_par_adresse_pas_par_nom(self):
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            s.analyze()
            s.decompile("sym.inject")
        seeks = [c for c in _LAST_R2["obj"].commands if c.startswith("s ")]
        self.assertIn("s 0x401500", seeks, f"attendu un seek par adresse, vu {seeks}")

    # --- FIX 4 : detection d'erreur de decompilation ---
    def test_pdg_valide_contenant_le_mot_cannot_est_conserve(self):
        class R:
            def cmd(self, c):
                if c.startswith("?v"): return "0x140001420"
                return 'void f(void){ puts("Cannot open handle"); }' if c == "pdg" else "PDC PAUVRE"
            def cmdj(self, c): return []
        s = wma.ReverseSession.__new__(wma.ReverseSession)
        s.r2, s.errors, s._functions_cache, s._name_to_offset = R(), [], None, {}
        self.assertIn("Cannot open handle", s.decompile("entry0"))

    def test_vraie_erreur_declenche_le_repli(self):
        class R:
            def cmd(self, c):
                if c.startswith("?v"): return "0x140001420"
                return "Cannot find function at 0x0" if c == "pdg" else "PDC OK"
            def cmdj(self, c): return []
        s = wma.ReverseSession.__new__(wma.ReverseSession)
        s.r2, s.errors, s._functions_cache, s._name_to_offset = R(), [], None, {}
        self.assertEqual(s.decompile("entry0"), "PDC OK")

    # --- FIX 5 : crash console sur champs None ---
    def test_console_ne_crashe_plus_sur_aflj_incomplet(self):
        wma._print_functions([{"name": "f", "offset": None, "size": None, "nbbs": None},
                              {"name": None, "offset": 1, "size": 2, "nbbs": 3},
                              "pas_un_dict"])

    # --- FIX 6 : FIFO / device ---
    def test_fifo_refuse_sans_blocage(self):
        import signal
        fifo = self.tmp / "pipe"
        if fifo.exists(): fifo.unlink()
        os.mkfifo(fifo)
        signal.signal(signal.SIGALRM, lambda *a: (_ for _ in ()).throw(TimeoutError()))
        signal.alarm(5)
        try:
            with self.assertRaises(ValueError):
                wma.safe_read_binary(fifo, 1000)
        finally:
            signal.alarm(0)

    def test_repertoire_refuse(self):
        with self.assertRaises((ValueError, IsADirectoryError, PermissionError, OSError)):
            wma.safe_read_binary(self.tmp, 1000)

    # --- FIX 7 : symlink refuse pour LIEF/rizin aussi ---
    def test_symlink_refuse_en_phase_statique(self):
        link = self.tmp / "lien.exe"
        if link.exists() or link.is_symlink(): link.unlink()
        link.symlink_to(self.sample)
        res = wma.run_static_phase(link, self.opts, self.logger)
        self.assertFalse(res["success"])
        self.assertIn("symbolique", res["error"])

    def test_symlink_ignore_en_batch(self):
        link = self.tmp / "lien2.exe"
        if link.exists() or link.is_symlink(): link.unlink()
        link.symlink_to(self.sample)
        found = wma.discover_targets(self.tmp, recursive=False)
        self.assertNotIn(link, found)
        self.assertIn(self.sample, found)

    # --- FIX 8 : ecriture atomique / serialisation defensive ---
    def test_objet_non_serialisable_ne_corrompt_pas_le_json(self):
        class Weird:
            def __str__(self): return "objet-lief-exotique"
        out = self.tmp / "o"; out.mkdir(exist_ok=True)
        p = wma._dump_json({"a": Weird()}, out / "r.json", self.sample, self.logger)
        self.assertIsNotNone(p)
        self.assertEqual(json.loads(p.read_text())["a"], "objet-lief-exotique")

    def test_surrogates_ne_cassent_pas_l_ecriture(self):
        out = self.tmp / "o"; out.mkdir(exist_ok=True)
        p = wma._dump_json({"code": "abc\udcff def"}, out / "s.json", self.sample, self.logger)
        self.assertIsNotNone(p)
        json.loads(p.read_text())

    def test_aucun_fichier_temporaire_laisse(self):
        out = self.tmp / "o"; out.mkdir(exist_ok=True)
        wma._dump_json({"x": 1}, out / "t.json", self.sample, self.logger)
        self.assertEqual(list(out.glob(".wma_*")), [])

    def test_echec_total_d_ecriture_ne_leve_pas(self):
        with mock.patch.object(wma.tempfile, "mkstemp", side_effect=OSError("disque plein")):
            r = wma._dump_json({"x": 1}, self.tmp / "o" / "z.json", self.sample, self.logger)
        self.assertIsNone(r)

    # --- FIX 9 : timeout rizin (nom de variable) ---
    def test_timeout_pose_sur_les_deux_moteurs(self):
        with wma.ReverseSession(str(self.sample), 42, self.logger):
            pass
        cmds = " ".join(_LAST_R2["obj"].commands)
        self.assertIn("analysis.timeout=42", cmds)   # rizin
        self.assertIn("anal.timeout=42", cmds)       # radare2
        self.assertIn("scr.color=0", cmds)           # pas d'ANSI dans le JSON

    # --- FIX 10 : rizin absent ---
    def test_message_clair_si_rizin_absent(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        with mock.patch.object(wma, "_rizin_available", return_value=False):
            rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, False)
        self.assertFalse(rev["success"])
        self.assertIn("PATH", rev["error"])

    # --- FIX 11 : TOCTOU entre les deux phases ---
    def test_modification_du_fichier_entre_les_phases_bloque_le_reverse(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        self.sample.write_bytes(self.sample.read_bytes() + b"MODIFIE_APRES_HASH")
        rev = wma.run_reverse_phase(self.sample, static, self.opts, self.logger, False)
        self.assertFalse(rev["success"])
        self.assertIn("modifie", rev["error"])

    # --- FIX 12 : memoire / plafonds ---
    def test_plafond_ioc_respecte(self):
        blob = b"MZ" + b"\x00".join(f"http://h{i}.example.test/a".encode() for i in range(3000))
        iocs = wma.extract_iocs(blob)
        self.assertLessEqual(len(iocs["urls"]), wma.MAX_IOCS_PER_TYPE)
        self.assertTrue(iocs["extraction_stats"]["truncated"])

    def test_stats_extraction_presentes(self):
        iocs = wma.extract_iocs(b"MZ\x00 http://a.example.test/x \x00")
        self.assertIn("extraction_stats", iocs)
        self.assertFalse(iocs["extraction_stats"]["truncated"])

    # --- FIX 13 : horodatage UTC ---
    def test_horodatage_utc(self):
        res = wma.run_static_phase(self.sample, self.opts, self.logger)
        self.assertTrue(res["report"]["generated_at"].endswith("Z"))

    # --- FIX 14 : batch determine par la nature de la cible ---
    def test_dossier_avec_un_seul_fichier_ne_demande_rien(self):
        d = self.tmp / "un"; d.mkdir(exist_ok=True)
        make_pe_file(d / "seul.exe")
        opts_dir = dict(self.opts, output_dir=str(self.tmp/"o2"), reverse_mode="ask")
        with mock.patch.object(sys.stdin, "isatty", lambda: True), \
             mock.patch("builtins.input", side_effect=AssertionError("invite interdite sur un dossier")):
            rc = wma.main([str(d), "-o", str(self.tmp/"o2"), "--reverse", "ask", "--no-interactive"])
        self.assertEqual(rc, 0)

    # --- FIX 15 : code retour sur interruption ---
    def test_code_retour_130_si_interrompu(self):
        with mock.patch.object(wma, "run_auto_reverse", side_effect=KeyboardInterrupt):
            rc = wma.main([str(self.sample), "-o", str(self.tmp/"o3"), "--reverse", "auto", "--no-interactive"])
        self.assertEqual(rc, 130)

    # --- FIX 16 : cle VT via l'environnement ---
    def test_cle_vt_lue_dans_l_environnement(self):
        captured = {}
        def fake_static(t, opts, log):
            captured["vt"] = opts["vt_key"]
            return {"success": False, "report": None, "error": "stop", "sha256": None,
                    "flagged_apis": [], "target": str(t), "stat": None}
        with mock.patch.dict(os.environ, {"VT_API_KEY": "cle-secrete"}), \
             mock.patch.object(wma, "run_static_phase", side_effect=fake_static):
            wma.main([str(self.sample), "-o", str(self.tmp/"o4")])
        self.assertEqual(captured["vt"], "cle-secrete")

    # --- FIX 17 : chaines marquees bornees ---
    def test_plafond_chaines_marquees(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        script = [f"pick chaine{i}" for i in range(wma.MAX_SELECTED_STRINGS + 20)] + ["q"]
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            s.analyze()
            with mock.patch("builtins.input", side_effect=script):
                log = wma.interactive_reverse_console(s, static["report"], self.logger)
        self.assertEqual(len(log["selected_strings"]), wma.MAX_SELECTED_STRINGS)

    # --- FIX 18 : commande top ---
    def test_commande_top(self):
        static = wma.run_static_phase(self.sample, self.opts, self.logger)
        with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
            auto = wma.run_auto_reverse(s, static["flagged_apis"], 5, True, self.logger)
            with mock.patch("builtins.input", side_effect=["top", "q"]):
                log = wma.interactive_reverse_console(s, static["report"], self.logger, auto_result=auto)
        self.assertTrue(log["commands"][0]["ok"])

    # --- checksum par blocs ---
    def test_checksum_blocs_taille_impaire(self):
        d = bytearray(os.urandom(0x501)); d[0:2] = b"MZ"
        struct.pack_into("<I", d, 0x3C, 0x80); d[0x80:0x84] = b"PE\x00\x00"
        self.assertIsInstance(wma.compute_pe_checksum(bytes(d)), int)

    def test_plafond_chaines_signale_dans_le_rapport(self):
        # Chaines DISTINCTES : les doublons ne consomment plus le budget (v7.8).
        limite = wma.MAX_STRINGS_SCANNED // 3
        blob = b"MZ" + b"\x00".join(
            f"chaine_unique_{i:08d}".encode() for i in range(limite + 500))
        stats = wma.extract_iocs(blob)["extraction_stats"]
        self.assertTrue(stats["string_limit_reached"])
        self.assertTrue(stats["truncated"], "truncated doit refleter le plafond de chaines")

    def test_doublons_ne_consomment_pas_le_budget(self):
        """Un binaire tres repetitif ne doit pas epuiser le budget en doublons :
        un IOC place APRES eux doit rester detecte."""
        blob = b"MZ" + b"\x00".join(b"AAAA" for _ in range(wma.MAX_STRINGS_SCANNED + 5000))
        blob += b"\x00http://c2-en-fin-de-fichier.example.test/gate\x00"
        iocs = wma.extract_iocs(blob)
        self.assertIn("http://c2-en-fin-de-fichier.example.test/gate", iocs["urls"])
        self.assertFalse(iocs["extraction_stats"]["truncated"])

class TestSchemaRizin(unittest.TestCase):
    """Regression : rizin nomme l'adresse `addr` la ou radare2 utilise `offset`.
    Constate en production, toutes les fonctions s'affichaient a 0x0."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_schema"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)

    def test_extraction_adresse_multi_schema(self):
        self.assertEqual(wma._extract_address({"offset": 0x401000}), 0x401000)
        self.assertEqual(wma._extract_address({"addr": 0x401000}), 0x401000)
        self.assertEqual(wma._extract_address({"vaddr": 0x401000}), 0x401000)
        self.assertEqual(wma._extract_address({"addr": "0x401000"}), 0x401000)

    def test_adresse_absente_ou_nulle_nest_pas_zero(self):
        for entry in ({}, {"offset": 0}, {"offset": None}, {"addr": False}, "pas_un_dict"):
            self.assertIsNone(wma._extract_address(entry), entry)

    def test_seek_par_adresse_avec_schema_rizin(self):
        """Le durcissement (ciblage par adresse) doit rester actif sur rizin."""
        class RizinR2(_FakeR2):
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.cible", "addr": 0x401500, "size": 900, "nbbs": 30}]
                return super().cmdj(c) if not c.startswith("aflj") else []
        r = RizinR2(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                s.decompile("sym.cible")
        self.assertIn("s 0x401500", r.commands,
                      "la fonction doit etre ciblee par adresse, pas par nom")

    def test_jamais_de_seek_en_0x0(self):
        """Une cible non resolue doit echouer franchement, jamais seeker en 0x0
        ni decompiler la position courante sous le nom demande (v7.6)."""
        class NoAddrR2(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v"):
                    return "0x0"          # symbole inconnu cote rizin
                if c == "pdg":
                    return "void autre_fonction(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.sansadresse", "size": 100, "nbbs": 3}]
                return []
        r = NoAddrR2(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                with self.assertRaises(ValueError):
                    s.decompile("sym.sansadresse")
        self.assertNotIn("s 0x0", r.commands, "seek en 0x0 : la fonction analysee serait fausse")
        self.assertNotIn("s sym.sansadresse", r.commands)

    def test_affichage_sans_adresse_montre_interrogation(self):
        wma._print_functions([{"name": "f", "size": 10, "nbbs": 2}])

class TestRetoursTerrain(unittest.TestCase):
    """Regressions issues des rapports JSON d'une analyse reelle (v7.3)."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_terrain"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)

    # --- xrefs CODE/ICOD ignorees : le bug le plus grave ---
    def test_xrefs_de_type_code_et_icod_conservees(self):
        class R(_FakeR2):
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.appel", "addr": 0x401500, "size": 100, "nbbs": 4}]
                if c.startswith("axtj"):
                    return [{"type": "CODE", "from": 0x401510, "fcn_name": "sym.appel"},
                            {"type": "ICOD", "from": 0x401520, "fcn_name": "sym.appel"}]
                return []
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                refs = s.xrefs_to("SetThreadContext")
        self.assertEqual(len(refs), 2, "les xrefs CODE/ICOD doivent etre conservees")
        self.assertEqual({x["xref_type"] for x in refs}, {"CODE", "ICOD"})

    def test_priority_functions_non_vide_avec_xrefs_code(self):
        class R(_FakeR2):
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.appel", "addr": 0x401500, "size": 100, "nbbs": 4}]
                if c.startswith("axtj"):
                    return [{"type": "CODE", "from": 0x401510, "fcn_name": "sym.appel"}]
                if c.startswith("agfj"):
                    return [{"blocks": [{"addr": 0x401500, "jump": 0x401520}]}]
                return []
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                auto = wma.run_auto_reverse(s, ["SetThreadContext"], 5, True, self.logger)
        self.assertTrue(auto["suspicious_api_xrefs"], "xrefs vides malgre une capacite detectee")
        self.assertTrue(auto["priority_functions"], "aucune fonction prioritaire selectionnee")

    # --- chaines de bruit dans la regle YARA ---
    def test_stub_dos_et_prologues_exclus(self):
        for bruit in ["!This program cannot be run in DOS mode.", "AWAVAUATUWVSH",
                      "X[^_]A\\A]A^A_", "_GLOBAL_H9", "UAWAVAUATWVSH"]:
            self.assertFalse(wma._is_meaningful_string(bruit), bruit)

    def test_vraies_chaines_conservees(self):
        for utile in ["SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Run",
                      "http://c2.example.test/gate",
                      "C:\\Users\\victim\\AppData\\Roaming\\svc.exe"]:
            self.assertTrue(wma._is_meaningful_string(utile), utile)

    def test_yara_ne_contient_pas_le_stub_dos(self):
        blob = b"MZ" + b"\x00!This program cannot be run in DOS mode.\x00" * 3
        blob += b"\x00 http://reel.example.test/gate \x00"
        iocs = wma.extract_iocs(blob)
        rule = wma.build_yara_rule("s.exe", "a"*64, iocs)
        self.assertNotIn("DOS mode", rule, "regle YARA qui matcherait tous les PE")

    # --- faux positifs XOR ---
    def test_xor_sur_plage_uniforme_ne_produit_pas_de_domaines(self):
        # 0x00 ^ 0x6c = 'l' : du padding devient "llllll..."
        blob = b"MZ" + b"\x00" * 20000 + b"http://vrai.example.test/x"
        res = wma._brute_force_xor_iocs(blob)
        for entry in res:
            for values in entry["iocs"].values():
                for v in values:
                    self.assertLessEqual(wma._dominant_char_ratio(v), 0.4, v)

    # --- faux positifs IOC ---
    def test_extension_source_pas_un_domaine(self):
        self.assertNotIn("main.cpp", wma._filter_domain_candidates(["main.cpp", "evil.test"]))
        self.assertIn("evil.test", wma._filter_domain_candidates(["main.cpp", "evil.test"]))

    def test_ipv4_usage_special_ecartee(self):
        garde = wma._filter_ipv4_candidates(["3.0.0.0", "127.0.0.1", "0.0.0.0",
                                             "224.0.0.1", "185.220.101.5"])
        self.assertEqual(garde, ["185.220.101.5"])

    # --- score proportionnel ---
    def _rapport(self, caps):
        return {"file_metadata": {"is_packed": False},
                "pe_structure": {"behavioral_capabilities": caps,
                                 "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
                                 "authenticode": {"has_signature": False}, "resources": {},
                                 "checksum": {}},
                "iocs": {}, "reverse_engineering": {}}

    def test_une_seule_api_pese_moins_que_plusieurs(self):
        faible = wma.compute_suspicion_score(
            self._rapport({"Process_Injection_Hoisting": ["SetThreadContext"]}))["score"]
        fort = wma.compute_suspicion_score(self._rapport({"Process_Injection_Hoisting": [
            "VirtualAllocEx", "WriteProcessMemory", "CreateRemoteThread", "QueueUserAPC"]}))["score"]
        self.assertLess(faible, fort)
        self.assertIn("1/7 API", " ".join(wma.compute_suspicion_score(
            self._rapport({"Process_Injection_Hoisting": ["SetThreadContext"]}))["reasons"]))

    def test_injecteur_reel_reste_eleve(self):
        r = wma.compute_suspicion_score(self._rapport({
            "Process_Injection_Hoisting": ["VirtualAllocEx", "WriteProcessMemory",
                                           "CreateRemoteThread", "QueueUserAPC"],
            "Anti_Debugging_Evasion": ["IsDebuggerPresent", "CheckRemoteDebuggerPresent",
                                       "NtQueryInformationProcess"],
            "Ransomware_Crypto": ["CryptEncrypt", "CryptGenKey", "CryptAcquireContextA"]}))
        self.assertEqual(r["level"], "eleve")

    def test_elevation_de_privileges_scoree(self):
        rap = self._rapport({})
        rap["pe_structure"]["resources"] = {"requested_execution_level": "requireAdministrator"}
        self.assertIn("elevation", " ".join(wma.compute_suspicion_score(rap)["reasons"]))

    # --- divers ---
    def test_entropie_nulle_pas_negative(self):
        self.assertEqual(str(wma.calculate_entropy(b"\x00" * 100)), "0.0")

    def test_version_info_sans_repr_objet(self):
        class Items:
            items = {"CompanyName": "ACME", "FileVersion": "1.0.0"}
        class SFI:
            langcode_items = [Items()]
        class Ver:
            string_file_info = SFI()
        class RM:
            version = Ver()
        out = wma._extract_version_info(RM())
        self.assertEqual(out["CompanyName"], "ACME")
        self.assertNotIn("object at 0x", str(out), "fuite d'adresse memoire dans le rapport")

    def test_version_info_robuste_si_api_absente(self):
        class RM:
            @property
            def version(self): raise AttributeError("indisponible")
        self.assertEqual(wma._extract_version_info(RM()), {})

class TestBinaireSainConnu(unittest.TestCase):
    """Regressions v7.4, calees sur un binaire GUI MinGW certifie sain
    (source disponible) : aucun de ces faux positifs ne doit revenir."""

    def _iocs_vides(self):
        d = {k: [] for k in wma.IOC_PATTERNS}
        d.update({"obfuscated_b64_iocs": [], "obfuscated_xor_iocs": [],
                  "notable_strings": [],
                  "extraction_stats": {"strings_scanned": 10232, "truncated": False}})
        return d

    def _rapport(self, caps=None, **kw):
        return {"file_metadata": {"is_packed": kw.get("packed", False)},
                "pe_structure": {"behavioral_capabilities": caps or {},
                                 "tls_callbacks": {"count": kw.get("tls", 0)},
                                 "overlay": kw.get("ov", {"size": 0}),
                                 "authenticode": {"has_signature": False},
                                 "resources": kw.get("res", {}), "checksum": {},
                                 "delay_imported_dlls": []},
                "iocs": kw.get("iocs", self._iocs_vides()), "reverse_engineering": {}}

    # --- extraction_stats compte comme un IOC ---
    def test_extraction_stats_nest_pas_un_ioc(self):
        r = wma.compute_suspicion_score(self._rapport())
        self.assertNotIn("type(s) d'IOC", " ".join(r["reasons"]),
                         "un bloc de metadonnees est compte comme un IOC")

    def test_ioc_reel_toujours_compte(self):
        iocs = self._iocs_vides(); iocs["urls"] = ["http://x.test/a"]
        r = wma.compute_suspicion_score(self._rapport(iocs=iocs))
        self.assertIn("type(s) d'IOC", " ".join(r["reasons"]))

    # --- faux positifs XOR observes sur le binaire sain ---
    def test_faux_positifs_xor_reels_ecartes(self):
        for faux in ["4Dx.Dx", "IV.ml", "Pioj.imj", "S.cl", "VV.wwLbb", "W.xc",
                     "aV.ml", "bd.gl", "h.bD", "i.jmi", "ij.Piijd", "ioj.yxj",
                     "dlltellnmsnchXellmWellm.elln"]:
            self.assertFalse(wma._is_plausible_xor_domain(faux), faux)

    def test_vrais_c2_conserves_apres_xor(self):
        for vrai in ["evil-panel.xyz", "c2server.top", "update.microsoft.com",
                     "gate.onion", "botnet-cc.ru"]:
            self.assertTrue(wma._is_plausible_xor_domain(vrai), vrai)

    # --- boilerplate runtime C++ ---
    def test_runtime_cpp_exclu_des_chaines_notables(self):
        for bruit in ["std::bad_cast", "std::bad_array_new_length",
                      "__gnu_cxx::__concurrence_lock_error", "glibcxx.eh_pool",
                      "terminate called recursively", "std::exception"]:
            self.assertFalse(wma._is_meaningful_string(bruit), bruit)

    def test_chaines_applicatives_conservees(self):
        for utile in ["Lancement de winget...", "Erreur: Impossible de lancer winget."]:
            self.assertTrue(wma._is_meaningful_string(utile), utile)

    # --- overlay qualifie ---
    def test_overlay_basse_entropie_moins_penalise(self):
        anodin = wma.compute_suspicion_score(self._rapport(
            ov={"size": 134714, "entropy": 4.07, "embedded_containers": [], "is_suspicious": False}))
        suspect = wma.compute_suspicion_score(self._rapport(
            ov={"size": 134714, "entropy": 7.9, "embedded_containers": [], "is_suspicious": True}))
        self.assertLess(anodin["score"], suspect["score"])

    def test_pe_embarque_en_overlay_detecte(self):
        self.assertIn("pe_executable", wma._detect_embedded_containers(b"MZ\x90\x00" + b"\x00" * 200))
        self.assertIn("zip", wma._detect_embedded_containers(b"junk" + b"PK\x03\x04" + b"\x00" * 50))
        self.assertEqual(wma._detect_embedded_containers(b"\x01\x02\x03" * 100), [])

    # --- calibration d'ensemble ---
    def test_binaire_sain_ne_depasse_pas_faible(self):
        r = wma.compute_suspicion_score(self._rapport(
            caps={"Process_Injection_Hoisting": ["SetThreadContext"],
                  "Anti_Debugging_Evasion": ["OutputDebugStringA"]},
            tls=3,
            ov={"size": 134714, "entropy": 4.07, "embedded_containers": [], "is_suspicious": False},
            res={"requested_execution_level": "requireAdministrator"}))
        self.assertLessEqual(r["score"], 34, f"faux positif : {r['score']} — {r['reasons']}")
        self.assertIn(r["level"], ("minimal", "faible"))

    def test_dropper_reste_eleve(self):
        r = wma.compute_suspicion_score(self._rapport(
            caps={"Network_C2_Communication": ["InternetOpenA", "URLDownloadToFileA", "WinHttpOpen"],
                  "Persistence_Mechanisms": ["RegSetValueExA", "RegCreateKeyExA", "CreateServiceA"]},
            packed=True,
            ov={"size": 500000, "entropy": 7.9, "embedded_containers": ["pe_executable"],
                "is_suspicious": True},
            iocs={"urls": ["http://x.test"], "domains": ["x.test"]}))
        self.assertEqual(r["level"], "eleve", f"{r['score']} — {r['reasons']}")

    # --- toolchain ---
    def test_detection_mingw(self):
        pe = {"sections": [{"name": ".text"}, {"name": "/4"}, {"name": "/19"}], "rich_header": {}}
        self.assertIn("gcc_mingw", wma._detect_toolchain(pe))

    def test_detection_msvc(self):
        pe = {"sections": [{"name": ".text"}], "rich_header": {"key": 123}}
        self.assertIn("msvc", wma._detect_toolchain(pe))

class TestSessionReverseReelle(unittest.TestCase):
    """Regressions v7.5, issues d'une session reverse reelle sur rizin."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_rev75"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)

    def _rizin(self, extra=None):
        """Doublure reproduisant le nommage reel de rizin pour un PE :
        sym.imp.<DLL>.dll_<API>, et non sym.imp.<API>."""
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("f~"):
                    return "0x14002e1a0 8 sym.imp.KERNEL32.dll_SetThreadContext\n"
                if c == "pdg":
                    return "void f(void) { /* code */ }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.handler", "addr": 0x140010940, "size": 672, "nbbs": 19}]
                if c == "iij":
                    return [{"name": "KERNEL32.dll_SetThreadContext",
                             "libname": "KERNEL32.dll", "plt": 0x14002e1a0}]
                if c.startswith("axtj 0x14002e1a0"):
                    return [{"type": "CODE", "from": 0x140010a12, "fcn_name": "sym.handler"}]
                if c.startswith("axtj"):
                    return []
                if c == "izj":
                    return [{"vaddr": 0x140024000, "string": "Lancement de winget...",
                             "section": ".rdata", "size": 22}]
                if c == "izzj":
                    return [{"vaddr": 0x4d, "string": "!This program cannot be run in DOS mode.",
                             "section": None, "size": 40}]
                return []
        r = R(); _LAST_R2["obj"] = r
        return r

    # --- LE bug : schema de nommage des imports ---
    def test_xrefs_trouvees_avec_nommage_dll_qualifie(self):
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                refs = s.xrefs_to("SetThreadContext")
        self.assertTrue(refs, "aucune xref alors que l'import existe (sym.imp.<DLL>.dll_<API>)")
        self.assertEqual(refs[0]["calling_function"], "sym.handler")

    def test_resolution_par_adresse_pas_par_nom(self):
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                refs = s.xrefs_to("SetThreadContext")
        self.assertTrue(refs[0]["resolved_via"].startswith("0x"),
                        "la resolution doit passer par une adresse d'import")

    def test_priority_functions_alimentees_de_bout_en_bout(self):
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                auto = wma.run_auto_reverse(s, ["SetThreadContext"], 5, False, self.logger)
        self.assertTrue(auto["suspicious_api_xrefs"])
        self.assertTrue(auto["priority_functions"])
        self.assertTrue(auto["decompiled_functions"])

    def test_repli_si_table_imports_absente(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.x", "addr": 0x401000, "size": 10, "nbbs": 1}]
                if c == "axtj sym.imp.VirtualAllocEx":
                    return [{"type": "CALL", "from": 0x401010, "fcn_name": "sym.x"}]
                return []
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                refs = s.xrefs_to("VirtualAllocEx")
        self.assertTrue(refs, "le repli sur les anciens schemas doit rester fonctionnel")

    # --- journal d'audit ---
    def test_resultat_vide_nest_pas_une_erreur(self):
        static = wma.run_static_phase(self.sample, {
            "output_dir": str(self.tmp), "reverse_mode": "auto", "no_interactive": True,
            "max_functions": 3, "no_cfg": True, "yara": False, "vt_key": None,
            "timeout": 10, "max_file_size": 10**8, "verbosity": 0}, self.logger)
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                with mock.patch("builtins.input", side_effect=["x ApiQuiNexistePas", "q"]):
                    log = wma.interactive_reverse_console(s, static["report"], self.logger)
        entry = log["commands"][0]
        self.assertTrue(entry["ok"], "une commande sans resultat n'est pas une commande en erreur")
        self.assertIn("aucune reference", entry["detail"])

    def test_vraie_erreur_reste_ok_false(self):
        static = wma.run_static_phase(self.sample, {
            "output_dir": str(self.tmp), "reverse_mode": "auto", "no_interactive": True,
            "max_functions": 3, "no_cfg": True, "yara": False, "vt_key": None,
            "timeout": 10, "max_file_size": 10**8, "verbosity": 0}, self.logger)
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                with mock.patch("builtins.input", side_effect=["d evil; !id", "q"]):
                    log = wma.interactive_reverse_console(s, static["report"], self.logger)
        self.assertFalse(log["commands"][0]["ok"])

    # --- chaines UTF-8 ---
    def test_chaine_accentuee_non_fragmentee(self):
        blob = b"MZ\x00" + "Verification des mises a jour terminee.".replace("e", "é").encode() + b"\x00"
        notables = wma.extract_iocs(blob)["notable_strings"]
        self.assertTrue(any(len(x) > 30 for x in notables), notables)

    def test_fragments_absorbes_par_la_chaine_complete(self):
        blob = b"MZ\x00" + "Vérification des mises à jour terminée.".encode() + b"\x00"
        notables = wma.extract_iocs(blob)["notable_strings"]
        self.assertNotIn("jour termin", notables)
        self.assertNotIn("rification des mises", notables)

    def test_ascii_toujours_extrait(self):
        blob = b"MZ\x00 http://c2.example.test/gate \x00"
        self.assertIn("http://c2.example.test/gate", wma.extract_iocs(blob)["urls"])

    # --- demangleur C++ ---
    def test_boilerplate_demangleur_exclu(self):
        for bruit in ["(anonymous namespace)", "decltype(auto)", "vtable for",
                      "typeinfo name for", "construction vtable for",
                      "template parameter object for", "string literal"]:
            self.assertFalse(wma._is_meaningful_string(bruit), bruit)

    # --- commande str ---
    def test_str_utilise_les_sections_de_donnees(self):
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                items = s.strings(10)
        self.assertIn("izj", r.commands)
        self.assertEqual(items[0]["string"], "Lancement de winget...")

    def test_str_all_ratisse_tout_le_fichier(self):
        r = self._rizin()
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                items = s.strings(10, whole_file=True)
        self.assertIn("izzj", r.commands)
        self.assertIn("DOS mode", items[0]["string"])

    # --- contexte comportemental non score ---
    def test_contexte_comportemental_non_score(self):
        ctx = {"Process_Execution": ["CreateProcessW"],
               "Child_Process_IO_Capture": ["CreatePipe", "SetHandleInformation"],
               "Process_Control": ["TerminateProcess"]}
        rap = {"file_metadata": {}, "pe_structure": {
                   "behavioral_capabilities": {}, "behavioral_context": ctx,
                   "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
                   "authenticode": {"has_signature": False},
                   "resources": {}, "checksum": {}},
               "iocs": {}, "reverse_engineering": {}}
        r = wma.compute_suspicion_score(rap)
        self.assertEqual(r["score"], 3, "seul 'binaire non signe' doit peser")

    def test_contexte_comportemental_bien_rapporte(self):
        self.assertIn("CreateProcessW", wma.BEHAVIORAL_CONTEXT["Process_Execution"])
        self.assertIn("CreatePipe", wma.BEHAVIORAL_CONTEXT["Child_Process_IO_Capture"])
        for apis in wma.BEHAVIORAL_CONTEXT.values():
            for api in apis:
                for scored in list(wma.WIN32_CAPABILITIES.values()):
                    self.assertNotIn(api, scored,
                                     f"{api} present a la fois en contexte et en capacite scoree")

class TestDeuxBinairesSains(unittest.TestCase):
    """Regressions v7.6, issues de deux binaires sains de chaines differentes :
    un GUI MinGW/Win32 et une application Qt (0 detection VirusTotal)."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_v76"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)

    def _session(self, cls):
        r = cls(); _LAST_R2["obj"] = r
        return r

    # === A : falsification de rapport ===
    def test_nom_inconnu_ne_renvoie_pas_une_autre_fonction(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v"): return "0x0"
                if c == "pdg": return "void une_tout_autre_fonction(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [{"name": "sym.autre", "addr": 0x401000, "size": 10, "nbbs": 1}] if c == "aflj" else []
        r = self._session(R)
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                with self.assertRaises(ValueError):
                    s.decompile("fonction_qui_nexiste_pas")

    def test_console_nenregistre_pas_de_code_mal_attribue(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v"): return "0x0"
                if c == "pdg": return "void une_tout_autre_fonction(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [] if c != "aflj" else [{"name": "sym.a", "addr": 0x401000, "size": 5, "nbbs": 1}]
        static = wma.run_static_phase(self.sample, {
            "output_dir": str(self.tmp), "reverse_mode": "auto", "no_interactive": True,
            "max_functions": 3, "no_cfg": True, "yara": False, "vt_key": None,
            "timeout": 10, "max_file_size": 10**8, "verbosity": 0}, self.logger)
        r = self._session(R)
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                with mock.patch("builtins.input", side_effect=["d fonction_qui_nexiste_pas", "q"]):
                    log = wma.interactive_reverse_console(s, static["report"], self.logger)
        self.assertEqual(log["decompiled_functions"], {},
                         "du code a ete enregistre sous un nom qui n'existe pas dans le binaire")
        self.assertFalse(log["commands"][0]["ok"])

    # === B : noms symbole+decalage ===
    def test_symbole_plus_decalage_resolu(self):
        class R(_FakeR2):
            def cmdj(self, c):
                self.commands.append(c)
                return [{"name": "sym.sched_yield", "addr": 0x140020280, "size": 19, "nbbs": 1}] if c == "aflj" else []
        r = self._session(R)
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                self.assertEqual(s.resolve("sym.sched_yield+80"), "0x140020280")
                self.assertEqual(s.resolve("sym.sched_yield+152"), "0x140020280")

    def test_fonctions_prioritaires_dedupliquees(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                return "void f(void){}" if c == "pdg" else ""
            def cmdj(self, c):
                self.commands.append(c)
                if c == "aflj":
                    return [{"name": "sym.sched_yield", "addr": 0x140020280, "size": 19, "nbbs": 1}]
                if c == "iij":
                    return [{"name": "KERNEL32.dll_SetThreadContext", "plt": 0x14002e688}]
                if c.startswith("axtj 0x14002e688"):
                    return [{"type": "CODE", "from": 0x1400202d0, "fcn_name": "sym.sched_yield+80"},
                            {"type": "CODE", "from": 0x140020318, "fcn_name": "sym.sched_yield+152"}]
                return []
        r = self._session(R)
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                auto = wma.run_auto_reverse(s, ["SetThreadContext"], 5, False, self.logger)
        noms = [e["function"] for e in auto["priority_functions"]]
        self.assertEqual(noms, ["sym.sched_yield"], noms)
        self.assertNotIn("filtre de securite", " ".join(auto["errors"]))
        self.assertIn("sym.sched_yield", auto["decompiled_functions"])

    # === C : extensions de fichiers ===
    def test_cookies_sqlite_nest_pas_un_domaine(self):
        garde = wma._filter_domain_candidates(["cookies.sqlite", "evil-c2.test", "profil.ldb"])
        self.assertEqual(garde, ["evil-c2.test"])

    # === D/E/F : bruit dans les chaines notables ===
    def test_symboles_cpp_manges_exclus(self):
        for m in ["_ZN7QObject11connectImplEPKS_PPvS1_S3_PN9QtPrivate15QSlotObjectBaseE",
                  "_ZN11QMessageBox7warningEP7QWidgetRK7QStringS4_",
                  "_ZTVN10__cxxabiv117__class_type_infoE",
                  "?func@@YAXXZ"]:
            self.assertFalse(wma._is_meaningful_string(m), m)

    def test_donnees_de_pixels_exclues(self):
        for px in ["{{{uuuoooccc^^^ZYYUUUPPPMLLKJJGGGLLLLLLLLLGGG",
                   "!!!555EEEVVVaaaiiinnnsssssstttoookkkeee"]:
            self.assertFalse(wma._is_meaningful_string(px), px)

    def test_manifeste_et_runtime_mingw_exclus(self):
        for b in ['<assembly xmlns="urn:schemas-microsoft-com:asm.v1" manifestVersion="1.0">',
                  '<requestedExecutionLevel level="requireAdministrator" uiAccess="false"/>',
                  "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?>",
                  "%d bit pseudo relocation at %p out of range, targeting %p",
                  "VirtualQuery failed for %d bytes at address %p",
                  "The result is too small to be represented (UNDERFLOW)",
                  "std::basic_iostream<char, std::char_traits<char> >",
                  "<GNU C99 15.2.0 -m64 -masm=att -mtune=generic -march=nocona"]:
            self.assertFalse(wma._is_meaningful_string(b), b)

    def test_chaines_applicatives_des_deux_binaires_conservees(self):
        for utile in ["Hus-Clean - Nettoyage de fichiers",
                      "Le scan n'est pas terminé. Attendez la fin du scan avant de nettoyer.",
                      "Voulez-vous vraiment nettoyer ces fichiers ?",
                      "BraveSoftware/Brave-Browser", "Opera Software/Opera Stable",
                      "Lancement de winget...", "Vérification des mises à jour terminée."]:
            self.assertTrue(wma._is_meaningful_string(utile), utile)

    def test_yara_dune_app_qt_ne_contient_pas_de_symboles_qt(self):
        blob = b"MZ\x00" + b"\x00".join([
            b"_ZN7QObject11connectImplEPKS_PPvS1_S3_PN9QtPrivate15QSlotObjectBaseE",
            b"_ZN11QMessageBox7warningEP7QWidgetRK7QStringS4_",
            "Hus-Clean - Nettoyage de fichiers".encode(),
        ])
        rule = wma.build_yara_rule("hus.exe", "a" * 64, wma.extract_iocs(blob))
        self.assertNotIn("_ZN", rule, "regle YARA qui matcherait toute application Qt")
        self.assertIn("Hus-Clean", rule)

    # === G : xrefs issues du seul runtime ===
    def _rapport_xrefs(self, xrefs):
        return {"file_metadata": {"is_packed": False},
                "pe_structure": {"behavioral_capabilities": {
                        "Process_Injection_Hoisting": ["SetThreadContext"]},
                    "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
                    "authenticode": {"has_signature": False}, "resources": {},
                    "checksum": {}, "delay_imported_dlls": []},
                "iocs": {}, "reverse_engineering": {"suspicious_api_xrefs": xrefs}}

    def test_xrefs_runtime_seul_non_retenu_a_charge(self):
        r = wma.compute_suspicion_score(self._rapport_xrefs(
            {"SetThreadContext": [{"calling_function": "sym.pthread_cancel"},
                                  {"calling_function": "sym.sched_yield+80"}]}))
        self.assertIn("uniquement depuis le runtime", " ".join(r["reasons"]))

    def test_xrefs_applicatives_toujours_retenues(self):
        avec = wma.compute_suspicion_score(self._rapport_xrefs(
            {"SetThreadContext": [{"calling_function": "sym.InjectPayload"}]}))["score"]
        runtime = wma.compute_suspicion_score(self._rapport_xrefs(
            {"SetThreadContext": [{"calling_function": "sym.pthread_cancel"}]}))["score"]
        self.assertGreater(avec, runtime)

    def test_reconnaissance_symboles_runtime(self):
        for rt in ["sym.pthread_cancel", "sym.sched_yield+80", "dbg.__tmainCRTStartup",
                   "sym.__gxx_personality_seh0", "method.std::bad_cast._bad_cast__",
                   "sym._Unwind_Resume", "sym.imp.KERNEL32.dll_Sleep"]:
            self.assertTrue(wma._is_runtime_symbol(rt), rt)
        for app in ["sym.WindowProc", "sym.UpdateThread_void_", "fcn.140006630",
                    "sym.ExecuteWingetUpgrade_void_"]:
            self.assertFalse(wma._is_runtime_symbol(app), app)

    # === H : structure du JSON ===
    def test_note_hors_du_dictionnaire_de_capacites(self):
        res = wma.analyze_pe_structure(str(self.sample), self.sample.read_bytes(), self.logger)
        ctx = res.get("behavioral_context", {})
        self.assertNotIn("_note", ctx, "la note casse tout consommateur iterant sur les capacites")
        for value in ctx.values():
            self.assertIsInstance(value, list)

class TestCheminsMalveillants(unittest.TestCase):
    """v7.7 — chemins exe/sys malveillants, les moins exerces jusqu'ici.
    Calibration sur formes synthetiques : a confirmer sur echantillons reels."""

    def _sect(self, n, e=5.0, raw=4096, virt=4096, va=0x1000):
        return {"name": n, "entropy": e, "raw_size": raw, "virtual_size": virt,
                "virtual_address": hex(va), "is_suspicious": e > 7.1}

    def _score(self, caps, **kw):
        rap = {"file_metadata": {"is_packed": kw.get("packed", False)},
               "pe_structure": {"behavioral_capabilities": caps,
                                "is_driver": kw.get("driver", False),
                                "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
                                "authenticode": {"has_signature": False},
                                "resources": {}, "checksum": {}, "delay_imported_dlls": [],
                                "packing": kw.get("packing", {})},
               "iocs": kw.get("iocs", {}), "reverse_engineering": {}}
        return wma.compute_suspicion_score(rap)

    # --- packers ---
    def test_packers_connus_identifies(self):
        for secs, attendu in (
            (["UPX0", "UPX1"], "UPX"),
            ([".text", ".vmp0"], "VMProtect"),
            ([".themida"], "Themida"),
            ([".aspack", ".adata"], "ASPack"),
            (["PEC2", ".text"], "PECompact"),
        ):
            self.assertEqual(wma._detect_packer_sections(secs)[0], attendu, secs)

    def test_binaire_sain_pas_pris_pour_packe(self):
        self.assertEqual(wma._detect_packer_sections([".text", ".rdata", ".rsrc", "/19", "/4"]), [])
        pe = {"sections": [self._sect(".text", 6.1), self._sect(".rdata", 5.2, va=0x24000),
                           self._sect(".rsrc", 3.4, 0x190000, 0x190000, 0x31000)],
              "entrypoint": "0x140001420", "image_base": "0x140000000"}
        self.assertFalse(wma._detect_packing_indicators(pe, 140)["likely_packed"])

    def test_packe_malgre_une_entropie_globale_basse(self):
        """Le cas que l'entropie globale seule ne peut pas voir : petit stub
        packe noye dans de grosses ressources peu entropiques."""
        pe = {"sections": [self._sect("UPX0", 0.0, 0, 0x40000),
                           self._sect("UPX1", 7.9, 0x8000, 0x8000, 0x41000),
                           self._sect(".rsrc", 3.4, 0x400000, 0x400000, 0x50000)],
              "entrypoint": "0x140049000", "image_base": "0x140000000"}
        res = wma._detect_packing_indicators(pe, 6)
        self.assertEqual(res["packer_hint"], "UPX")
        self.assertTrue(res["likely_packed"])

    def test_stub_de_depaquetage_sans_nom_connu(self):
        """Packer inconnu : le faisceau doit suffire sans signature de nom."""
        pe = {"sections": [self._sect(".text", 7.4, 0x2000, 0x80000),
                           self._sect(".data", 7.6, 0x1000, 0x1000, 0x81000)],
              "entrypoint": "0x140081000", "image_base": "0x140000000"}
        res = wma._detect_packing_indicators(pe, 5)
        self.assertIsNone(res["packer_hint"])
        self.assertTrue(res["likely_packed"])
        self.assertGreaterEqual(len(res["indicators"]), 2)

    def test_packer_identifie_pese_dans_le_score(self):
        sans = self._score({})["score"]
        avec = self._score({}, packing={"packer_hint": "UPX", "likely_packed": True,
                                        "indicators": ["sections de packer connues : UPX"]})["score"]
        self.assertGreater(avec, sans)

    # --- drivers ---






    def test_section_contenant_une_rva(self):
        pe = {"sections": [self._sect(".text", va=0x1000), self._sect(".data", va=0x2000)]}
        self.assertEqual(wma._section_containing(pe, 0x1500), ".text")
        self.assertEqual(wma._section_containing(pe, 0x2500), ".data")
        self.assertIsNone(wma._section_containing(pe, 0x99999))

    def test_helpers_packing_robustes(self):
        for pe in ({}, {"sections": []}, {"sections": ["pas_un_dict"]},
                   {"sections": [{"name": None, "entropy": None, "raw_size": None,
                                  "virtual_size": None}]}):
            wma._detect_packing_indicators(pe, 0)
        self.assertEqual(wma._detect_packer_sections([None, 123, ""]), [])

class TestSastV78(unittest.TestCase):
    """v7.8 — passe SAST : securite, erreurs logiques, memoire, coherence."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_sast78"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.exe")
        self.logger = wma.setup_logger(0)

    # --- coherence des tables ---
    def test_aucune_api_dans_deux_categories_scorees(self):
        vues = {}
        for table in (wma.WIN32_CAPABILITIES, wma.DOTNET_CAPABILITIES):
            for cat, apis in table.items():
                for api in apis:
                    vues.setdefault(api, []).append(cat)
        doublons = {a: c for a, c in vues.items() if len(c) > 1}
        self.assertEqual(doublons, {}, f"double comptage possible : {doublons}")

    def test_toute_categorie_a_un_poids(self):
        for cat in list(wma.WIN32_CAPABILITIES) + list(wma.DOTNET_CAPABILITIES):
            self.assertIn(cat, wma._CAPABILITY_WEIGHTS, cat)

    # --- evasion par nommage de symboles ---
    def _score_xref(self, api, caller):
        rap = {"file_metadata": {}, "pe_structure": {
                   "behavioral_capabilities": {"Process_Injection_Hoisting": [api]},
                   "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
                   "authenticode": {"has_signature": False}, "resources": {},
                   "checksum": {}, "delay_imported_dlls": [], "packing": {}},
               "iocs": {}, "reverse_engineering": {
                   "suspicious_api_xrefs": {api: [{"calling_function": caller}]}}}
        return wma.compute_suspicion_score(rap)["score"]

    def test_exemption_runtime_non_contournable_par_le_nom(self):
        """Les noms de symboles viennent du binaire, donc de l'attaquant :
        se dire `pthread_*` ne doit pas suffire a annuler la penalite."""
        evade = self._score_xref("WriteProcessMemory", "sym.pthread_helper_interne")
        franc = self._score_xref("WriteProcessMemory", "sym.InjectPayload")
        self.assertEqual(evade, franc, "penalite annulee par un simple nommage")

    def test_exemption_runtime_legitime_maintenue(self):
        exempt = self._score_xref("SetThreadContext", "sym.pthread_cancel")
        charge = self._score_xref("SetThreadContext", "sym.InjectPayload")
        self.assertLess(exempt, charge)

    def test_api_implausible_pour_un_runtime(self):
        for api in ("CreateRemoteThread", "WriteProcessMemory", "CryptEncrypt",
                    "AmsiScanBuffer"):
            self.assertNotIn(api.lower(), wma._RUNTIME_PLAUSIBLE_APIS, api)

    # --- extraction du nom d'import ---
    def test_prefixe_de_module_retire_sans_mutiler_lapi(self):
        cas = {"KERNEL32.dll_SetThreadContext": "SetThreadContext",
               "msvcrt.dll___C_specific_handler": "__C_specific_handler",
               "msvcrt.dll__amsg_exit": "_amsg_exit",
               "api-ms-win-crt-runtime-l1-1-0.dll_exit": "exit",
               "fltmgr.sys_FltRegisterFilter": "FltRegisterFilter",
               "SetThreadContext": "SetThreadContext"}
        for brut, attendu in cas.items():
            self.assertEqual(wma._IMPORT_MODULE_PREFIX_RE.sub("", brut, count=1), attendu, brut)

    # --- budget de chaines ---
    def test_passe_utf8_atteinte_malgre_beaucoup_dascii(self):
        blob = b"MZ" + b"\x00".join(f"ascii_{i:06d}".encode() for i in range(50000))
        blob += b"\x00" + "Vérification des mises à jour terminée.".encode() + b"\x00"
        notables = wma.extract_iocs(blob)["notable_strings"]
        self.assertTrue(any("Vérification" in x for x in notables), notables[:3])

    def test_ioc_en_fin_de_binaire_repetitif(self):
        blob = b"MZ" + b"\x00".join(b"PAD!" for _ in range(300000))
        blob += b"\x00SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Run\x00"
        self.assertTrue(wma.extract_iocs(blob)["win_registry"])

    def test_fenetres_en_flux_pas_en_liste(self):
        import types
        self.assertIsInstance(wma._iter_string_windows("A" * 5000), types.GeneratorType)

    # --- fichier non PE ---
    def test_non_pe_sans_score(self):
        r = wma.compute_suspicion_score({"file_metadata": {},
                                         "pe_structure": {"error": "Fichier non PE"},
                                         "iocs": {}, "reverse_engineering": {}})
        self.assertIsNone(r["score"])
        self.assertEqual(r["level"], "non_applicable")

    def test_rapport_non_pe_ecrit_sans_planter(self):
        with mock.patch.object(wma, "analyze_pe_structure",
                               return_value={"error": "Fichier non PE"}):
            res = wma.run_static_phase(self.sample, {
                "output_dir": str(self.tmp), "reverse_mode": "skip", "no_interactive": True,
                "max_functions": 3, "no_cfg": True, "yara": True, "vt_key": None,
                "timeout": 5, "max_file_size": 10**8, "verbosity": 0}, self.logger)
            chemin = wma.write_static_report(res, {
                "output_dir": str(self.tmp), "yara": True}, self.logger)
        self.assertTrue(chemin.exists())
        self.assertIsNone(json.loads(chemin.read_text())["suspicion"]["score"])

    # --- verdict de packing unifie ---
    def test_is_packed_reflete_les_indices_structurels(self):
        pe = {"packing": {"likely_packed": True, "packer_hint": "UPX"}}
        with mock.patch.object(wma, "analyze_pe_structure", return_value=pe):
            res = wma.run_static_phase(self.sample, {
                "output_dir": str(self.tmp), "reverse_mode": "skip", "no_interactive": True,
                "max_functions": 3, "no_cfg": True, "yara": False, "vt_key": None,
                "timeout": 5, "max_file_size": 10**8, "verbosity": 0}, self.logger)
        meta = res["report"]["file_metadata"]
        self.assertTrue(meta["is_packed"], "entropie basse mais indices structurels present")
        self.assertEqual(meta["packing_evidence"], "indices_structurels")

    # --- bornes ---
    def test_budget_temps_xor_defini(self):
        self.assertGreater(wma.XOR_SCAN_BUDGET_S, 0)
        self.assertLessEqual(wma.XOR_SCAN_BUDGET_S, wma.IOC_EXTRACTION_BUDGET_S)

    def test_cibles_xref_bornees(self):
        appels = []
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("f~"):
                    return "\n".join(f"0x{0x400000 + i * 8:x} 8 sym.imp.X_{i}" for i in range(500))
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                if c.startswith("axtj"):
                    appels.append(c)
                    return []
                if c == "iij":
                    return [{"name": f"KERNEL32.dll_X", "plt": 0x401000 + i} for i in range(200)]
                return []
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.xrefs_to("X")
        self.assertLessEqual(len(appels), wma.MAX_XREF_TARGETS, len(appels))

    # --- point d'entree ---
    def test_pas_dindice_de_packing_sans_image_base(self):
        pe = {"sections": [{"name": ".text", "virtual_address": "0x1000",
                            "virtual_size": 4096, "raw_size": 4096, "entropy": 6.0}],
              "entrypoint_section": None}
        res = wma._detect_packing_indicators(pe, 100)
        self.assertFalse(any("point d'entree" in i for i in res["indicators"]))

    # --- permissions ---
    def test_rapport_non_lisible_par_tous(self):
        chemin = wma._dump_json({"x": 1}, self.tmp / "perm.json", self.sample, self.logger)
        self.assertEqual(os.stat(chemin).st_mode & 0o077, 0,
                         "un rapport contenant du code de malware ne doit pas etre world-readable")

class TestAnalyseDLL(unittest.TestCase):
    """v7.9 — analyse des DLL : exports, surface d'execution, sideloading."""

    def setUp(self):
        self.tmp = Path("/tmp/wma_dll"); self.tmp.mkdir(exist_ok=True)
        self.sample = make_pe_file(self.tmp / "s.dll")
        self.logger = wma.setup_logger(0)

    def _pe(self, **kw):
        return {"behavioral_capabilities": kw.get("caps", {}),
                "pe_type": kw.get("t", "dll"), "tls_callbacks": {"count": 0},
                "overlay": {"size": 0}, "authenticode": {"has_signature": False},
                "resources": {}, "checksum": {}, "delay_imported_dlls": [], "packing": {},
                "exports": kw.get("exports", {"count": 30}),
                "dll_sideloading": kw.get("sl", {})}

    def _score(self, **kw):
        return wma.compute_suspicion_score({"file_metadata": {}, "pe_structure": self._pe(**kw),
                                            "iocs": {}, "reverse_engineering": {}})

    # --- classification ---
    def test_extensions_famille_dll(self):
        for ext in (".dll", ".ocx", ".cpl", ".ax"):
            self.assertIn(ext, wma.DLL_FAMILY_EXTENSIONS, ext)


    def test_surface_execution_couvre_les_hotes_windows(self):
        tous = {a for v in wma.EXPORT_EXECUTION_SURFACE.values() for a in v}
        for nom in ("DllRegisterServer", "DllGetClassObject", "ServiceMain",
                    "CPlApplet", "DllInstall", "ReflectiveLoader"):
            self.assertIn(nom, tous, nom)

    def test_surface_execution_sans_chevauchement_avec_le_score(self):
        scorees = {a for t in (wma.WIN32_CAPABILITIES,) for v in t.values() for a in v}
        exports = {a for v in wma.EXPORT_EXECUTION_SURFACE.values() for a in v}
        self.assertEqual(scorees & exports, set())

    # --- sideloading ---
    def test_dll_proxy_detectee(self):
        exports = {"count": 40, "forwarded_ratio": 0.95,
                   "forwarded": [{"forwards_to": "version_orig.f"}] * 38}
        r = wma._detect_export_proxying(exports, "version.dll")
        self.assertTrue(r["is_proxy_like"])
        self.assertIn("version_orig", r["targets"])

    def test_redirecteur_microsoft_authentique_exclu(self):
        exports = {"count": 50, "forwarded_ratio": 1.0,
                   "forwarded": [{"forwards_to": "ucrtbase.f"}] * 50}
        r = wma._detect_export_proxying(exports, "api-ms-win-crt-math-l1-1-0.dll")
        self.assertFalse(r["is_proxy_like"])
        self.assertTrue(r["microsoft_forwarder"])

    def test_evasion_par_nom_de_redirecteur_fermee(self):
        """Le nom declare vient du binaire : se dire `api-ms-win-*` ne doit pas
        suffire a echapper a la detection de sideloading."""
        exports = {"count": 40, "forwarded_ratio": 0.95,
                   "forwarded": [{"forwards_to": "charge_deposee.f"}] * 38}
        r = wma._detect_export_proxying(exports, "api-ms-win-core-evasion-l1-1-0.dll")
        self.assertTrue(r["is_proxy_like"], "evasion par simple renommage")
        self.assertTrue(r.get("name_mimics_ms_forwarder"))

    def test_peu_dexports_pas_de_verdict_proxy(self):
        for n in (0, 1, 2):
            r = wma._detect_export_proxying(
                {"count": n, "forwarded_ratio": 1.0,
                 "forwarded": [{"forwards_to": "x.f"}] * n}, "a.dll")
            self.assertFalse(r["is_proxy_like"], n)

    def test_division_protegee_sur_table_vide(self):
        r = wma._detect_export_proxying({"count": 0, "forwarded": []}, "a.dll")
        self.assertEqual(r["ratio"], 0.0)

    # --- score ---
    def test_sideloading_pese_dans_le_score(self):
        sans = self._score()["score"]
        avec = self._score(sl={"is_proxy_like": True, "ratio": 0.95,
                               "targets": ["version_orig"]})["score"]
        self.assertGreater(avec, sans)

    def test_dll_sans_export_signalee(self):
        r = self._score(exports={"count": 0})
        self.assertIn("sans aucun export", " ".join(r["reasons"]))

    def test_exe_sans_export_non_penalise(self):
        r = self._score(t="executable", exports={"count": 0})
        self.assertNotIn("sans aucun export", " ".join(r["reasons"]))

    def test_dll_legitime_reste_basse(self):
        self.assertIn(self._score()["level"], ("minimal", "faible"))

    # --- robustesse ---
    def test_exports_absents_ou_en_erreur(self):
        for exports in ({}, {"count": 0}, {"error": "boum"}):
            wma._detect_export_proxying(exports, "a.dll")
            self._score(exports=exports)

    def test_nom_dexport_hostile_refuse(self):
        for nom in ("evil; !id", "a@0x41414141", "$(id)", "Dll`whoami`", "x" * 300):
            self.assertIsNone(wma._sanitize_r2_symbol(nom), nom)

    def test_empreinte_porte_sur_tous_les_exports(self):
        noms = [f"Export{i:05d}" for i in range(wma.MAX_EXPORT_NAMES + 700)]
        complete = hashlib.sha256("|".join(sorted(n.lower() for n in noms)).encode()).hexdigest()
        tronquee = hashlib.sha256(
            "|".join(sorted(n.lower() for n in noms[:wma.MAX_EXPORT_NAMES])).encode()).hexdigest()
        self.assertNotEqual(complete, tronquee,
                            "une empreinte tronquee regrouperait a tort des DLL distinctes")

    # --- reverse : decompilation des exports ---
    def test_exports_decompiles_par_adresse(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v"): return "0x0"
                if c == "pdg": return "void DllRegisterServer(void) { /* charge */ }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [] if c != "aflj" else [{"name": "sym.x", "addr": 0x401000,
                                                "size": 10, "nbbs": 1}]
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                auto = wma.run_auto_reverse(
                    s, [], 5, False, self.logger,
                    exported_functions=[{"name": "DllRegisterServer", "address": "0x140002000"}])
        self.assertIn("DllRegisterServer", auto["decompiled_exports"])
        self.assertIn("s 0x140002000", r.commands, "l'export doit etre cible par adresse")

    def test_repli_sur_le_nom_si_adresse_inconnue(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v sym.Start"): return "0x140003000"
                if c.startswith("?v"): return "0x0"
                if c == "pdg": return "void Start(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [] if c != "aflj" else [{"name": "sym.a", "addr": 0x401000,
                                                "size": 5, "nbbs": 1}]
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                auto = wma.run_auto_reverse(s, [], 5, False, self.logger,
                                            exported_functions=[{"name": "Start", "address": None}])
        self.assertIn("Start", auto["decompiled_exports"])

    def test_nombre_dexports_decompiles_borne(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if c.startswith("?v"): return "0x140002000"
                if c == "pdg": return "void f(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [] if c != "aflj" else [{"name": "sym.a", "addr": 0x401000,
                                                "size": 5, "nbbs": 1}]
        r = R(); _LAST_R2["obj"] = r
        cibles = [{"name": f"Export{i}", "address": None} for i in range(50)]
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                auto = wma.run_auto_reverse(s, [], 5, False, self.logger,
                                            exported_functions=cibles)
        self.assertLessEqual(len(auto["decompiled_exports"]), wma.MAX_EXPORTS_DECOMPILED)

    def test_export_irresoluble_nempeche_pas_les_suivants(self):
        class R(_FakeR2):
            def cmd(self, c):
                self.commands.append(c)
                if "Introuvable" in c: return "0x0"
                if c.startswith("?v"): return "0x140002000"
                if c == "pdg": return "void f(void) { }"
                return ""
            def cmdj(self, c):
                self.commands.append(c)
                return [] if c != "aflj" else [{"name": "sym.a", "addr": 0x401000,
                                                "size": 5, "nbbs": 1}]
        r = R(); _LAST_R2["obj"] = r
        with mock.patch.object(wma.r2pipe, "open", return_value=r):
            with wma.ReverseSession(str(self.sample), 10, self.logger) as s:
                s.analyze()
                auto = wma.run_auto_reverse(
                    s, [], 5, False, self.logger,
                    exported_functions=[{"name": "Introuvable", "address": None},
                                        {"name": "Valide", "address": None}])
        self.assertIn("Valide", auto["decompiled_exports"])
        self.assertNotIn("Introuvable", auto["decompiled_exports"])

class TestEchantillonsReels(unittest.TestCase):
    """v8.0 — regressions issues d'un lot reel : .NET, binaires signes,
    packers, installeur legitime, deux binaires sains de reference."""

    def _score(self, pe_extra=None, iocs=None):
        pe = {"behavioral_capabilities": {}, "behavioral_capabilities_inferred": {},
              "is_driver": False, "pe_type": "executable", "tls_callbacks": {"count": 0},
              "overlay": {"size": 0}, "authenticode": {"has_signature": False},
              "resources": {}, "checksum": {}, "delay_imported_dlls": [], "packing": {},
              "exports": {"count": 0}, "dll_sideloading": {}}
        pe.update(pe_extra or {})
        return wma.compute_suspicion_score({"file_metadata": {}, "pe_structure": pe,
                                            "iocs": iocs or {}, "reverse_engineering": {}})

    # === .NET : le faux negatif majeur ===
    def test_capacites_recuperees_dans_les_chaines(self):
        raw = b"MZ\x00" + b"\x00".join([b"amsi.dllAmsiScanBuffer",
                                        b"AddVectoredExceptionHandler",
                                        b"GetFunctionPointerForDelegate"])
        tables = dict(wma.WIN32_CAPABILITIES); tables.update(wma.DOTNET_CAPABILITIES)
        trouvees = wma._capabilities_from_strings(raw, tables)
        self.assertIn("Defense_Evasion_AMSI_ETW", trouvees)
        self.assertIn("Hooking_VEH", trouvees)

    def test_patcheur_amsi_dotnet_nest_plus_minimal(self):
        inf = {"Defense_Evasion_AMSI_ETW": ["AmsiScanBuffer", "amsi.dll"],
               "Hooking_VEH": ["AddVectoredExceptionHandler",
                               "RemoveVectoredExceptionHandler"]}
        r = self._score({"behavioral_capabilities_inferred": inf, "is_dotnet": True,
                         "pe_type": "dll", "exports": {"count": 0}})
        self.assertGreaterEqual(r["score"], 30, f"{r['score']} — {r['reasons']}")
        self.assertNotEqual(r["level"], "minimal")

    def test_api_hautement_specifique_pleine_confiance(self):
        seul = self._score({"behavioral_capabilities": {
            "Defense_Evasion_AMSI_ETW": ["AmsiScanBuffer"]}})
        banal = self._score({"behavioral_capabilities": {
            "Discovery_Recon": ["GetComputerNameA"]}})
        self.assertGreater(seul["score"], banal["score"])

    def test_veh_couvert(self):
        toutes = {a for v in wma.WIN32_CAPABILITIES.values() for a in v}
        self.assertIn("AddVectoredExceptionHandler", toutes)

    # === binaires signes ===
    def test_signature_non_valide_plus_grave_que_labsence(self):
        invalide = self._score({"authenticode": {
            "has_signature": True,
            "verification_status": "VERIFICATION_FLAGS.CERT_NOT_FOUND|BAD_SIGNATURE"}})
        non_signe = self._score()
        self.assertGreater(invalide["score"], non_signe["score"])
        self.assertIn("NON VALIDE", " ".join(invalide["reasons"]))

    def test_signature_valide_dun_editeur_etabli_a_decharge(self):
        r = self._score({"authenticode": {
            "has_signature": True, "verification_status": "VERIFICATION_FLAGS.OK",
            "signers": [{"subject": "C=US, O=Microsoft Corporation, CN=Microsoft Corporation",
                         "issuer": "C=US, O=Microsoft Corporation, CN=Microsoft Code Signing PCA 2011"}]}})
        self.assertEqual(r["score"], 0, r["reasons"])
        self.assertIn("editeur etabli", " ".join(r["reasons"]))

    # === packing : faux positifs .NET ===
    def test_dotnet_signe_nest_pas_pris_pour_packe(self):
        pe = {"sections": [{"name": ".text", "entropy": 7.72, "raw_size": 33525248,
                            "virtual_size": 33525060, "virtual_address": "0x2000"},
                           {"name": ".rsrc", "entropy": 5.25, "raw_size": 372224,
                            "virtual_size": 372024, "virtual_address": "0x1ffc000"}],
              "is_dotnet": True, "entrypoint_section": ".text"}
        self.assertFalse(wma._detect_packing_indicators(pe, 1)["likely_packed"])

    def test_packer_natif_toujours_detecte(self):
        pe = {"sections": [{"name": "", "entropy": 6.7, "raw_size": 319488,
                            "virtual_size": 319100, "virtual_address": "0x4d000"},
                           {"name": ".data", "entropy": 1.14, "raw_size": 512,
                            "virtual_size": 101968, "virtual_address": "0x31000"}],
              "is_dotnet": False, "entrypoint_section": ".text"}
        res = wma._detect_packing_indicators(pe, 1)
        self.assertTrue(res["likely_packed"])
        self.assertTrue(any("sans nom" in i for i in res["indicators"]))

    def test_charge_embarquee_dans_une_section(self):
        pe = {"sections": [{"name": ".rsrc", "entropy": 8.0, "raw_size": 10567680,
                            "virtual_size": 10567396, "virtual_address": "0x100000"}],
              "is_dotnet": False}
        res = wma._detect_packing_indicators(pe, 5)
        self.assertTrue(any("charge embarquee" in i for i in res["indicators"]))

    # === IOC : faux positifs reels ===
    def test_namespaces_dotnet_ne_sont_pas_des_domaines(self):
        rejets = ["System.Windows.Controls", "MinecraftInstaller.Helpers",
                  "amsi.dllAmsiScanBuffer", "3YYw.EH", "Dv40H.GWb", "QM9N.tr",
                  "O.QChAh", "h.reloc", "0.1.vlg"]
        self.assertEqual(wma._filter_domain_candidates(rejets), [])

    def test_vrais_domaines_conserves(self):
        gardes = ["ca.priceandsons.tech", "ns.adobe.com", "www.w3.org",
                  "schemas.microsoft.com", "test1.playfabapi.com"]
        self.assertEqual(wma._filter_domain_candidates(gardes), gardes)

    def test_horodatage_nest_pas_de_lipv6(self):
        self.assertEqual(wma.IOC_PATTERNS["ipv6"].findall("01:07:22"), [])
        self.assertTrue(wma.IOC_PATTERNS["ipv6"].findall("2001:0db8:85a3:0000:8a2e:0370:7334"))

    def test_oid_de_certificat_nest_pas_une_ip(self):
        self.assertEqual(wma._filter_ipv4_candidates(["1.3.6.1", "1.2.840.113549"]), [])
        self.assertEqual(wma._filter_ipv4_candidates(["185.220.101.5"]), ["185.220.101.5"])

    def test_guid_nest_pas_une_adresse_bitcoin(self):
        self.assertEqual(
            wma._filter_wallet_candidates(["14135847554A11E4BBF1D82D68481CF5"]), [])
        for vrai in ("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa", "0x" + "a" * 40):
            self.assertIn(vrai, wma._filter_wallet_candidates([vrai]))

    def test_ioc_base64_filtres_comme_le_clair(self):
        blob = b"MZ\x00/Admin/GetRandomResultTables\x00"
        for entree in wma.extract_iocs(blob)["obfuscated_b64_iocs"]:
            if entree.get("type") == "domains":
                self.assertEqual(wma._filter_domain_candidates(entree["decoded"]),
                                 entree["decoded"])

    # === qualite YARA ===
    def test_api_importees_exclues_de_yara(self):
        iocs = {"notable_strings": ["SetUnhandledExceptionFilter", "QueryPerformanceCounter",
                                    "Elevation:Administrator!new:"]}
        regle = wma.build_yara_rule("s.exe", "a" * 64, iocs,
                                    imported_apis=["SetUnhandledExceptionFilter",
                                                   "QueryPerformanceCounter"])
        self.assertNotIn("QueryPerformanceCounter", regle)
        self.assertIn("Elevation:Administrator", regle)

    def test_symboles_comdat_exclus(self):
        for c in (".xdata$_ZL23__gxx_exception_cleanup19_Unwind_Reason_Code",
                  ".text$_ZL17parse_lsda_headerP15_Unwind_Context",
                  ".data$__emutls_v._ZZN12_GLOBAL__N_1L10get_globalEvE6global"):
            self.assertFalse(wma._is_meaningful_string(c), c)

    def test_boilerplate_dotnet_exclu(self):
        for c in ("CompilationRelaxationsAttribute", "RuntimeCompatibilityAttribute",
                  "CompilerGeneratedAttribute", "WrapNonExceptionThrows",
                  "__StaticArrayInitTypeSize=28", "$$method0x600000a-2"):
            self.assertFalse(wma._is_meaningful_string(c), c)

    def test_chaines_discriminantes_conservees(self):
        for c in ("winget upgrade --all --accept-source-agreements",
                  "Elevation:Administrator!new:",
                  "Hus-Clean - Nettoyage de fichiers"):
            self.assertTrue(wma._is_meaningful_string(c), c)

class TestLotFinal(unittest.TestCase):
    """v8.1 — retrait (puis v8.4 : reintroduction) du support .sys, et
    regressions du dernier lot reel : installeur signe legitime, DLL SAP
    detournee, assembly .NET."""

    def _score(self, **kw):
        pe = {"behavioral_capabilities": kw.get("caps", {}),
              "behavioral_capabilities_inferred": kw.get("inf", {}),
              "pe_type": kw.get("t", "executable"), "is_dotnet": kw.get("net", False),
              "tls_callbacks": {"count": kw.get("tls", 0)},
              "overlay": kw.get("ov", {"size": 0}),
              "authenticode": kw.get("sig", {"has_signature": False}),
              "resources": {}, "checksum": {}, "delay_imported_dlls": [],
              "packing": kw.get("pk", {}), "exports": kw.get("exp", {"count": 0}),
              "dll_sideloading": kw.get("sl", {})}
        return wma.compute_suspicion_score(
            {"file_metadata": {"is_packed": kw.get("packed", False)},
             "pe_structure": pe, "iocs": kw.get("iocs", {}), "reverse_engineering": {}})

    # === support .sys reintroduit en v8.4 (voir TestSupportDrivers) ===

    def test_types_pe_restants(self):
        self.assertEqual(self._score(t="dll", exp={"count": 5})["score"], 3)
        self.assertEqual(self._score(t="executable")["score"], 3)

    def test_scoring_dll_preserve_apres_retrait(self):
        """Le scoring DLL ne doit pas avoir ete emporte avec le code driver."""
        sans = self._score(t="dll", exp={"count": 5})["score"]
        sans_export = self._score(t="dll", exp={"count": 0})
        self.assertIn("sans aucun export", " ".join(sans_export["reasons"]))
        self.assertGreater(sans_export["score"], sans)
        proxy = self._score(t="dll", exp={"count": 40},
                            sl={"is_proxy_like": True, "ratio": 0.9, "targets": ["orig"]})
        self.assertIn("sideloading", " ".join(proxy["reasons"]))

    # === signature valide a decharge ===
    def test_signe_valide_annule_le_bruit_heuristique(self):
        r = self._score(net=True, packed=True,
                        sig={"has_signature": True,
                             "verification_status": "VERIFICATION_FLAGS.OK",
                             "signers": [{"subject": "CN=Microsoft Corporation",
                                          "issuer": "CN=Microsoft Code Signing PCA 2011"}]},
                        iocs={"urls": ["http://x"], "domains": ["a.b.com"]})
        self.assertEqual(r["level"], "minimal", f"{r['score']} — {r['reasons']}")
        self.assertIn("Microsoft Corporation", " ".join(r["reasons"]))

    def test_entropie_globale_ignoree_pour_dotnet(self):
        avec = self._score(packed=True, net=False)["score"]
        sans = self._score(packed=True, net=True)["score"]
        self.assertGreater(avec, sans)

    # === sections aleatoires (DLL detournee) ===
    def test_sections_aleatoires_detectees(self):
        noms = [".szgpcfh", ".planpm", ".apwhamw", ".ynp", ".gamjck", ".xer",
                ".ujzhqjt", ".fjqg", ".txfdv", ".rsrc"]
        self.assertEqual(len(wma._detect_randomized_sections(noms)), 9)

    def test_binaires_sains_sans_faux_positif_de_sections(self):
        for noms in ([".text", ".data", ".rdata", ".pdata", ".xdata", ".bss",
                      ".idata", ".tls", ".rsrc", ".reloc", "/4", "/19", "/31"],
                     [".text", ".data", ".rdata", ".qtversi", ".pdata", ".xdata",
                      ".bss", ".idata", ".CRT", ".tls", ".rsrc", ".reloc"],
                     [".text", ".rsrc", ".reloc"]):
            self.assertEqual(wma._detect_randomized_sections(noms), [], noms)

    def test_dll_detournee_reste_elevee(self):
        r = self._score(t="dll", tls=2, exp={"count": 2},
                        caps={"Process_Injection_Hoisting": ["SetThreadContext"],
                              "Hooking_VEH": ["AddVectoredExceptionHandler"]},
                        sig={"has_signature": True,
                             "verification_status": "VERIFICATION_FLAGS.BAD_DIGEST|BAD_SIGNATURE"},
                        pk={"likely_packed": True, "indicators": ["sections aleatoires"]},
                        iocs={"domains": ["a.b.com"], "urls": ["http://x"]})
        self.assertEqual(r["level"], "eleve", f"{r['score']} — {r['reasons']}")

    # === bruit PKI ===
    def test_infrastructure_pki_ecartee(self):
        for pki in ("cacerts.digicert.com", "crl3.digicert.com", "s1.symcb.com",
                    "www.symauth.com", "http://ocsp.digicert.com0A"):
            self.assertTrue(wma._is_pki_infrastructure(pki), pki)

    def test_fausse_autorite_conservee(self):
        """Le domaine d'une AC contrefaite est le signal recherche."""
        for faux in ("ca.priceandsons.tech", "priceandsons.tech",
                     "intermediate.priceandsons.tech"):
            self.assertFalse(wma._is_pki_infrastructure(faux), faux)

    def test_pki_absente_des_iocs(self):
        blob = (b"MZ\x00http://crl3.digicert.com/Root.crl\x00"
                b"cacerts.digicert.com\x00ca.priceandsons.tech\x00")
        iocs = wma.extract_iocs(blob)
        self.assertNotIn("cacerts.digicert.com", iocs["domains"])
        self.assertIn("ca.priceandsons.tech", iocs["domains"])
        self.assertEqual([u for u in iocs["urls"] if "digicert" in u], [])

    # === double comptage AMSI ===
    def test_amsi_compte_une_seule_fois(self):
        vues = {}
        for table in (wma.WIN32_CAPABILITIES, wma.DOTNET_CAPABILITIES):
            for cat, apis in table.items():
                for api in apis:
                    vues.setdefault(api, []).append(cat)
        self.assertEqual({a: c for a, c in vues.items() if len(c) > 1}, {})

    # === chaines : RTTI MSVC, jeux de caracteres, extensions de donnees ===
    def test_descripteurs_rtti_msvc_exclus(self):
        for c in (".?AV?$CMap@V?$CStringT@_WV?$StrTraitMFC@_W@ATL@@@@",
                  ".?AV?$CFixedStringT@V?$CStringT@_W@ATL@@"):
            self.assertFalse(wma._is_meaningful_string(c), c)

    def test_table_de_caracteres_exclue(self):
        self.assertFalse(wma._is_meaningful_string(
            "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ"
            "[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"))

    def test_extensions_de_donnees_pas_des_domaines(self):
        self.assertEqual(wma._filter_domain_candidates(
            ["cry.dbf", "tbl.dbf", "crystalreports.dataaccess"]), [])

class TestGrandLotReel(unittest.TestCase):
    """v8.2 — regressions issues d'un lot de 25 binaires reels
    (22 malveillants : rancongiciels, droppers, loaders, RAT, charges Go
    signees, agents packes ; 3 sains dont un installeur Microsoft signe)."""

    def _score(self, **kw):
        pe = {"behavioral_capabilities": kw.get("caps", {}),
              "behavioral_capabilities_inferred": kw.get("inf", {}),
              "pe_type": kw.get("t", "executable"), "is_dotnet": kw.get("net", False),
              "tls_callbacks": {"count": kw.get("tls", 0)},
              "overlay": kw.get("ov", {"size": 0}),
              "authenticode": kw.get("sig", {"has_signature": False}),
              "resources": {}, "checksum": {}, "delay_imported_dlls": [],
              "packing": kw.get("pk", {}), "exports": kw.get("exp", {"count": 0}),
              "dll_sideloading": {}}
        return wma.compute_suspicion_score(
            {"file_metadata": {"is_packed": kw.get("packed", False)},
             "pe_structure": pe, "iocs": kw.get("iocs", {}), "reverse_engineering": {}})

    def _sect(self, nom, ent, raw, virt=None):
        return {"name": nom, "entropy": ent, "raw_size": raw,
                "virtual_size": virt or raw, "virtual_address": "0x1000"}

    # === reputation de signature ===
    def test_editeur_etabli_via_ac_reconnue_a_decharge(self):
        pts, _ = wma._signature_reputation({"signers": [{
            "subject": "C=US, O=Microsoft Corporation, CN=Microsoft Corporation",
            "issuer": "CN=Microsoft Code Signing PCA 2011", "issued_days_ago": 700}]})
        self.assertEqual(pts, -20)

    def test_certificat_recent_dune_societe_ecran_a_charge(self):
        """Trois charges Go du lot portaient des certificats DigiCert valides,
        emis a des coquilles une semaine avant l'analyse."""
        pts, motif = wma._signature_reputation({"signers": [{
            "subject": "C=US, O=NexaForge Solutions, CN=NexaForge Solutions",
            "issuer": "C=US, O=DigiCert, Inc., CN=DigiCert Trusted G4 Code Signing CA",
            "issued_days_ago": 7}]})
        self.assertGreater(pts, 0)
        self.assertIn("RECENT", motif)

    def test_evasion_par_sujet_contrefait_fermee(self):
        """Le sujet est ecrit dans le binaire : se declarer editeur connu dans
        un certificat auto-signe ne doit donner aucune remise."""
        pts, _ = wma._signature_reputation({"signers": [{
            "subject": "O=Microsoft Corporation Ltd",
            "issuer": "CN=Microsoft Corporation Ltd", "issued_days_ago": 3}]})
        self.assertGreater(pts, 0, "remise accordee a un certificat auto-signe usurpateur")

    def test_editeur_inconnu_certificat_ancien_neutre(self):
        pts, _ = wma._signature_reputation({"signers": [{
            "subject": "O=PetiteBoite SARL",
            "issuer": "O=Sectigo Limited, CN=Sectigo RSA Code Signing CA",
            "issued_days_ago": 730}]})
        self.assertEqual(pts, 0)

    def test_age_de_certificat(self):
        self.assertIsNone(wma._certificate_age_days(None))
        self.assertIsNone(wma._certificate_age_days("bruit"))
        self.assertGreater(wma._certificate_age_days("[2020, 1, 1, 0, 0, 0]"), 1000)

    # === indices de packing conclusifs ===
    def test_zero_import_est_conclusif(self):
        pe = {"sections": [self._sect(".text", 6.0, 40960)], "is_dotnet": False}
        res = wma._detect_packing_indicators(pe, 0)
        self.assertTrue(res["conclusive_indicators"])
        self.assertTrue(res["likely_packed"])

    def test_section_unique_conclusive(self):
        pe = {"sections": [self._sect(".text", 7.95, 45000)], "is_dotnet": False}
        res = wma._detect_packing_indicators(pe, 5)
        self.assertTrue(any("section unique" in i for i in res["conclusive_indicators"]))

    def test_code_chiffre_conclusif(self):
        pe = {"sections": [self._sect(".text", 8.0, 104960),
                           self._sect(".rdata", 4.16, 1536)], "is_dotnet": False}
        res = wma._detect_packing_indicators(pe, 20)
        self.assertTrue(any("code" in i and "chiffre" in i
                            for i in res["conclusive_indicators"]))

    def test_proportion_chiffree_conclusive(self):
        """Un packer Delphi du lot logeait 100% de son contenu a 8,0 d'entropie."""
        pe = {"sections": [self._sect(".text", 8.0, 104960), self._sect(".itext", 6.33, 1536),
                           self._sect(".rdata", 4.16, 1536), self._sect(".data", 8.0, 40960),
                           self._sect(".pdata", 7.97, 7168)], "is_dotnet": False}
        res = wma._detect_packing_indicators(pe, 20)
        self.assertTrue(any("essentiellement chiffre" in i
                            for i in res["conclusive_indicators"]))

    def test_binaires_sains_sans_indice_conclusif(self):
        for pe in ({"sections": [self._sect(".text", 6.1, 138240),
                                 self._sect(".rsrc", 3.47, 1649744)], "is_dotnet": False},
                   {"sections": [self._sect(".text", 5.83, 32256),
                                 self._sect(".rdata", 5.22, 19968)], "is_dotnet": False},
                   {"sections": [self._sect(".text", 7.72, 33525248),
                                 self._sect(".rsrc", 5.25, 372224)], "is_dotnet": True}):
            res = wma._detect_packing_indicators(pe, 150)
            self.assertEqual(res["conclusive_indicators"], [], pe["sections"][0])

    def test_indices_conclusifs_cumules(self):
        un = self._score(pk={"conclusive_indicators": ["a"], "likely_packed": True})["score"]
        trois = self._score(pk={"conclusive_indicators": ["a", "b", "c"],
                                "likely_packed": True})["score"]
        self.assertGreater(trois, un)

    # === note de rancon multilingue ===
    def test_note_de_rancon_multilingue(self):
        import base64 as b64
        textes = ["فایل های شما رمزگذاری شده است لطفا با ما تماس بگیرید",
                  "הקבצים שלך מוצפנים אנא צור איתנו קשר עכשיו",
                  "आपकी फ़ाइलें एन्क्रिप्ट की गई हैं कृपया संपर्क करें"]
        res = wma._detect_multilingual_base64(
            [b64.b64encode(t.encode()).decode() for t in textes])
        self.assertTrue(res["is_ransom_note_like"], res)
        self.assertGreaterEqual(len(res["scripts"]), 3)

    def test_une_seule_langue_nest_pas_une_note(self):
        import base64 as b64
        self.assertFalse(wma._detect_multilingual_base64(
            [b64.b64encode("Bienvenue dans notre application".encode()).decode()]
        )["is_ransom_note_like"])

    def test_blob_base64_geant_borne(self):
        import time as t
        t0 = t.perf_counter()
        wma._detect_multilingual_base64(["Q" * 40_000_000])
        self.assertLess(t.perf_counter() - t0, 3.0, "blob geant non borne")

    # === chaines d'outils ===
    def test_detection_langages(self):
        self.assertEqual(wma._detect_language_toolchain(b"xx Go build ID: abc"), "go")
        self.assertEqual(wma._detect_language_toolchain(b"xx cargo/registry yy"), "rust")
        self.assertEqual(wma._detect_language_toolchain(b"a node_modules b"), "electron_node")
        self.assertIsNone(wma._detect_language_toolchain(b"binaire quelconque"))

    def test_depots_filtres_seulement_si_le_langage_lexplique(self):
        """GitHub heberge aussi des charges : ne filtrer que si le binaire est
        manifestement Go, Rust ou Node."""
        go = b"MZ\x00Go build ID: x\x00github.com/evil/pkg\x00"
        autre = b"MZ\x00raw.github.com/evil/payload\x00"
        self.assertNotIn("github.com", " ".join(wma.extract_iocs(go)["domains"]))
        self.assertTrue(any("github.com" in d for d in wma.extract_iocs(autre)["domains"]))

    # === coherence ===
    def test_overlay_non_suspect_non_score(self):
        sans = self._score()["score"]
        avec = self._score(ov={"size": 134714, "entropy": 4.07,
                               "embedded_containers": [], "is_suspicious": False})["score"]
        self.assertEqual(sans, avec, "un overlay juge non suspect ne doit pas etre compte")

    def test_archive_en_overlay_est_un_dropper(self):
        r = self._score(ov={"size": 1010092, "entropy": 8.0,
                            "embedded_containers": ["7zip"], "is_suspicious": True})
        self.assertIn("dropper", " ".join(r["reasons"]))

    def test_artefacts_c2_couverts(self):
        # Categorie applicable a TOUT binaire a imports muets, pas seulement .NET.
        self.assertIn("Network_C2_Artifacts", wma.MUTED_IMPORTS_CAPABILITIES)
        self.assertNotIn("Network_C2_Artifacts", wma.DOTNET_CAPABILITIES)

class TestDernierLot(unittest.TestCase):
    """v8.3 — quatre defauts revelés par le rejeu du lot complet en v8.2."""

    # === artefacts C2 rangés dans la mauvaise table ===
    def test_artefacts_c2_sappliquent_aux_binaires_natifs(self):
        """Un agent natif packe n'expose que son user-agent : la categorie ne
        doit pas dependre de la detection .NET."""
        raw = b"MZ\x00Mozilla/5.0 (Windows NT 6.1)\x00Content-Type: */*\x00demon.x64.exe\x00"
        tables = dict(wma.WIN32_CAPABILITIES)
        tables.update(wma.MUTED_IMPORTS_CAPABILITIES)
        trouvees = wma._capabilities_from_strings(raw, tables)
        self.assertIn("Network_C2_Artifacts", trouvees)

    def test_categorie_c2_a_un_poids(self):
        self.assertIn("Network_C2_Artifacts", wma._CAPABILITY_WEIGHTS)
        self.assertIn("Network_C2_Artifacts", wma._ALL_CAPABILITY_TABLES)

    # === base64 en UTF-16LE (.NET) ===
    def test_note_de_rancon_en_utf16(self):
        """Un assembly .NET stocke ses chaines en UTF-16LE : la recherche sur
        les octets bruts ne voyait aucun blob."""
        import base64 as b64
        notes = ["برایانه خود را خاموش نکنید ممکن است مدتی طول بکشد",
                 "אל תכבה את המחשב פעולה זו עשויה להימשך זמן מה",
                 "आपका पीसी कई बार पुनरारंभ होगा कृपया प्रतीक्षा करें"]
        blob = b"MZ" + b"".join(
            b64.b64encode(n.encode()).decode().encode("utf-16le") + b"\x00\x00" for n in notes)
        res = wma.extract_iocs(blob)["multilingual_base64"]
        self.assertTrue(res["is_ransom_note_like"], res)

    def test_note_de_rancon_en_ascii_toujours_detectee(self):
        import base64 as b64
        notes = ["برایانه خود را خاموش نکنید ممکن است مدتی طول بکشد",
                 "אל תכבה את המחשב פעולה זו עשויה להימשך זמן מה",
                 "आपका पीसी कई बार पुनरारंभ होगा कृपया प्रतीक्षा करें"]
        blob = b"MZ\x00" + b"\x00".join(b64.b64encode(n.encode()) for n in notes)
        self.assertTrue(wma.extract_iocs(blob)["multilingual_base64"]["is_ransom_note_like"])

    # === fenetre de detection du langage ===
    def test_langage_detecte_en_fin_de_gros_fichier(self):
        """Sur une application Electron de 37 Mo, les marqueurs se trouvaient
        au-dela des 8 premiers Mo."""
        gros = b"\x00" * (20 * 1024 * 1024) + b"node_modules/electron"
        self.assertEqual(wma._detect_language_toolchain(gros), "electron_node")

    def test_langage_detecte_en_tete(self):
        self.assertEqual(wma._detect_language_toolchain(b"Go build ID: x" + b"\x00" * 1000), "go")

    # === symboles pointes pris pour des domaines ===
    def test_symboles_go_et_js_ne_sont_pas_des_domaines(self):
        faux = ["cpu.cpuid", "bucket.mpbad", "chacha8rand.block", "addressmspan.sweep",
                "barrierwork.nwait", "3pxjquery.after",
                "ApplyCodeActionCommand.SourceAddMissingImports.command"]
        self.assertEqual(wma._filter_domain_candidates(faux), [])

    def test_vrais_domaines_toujours_conserves(self):
        vrais = ["ca.priceandsons.tech", "ns.adobe.com", "www.w3.org",
                 "schemas.microsoft.com", "test1.playfabapi.com", "gate.onion",
                 "botnet.xyz", "mariano.biz", "c2.duckdns.org", "cdn.jsdelivr.net"]
        self.assertEqual(wma._filter_domain_candidates(vrais), vrais)

    def test_tld_inhabituel_accepte_sil_apparait_dans_une_url(self):
        self.assertEqual(
            wma._filter_domain_candidates(["panel.weirdtld"], ["http://panel.weirdtld/a"]),
            ["panel.weirdtld"])
        self.assertEqual(wma._filter_domain_candidates(["panel.weirdtld"], []), [])


class TestSupportDrivers(unittest.TestCase):
    """v8.4 — reintroduction du support .sys, cette fois etaye par un lot reel
    (voir README) : classification par import kernel-only, capacites noyau,
    recoupement BYOVD."""

    # === classification ===
    def test_import_ntoskrnl_classe_driver_quelle_que_soit_lextension(self):
        """Le signal primaire est l'import, pas l'extension sur disque : un
        driver renomme en .bin reste detecte (sous-systeme NATIVE corrobore
        ici l'import, a defaut du drapeau IMAGE_FILE_DLL)."""
        self.assertTrue(wma._is_kernel_driver(["ntoskrnl.exe", "hal.dll"], ".bin", "NATIVE", False))
        self.assertTrue(wma._is_kernel_driver(["FLTMGR.SYS"], ".dll", "WINDOWS_GUI", True))

    def test_extension_sys_seule_sans_import_kernel_ne_suffit_pas_hors_native(self):
        self.assertFalse(wma._is_kernel_driver(["kernel32.dll"], ".sys", "WINDOWS_GUI", False))

    def test_extension_sys_plus_native_sans_imports_est_un_repli_valide(self):
        """Repli utile quand la table d'imports est absente (driver packe)."""
        self.assertTrue(wma._is_kernel_driver([], ".sys", "NATIVE", False))

    def test_ntdll_seul_ne_suffit_pas_a_conclure_driver(self):
        """Non-regression du piege documente plus haut dans le fichier :
        ntdll.dll est NATIVE tout en etant une DLL usermode."""
        self.assertFalse(wma._is_kernel_driver(["ntdll.dll", "kernel32.dll"], ".dll", "NATIVE", True))

    def test_import_ntoskrnl_seul_sans_corroboration_ne_suffit_plus(self):
        """DURCISSEMENT SAST v8.4 : ce script analyse des octets, il ne charge
        jamais le binaire — un import 'ntoskrnl.exe' FORGE dans un executable
        ordinaire (drapeau IMAGE_FILE_DLL absent, sous-systeme usermode) ne
        suffit plus a se faire classer 'driver'. Sans ce garde-fou, cette
        evasion aurait desactive les regles de score specifiques aux
        executables/DLL (export anormal sur un .exe, DLL sans export...)."""
        self.assertFalse(wma._is_kernel_driver(["ntoskrnl.exe"], ".exe", "WINDOWS_CUI", False))

    def test_vrai_driver_reste_detecte_malgre_le_durcissement(self):
        """Non-regression : un vrai driver porte TOUJOURS IMAGE_FILE_DLL."""
        self.assertTrue(wma._is_kernel_driver(["ntoskrnl.exe", "hal.dll"], ".sys", "NATIVE", True))

    # === capacites noyau : gardees derriere is_driver ===
    def test_capacites_noyau_dans_la_table_globale(self):
        for cap in ("Kernel_Notify_Routine_Tamper", "Kernel_Handle_Callback_Protection",
                    "Kernel_Arbitrary_Memory_Access"):
            self.assertIn(cap, wma.KERNEL_CAPABILITIES)
            self.assertIn(cap, wma._ALL_CAPABILITY_TABLES)
            self.assertIn(cap, wma._CAPABILITY_WEIGHTS)

    # === recoupement BYOVD ===
    def test_pilote_vulnerable_reconnu_via_metadonnees_internes(self):
        """Le nom sur disque est renomme (evasion triviale) ; seul le
        VERSIONINFO interne trahit encore l'identite du pilote."""
        pe_info = {"resources": {"version_info": {"OriginalFilename": "RTCore64.sys"}},
                   "pdb_path": None}
        r = wma._check_known_vulnerable_driver(pe_info, "totalement_innocent.sys")
        self.assertIsNotNone(r)
        self.assertEqual(r["matched_name"], "rtcore64.sys")
        self.assertIn("version_info", r["matched_via"])

    def test_pilote_inconnu_ne_declenche_rien(self):
        pe_info = {"resources": {"version_info": {"OriginalFilename": "monpilote.sys"}},
                   "pdb_path": None}
        self.assertIsNone(wma._check_known_vulnerable_driver(pe_info, "monpilote.sys"))

    def test_byovd_pese_lourd_dans_le_score(self):
        pe = {"behavioral_capabilities": {}, "behavioral_capabilities_inferred": {},
              "pe_type": "driver", "is_dotnet": False, "tls_callbacks": {"count": 0},
              "overlay": {"size": 0}, "authenticode": {"has_signature": False},
              "resources": {}, "checksum": {}, "delay_imported_dlls": [], "packing": {},
              "exports": {"count": 0}, "dll_sideloading": {},
              "known_vulnerable_driver": {"matched_name": "gdrv.sys",
                                          "matched_via": "version_info.OriginalFilename"}}
        r = wma.compute_suspicion_score(
            {"file_metadata": {"is_packed": False}, "pe_structure": pe,
             "iocs": {}, "reverse_engineering": {}})
        self.assertGreaterEqual(r["score"], 35)
        self.assertTrue(any("BYOVD" in raison for raison in r["reasons"]))

    # === IOC des chemins de device NT ===
    def test_ioc_device_path_noyau_et_usermode(self):
        contenu = b"\\Device\\MyDriver\x00\\\\.\\RTCore64\x00"
        trouves = wma.IOC_PATTERNS["nt_device_paths"].findall(contenu.decode("latin-1"))
        self.assertTrue(any("Device" in t for t in trouves))
        self.assertTrue(any(t.startswith("\\\\.\\") for t in trouves))

    # === filet de securite du reverse sur echantillon packe ===
    def test_selection_priorite_complete_par_la_complexite_si_peu_dapi(self):
        fallback = [{"name": "fcn.gros_bloc", "nbbs": 40},
                    {"name": "fcn.petit_bloc", "nbbs": 2},
                    {"name": "entry0", "nbbs": 99}]
        selection = wma._select_priority_functions({}, budget=2, complexity_fallback=fallback)
        self.assertEqual(len(selection), 2)
        noms = [s["function"] for s in selection]
        self.assertIn("fcn.gros_bloc", noms)
        self.assertNotIn("entry0", noms)          # deja decompile a part, jamais reselectionne
        self.assertTrue(all(s["reason"] == "complexity_fallback" for s in selection))

    def test_selection_priorite_api_xref_prioritaire_sur_la_complexite(self):
        counts = {"fcn.suspecte": {"WriteProcessMemory", "VirtualAllocEx"}}
        fallback = [{"name": "fcn.norme", "nbbs": 500}]
        selection = wma._select_priority_functions(counts, budget=1, complexity_fallback=fallback)
        self.assertEqual(selection, [{"function": "fcn.suspecte", "reason": "api_xref"}])


class TestReverseTimeoutHardening(unittest.TestCase):
    """v8.4 — sur un vrai binaire Go du lot de test reel, une fonction de
    550+ blocs basiques (inlining massif du runtime Go, pas une logique
    malveillante dense) faisait tourner `pdg` plus de 90s sans jamais
    aboutir, jusqu'a l'arret force du processus par l'OOM-killer du
    conteneur en conditions batch. `r2ghidra.timeout` — le mecanisme natif
    prevu pour ce role — s'est revele peu fiable en conteneur : une fois
    positionne a une valeur non nulle, il rendait TOUTE decompilation muette,
    y compris celle d'une fonction triviale. Le garde-fou est donc implemente
    cote Python (ReverseSession._run_bounded), independant du moteur."""

    def test_fonction_trop_volumineuse_exclue_du_filet_de_secours(self):
        fallback = [{"name": "fcn.monstre_go", "nbbs": 550},
                    {"name": "fcn.correcte", "nbbs": 120}]
        selection = wma._select_priority_functions({}, budget=1, complexity_fallback=fallback)
        self.assertEqual([s["function"] for s in selection], ["fcn.correcte"])

    def test_run_bounded_tue_le_processus_et_marque_la_session_morte(self):
        import time as _time

        class FakeProcess:
            def __init__(self):
                self.killed = False

            def kill(self):
                self.killed = True

        class SlowR2:
            def __init__(self):
                self.process = FakeProcess()

            def cmd(self, c):
                _time.sleep(2)
                return "trop tard"

        s = wma.ReverseSession.__new__(wma.ReverseSession)
        s.r2 = SlowR2()
        s.errors = []
        s.logger = types.SimpleNamespace(debug=lambda *a, **k: None)
        s._dead = False
        with self.assertRaises(TimeoutError):
            s._run_bounded(lambda: s.r2.cmd("pdg"), timeout_s=0.2)
        self.assertTrue(s.r2.process.killed)
        self.assertTrue(s._dead)

    def test_session_morte_refuse_toute_commande_ulterieure_sans_tenter_le_pipe(self):
        """Une fois la session marquee morte, aucune commande n'est renvoyee
        au processus tue : lire sur un pipe casse pourrait re-bloquer."""
        s = wma.ReverseSession.__new__(wma.ReverseSession)
        s.r2, s._dead = mock.Mock(), True
        with self.assertRaises(RuntimeError):
            s._cmd("i")
        s.r2.cmd.assert_not_called()

    def test_ancienne_construction_de_test_sans_attribut_dead_ne_casse_pas(self):
        """Non-regression : les doublures construites via __new__ sans passer
        par __init__ (comme dans TestCorrectifsV71) n'ont pas cet attribut."""
        class R:
            def cmd(self, c):
                return "ok"

        s = wma.ReverseSession.__new__(wma.ReverseSession)
        s.r2 = R()
        self.assertEqual(s._cmd("i"), "ok")


class TestLotReelV84(unittest.TestCase):
    """v8.4 — deux faux negatifs confirmes sur le lot reel de 26 echantillons
    MalwareBazaar (dont un binaire Go, plusieurs .sys, une DLL Lazarus) :
    malware .NET sans le moindre P/Invoke (score 9/100 "minimal", zero
    capacite detectee) et DLL deguisee en .exe (dispatcheur DllMain visible
    en reverse, score 7/100, aucune regle ne comptait l'anomalie)."""

    def _score(self, **kw):
        pe = {"behavioral_capabilities": kw.get("caps", {}),
              "behavioral_capabilities_inferred": kw.get("inf", {}),
              "pe_type": kw.get("t", "executable"), "is_dotnet": kw.get("net", False),
              "tls_callbacks": {"count": 0}, "overlay": {"size": 0},
              "authenticode": kw.get("sig", {"has_signature": False}),
              "resources": {}, "checksum": {}, "delay_imported_dlls": [],
              "packing": {}, "exports": kw.get("exp", {"count": 0}),
              "dll_sideloading": {}}
        return wma.compute_suspicion_score(
            {"file_metadata": {"is_packed": kw.get("packed", False)},
             "pe_structure": pe, "iocs": kw.get("iocs", {}), "reverse_engineering": {}})

    def test_capacites_dotnet_managees_dans_la_table_globale(self):
        for cap in ("Dotnet_Network_Exfiltration", "Dotnet_Registry_Persistence",
                    "Dotnet_Crypto_Operations", "Dotnet_Anti_Analysis"):
            self.assertIn(cap, wma.DOTNET_MANAGED_CAPABILITIES)
            self.assertIn(cap, wma._ALL_CAPABILITY_TABLES)
            self.assertIn(cap, wma._CAPABILITY_WEIGHTS)

    def test_malware_dotnet_sans_pinvoke_desormais_detecte_dans_les_chaines(self):
        """Reproduit le cas reel : mscoree.dll seul importe, exfiltration et
        chiffrement geres entierement en managed."""
        contenu = ("System.Net.WebClient DownloadString RijndaelManaged "
                   "System.Security.Cryptography.Aes").encode("utf-8")
        trouvees = wma._capabilities_from_strings(contenu, wma.DOTNET_MANAGED_CAPABILITIES)
        self.assertIn("Dotnet_Network_Exfiltration", trouvees)
        self.assertIn("Dotnet_Crypto_Operations", trouvees)

    def test_process_start_seul_nest_pas_dans_la_table_managee(self):
        """Trop courant dans un installeur/lanceur .NET legitime — meme choix
        que BEHAVIORIAL_CONTEXT cote natif : deliberement absent."""
        toutes = {a.lower() for apis in wma.DOTNET_MANAGED_CAPABILITIES.values() for a in apis}
        self.assertNotIn("process.start", toutes)
        self.assertNotIn("processstartinfo".lower(), {a.lower() for a in toutes})

    def test_executable_avec_exports_est_anormal_et_score(self):
        """DLL deguisee en .exe : pe_type reste 'executable' (drapeau IMAGE_FILE_DLL
        absent), mais exporter des fonctions est desormais compte a charge."""
        sans_export = self._score(t="executable", exp={"count": 0})["score"]
        avec_exports = self._score(t="executable", exp={"count": 2})
        self.assertGreater(avec_exports["score"], sans_export)
        self.assertTrue(any("Executable exposant" in r for r in avec_exports["reasons"]))

    def test_boilerplate_xml_dotnet_ecarte_des_iocs(self):
        """Reproduit le cas reel : une appli .NET dont les deux SEULS 'IOC'
        etaient le placeholder DataSet type Visual Studio et l'espace de noms
        XML Schema standard, sans rapport avec le comportement du binaire."""
        contenu = b"http://tempuri.org/DataSet1.xsd http://www.w3.org/2001/XMLSchema\x00"
        r = wma.extract_iocs(contenu)
        self.assertEqual(r["urls"], [])

    def test_domaine_boilerplate_ne_matche_pas_par_sous_chaine(self):
        """DURCISSEMENT SAST v8.4 : un test `domaine in valeur` matcherait
        aussi bien un sous-domaine cree pour l'occasion qu'un vrai C2 deguise
        pour se faire ecarter a tort comme boilerplate."""
        self.assertFalse(wma._is_dotnet_xml_boilerplate("tempuri.org.attacker.net"))
        self.assertFalse(wma._is_dotnet_xml_boilerplate("http://evil-w3.org-c2.net/beacon"))
        self.assertTrue(wma._is_dotnet_xml_boilerplate("http://www.w3.org/2001/XMLSchema"))
        self.assertTrue(wma._is_dotnet_xml_boilerplate("sub.tempuri.org"))

    def test_schemas_microsoft_com_pas_ecarte_comme_boilerplate_xml(self):
        """Non-regression : contrairement a tempuri.org/w3.org, cette exclusion
        reste volontairement etroite."""
        self.assertFalse(wma._is_dotnet_xml_boilerplate("schemas.microsoft.com"))

    def test_dll_normale_avec_exports_non_affectee_par_la_nouvelle_regle(self):
        """Non-regression : la regle ne s'applique qu'aux pe_type='executable',
        pas aux vraies DLL qui exportent normalement."""
        r = self._score(t="dll", exp={"count": 40})
        self.assertFalse(any("Executable exposant" in raison for raison in r["reasons"]))


if __name__ == "__main__":
    unittest.main(verbosity=2)