Async Mode Meterpreter

lib/msf/base/serializer/readable_text.rb

@@ -864,6 +864,10 @@ def self.create_msf_session_row(session, show_extended)
       row[-1] << " #{session.platform}"
     end
 
+    if session.respond_to?(:async_mode_enabled?) && session.async_mode_enabled?
+      row[-1] << " (async)"
+    end
+
     if show_extended
       if session.respond_to?(:last_checkin) && session.last_checkin
         row << "#{(Time.now.to_i - session.last_checkin.to_i)}s ago"

lib/msf/base/sessions/meterpreter.rb

@@ -97,6 +97,17 @@ def initialize(rstream, opts={})
 
   def exit
     begin
+      # If async mode is active, warn the operator that shutdown will
+      # block until the implant next polls (up to poll_interval + jitter),
+      # and stop the async worker so no queued work is left dangling.
+      if respond_to?(:async_mode_enabled?) && async_mode_enabled?
+        cfg = async_config
+        poll = cfg[:poll_interval].to_i
+        jitter_pct = cfg[:jitter].to_i
+        worst_case = poll + (poll * jitter_pct / 100)
+        print_status("Async mode is on - waiting up to #{worst_case + 10}s for the implant's next check-in to deliver shutdown...")
+        async_store.stop_worker if respond_to?(:async_store)
+      end
       self.core.shutdown
     rescue StandardError
       nil

lib/msf/core/session_compatibility.rb

@@ -40,6 +40,17 @@ def setup
       # for its platform, capabilities, etc.
       check_for_session_readiness if session.type == "meterpreter"
 
+      # Block post modules from running against sessions in async mode.
+      # Async mode uses long polling intervals making multi-step post modules
+      # impractical or broken (each send_request blocks for a full poll cycle).
+      # Exception: when dispatched from an async worker thread (via 'async run
+      # run post/...'), the module is allowed to execute since it's already
+      # off the interactive shell and smart-sync makes multi-request chains
+      # feasible.
+      if session.type == 'meterpreter' && session.respond_to?(:async_mode_enabled?) && session.async_mode_enabled? && !::Thread.current[:msf_async_bypass_post]
+        raise Msf::ValidationError, "Session #{session.sid} is in async mode. Post modules cannot run against async sessions. Use 'async_mode off' in the session first, or dispatch via 'async run run post/...'."
+      end
+
       incompatibility_reasons = session_incompatibility_reasons(session)
       if incompatibility_reasons.any?
         print_warning('SESSION may not be compatible with this module:')

lib/msf/ui/console/command_dispatcher/core.rb

@@ -1753,6 +1753,11 @@ def cmd_sessions(*args)
         if session
           if session.respond_to?(:response_timeout)
             last_known_timeout = session.response_timeout
+            # Don't lower the timeout if the session is in async mode
+            # async sessions need longer timeouts to accommodate poll intervals.
+            if session.respond_to?(:async_mode_enabled?) && session.async_mode_enabled?
+              response_timeout = [response_timeout, last_known_timeout].max
+            end
             session.response_timeout = response_timeout
             session.on_run_command_error_proc = log_on_timeout_error("Send timed out. Timeout currently #{session.response_timeout} seconds, you can configure this with %grnsessions --interact <id> --timeout <value>%clr")
           end

lib/rex/post/meterpreter/async_result_store.rb

@@ -0,0 +1,256 @@
+# -*- coding: binary -*-
+
+require 'rex/thread_factory'
+
+module Rex
+module Post
+module Meterpreter
+
+###
+#
+# Thread-safe store for tracking asynchronously dispatched commands
+# and their results. Used when async mode is enabled to allow
+# queuing multiple commands without blocking the console.
+#
+###
+class AsyncResultStore
+
+  # Entry states
+  STATUS_PENDING   = :pending
+  STATUS_RUNNING   = :running
+  STATUS_COMPLETE  = :complete
+  STATUS_ERROR     = :error
+
+  def initialize
+    @results = {}
+    @mutex = ::Mutex.new
+    @work_queue = ::Queue.new
+    @worker = nil
+    @worker_mutex = ::Mutex.new
+  end
+
+  #
+  # Enqueue a unit of work to be executed serially by the worker thread.
+  # The worker is started lazily on first enqueue. The provided block is
+  # invoked in the worker with (rid, label) and is responsible for calling
+  # {#complete} or {#error} when done.
+  #
+  # @param rid [String] the request ID
+  # @param label [String] human-readable command label
+  # @yieldparam rid [String]
+  # @yieldparam label [String]
+  # @return [void]
+  #
+  def enqueue_work(rid, label, &executor)
+    queue(rid, label)
+    ensure_worker_started
+    @work_queue.push([rid, label, executor])
+  end
+
+  #
+  # Ensure the worker thread is running.
+  #
+  # @return [void]
+  #
+  def ensure_worker_started
+    @worker_mutex.synchronize do
+      return if @worker && @worker.alive?
+
+      @worker = Rex::ThreadFactory.spawn('AsyncCommandWorker', false) do
+        loop do
+          item = @work_queue.pop
+          break if item == :stop
+
+          rid, label, executor = item
+          short = rid[0..7]
+          started = ::Time.now
+          begin
+            dlog("async worker: picked up #{short} (#{label.inspect})", 'meterpreter/async')
+            mark_running(rid)
+            executor.call(rid)
+            elapsed = (::Time.now - started).round(1)
+            dlog("async worker: completed #{short} in #{elapsed}s", 'meterpreter/async')
+          rescue ::Exception => e
+            elapsed = (::Time.now - started).round(1)
+            elog("async worker: #{short} raised #{e.class} after #{elapsed}s: #{e.message}", 'meterpreter/async', error: e)
+            error(rid, "#{e.class}: #{e.message}")
+          end
+        end
+      end
+    end
+  end
+
+  #
+  # Signal the worker to stop after draining its current item.
+  # Safe to call even if the worker was never started.
+  #
+  # @return [void]
+  #
+  def stop_worker
+    @worker_mutex.synchronize do
+      return unless @worker && @worker.alive?
+
+      @work_queue.push(:stop)
+      @worker.join(5)
+      @worker = nil
+    end
+  end
+
+  #
+  # Register a command as pending delivery.
+  #
+  # @param rid [String] the request ID
+  # @param label [String] human-readable command label (e.g. "ls /tmp")
+  # @return [void]
+  #
+  def queue(rid, label)
+    @mutex.synchronize do
+      @results[rid] = {
+        label: label,
+        status: STATUS_PENDING,
+        queued_at: ::Time.now,
+        started_at: nil,
+        completed_at: nil,
+        response: nil,
+        output: nil
+      }
+    end
+  end
+
+  #
+  # Mark a queued command as currently running (worker picked it up).
+  #
+  # @param rid [String] the request ID
+  # @return [void]
+  #
+  def mark_running(rid)
+    @mutex.synchronize do
+      return unless @results.key?(rid)
+
+      @results[rid][:status] = STATUS_RUNNING
+      @results[rid][:started_at] = ::Time.now
+    end
+  end
+
+  #
+  # Mark a command as complete with its response.
+  #
+  # @param rid [String] the request ID
+  # @param response [Rex::Post::Meterpreter::Packet, nil] the response packet
+  # @param output [String, nil] captured console output
+  # @return [void]
+  #
+  def complete(rid, response, output = nil)
+    @mutex.synchronize do
+      return unless @results.key?(rid)
+
+      @results[rid][:status] = STATUS_COMPLETE
+      @results[rid][:completed_at] = ::Time.now
+      @results[rid][:response] = response
+      @results[rid][:output] = output
+    end
+  end
+
+  #
+  # Mark a command as errored.
+  #
+  # @param rid [String] the request ID
+  # @param error_message [String] the error description
+  # @return [void]
+  #
+  def error(rid, error_message)
+    @mutex.synchronize do
+      return unless @results.key?(rid)
+
+      @results[rid][:status] = STATUS_ERROR
+      @results[rid][:completed_at] = ::Time.now
+      @results[rid][:output] = error_message
+    end
+  end
+
+  #
+  # Return all pending entries.
+  #
+  # @return [Hash] rid => entry hash
+  #
+  def pending
+    @mutex.synchronize do
+      @results.select { |_rid, entry| entry[:status] == STATUS_PENDING }
+    end
+  end
+
+  #
+  # Return all completed entries.
+  #
+  # @return [Hash] rid => entry hash
+  #
+  def completed
+    @mutex.synchronize do
+      @results.select { |_rid, entry| entry[:status] == STATUS_COMPLETE }
+    end
+  end
+
+  #
+  # Return all entries regardless of status.
+  #
+  # @return [Hash] rid => entry hash
+  #
+  def all
+    @mutex.synchronize do
+      @results.dup
+    end
+  end
+
+  #
+  # Fetch a specific result by rid.
+  #
+  # @param rid [String] the request ID
+  # @return [Hash, nil] the entry or nil if not found
+  #
+  def fetch(rid)
+    @mutex.synchronize do
+      @results[rid]&.dup
+    end
+  end
+
+  #
+  # Remove a specific entry.
+  #
+  # @param rid [String] the request ID
+  # @return [void]
+  #
+  def delete(rid)
+    @mutex.synchronize do
+      @results.delete(rid)
+    end
+  end
+
+  #
+  # Clear all completed and errored entries.
+  #
+  # @return [Integer] number of entries cleared
+  #
+  def clear_completed
+    @mutex.synchronize do
+      before = @results.size
+      @results.reject! { |_rid, entry| entry[:status] != STATUS_PENDING }
+      before - @results.size
+    end
+  end
+
+  #
+  # Return the total number of tracked entries.
+  #
+  # @return [Integer]
+  #
+  def size
+    @mutex.synchronize do
+      @results.size
+    end
+  end
+
+end
+
+end
+end
+end

lib/rex/post/meterpreter/client.rb

@@ -503,6 +503,72 @@ def unicode_filter_decode(str)
   # Whether or not to use a debug build for loaded extensions
   #
   attr_accessor :debug_build
+  #
+  # Whether async mode is currently enabled on this session
+  #
+  attr_accessor :async_mode_enabled
+
+  # Locally stored async config values (applied when async mode is enabled)
+  def async_config
+    @async_config ||= { poll_interval: 60, jitter: 0, work_start: 0, work_end: 24, work_days: 0x7F, smart_sync: 0 }
+  end
+
+  def async_mode_enabled?
+    !!self.async_mode_enabled
+  end
+
+  #
+  # A dedicated console used by the async worker thread to execute queued
+  # commands. It has its own output buffer and dispatcher stack so nothing
+  # it does can race with the operator's interactive shell. Rebuild it if
+  # the main shell's dispatcher stack (extensions) has changed since the
+  # last call.
+  #
+  # @param main_shell [Rex::Post::Meterpreter::Ui::Console] the interactive
+  #   shell whose dispatcher stack should be mirrored.
+  # @return [Rex::Post::Meterpreter::Ui::Console]
+  #
+  def async_shell(main_shell)
+    core_klass = Rex::Post::Meterpreter::Ui::Console::CommandDispatcher::Core
+    main_core = main_shell.dispatcher_stack.find { |d| d.is_a?(core_klass) }
+    main_extensions = main_core ? main_core.instance_variable_get(:@extensions).dup : []
+
+    if @async_shell.nil? || @async_shell_extensions != main_extensions
+      shell = Rex::Post::Meterpreter::Ui::Console.new(self)
+
+      # Wire the async shell to a BidirectionalPipe for both input and output.
+      # A single Pipe object serves both roles (borrowed from Msf::Ui::Web),
+      # which means:
+      #   - modules that expect a non-nil user_input (e.g. reading via gets)
+      #     get a valid IO instead of nil
+      #   - the same pipe collects everything the module or command prints
+      #     via a named subscriber ("async"), which we drain per run
+      # This is safer than a bare Output::Buffer + nil input, especially when
+      # Msf::SessionCompatibility#setup calls session.init_ui(input, output)
+      # during post-module execution.
+      pipe = Rex::Ui::Text::BidirectionalPipe.new
+      # Msf module runners (e.g. cmd_run's run_simple path) check
+      # `LocalOutput.prompting?` before writing status. BidirectionalPipe
+      # doesn't define it - WebConsole subclasses to add it. Add it here
+      # as a singleton method to avoid defining a whole subclass.
+      pipe.define_singleton_method(:prompting?) { false }
+      pipe.create_subscriber('async')
+      shell.init_ui(pipe, pipe)
+      shell.instance_variable_set(:@async_bypass, true)
+      shell.instance_variable_set(:@async_pipe, pipe)
+
+      # Re-load each extension on the async shell using the same code path as
+      # the main shell. Each extension class's initialize enstacks any child
+      # dispatchers itself (e.g. Stdapi enstacks Fs, Net, Sys, ...), so we
+      # must NOT enstack children directly or we get duplicates.
+      async_core = shell.dispatcher_stack.find { |d| d.is_a?(core_klass) }
+      main_extensions.each { |mod| async_core.send(:add_extension_client, mod) }
+
+      @async_shell = shell
+      @async_shell_extensions = main_extensions
+    end
+    @async_shell
+  end
 
 protected
   attr_accessor :parser, :ext_aliases # :nodoc: