metasploit-framework/lib/msf/core/thread_manager.rb

# -*- coding: binary -*-
require 'msf/core/plugin'

=begin
require 'active_record'
#
# This monkeypatch can help to diagnose errors involving connection pool
# exhaustion and other strange ActiveRecord including errors like:
#
#   DEPRECATION WARNING: Database connections will not be closed automatically, please close your
#   database connection at the end of the thread by calling `close` on your
#   connection.  For example: ActiveRecord::Base.connection.close
#
# and
#
#   ActiveRecord::StatementInvalid NoMethodError: undefined method `fields' for nil:NilClass: SELECT  "workspaces".* FROM "workspaces"  WHERE "workspaces"."id" = 24 LIMIT 1
#
#
# Based on this code: https://gist.github.com/1364551 linked here:
# http://bibwild.wordpress.com/2011/11/14/multi-threading-in-rails-activerecord-3-0-3-1/
module ActiveRecord
  class Base
    class << self
      def connection
        unless connection_pool.active_connection?
          $stdout.puts("AR::B.connection implicit checkout")
          $stdout.puts(caller.join("\n"))
          raise ImplicitConnectionForbiddenError.new("Implicit ActiveRecord checkout attempted!")
        end
        retrieve_connection
      end
    end
  end
  class ImplicitConnectionForbiddenError < ActiveRecord::ConnectionTimeoutError ; end
end
=end


module Msf

###
#
# This class manages the threads spawned by the framework object, this provides some additional
# features over standard ruby threads.
#
###
class ThreadManager < Array

  include Framework::Offspring

  attr_accessor :monitor

  #
  # Initializes the thread manager.
  #
  def initialize(framework)
    self.framework = framework
    self.monitor   = spawn_monitor
  end

  #
  # Spawns a monitor thread for removing dead threads
  #
  def spawn_monitor
    ::Thread.new do
      begin

      ::Thread.current[:tm_name] = "Thread Monitor"
      ::Thread.current[:tm_crit] = true

      while true
        ::IO.select(nil, nil, nil, 1.0)
        self.each_index do |i|
          state = self[i].alive? rescue false
          self[i] = nil if not state
        end
        self.delete(nil)
      end

      rescue ::Exception => e
        elog("thread monitor: #{e} #{e.backtrace} source:#{self[:tm_call].inspect}")
      end
    end
  end

  #
  # Spawns a new thread
  #
  def spawn(name, crit, *args, &block)
    t = nil

    if block
      t = ::Thread.new(name, crit, caller, block, *args) do |*argv|
        ::Thread.current[:tm_name] = argv.shift.to_s
        ::Thread.current[:tm_crit] = argv.shift
        ::Thread.current[:tm_call] = argv.shift
        ::Thread.current[:tm_time] = Time.now

        begin
          argv.shift.call(*argv)
        rescue ::Exception => e
          elog("thread exception: #{::Thread.current[:tm_name]}  critical=#{::Thread.current[:tm_crit]}  error:#{e.class} #{e} source:#{::Thread.current[:tm_call].inspect}")
          elog("Call Stack\n#{e.backtrace.join("\n")}")
          raise e
        ensure
          if framework.db and framework.db.active
            # NOTE: despite the Deprecation Warning's advice, this should *NOT*
            # be ActiveRecord::Base.connection.close which causes unrelated
            # threads to raise ActiveRecord::StatementInvalid exceptions at
            # some point in the future, presumably due to the pool manager
            # believing that the connection is still usable and handing it out
            # to another thread.
            ::ActiveRecord::Base.connection_pool.release_connection
          end
        end
      end
    else
      t = ::Thread.new(name, crit, caller, *args) do |*argv|
        ::Thread.current[:tm_name] = argv.shift
        ::Thread.current[:tm_crit] = argv.shift
        ::Thread.current[:tm_call] = argv.shift
        ::Thread.current[:tm_time] = Time.now
        # Calling spawn without a block means we cannot force a database
        # connection release when the thread completes, so doing so can
        # potentially use up all database resources and starve all subsequent
        # threads that make use of the database. Log a warning so we can track
        # down this kind of usage.
        dlog("Thread spawned without a block!")
        dlog("Call stack: \n#{::Thread.current[:tm_call].join("\n")}")
      end
    end

    self << t
    t
  end

  #
  # Registers an existing thread
  #
  def register(t, name, crit)
    t[:tm_name] = name
    t[:tm_crit] = crit
    t[:tm_call] = caller
    t[:tm_time] = Time.now
    self << t
    t
  end

  #
  # Updates an existing thread
  #
  def update(ut, name, crit)
    ti = nil
    self.each_index do |i|
      tt = self[i]
      next if not tt
      if ut.__id__ == tt.__id__
        ti = i
        break
      end
    end

    t = self[ti]
    if not t
      raise RuntimeError, "Thread not found"
    end

    t[:tm_name] = name
    t[:tm_crit] = crit
    t
  end

  #
  # Kills a thread by index
  #
  def kill(idx)
    self[idx].kill rescue false
  end

end

end