diff --git a/lib/msf/base/serializer/readable_text.rb b/lib/msf/base/serializer/readable_text.rb index e21fa35f39d1d..6204a2ee3f7e8 100644 --- a/lib/msf/base/serializer/readable_text.rb +++ b/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" diff --git a/lib/msf/base/sessions/meterpreter.rb b/lib/msf/base/sessions/meterpreter.rb index d072d2ad925e2..705fed92afb0f 100644 --- a/lib/msf/base/sessions/meterpreter.rb +++ b/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 diff --git a/lib/msf/core/session_compatibility.rb b/lib/msf/core/session_compatibility.rb index 147bb3dba1ebe..1adc231ff9722 100644 --- a/lib/msf/core/session_compatibility.rb +++ b/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:') diff --git a/lib/msf/ui/console/command_dispatcher/core.rb b/lib/msf/ui/console/command_dispatcher/core.rb index 16e996968d47c..968113b449141 100644 --- a/lib/msf/ui/console/command_dispatcher/core.rb +++ b/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 --timeout %clr") end diff --git a/lib/rex/post/meterpreter/async_result_store.rb b/lib/rex/post/meterpreter/async_result_store.rb new file mode 100644 index 0000000000000..5a4dd0ef72989 --- /dev/null +++ b/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 diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index 51c94ae12bfd0..ecb43325bdbb8 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/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: diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index b19c784f09634..ca6c1a6da6561 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -537,6 +537,74 @@ def transport_sleep(seconds) return true end + # + # Enable or disable async mode on the implant. + # When enabled, the implant polls at the configured interval + # and only during business hours. + # + # @param opts [Hash] configuration options + # @option opts [Boolean] :enabled enable/disable async mode + # @option opts [Integer] :poll_interval seconds between check-ins + # @option opts [Integer] :jitter jitter percentage (0-99) + # @option opts [Integer] :work_start business hours start (0-23) + # @option opts [Integer] :work_end business hours end (0-23) + # @option opts [Integer] :work_days bitmask of active days (bit0=Sun..bit6=Sat) + # @option opts [Integer] :smart_sync seconds to keep polling rapidly after any + # request/response activity, allowing multi-request commands and post modules + # to complete in a single burst window (0 disables) + # @return [Rex::Post::Meterpreter::Packet] response packet + # + def async_mode(opts = {}) + request = Packet.create_request(COMMAND_ID_CORE_ASYNC_MODE) + request.add_tlv(TLV_TYPE_ASYNC_ENABLED, opts[:enabled]) + request.add_tlv(TLV_TYPE_ASYNC_POLL_INTERVAL, opts[:poll_interval]) if opts[:poll_interval] + request.add_tlv(TLV_TYPE_ASYNC_POLL_JITTER, opts[:jitter]) if opts[:jitter] + request.add_tlv(TLV_TYPE_ASYNC_WORK_START, opts[:work_start]) if opts[:work_start] + request.add_tlv(TLV_TYPE_ASYNC_WORK_END, opts[:work_end]) if opts[:work_end] + request.add_tlv(TLV_TYPE_ASYNC_WORK_DAYS, opts[:work_days]) if opts[:work_days] + request.add_tlv(TLV_TYPE_ASYNC_SMART_SYNC, opts[:smart_sync]) if opts[:smart_sync] + response = client.send_request(request) + client.async_mode_enabled = response.get_tlv_value(TLV_TYPE_ASYNC_ENABLED) + + # Adjust response_timeout to accommodate the poll interval. + # Commands need to wait at least poll_interval + jitter for the implant + # to check in, plus time to execute and respond. + if client.async_mode_enabled + poll = opts[:poll_interval] || 60 + jitter_pct = opts[:jitter] || 0 + # Timeout = 3x worst-case poll interval (poll + max jitter) + worst_case = poll + (poll * jitter_pct / 100) + new_timeout = [worst_case * 3, client.response_timeout].max + @pre_async_response_timeout ||= client.response_timeout + client.response_timeout = new_timeout + + # Install a floor on response_timeout so downstream framework helpers + # (e.g. Msf::Post::Common#cmd_exec) can't silently lower it below the + # async poll window. Without this, cmd_exec's `session.response_timeout + # = time_out` (default 15s) causes every send_request to raise + # Rex::TimeoutError before the target has a chance to check in. + floor = worst_case + 10 + client.instance_variable_set(:@async_timeout_floor, floor) + unless client.singleton_class.instance_methods(false).include?(:response_timeout=) + client.define_singleton_method(:response_timeout=) do |val| + floor_val = instance_variable_get(:@async_timeout_floor).to_i + @response_timeout = [val.to_i, floor_val].max + end + end + elsif defined?(@pre_async_response_timeout) && @pre_async_response_timeout + # Remove the singleton floor before restoring the original timeout, + # otherwise the floor would clamp us back up. + if client.singleton_class.instance_methods(false).include?(:response_timeout=) + client.singleton_class.send(:remove_method, :response_timeout=) + end + client.instance_variable_set(:@async_timeout_floor, nil) + client.response_timeout = @pre_async_response_timeout + @pre_async_response_timeout = nil + end + + response + end + # # Change the active transport to the next one in the transport list. # @@ -746,7 +814,22 @@ def shutdown # otherwise the session may not receive the command before we # kill the handler. This could be improved by the server side # sending a reply to shutdown first. - self.client.send_packet_wait_response(request, 10) + # + # When async mode is enabled, the target only checks in every + # poll_interval seconds, so a fixed 10s wait would tear down the + # handler before the implant ever sees the shutdown packet - + # leaving an orphan payload that reconnects on next msf launch. + # Scale the wait to cover at least one worst-case poll window + # (interval + jitter) plus a small buffer for the C side to react. + wait = 10 + if client.respond_to?(:async_mode_enabled?) && client.async_mode_enabled? + cfg = client.async_config + poll = cfg[:poll_interval].to_i + jitter_pct = cfg[:jitter].to_i + worst_case = poll + (poll * jitter_pct / 100) + wait = [worst_case + 10, wait].max + end + self.client.send_packet_wait_response(request, wait) else # If this is a standard TCP session, send and forget. self.client.send_packet(request) diff --git a/lib/rex/post/meterpreter/core_ids.rb b/lib/rex/post/meterpreter/core_ids.rb index a7113ba872838..18adab62c071b 100644 --- a/lib/rex/post/meterpreter/core_ids.rb +++ b/lib/rex/post/meterpreter/core_ids.rb @@ -46,6 +46,7 @@ module Meterpreter COMMAND_ID_CORE_TRANSPORT_SET_TIMEOUTS = EXTENSION_ID_CORE + 32 COMMAND_ID_CORE_TRANSPORT_SLEEP = EXTENSION_ID_CORE + 33 COMMAND_ID_CORE_PIVOT_SESSION_NEW = EXTENSION_ID_CORE + 34 +COMMAND_ID_CORE_ASYNC_MODE = EXTENSION_ID_CORE + 35 end end diff --git a/lib/rex/post/meterpreter/packet.rb b/lib/rex/post/meterpreter/packet.rb index a3823579da95a..06b2f50966c78 100644 --- a/lib/rex/post/meterpreter/packet.rb +++ b/lib/rex/post/meterpreter/packet.rb @@ -121,6 +121,17 @@ module Meterpreter TLV_TYPE_PIVOT_STAGE_DATA = TLV_META_TYPE_RAW | 651 TLV_TYPE_PIVOT_NAMED_PIPE_NAME = TLV_META_TYPE_STRING | 653 +# +# Async mode +# +TLV_TYPE_ASYNC_ENABLED = TLV_META_TYPE_BOOL | 700 +TLV_TYPE_ASYNC_POLL_INTERVAL = TLV_META_TYPE_UINT | 701 +TLV_TYPE_ASYNC_POLL_JITTER = TLV_META_TYPE_UINT | 702 +TLV_TYPE_ASYNC_WORK_START = TLV_META_TYPE_UINT | 703 +TLV_TYPE_ASYNC_WORK_END = TLV_META_TYPE_UINT | 704 +TLV_TYPE_ASYNC_WORK_DAYS = TLV_META_TYPE_UINT | 705 +TLV_TYPE_ASYNC_SMART_SYNC = TLV_META_TYPE_UINT | 706 + # # Core flags diff --git a/lib/rex/post/meterpreter/packet_dispatcher.rb b/lib/rex/post/meterpreter/packet_dispatcher.rb index 37f8d01abce15..70bb5f090d95f 100644 --- a/lib/rex/post/meterpreter/packet_dispatcher.rb +++ b/lib/rex/post/meterpreter/packet_dispatcher.rb @@ -2,6 +2,7 @@ require 'rex/post/meterpreter/command_mapper' require 'rex/post/meterpreter/packet_response_waiter' +require 'rex/post/meterpreter/async_result_store' require 'rex/exceptions' require 'pathname' @@ -84,6 +85,15 @@ def initialize_passive_dispatcher self.alive = true end + # + # Returns the async result store, creating it if needed. + # + # @return [AsyncResultStore] + # + def async_store + @async_store ||= AsyncResultStore.new + end + def shutdown_passive_dispatcher self.alive = false self.send_queue = [] diff --git a/lib/rex/post/meterpreter/ui/console.rb b/lib/rex/post/meterpreter/ui/console.rb index f8c9d3fd927f0..216ef4d75cb28 100644 --- a/lib/rex/post/meterpreter/ui/console.rb +++ b/lib/rex/post/meterpreter/ui/console.rb @@ -93,11 +93,27 @@ def queue_cmd(cmd) self.commands << cmd end + # Commands that are allowed when async mode is enabled. + # Everything else is blocked since direct commands would block + # on the slow poll interval. + ASYNC_ALLOWED_COMMANDS = %w[ + background bg exit quit help + async + ].freeze + # # Runs the specified command wrapper in something to catch meterpreter # exceptions. # def run_command(dispatcher, method, arguments) + # In async mode, only allow async-related and session management commands. + # All others must be run via 'async run '. + # @async_bypass is set by async_subcmd_run to allow dispatched commands through. + if client.async_mode_enabled? && !@async_bypass && !ASYNC_ALLOWED_COMMANDS.include?(method) + log_error("Cannot run '#{method}' directly in async mode. Use 'async run #{method}' or 'async mode off' first.") + return + end + begin super rescue Exception => e @@ -127,7 +143,7 @@ def log_error(msg) elog(msg, 'meterpreter') - dlog("Call stack:\n#{$@.join("\n")}", 'meterpreter') + dlog("Call stack:\n#{$@&.join("\n")}", 'meterpreter') if $@ end attr_reader :client # :nodoc: diff --git a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb index f507abb91b08f..5ab7342a2c459 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -1,5 +1,6 @@ # -*- coding: binary -*- require 'set' +require 'securerandom' require 'rex/post/meterpreter' require 'rex' @@ -77,7 +78,9 @@ def commands 'transport' => 'Manage the transport mechanisms', 'get_timeouts' => 'Get the current session timeout values', 'set_timeouts' => 'Set the current session timeout values', - 'ssl_verify' => 'Modify the SSL certificate verification setting' + 'ssl_verify' => 'Modify the SSL certificate verification setting', + # async mode commands + 'async' => 'Manage async polling mode (mode, config, run, queue)', } if msf_loaded? @@ -107,6 +110,8 @@ def commands ], 'get_timeouts' => [COMMAND_ID_CORE_TRANSPORT_SET_TIMEOUTS], 'set_timeouts' => [COMMAND_ID_CORE_TRANSPORT_SET_TIMEOUTS], + # async mode + 'async' => [COMMAND_ID_CORE_ASYNC_MODE], } # XXX: Remove this line once the payloads gem has had another major version bump from 2.x to 3.x and @@ -731,6 +736,376 @@ def cmd_sleep(*args) end end + # + # Display help for async command. + # + def cmd_async_help + print(<<~HELP + Usage: async [options] + + Manage async polling mode for HTTP transport. + + Subcommands: + mode [on|off] Toggle async mode on/off, or show current status + config [options] Configure polling interval and business hours + run Enqueue a command for async execution + queue [rid] View queued commands or a specific result + queue -c Clear completed results + + Config options: + -i Poll interval in seconds (default: 60) + -j Jitter percentage 0-99 (default: 0) + -s Work hours start 0-23 (default: 0) + -e Work hours end 0-23 (default: 24) + -d Work days: sun,mon,tue,wed,thu,fri,sat or mon-fri (default: all) + -y Smart-sync burst window: seconds to keep polling rapidly + after any request/response activity, then fall back to the + normal interval. 0 disables (default: 0) + + Examples: + async mode on + async config -i 300 -j 20 -s 8 -e 17 -d mon-fri + async config -i 600 -y 30 + async run ls + async run execute -f cmd.exe -a "/c whoami" -H + async queue + async queue a1b2c3d4 + + HELP + ) + end + + DAY_NAMES = { 'sun' => 0, 'mon' => 1, 'tue' => 2, 'wed' => 3, 'thu' => 4, 'fri' => 5, 'sat' => 6 }.freeze + + # + # Parse a day specification into a bitmask. + # + def parse_work_days(spec) + return spec.to_i if spec =~ /\A0x[0-9a-f]+\z/i || spec =~ /\A\d+\z/ + + if spec == 'mon-fri' + return 0x3E # bits 1-5 + elsif spec == 'all' + return 0x7F + end + + mask = 0 + spec.split(',').each do |day| + day = day.strip.downcase[0..2] + bit = DAY_NAMES[day] + mask |= (1 << bit) if bit + end + mask + end + + # + # Format a work days bitmask into a human-readable string. + # + def format_work_days(mask) + return 'all' if mask == 0x7F + return 'mon-fri' if mask == 0x3E + + names = DAY_NAMES.sort_by { |_, v| v }.select { |_, v| (mask & (1 << v)) != 0 }.map(&:first) + names.empty? ? 'none' : names.join(', ') + end + + # + # Handle the async command with subcommands. + # + def cmd_async(*args) + if args.empty? || args.include?('-h') + cmd_async_help + return + end + + subcmd = args.shift + case subcmd + when 'mode' + async_subcmd_mode(args) + when 'config' + async_subcmd_config(args) + when 'run' + async_subcmd_run(args) + when 'queue' + async_subcmd_queue(args) + else + cmd_async_help + end + end + + # + # Tab completion for async. + # + def cmd_async_tabs(str, words) + if words.length == 1 + %w[mode config run queue].select { |o| o.start_with?(str) } + elsif words.length == 2 && words[0] == 'mode' + %w[on off].select { |o| o.start_with?(str) } + else + [] + end + end + + # + # async mode [on|off] + # + def async_subcmd_mode(args) + subcmd = args.shift + if subcmd.nil? + if client.async_mode_enabled? + print_good('Async mode: enabled') + else + print_status('Async mode: disabled') + end + return + end + + case subcmd + when 'on' + cfg = client.async_config + print_status("Enabling async mode (poll #{cfg[:poll_interval]}s, jitter #{cfg[:jitter]}%)...") + client.core.async_mode(enabled: true, **cfg) + print_good('Async mode enabled. Use "async run " to enqueue commands.') + print_warning('Channels, port forwards, interactive shell, and post modules are unavailable in async mode.') + if cfg[:work_start] != 0 || cfg[:work_end] != 24 + print_warning("Business hours #{cfg[:work_start]}:00-#{cfg[:work_end]}:00 use the TARGET's local time.") + end + when 'off' + print_status('Disabling async mode...') + client.core.async_mode(enabled: false) + client.async_store.stop_worker + print_good('Async mode disabled. Session is now interactive.') + else + print_error("Unknown mode: #{subcmd}. Use 'on' or 'off'.") + end + end + + # + # async config -i -j -s -e -d + # + def async_subcmd_config(args) + cfg = client.async_config + + if args.empty? + # Dump current config values + print(<<~CONFIG + + Async Configuration: + Poll interval : #{cfg[:poll_interval]}s + Jitter : #{cfg[:jitter]}% + Work hours : #{cfg[:work_start]}:00 - #{cfg[:work_end]}:00 + Work days : #{format_work_days(cfg[:work_days])} + Smart-sync : #{cfg[:smart_sync].to_i > 0 ? "#{cfg[:smart_sync]}s burst window" : 'disabled'} + Mode : #{client.async_mode_enabled? ? 'enabled' : 'disabled'} + + CONFIG + ) + return + end + + opts = Rex::Parser::Arguments.new( + '-i' => [true, 'Poll interval (seconds)'], + '-j' => [true, 'Jitter percent (0-99)'], + '-s' => [true, 'Work start hour (0-23)'], + '-e' => [true, 'Work end hour (0-23)'], + '-d' => [true, 'Work days'], + '-y' => [true, 'Smart-sync burst window (seconds, 0 disables)'] + ) + opts.parse(args) do |opt, _idx, val| + case opt + when '-i' + cfg[:poll_interval] = val.to_i + when '-j' + cfg[:jitter] = val.to_i + when '-s' + cfg[:work_start] = val.to_i + when '-e' + cfg[:work_end] = val.to_i + when '-d' + cfg[:work_days] = parse_work_days(val) + when '-y' + cfg[:smart_sync] = val.to_i + end + end + + smart_sync_note = cfg[:smart_sync].to_i > 0 ? ", smart-sync #{cfg[:smart_sync]}s" : '' + print_good("Async configuration updated (poll #{cfg[:poll_interval]}s, jitter #{cfg[:jitter]}%, hours #{cfg[:work_start]}:00-#{cfg[:work_end]}:00#{smart_sync_note}).") + if client.async_mode_enabled? + print_status('Async mode is active. Sending updated config to target...') + client.core.async_mode(enabled: true, **cfg) + print_good('Config applied to active session.') + else + print_status('Config saved locally. Use "async mode on" to activate.') + end + end + + # + # async run + # + def async_subcmd_run(args) + if args.empty? + print_error('Usage: async run [arguments]') + return + end + + # The dispatcher shell strips quotes when parsing the command line before + # passing us *args. Rejoin so the async shell can re-parse it correctly. + # Wrap in double quotes only when strictly necessary (whitespace or embedded + # quotes); avoid Shellwords.shelljoin which aggressively backslash-escapes + # characters like '=' and '/' that MSF option parsing needs to see raw + # (e.g. CMD=whoami must stay CMD=whoami, not CMD\=whoami). + cmd_line = args.map do |arg| + if arg =~ /[\s"']/ + %("#{arg.gsub('"', '\\"')}") + else + arg + end + end.join(' ') + rid = SecureRandom.hex(16) + + # Capture the output handle now so completion notifications go to the + # operator's console (not the async shell's buffer). + notify_output = shell.output + main_shell = shell + + client.async_store.enqueue_work(rid, cmd_line) do |work_rid| + async_shell = client.async_shell(main_shell) + # Drain any leftover output from previous runs on this shell so the + # captured output for this rid is fresh. + async_pipe = async_shell.instance_variable_get(:@async_pipe) + async_pipe.read_subscriber('async') if async_pipe + + # Post modules invoke Msf::SessionCompatibility#setup which calls + # @session.init_ui(user_input, user_output). session.init_ui cascades + # to session.console.init_ui(...) which would clobber the operator's + # main console readline input on the interactive thread (crashing + # get_input_line with "undefined method 'pgets' for nil"). Redirect + # session.console to our async_shell for the duration of the run, so + # the compat setup lands on our private shell and never touches + # main_shell. Also snapshot user_input/user_output so we can restore + # them cleanly afterwards. + swap_console = client.respond_to?(:console=) && client.console.equal?(main_shell) + saved_console = swap_console ? client.console : nil + saved_user_input = client.respond_to?(:user_input) ? client.user_input : nil + saved_user_output = client.respond_to?(:user_output) ? client.user_output : nil + client.console = async_shell if swap_console + + # Signal to Msf::SessionCompatibility that post modules dispatched from + # this worker thread are allowed to run against the async session. + ::Thread.current[:msf_async_bypass_post] = true + begin + async_shell.run_single(cmd_line) + captured = async_pipe ? async_pipe.read_subscriber('async') : '' + client.async_store.complete(work_rid, nil, captured.empty? ? '(no output)' : captured) + ensure + ::Thread.current[:msf_async_bypass_post] = nil + begin + client.console = saved_console if swap_console && saved_console + if client.respond_to?(:init_ui) && (saved_user_input || saved_user_output) + client.init_ui(saved_user_input, saved_user_output) + end + rescue ::Exception + # Best-effort restoration; don't mask the original error + end + begin + notify_output.print_good("Async result ready: #{cmd_line} (rid: #{work_rid[0..7]}). Use 'async queue #{work_rid[0..7]}' to view.") + # rb-readline captures $stdout while blocked in readline(). Force a + # display refresh so the notification appears immediately instead of + # waiting for the next user input. + if defined?(::RbReadline) && ::RbReadline.respond_to?(:rl_forced_update_display) + ::RbReadline.rl_forced_update_display + end + rescue ::Exception + # Notification delivery may fail if the session is closing + end + end + end + + print_status("Queued: #{cmd_line} (rid: #{rid[0..7]})") + end + + # + # async queue [rid] [-c] + # + def async_subcmd_queue(args) + store = client.async_store + + if args.include?('-c') + cleared = store.clear_completed + print_good("Cleared #{cleared} completed result(s).") + return + end + + # If a specific rid is given, show its output + if args.length > 0 && !args[0].start_with?('-') + rid = args[0] + entry = store.fetch(rid) + if entry.nil? + # Try partial match + all = store.all + matches = all.keys.select { |k| k.start_with?(rid) } + if matches.length == 1 + rid = matches.first + entry = store.fetch(rid) + elsif matches.length > 1 + print_error("Ambiguous rid '#{rid}' matches #{matches.length} entries.") + return + else + print_error("No result found for rid '#{rid}'.") + return + end + end + + print_line("Command: #{entry[:label]}") + print_line("Status: #{entry[:status]}") + print_line("Queued: #{entry[:queued_at]}") + if entry[:completed_at] + elapsed = entry[:completed_at] - entry[:queued_at] + print_line("Done: #{entry[:completed_at]} (#{elapsed.round(1)}s)") + end + print_line + if entry[:output] + print_line(entry[:output]) + elsif entry[:response] + print_line(entry[:response].inspect) + else + print_status('No output captured.') + end + return + end + + # Show summary table + results = store.all + if results.empty? + print_status('No async commands queued.') + return + end + + tbl = Rex::Text::Table.new( + 'Header' => 'Async Command Queue', + 'Indent' => 2, + 'Columns' => ['RID (short)', 'Command', 'Status', 'Age'] + ) + + results.each do |rid, entry| + # For running entries show elapsed-since-start (how long this item has + # been executing) so operators can distinguish a slow-but-progressing + # command from a hung one. For everything else show elapsed-since-queued. + reference = entry[:status] == Rex::Post::Meterpreter::AsyncResultStore::STATUS_RUNNING && entry[:started_at] ? entry[:started_at] : entry[:queued_at] + age = ::Time.now - reference + age_str = if age < 60 + "#{age.round(0)}s" + elsif age < 3600 + "#{(age / 60).round(0)}m" + else + "#{(age / 3600).round(1)}h" + end + tbl << [rid[0..7], entry[:label] || '(unknown)', entry[:status].to_s, age_str] + end + + print_line(tbl.to_s) + end + # # Arguments for transport switching #