procbuster.py

#!/usr/bin/env python3
import argparse
import shlex
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

DEFAULT_MAX_PID  = 65535
DEFAULT_WORKERS  = 50
DEFAULT_TIMEOUT  = 5          # seconds per subprocess call
COL_PID          = 7
COL_USER         = 20

def make_cmd_reader(cmd: str, timeout: int):
    """
    Factory: returns a reader that invokes an external command for every path.
    The command receives the remote path as its sole argument.

        $ ./exploit.sh /proc/1/status
    """
    def read_via_cmd(path: str, _timeout: int) -> str | None:
        try:
            parts = shlex.split(cmd) + [path]
            result = subprocess.run(
                parts,
                capture_output=True,
                timeout=timeout,
            )
            # Empty output (for any reason) → file missing / not found.
            # Non-empty → return content as-is.
            if not result.stdout.strip():
                return None
            return result.stdout.decode("utf-8", errors="replace")
        except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
            return None

    return read_via_cmd


def fetch_uid_map(reader, timeout: int) -> dict[str, str]:
    text = reader("/etc/passwd", timeout)
    if not text:
        return {}
    uid_map: dict[str, str] = {}
    for line in text.splitlines():
        # username:password:uid:gid:gecos:home:shell
        parts = line.split(":")
        if len(parts) >= 3:
            uid_map[parts[2].strip()] = parts[0].strip()
    return uid_map


def parse_cmdline(text: str) -> str:
    cmd = text.replace("\x00", " ").replace("\n", " ").strip()
    return cmd if cmd else ""


def probe_pid(pid: int, reader, timeout: int, uid_map: dict[str, str]) -> dict | None:
    cmdline_text = reader(f"/proc/{pid}/cmdline", timeout)
    if not cmdline_text:
        return None

    cmd = parse_cmdline(cmdline_text)

    # Try status for Name + UID; if unavailable user stays "?"
    user = "?"
    status_text = reader(f"/proc/{pid}/status", timeout)
    if status_text:
        fields  = {}
        for line in status_text.splitlines():
            if ":" in line:
                key, _, val = line.partition(":")
                fields[key.strip()] = val.strip()
        name = fields.get("Name", "")
        uid  = fields.get("Uid", "").split()
        user = uid_map.get(uid[0], uid[0]) if uid else "?"
        cmd  = cmd or f"[{name}]"
    else:
        cmd = cmd or "[unknown]"

    return {"pid": pid, "user": user, "cmd": cmd}


def banner() -> None:
    print(
        f"{'PID':<{COL_PID}} {'USER':<{COL_USER}} CMD",
        flush=True,
    )


def print_entry(entry: dict) -> None:
    print(
        f"{entry['pid']:<{COL_PID}} "
        f"{entry['user']:<{COL_USER}} "
        f"{entry['cmd']}",
        flush=True,
    )


def check_reader(reader, timeout: int) -> None:
    result = reader("/proc/self/cmdline", timeout)
    if not result:
        print(
            "[!] file-read-cmd check failed: "
            "no output received for /proc/self/cmdline.\n"
            "    Verify the command works and the target is reachable.",
            file=sys.stderr,
        )
        sys.exit(1)
    cmd = result.replace("\x00", " ").strip()
    print(f"[+] file-read-cmd OK (self: {cmd})", flush=True)


def run(args: argparse.Namespace) -> None:
    reader = make_cmd_reader(args.file_read_cmd, args.timeout)
    check_reader(reader, args.timeout)
    uid_map = fetch_uid_map(reader, args.timeout)

    banner()

    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {
            pool.submit(probe_pid, pid, reader, args.timeout, uid_map): pid
            for pid in range(1, args.max_pid + 1)
        }
        for future in as_completed(futures):
            entry = future.result()
            if entry:
                print_entry(entry)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Lists processes by brute-forcing /proc PIDs and reading "
            "status and cmdline."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "--file-read-cmd",
        metavar="CMD",
        required=True,
        help=(
            "Command used to read remote files. "
            "The target path is appended as the last argument. "
            "Example: ./exploit.sh"
        ),
    )
    parser.add_argument(
        "--max-pid",
        metavar="MAX_PID",
        type=int,
        default=DEFAULT_MAX_PID,
        help=f"Maximum PID to check (default: {DEFAULT_MAX_PID})",
    )
    parser.add_argument(
        "--workers",
        metavar="N",
        type=int,
        default=DEFAULT_WORKERS,
        help=f"Concurrent worker threads (default: {DEFAULT_WORKERS})",
    )
    parser.add_argument(
        "--timeout",
        metavar="S",
        type=int,
        default=DEFAULT_TIMEOUT,
        help=f"Per-request timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    return parser


if __name__ == "__main__":
    parser = build_parser()
    args   = parser.parse_args()
    try:
        run(args)
    except KeyboardInterrupt:
        print("\n[!] Interrupted.", file=sys.stderr)
        sys.exit(1)