From 2a9faa9385153bd749e7ee1db476c4d218ae3bfa Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:05:09 -0400 Subject: [PATCH 01/30] feat(meterpreter): add COMMAND_ID_CORE_ASYNC_MODE command identifier --- lib/rex/post/meterpreter/core_ids.rb | 1 + 1 file changed, 1 insertion(+) 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 From 96d2b51b581ac10f8dac9505880b2d46958ec8d2 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:05:22 -0400 Subject: [PATCH 02/30] feat(meterpreter): add TLV types for async mode configuration --- lib/rex/post/meterpreter/packet.rb | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/lib/rex/post/meterpreter/packet.rb b/lib/rex/post/meterpreter/packet.rb index a3823579da95a..78a4b7b3d6d1c 100644 --- a/lib/rex/post/meterpreter/packet.rb +++ b/lib/rex/post/meterpreter/packet.rb @@ -121,6 +121,16 @@ 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 + # # Core flags From 4dd791be44766ea6128d210b9b321a8a0bf91757 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:05:32 -0400 Subject: [PATCH 03/30] feat(meterpreter): add AsyncResultStore for tracking async command results --- .../post/meterpreter/async_result_store.rb | 167 ++++++++++++++++++ 1 file changed, 167 insertions(+) create mode 100644 lib/rex/post/meterpreter/async_result_store.rb 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..30180d6583c1f --- /dev/null +++ b/lib/rex/post/meterpreter/async_result_store.rb @@ -0,0 +1,167 @@ +# -*- coding: binary -*- + +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_COMPLETE = :complete + STATUS_ERROR = :error + + def initialize + @results = {} + @mutex = ::Mutex.new + 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, + completed_at: nil, + response: nil, + output: nil + } + 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 From 078660f68af40956a068a02fc23ecab559e030a8 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:05:40 -0400 Subject: [PATCH 04/30] feat(meterpreter): add async mode state and config accessors to client --- lib/rex/post/meterpreter/client.rb | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index 51c94ae12bfd0..d8d1173f2aa0d 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/lib/rex/post/meterpreter/client.rb @@ -503,6 +503,19 @@ 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 } + end + + def async_mode_enabled? + !!self.async_mode_enabled + end protected attr_accessor :parser, :ext_aliases # :nodoc: From b7e0ae507fccf9802b5952f9b4fb2930d51cf5aa Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:05:50 -0400 Subject: [PATCH 05/30] feat(meterpreter): add send_request_async for non-blocking command dispatch --- lib/rex/post/meterpreter/packet_dispatcher.rb | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/lib/rex/post/meterpreter/packet_dispatcher.rb b/lib/rex/post/meterpreter/packet_dispatcher.rb index 37f8d01abce15..2737163964722 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,43 @@ 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 + + # + # Sends a request asynchronously without blocking. The response will be + # captured via a completion_routine callback and stored in the async_store. + # + # @param packet [Packet] the request packet to send + # @param label [String] human-readable label for the command + # @return [String] the request ID (rid) for later retrieval + # + def send_request_async(packet, label: nil) + rid = packet.rid + async_store.queue(rid, label) + + send_packet(packet, + completion_routine: Proc.new { |response, param| + if response && response.result == 0 + async_store.complete(param[:rid], response) + elsif response + einfo = lookup_error(response.result) + async_store.error(param[:rid], einfo) + else + async_store.error(param[:rid], 'No response received') + end + }, + completion_param: { rid: rid } + ) + rid + end + def shutdown_passive_dispatcher self.alive = false self.send_queue = [] From c4bcaf4b0fcbf16c6199be82ab4f3e3f95e894a0 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:06:00 -0400 Subject: [PATCH 06/30] feat(meterpreter): implement async_mode method in ClientCore --- lib/rex/post/meterpreter/client_core.rb | 44 +++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index b19c784f09634..be90499968a51 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -537,6 +537,50 @@ 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) + # @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] + 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 = 3× 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 + elsif defined?(@pre_async_response_timeout) && @pre_async_response_timeout + 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. # From f0fd34c34a922f4d6fffa8185bdf4ddde3584fd1 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:06:13 -0400 Subject: [PATCH 07/30] feat(meterpreter): enforce command restrictions when async mode is active --- lib/rex/post/meterpreter/ui/console.rb | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/lib/rex/post/meterpreter/ui/console.rb b/lib/rex/post/meterpreter/ui/console.rb index f8c9d3fd927f0..810039ca2bf59 100644 --- a/lib/rex/post/meterpreter/ui/console.rb +++ b/lib/rex/post/meterpreter/ui/console.rb @@ -93,11 +93,26 @@ 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 '. + if client.async_mode_enabled? && !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 +142,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: From eb82a9e615cb74fda82b8260bde05ce6a45dee8f Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:06:24 -0400 Subject: [PATCH 08/30] feat(meterpreter): add async command with mode, config, run, and queue subcommands --- .../ui/console/command_dispatcher/core.rb | 302 +++++++++++++++++- 1 file changed, 301 insertions(+), 1 deletion(-) 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..f944c355d9f77 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,301 @@ def cmd_sleep(*args) end end + # + # Display help for async command. + # + def cmd_async_help + print_line('Usage: async [options]') + print_line + print_line('Manage async polling mode for HTTP transport.') + print_line + print_line('Subcommands:') + print_line(' mode [on|off] Toggle async mode on/off, or show current status') + print_line(' config [options] Configure polling interval and business hours') + print_line(' run Enqueue a command for async execution') + print_line(' queue [rid] View queued commands or a specific result') + print_line(' queue -c Clear completed results') + print_line + print_line('Config options:') + print_line(' -i Poll interval in seconds (default: 60)') + print_line(' -j Jitter percentage 0-99 (default: 0)') + print_line(' -s Work hours start 0-23 (default: 0)') + print_line(' -e Work hours end 0-23 (default: 24)') + print_line(' -d Work days: sun,mon,tue,wed,thu,fri,sat or mon-fri (default: all)') + print_line + print_line('Examples:') + print_line(' async mode on') + print_line(' async config -i 300 -j 20 -s 8 -e 17 -d mon-fri') + print_line(' async run ls') + print_line(' async run execute -f cmd.exe -a "/c whoami" -H') + print_line(' async queue') + print_line(' async queue a1b2c3d4') + print_line + 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) + 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_line + print_line("Async Configuration:") + print_line(" Poll interval : #{cfg[:poll_interval]}s") + print_line(" Jitter : #{cfg[:jitter]}%") + print_line(" Work hours : #{cfg[:work_start]}:00 - #{cfg[:work_end]}:00") + print_line(" Work days : #{format_work_days(cfg[:work_days])}") + print_line(" Mode : #{client.async_mode_enabled? ? 'enabled' : 'disabled'}") + print_line + 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'] + ) + 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) + end + end + + print_good("Async configuration updated (poll #{cfg[:poll_interval]}s, jitter #{cfg[:jitter]}%, hours #{cfg[:work_start]}:00-#{cfg[:work_end]}:00).") + 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 + + cmd_line = args.join(' ') + rid = SecureRandom.hex(16) + client.async_store.queue(rid, cmd_line) + + Rex::ThreadFactory.spawn("AsyncCmd-#{rid[0..7]}", false) do + output_buf = +'' + original_print_proc = shell.on_print_proc + shell.on_print_proc = proc { |msg| output_buf << msg.to_s } + begin + shell.run_single(cmd_line) + client.async_store.complete(rid, nil, output_buf.empty? ? '(no output)' : output_buf) + print_status("Async command completed: #{cmd_line} (rid: #{rid[0..7]})") + rescue ::Exception => e + client.async_store.error(rid, "#{e.class}: #{e.message}") + print_error("Async command failed: #{cmd_line} - #{e.message}") + ensure + shell.on_print_proc = original_print_proc + 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]) + 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| + age = ::Time.now - entry[:queued_at] + 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 # From f835ace8cdb4fa994fcb2633716278c683b9344e Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:06:52 -0400 Subject: [PATCH 09/30] feat(sessions): display async mode indicator in session listing --- lib/msf/base/serializer/readable_text.rb | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/lib/msf/base/serializer/readable_text.rb b/lib/msf/base/serializer/readable_text.rb index e21fa35f39d1d..94b660c62a341 100644 --- a/lib/msf/base/serializer/readable_text.rb +++ b/lib/msf/base/serializer/readable_text.rb @@ -864,6 +864,14 @@ 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 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" From 67309a87d91a54d1bf6cfee33bddd5b94e9dab39 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:07:04 -0400 Subject: [PATCH 10/30] feat(sessions): block post modules from running against async sessions --- lib/msf/core/session_compatibility.rb | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/lib/msf/core/session_compatibility.rb b/lib/msf/core/session_compatibility.rb index 147bb3dba1ebe..d43dc0d647112 100644 --- a/lib/msf/core/session_compatibility.rb +++ b/lib/msf/core/session_compatibility.rb @@ -40,6 +40,13 @@ 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). + if session.type == 'meterpreter' && session.respond_to?(:async_mode_enabled?) && session.async_mode_enabled? + 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." + end + incompatibility_reasons = session_incompatibility_reasons(session) if incompatibility_reasons.any? print_warning('SESSION may not be compatible with this module:') From b1b9761d6b1762f835629cde172c4bce9b5b02fd Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:07:13 -0400 Subject: [PATCH 11/30] fix(sessions): preserve async timeout when interacting with async sessions --- lib/msf/ui/console/command_dispatcher/core.rb | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/lib/msf/ui/console/command_dispatcher/core.rb b/lib/msf/ui/console/command_dispatcher/core.rb index 16e996968d47c..4594dd0a460cf 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 From 8472f682509b56803d5942044cdf9ae0637c8aaf Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 16:21:50 -0400 Subject: [PATCH 12/30] perf(meterpreter): consolidate async output into single print calls --- .../ui/console/command_dispatcher/core.rb | 74 ++++++++++--------- 1 file changed, 40 insertions(+), 34 deletions(-) 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 f944c355d9f77..9ceab8ce67217 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -740,32 +740,35 @@ def cmd_sleep(*args) # Display help for async command. # def cmd_async_help - print_line('Usage: async [options]') - print_line - print_line('Manage async polling mode for HTTP transport.') - print_line - print_line('Subcommands:') - print_line(' mode [on|off] Toggle async mode on/off, or show current status') - print_line(' config [options] Configure polling interval and business hours') - print_line(' run Enqueue a command for async execution') - print_line(' queue [rid] View queued commands or a specific result') - print_line(' queue -c Clear completed results') - print_line - print_line('Config options:') - print_line(' -i Poll interval in seconds (default: 60)') - print_line(' -j Jitter percentage 0-99 (default: 0)') - print_line(' -s Work hours start 0-23 (default: 0)') - print_line(' -e Work hours end 0-23 (default: 24)') - print_line(' -d Work days: sun,mon,tue,wed,thu,fri,sat or mon-fri (default: all)') - print_line - print_line('Examples:') - print_line(' async mode on') - print_line(' async config -i 300 -j 20 -s 8 -e 17 -d mon-fri') - print_line(' async run ls') - print_line(' async run execute -f cmd.exe -a "/c whoami" -H') - print_line(' async queue') - print_line(' async queue a1b2c3d4') - print_line + 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) + + Examples: + async mode on + async config -i 300 -j 20 -s 8 -e 17 -d mon-fri + 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 @@ -880,14 +883,17 @@ def async_subcmd_config(args) if args.empty? # Dump current config values - print_line - print_line("Async Configuration:") - print_line(" Poll interval : #{cfg[:poll_interval]}s") - print_line(" Jitter : #{cfg[:jitter]}%") - print_line(" Work hours : #{cfg[:work_start]}:00 - #{cfg[:work_end]}:00") - print_line(" Work days : #{format_work_days(cfg[:work_days])}") - print_line(" Mode : #{client.async_mode_enabled? ? 'enabled' : 'disabled'}") - print_line + 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])} + Mode : #{client.async_mode_enabled? ? 'enabled' : 'disabled'} + + CONFIG + ) return end From 2bfd601157d755a536a1a55f56fc76b77b076c50 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 17:14:52 -0400 Subject: [PATCH 13/30] fix(sessions): remove duplicate async indicator in session listing --- lib/msf/base/serializer/readable_text.rb | 4 ---- 1 file changed, 4 deletions(-) diff --git a/lib/msf/base/serializer/readable_text.rb b/lib/msf/base/serializer/readable_text.rb index 94b660c62a341..6204a2ee3f7e8 100644 --- a/lib/msf/base/serializer/readable_text.rb +++ b/lib/msf/base/serializer/readable_text.rb @@ -868,10 +868,6 @@ def self.create_msf_session_row(session, show_extended) row[-1] << " (async)" 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" From 7cd11d5df6583c41f0f62159ab9e5cd63e8b7345 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 17 Jul 2026 17:15:04 -0400 Subject: [PATCH 14/30] fix(meterpreter): allow async run to bypass command restriction check --- lib/rex/post/meterpreter/ui/console.rb | 3 ++- lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/lib/rex/post/meterpreter/ui/console.rb b/lib/rex/post/meterpreter/ui/console.rb index 810039ca2bf59..216ef4d75cb28 100644 --- a/lib/rex/post/meterpreter/ui/console.rb +++ b/lib/rex/post/meterpreter/ui/console.rb @@ -108,7 +108,8 @@ def queue_cmd(cmd) 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 '. - if client.async_mode_enabled? && !ASYNC_ALLOWED_COMMANDS.include?(method) + # @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 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 9ceab8ce67217..ee9e18c27ed79 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -947,6 +947,7 @@ def async_subcmd_run(args) original_print_proc = shell.on_print_proc shell.on_print_proc = proc { |msg| output_buf << msg.to_s } begin + shell.instance_variable_set(:@async_bypass, true) shell.run_single(cmd_line) client.async_store.complete(rid, nil, output_buf.empty? ? '(no output)' : output_buf) print_status("Async command completed: #{cmd_line} (rid: #{rid[0..7]})") @@ -954,6 +955,7 @@ def async_subcmd_run(args) client.async_store.error(rid, "#{e.class}: #{e.message}") print_error("Async command failed: #{cmd_line} - #{e.message}") ensure + shell.instance_variable_set(:@async_bypass, false) shell.on_print_proc = original_print_proc end end From fc8df895cfbf50b111321c7308a0f5b2f81cbc08 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 05:55:09 -0400 Subject: [PATCH 15/30] refactor(meterpreter): remove unused send_request_async helper --- lib/rex/post/meterpreter/packet_dispatcher.rb | 28 ------------------- 1 file changed, 28 deletions(-) diff --git a/lib/rex/post/meterpreter/packet_dispatcher.rb b/lib/rex/post/meterpreter/packet_dispatcher.rb index 2737163964722..70bb5f090d95f 100644 --- a/lib/rex/post/meterpreter/packet_dispatcher.rb +++ b/lib/rex/post/meterpreter/packet_dispatcher.rb @@ -94,34 +94,6 @@ def async_store @async_store ||= AsyncResultStore.new end - # - # Sends a request asynchronously without blocking. The response will be - # captured via a completion_routine callback and stored in the async_store. - # - # @param packet [Packet] the request packet to send - # @param label [String] human-readable label for the command - # @return [String] the request ID (rid) for later retrieval - # - def send_request_async(packet, label: nil) - rid = packet.rid - async_store.queue(rid, label) - - send_packet(packet, - completion_routine: Proc.new { |response, param| - if response && response.result == 0 - async_store.complete(param[:rid], response) - elsif response - einfo = lookup_error(response.result) - async_store.error(param[:rid], einfo) - else - async_store.error(param[:rid], 'No response received') - end - }, - completion_param: { rid: rid } - ) - rid - end - def shutdown_passive_dispatcher self.alive = false self.send_queue = [] From c57455a65b5893831b334de181b7bc01f448938b Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 05:55:21 -0400 Subject: [PATCH 16/30] feat(meterpreter/async): add worker thread and work queue to AsyncResultStore --- .../post/meterpreter/async_result_store.rb | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/lib/rex/post/meterpreter/async_result_store.rb b/lib/rex/post/meterpreter/async_result_store.rb index 30180d6583c1f..a976d331f445d 100644 --- a/lib/rex/post/meterpreter/async_result_store.rb +++ b/lib/rex/post/meterpreter/async_result_store.rb @@ -1,5 +1,7 @@ # -*- coding: binary -*- +require 'rex/thread_factory' + module Rex module Post module Meterpreter @@ -21,6 +23,68 @@ class AsyncResultStore 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 + begin + executor.call(rid) + rescue ::Exception => 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 # From 40f1b45dcf8d634937cf9b6c58b17866ccb99fcc Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 05:55:36 -0400 Subject: [PATCH 17/30] feat(meterpreter/async): add dedicated async_shell factory on client --- lib/rex/post/meterpreter/client.rb | 34 ++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index d8d1173f2aa0d..256abe0d5b8ed 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/lib/rex/post/meterpreter/client.rb @@ -517,6 +517,40 @@ 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) + shell.init_ui(nil, Rex::Ui::Text::Output::Buffer.new) + shell.instance_variable_set(:@async_bypass, true) + + # 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: attr_writer :ext, :sock # :nodoc: From 54e81725ec55cf8edff5f95709004472873982af Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 05:55:50 -0400 Subject: [PATCH 18/30] feat(meterpreter/async): route async run through worker queue and dedicated shell --- .../ui/console/command_dispatcher/core.rb | 39 ++++++++++++------- 1 file changed, 25 insertions(+), 14 deletions(-) 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 ee9e18c27ed79..ebc5fca60926a 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -869,6 +869,7 @@ def async_subcmd_mode(args) 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'.") @@ -940,23 +941,31 @@ def async_subcmd_run(args) cmd_line = args.join(' ') rid = SecureRandom.hex(16) - client.async_store.queue(rid, cmd_line) - Rex::ThreadFactory.spawn("AsyncCmd-#{rid[0..7]}", false) do - output_buf = +'' - original_print_proc = shell.on_print_proc - shell.on_print_proc = proc { |msg| output_buf << msg.to_s } + # 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) + async_shell.output.reset begin - shell.instance_variable_set(:@async_bypass, true) - shell.run_single(cmd_line) - client.async_store.complete(rid, nil, output_buf.empty? ? '(no output)' : output_buf) - print_status("Async command completed: #{cmd_line} (rid: #{rid[0..7]})") - rescue ::Exception => e - client.async_store.error(rid, "#{e.class}: #{e.message}") - print_error("Async command failed: #{cmd_line} - #{e.message}") + async_shell.run_single(cmd_line) + captured = async_shell.output.dump_buffer + client.async_store.complete(work_rid, nil, captured.empty? ? '(no output)' : captured) ensure - shell.instance_variable_set(:@async_bypass, false) - shell.on_print_proc = original_print_proc + 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 @@ -1005,6 +1014,8 @@ def async_subcmd_queue(args) print_line if entry[:output] print_line(entry[:output]) + elsif entry[:response] + print_line(entry[:response].inspect) else print_status('No output captured.') end From 1b38b17e1e5f19e35df8798965fb01818b187286 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 08:18:19 -0400 Subject: [PATCH 19/30] feat(meterpreter/async): define TLV_TYPE_ASYNC_SMART_SYNC --- lib/rex/post/meterpreter/packet.rb | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/rex/post/meterpreter/packet.rb b/lib/rex/post/meterpreter/packet.rb index 78a4b7b3d6d1c..06b2f50966c78 100644 --- a/lib/rex/post/meterpreter/packet.rb +++ b/lib/rex/post/meterpreter/packet.rb @@ -130,6 +130,7 @@ module Meterpreter 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 # From 0102d5a78f448f86b78c9e94a57f17459045b021 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 08:18:42 -0400 Subject: [PATCH 20/30] feat(meterpreter/async): default smart_sync to 0 in async_config --- lib/rex/post/meterpreter/client.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index 256abe0d5b8ed..65bb423bd17cc 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/lib/rex/post/meterpreter/client.rb @@ -510,7 +510,7 @@ def unicode_filter_decode(str) # 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 } + @async_config ||= { poll_interval: 60, jitter: 0, work_start: 0, work_end: 24, work_days: 0x7F, smart_sync: 0 } end def async_mode_enabled? From e27e77f3389284c557bf3470aec8f787971219a4 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 08:19:03 -0400 Subject: [PATCH 21/30] feat(meterpreter/async): send smart_sync TLV in core_async_mode request --- lib/rex/post/meterpreter/client_core.rb | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index be90499968a51..dfc56bf41be9d 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -549,6 +549,9 @@ def transport_sleep(seconds) # @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 = {}) @@ -559,6 +562,7 @@ def async_mode(opts = {}) 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) From 85e958f7c70009314415a12de1ffbe76f6daa46d Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 08:19:33 -0400 Subject: [PATCH 22/30] feat(meterpreter/async): expose smart-sync burst window via 'async config -y' --- .../ui/console/command_dispatcher/core.rb | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) 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 ebc5fca60926a..368dc0b9b3c27 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -758,10 +758,14 @@ def cmd_async_help -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 @@ -891,6 +895,7 @@ def async_subcmd_config(args) 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 @@ -903,7 +908,8 @@ def async_subcmd_config(args) '-j' => [true, 'Jitter percent (0-99)'], '-s' => [true, 'Work start hour (0-23)'], '-e' => [true, 'Work end hour (0-23)'], - '-d' => [true, 'Work days'] + '-d' => [true, 'Work days'], + '-y' => [true, 'Smart-sync burst window (seconds, 0 disables)'] ) opts.parse(args) do |opt, _idx, val| case opt @@ -917,10 +923,13 @@ def async_subcmd_config(args) 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 - 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 = 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) From 1daf91adb0ddb86757a4bb413922cccbbbc31c79 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 09:00:32 -0400 Subject: [PATCH 23/30] fix(meterpreter/async): scale shutdown wait to poll interval --- lib/rex/post/meterpreter/client_core.rb | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index dfc56bf41be9d..ad025b6f87d54 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -794,7 +794,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) From edcd938b351a258a2695941589e8eec689ce5216 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 09:00:45 -0400 Subject: [PATCH 24/30] fix(meterpreter/async): warn and stop async worker on session exit --- lib/msf/base/sessions/meterpreter.rb | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/lib/msf/base/sessions/meterpreter.rb b/lib/msf/base/sessions/meterpreter.rb index d072d2ad925e2..4a43acbf96494 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 From 4760d358bc32436d85042296ee73c0383da7438f Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 10:18:28 -0400 Subject: [PATCH 25/30] feat(meterpreter/async): dispatch post modules through dedicated async shell --- lib/msf/core/session_compatibility.rb | 8 ++- lib/rex/post/meterpreter/client.rb | 21 +++++++- .../ui/console/command_dispatcher/core.rb | 49 +++++++++++++++++-- 3 files changed, 72 insertions(+), 6 deletions(-) diff --git a/lib/msf/core/session_compatibility.rb b/lib/msf/core/session_compatibility.rb index d43dc0d647112..1adc231ff9722 100644 --- a/lib/msf/core/session_compatibility.rb +++ b/lib/msf/core/session_compatibility.rb @@ -43,8 +43,12 @@ def setup # 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). - if session.type == 'meterpreter' && session.respond_to?(:async_mode_enabled?) && session.async_mode_enabled? - 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." + # 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) diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index 65bb423bd17cc..ecb43325bdbb8 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/lib/rex/post/meterpreter/client.rb @@ -535,8 +535,27 @@ def async_shell(main_shell) if @async_shell.nil? || @async_shell_extensions != main_extensions shell = Rex::Post::Meterpreter::Ui::Console.new(self) - shell.init_ui(nil, Rex::Ui::Text::Output::Buffer.new) + + # 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 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 368dc0b9b3c27..65b630dae3fcd 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -948,7 +948,19 @@ def async_subcmd_run(args) return end - cmd_line = args.join(' ') + # 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 @@ -958,12 +970,43 @@ def async_subcmd_run(args) client.async_store.enqueue_work(rid, cmd_line) do |work_rid| async_shell = client.async_shell(main_shell) - async_shell.output.reset + # 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_shell.output.dump_buffer + 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 From 8292c36e306485277406c4b48019cf08a6e734a3 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 10:19:17 -0400 Subject: [PATCH 26/30] feat(meterpreter/async): worker debug logging and running-state visibility --- .../post/meterpreter/async_result_store.rb | 27 ++++++++++++++++++- .../ui/console/command_dispatcher/core.rb | 6 ++++- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/lib/rex/post/meterpreter/async_result_store.rb b/lib/rex/post/meterpreter/async_result_store.rb index a976d331f445d..5a4dd0ef72989 100644 --- a/lib/rex/post/meterpreter/async_result_store.rb +++ b/lib/rex/post/meterpreter/async_result_store.rb @@ -17,6 +17,7 @@ class AsyncResultStore # Entry states STATUS_PENDING = :pending + STATUS_RUNNING = :running STATUS_COMPLETE = :complete STATUS_ERROR = :error @@ -60,10 +61,18 @@ def ensure_worker_started item = @work_queue.pop break if item == :stop - rid, _label, executor = item + 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 @@ -100,6 +109,7 @@ def queue(rid, label) label: label, status: STATUS_PENDING, queued_at: ::Time.now, + started_at: nil, completed_at: nil, response: nil, output: nil @@ -107,6 +117,21 @@ def queue(rid, label) 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. # 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 65b630dae3fcd..5ab7342a2c459 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -1088,7 +1088,11 @@ def async_subcmd_queue(args) ) results.each do |rid, entry| - age = ::Time.now - entry[:queued_at] + # 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 From 057fe06ba60b4dd23058299b566541fc4cea4abc Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Mon, 20 Jul 2026 10:55:48 -0400 Subject: [PATCH 27/30] fix(meterpreter/async): floor response_timeout in async mode to survive cmd_exec --- lib/rex/post/meterpreter/client_core.rb | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index ad025b6f87d54..5fd441e9f7a9e 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -577,7 +577,27 @@ def async_mode(opts = {}) 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 From fe46da5a8b3940ead16830eadab93cf77e2907ef Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Thu, 23 Jul 2026 12:11:39 +0200 Subject: [PATCH 28/30] fix: msftidy fix --- lib/msf/base/sessions/meterpreter.rb | 2 +- lib/msf/ui/console/command_dispatcher/core.rb | 2 +- lib/rex/post/meterpreter/client_core.rb | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/msf/base/sessions/meterpreter.rb b/lib/msf/base/sessions/meterpreter.rb index 4a43acbf96494..705fed92afb0f 100644 --- a/lib/msf/base/sessions/meterpreter.rb +++ b/lib/msf/base/sessions/meterpreter.rb @@ -105,7 +105,7 @@ def exit 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...") + 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 diff --git a/lib/msf/ui/console/command_dispatcher/core.rb b/lib/msf/ui/console/command_dispatcher/core.rb index 4594dd0a460cf..968113b449141 100644 --- a/lib/msf/ui/console/command_dispatcher/core.rb +++ b/lib/msf/ui/console/command_dispatcher/core.rb @@ -1753,7 +1753,7 @@ 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 — + # 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 diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index 5fd441e9f7a9e..ca6c1a6da6561 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -572,7 +572,7 @@ def async_mode(opts = {}) if client.async_mode_enabled poll = opts[:poll_interval] || 60 jitter_pct = opts[:jitter] || 0 - # Timeout = 3× worst-case poll interval (poll + max jitter) + # 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 From 24a7780264579dde4618a62232de462d6160f149 Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Tue, 28 Jul 2026 12:57:48 +0200 Subject: [PATCH 29/30] feat(meterpreter/async): add checks for async mode transport requirements and command restrictions --- .../ui/console/command_dispatcher/core.rb | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) 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 8562bbad3b7ec..22e36930fefca 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -862,6 +862,11 @@ def async_subcmd_mode(args) case subcmd when 'on' + unless client.passive_service + print_error('Async mode requires an HTTP(S) transport. Current transport does not support async polling.') + print_status('Use "transport add" to add an HTTP(S) transport and switch to it before enabling async mode.') + return + end cfg = client.async_config print_status("Enabling async mode (poll #{cfg[:poll_interval]}s, jitter #{cfg[:jitter]}%)...") client.core.async_mode(enabled: true, **cfg) @@ -939,6 +944,13 @@ def async_subcmd_config(args) end end + # Commands that cannot be executed via 'async run' because they require + # an interactive channel or persistent socket that the async poll model + # cannot service. + ASYNC_RUN_BLOCKED_COMMANDS = %w[ + shell interact channel portfwd rportfwd + ].freeze + # # async run # @@ -948,6 +960,12 @@ def async_subcmd_run(args) return end + blocked = ASYNC_RUN_BLOCKED_COMMANDS.include?(args.first) + if blocked + print_error("Command '#{args.first}' cannot be run in async mode; it requires an interactive channel.") + 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 @@ -1185,6 +1203,16 @@ def cmd_transport(*args) return end + # Guardrail: transport switching is not safe while async mode is active. + # Changing/removing the current transport tears down the session and would + # strand queued async work. 'list' is read-only and always allowed; 'add' + # is allowed so operators can stage a new transport before disabling async. + if client.async_mode_enabled? && !%w[list add].include?(command) + print_error("Transport '#{command}' is not permitted while async mode is enabled.") + print_status("Disable async mode first with 'async mode off'.") + return + end + opts = { :uuid => client.payload_uuid, :transport => nil, From f8465e73b4cce6a949dcba6db990900c7ac3297d Mon Sep 17 00:00:00 2001 From: dledda-r7 Date: Fri, 28 Aug 2026 21:26:20 +0200 Subject: [PATCH 30/30] fix(async/metsrv): re-design async with TTL instead of smart-sync, some other edge case fix --- lib/msf/base/sessions/meterpreter.rb | 19 ++- lib/msf/core/session_compatibility.rb | 11 +- .../post/meterpreter/async_result_store.rb | 134 ++++++++++++---- lib/rex/post/meterpreter/async_window.rb | 76 +++++++++ lib/rex/post/meterpreter/client.rb | 144 +++++++++++++++++- lib/rex/post/meterpreter/client_core.rb | 113 +++++++------- lib/rex/post/meterpreter/core_ids.rb | 2 + lib/rex/post/meterpreter/packet.rb | 11 +- lib/rex/post/meterpreter/packet_dispatcher.rb | 35 ++++- lib/rex/post/meterpreter/ui/console.rb | 28 +++- .../ui/console/command_dispatcher/core.rb | 105 +++++++++---- .../meterpreter/async_result_store_spec.rb | 61 ++++++++ .../rex/post/meterpreter/async_window_spec.rb | 65 ++++++++ 13 files changed, 663 insertions(+), 141 deletions(-) create mode 100644 lib/rex/post/meterpreter/async_window.rb create mode 100644 spec/lib/rex/post/meterpreter/async_result_store_spec.rb create mode 100644 spec/lib/rex/post/meterpreter/async_window_spec.rb diff --git a/lib/msf/base/sessions/meterpreter.rb b/lib/msf/base/sessions/meterpreter.rb index a69a15aa6209b..3533dad6a61c0 100644 --- a/lib/msf/base/sessions/meterpreter.rb +++ b/lib/msf/base/sessions/meterpreter.rb @@ -97,15 +97,10 @@ 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. + # Stop the async worker so no queued work is left dangling, then wait + # for shutdown delivery at the next configured target check-in. 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...") + print_status("Async mode is on - waiting for the implant's next configured check-in to deliver shutdown...") async_store.stop_worker if respond_to?(:async_store) end self.core.shutdown @@ -521,6 +516,14 @@ def load_session_info begin ::Timeout.timeout(60) do update_session_info + begin + sample = core.get_target_time + self.target_time_sample = sample if sample + rescue ::Rex::Post::Meterpreter::RequestError => e + dlog("Payload does not support get_target_time: #{e.message}") + rescue ::StandardError => e + dlog("get_target_time sampling failed: #{e.message}") + end hobj = nil diff --git a/lib/msf/core/session_compatibility.rb b/lib/msf/core/session_compatibility.rb index 0e176b422105d..440b096b4fe8a 100644 --- a/lib/msf/core/session_compatibility.rb +++ b/lib/msf/core/session_compatibility.rb @@ -40,15 +40,10 @@ 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. + # Route Post modules through the private async worker so they run under + # an explicit rapid-poll lease without touching the interactive console. 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/...'." + raise Msf::ValidationError, "Session #{session.sid} is in async mode. Dispatch the Post module via 'async run run post/...' or disable async mode first." end incompatibility_reasons = session_incompatibility_reasons(session) diff --git a/lib/rex/post/meterpreter/async_result_store.rb b/lib/rex/post/meterpreter/async_result_store.rb index 5a4dd0ef72989..4cb2cad121125 100644 --- a/lib/rex/post/meterpreter/async_result_store.rb +++ b/lib/rex/post/meterpreter/async_result_store.rb @@ -1,4 +1,5 @@ # -*- coding: binary -*- +# frozen_string_literal: true require 'rex/thread_factory' @@ -20,6 +21,7 @@ class AsyncResultStore STATUS_RUNNING = :running STATUS_COMPLETE = :complete STATUS_ERROR = :error + STATUS_CANCELLED = :cancelled def initialize @results = {} @@ -42,9 +44,13 @@ def initialize # @return [void] # def enqueue_work(rid, label, &executor) - queue(rid, label) - ensure_worker_started - @work_queue.push([rid, label, executor]) + @worker_mutex.synchronize do + raise 'Async worker is stopping' if @stopping + + queue(rid, label) + ensure_worker_started_locked + @work_queue.push([rid, label, executor]) + end end # @@ -54,45 +60,55 @@ def enqueue_work(rid, label, &executor) # 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 + ensure_worker_started_locked end end # - # Signal the worker to stop after draining its current item. + # Cancel pending work and stop the worker. # Safe to call even if the worker was never started. # # @return [void] # def stop_worker + worker = nil @worker_mutex.synchronize do - return unless @worker && @worker.alive? + return unless @worker + + unless @worker.alive? + @worker = nil + return + end + + @stopping = true + worker = @worker + + begin + loop do + item = @work_queue.pop(true) + cancel(item[0], 'Cancelled before execution') unless item == :stop + end + rescue ::ThreadError + nil + end @work_queue.push(:stop) - @worker.join(5) + end + + cancel_running('Cancelled while running') + + unless worker.equal?(::Thread.current) + worker.join(5) + if worker.alive? + worker[:msf_async_forced_stop] = true + worker.kill + worker.join + end + end + + @worker_mutex.synchronize do @worker = nil + @stopping = false end end @@ -143,6 +159,7 @@ def mark_running(rid) def complete(rid, response, output = nil) @mutex.synchronize do return unless @results.key?(rid) + return if @results[rid][:status] == STATUS_CANCELLED @results[rid][:status] = STATUS_COMPLETE @results[rid][:completed_at] = ::Time.now @@ -161,6 +178,7 @@ def complete(rid, response, output = nil) def error(rid, error_message) @mutex.synchronize do return unless @results.key?(rid) + return if @results[rid][:status] == STATUS_CANCELLED @results[rid][:status] = STATUS_ERROR @results[rid][:completed_at] = ::Time.now @@ -168,6 +186,16 @@ def error(rid, error_message) end end + def cancel(rid, message = 'Cancelled') + @mutex.synchronize do + return unless @results.key?(rid) + + @results[rid][:status] = STATUS_CANCELLED + @results[rid][:completed_at] = ::Time.now + @results[rid][:output] = message + end + end + # # Return all pending entries. # @@ -233,7 +261,8 @@ def delete(rid) def clear_completed @mutex.synchronize do before = @results.size - @results.reject! { |_rid, entry| entry[:status] != STATUS_PENDING } + terminal = [STATUS_COMPLETE, STATUS_ERROR, STATUS_CANCELLED] + @results.reject! { |_rid, entry| terminal.include?(entry[:status]) } before - @results.size end end @@ -249,6 +278,51 @@ def size end end + private + + def ensure_worker_started_locked + 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 + execute = @worker_mutex.synchronize do + if @stopping + cancel(rid, 'Cancelled before execution') + false + else + mark_running(rid) + true + end + end + next unless execute + + short = rid[0..7] + started = ::Time.now + begin + dlog("async worker: picked up #{short} (#{label.inspect})", 'meterpreter/async') + executor.call(rid) + elapsed = (::Time.now - started).round(1) + dlog("async worker: completed #{short} in #{elapsed}s", 'meterpreter/async') + rescue ::StandardError => 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 + + def cancel_running(message) + running_ids = @mutex.synchronize do + @results.select { |_rid, entry| entry[:status] == STATUS_RUNNING }.keys + end + running_ids.each { |rid| cancel(rid, message) } + end + end end diff --git a/lib/rex/post/meterpreter/async_window.rb b/lib/rex/post/meterpreter/async_window.rb new file mode 100644 index 0000000000000..aaf2480d4d4e9 --- /dev/null +++ b/lib/rex/post/meterpreter/async_window.rb @@ -0,0 +1,76 @@ +# -*- coding: binary -*- +# frozen_string_literal: true + +module Rex + module Post + module Meterpreter + # + # Pure-function helpers for async meterpreter's work-hour / work-day + # window. Extracted from Client so the math can be unit-tested without + # loading the full meterpreter dependency graph. + # + module AsyncWindow + MAXIMUM_GAP = 7 * 86400 + + # Seconds until the next allowed poll window opens, given the current + # async work-hour config and a target-local "now" (a Time whose #hour + # and #wday read in the target's local frame — typically produced by + # Client#target_time_now). + # + # Returns 0 when polling is currently permitted (inside the window on + # a work day) or when the config is fully permissive. A zero work-day + # mask means all days, matching the target-side compatibility behavior. + # Equal start/end hours represent a full-day window. Overnight windows + # (for example 22-6) are supported. + # + # @param now [Time] target-local wall clock + # @param work_start [Integer] window start hour, 0-23 + # @param work_end [Integer] window end hour, 1-24 (exclusive upper bound) + # @param work_days [Integer] bitmask, bit0=Sun..bit6=Sat + # @return [Integer] seconds + def self.seconds_until_next_window(now, work_start, work_end, work_days) + work_start = work_start.to_i + work_end = work_end.to_i + work_days = work_days.to_i & 0x7F + work_days = 0x7F if work_days == 0 + + return 0 if work_days == 0x7F && (work_start == work_end || (work_start <= 0 && work_end >= 24)) + + hour = now.hour + wday = now.wday # 0=Sun..6=Sat + return 0 if allowed?(hour, wday, work_start, work_end, work_days) + + # Walk forward one hour at a time from the top of the next hour. + # Bounded by 7 days, so this loop is trivially finite. + delta = 3600 - (now.min * 60 + now.sec) + probe_hour = (hour + 1) % 24 + probe_wday = wday + ((hour + 1) / 24) + + while delta < MAXIMUM_GAP + wd = probe_wday % 7 + return delta if allowed?(probe_hour, wd, work_start, work_end, work_days) + + delta += 3600 + probe_hour = (probe_hour + 1) % 24 + probe_wday += 1 if probe_hour == 0 + end + MAXIMUM_GAP + end + + def self.allowed?(hour, wday, work_start, work_end, work_days) + return true if work_start == work_end + + if work_start < work_end + (work_days & (1 << wday)) != 0 && hour >= work_start && hour < work_end + elsif hour >= work_start + (work_days & (1 << wday)) != 0 + elsif hour < work_end + (work_days & (1 << ((wday - 1) % 7))) != 0 + else + false + end + end + end + end + end +end diff --git a/lib/rex/post/meterpreter/client.rb b/lib/rex/post/meterpreter/client.rb index 8f68f89163d73..36b5cd9f5db91 100644 --- a/lib/rex/post/meterpreter/client.rb +++ b/lib/rex/post/meterpreter/client.rb @@ -2,10 +2,12 @@ require 'socket' require 'openssl' +require 'rex/thread_factory' require 'rex/post/channel' require 'rex/post/meterpreter/extension_mapper' require 'rex/post/meterpreter/client_core' +require 'rex/post/meterpreter/async_window' require 'rex/post/meterpreter/channel' require 'rex/post/meterpreter/dependencies' require 'rex/post/meterpreter/object_aliases' @@ -35,6 +37,9 @@ module Extensions ### class Client + ASYNC_LEASE_POLL_INTERVAL = 1 + ASYNC_SCHEDULE_SAFETY_MARGIN = 2 * 3600 + include Rex::Post::Channel::Container include Rex::Post::Meterpreter::PacketDispatcher include Rex::Post::Meterpreter::PivotContainer @@ -542,13 +547,148 @@ def unicode_filter_decode(str) # 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 } + @async_config ||= { poll_interval: 60, jitter: 0, work_start: 0, work_end: 24, work_days: 0x7F, lease_ttl: 300, job_timeout: 86400 } end def async_mode_enabled? !!self.async_mode_enabled end + # Target-side wall-clock sample, populated at session bootstrap and refreshed + # opportunistically from async check-in responses. Shape: + # { target_unix_ts:, target_local_ts:, utc_offset:, sampled_at: } + # Nil when the payload doesn't support {ClientCore#get_target_time} or when + # the sample hasn't been taken yet. + attr_accessor :target_time_sample + + # Whether the target has acknowledged an active async job lease. + attr_reader :async_lease_enabled + + def async_lease_enabled? + return false unless @async_lease_enabled + return true unless @async_lease_deadline + + active = Process.clock_gettime(Process::CLOCK_MONOTONIC) < @async_lease_deadline + self.async_lease_enabled = false unless active + active + end + + def async_lease_renewed(ttl) + self.async_lease_enabled = true + @async_lease_deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + ttl + end + + def async_lease_enabled=(enabled) + @async_lease_enabled = enabled + @async_lease_deadline = nil unless enabled + end + + # Estimated target-local wall clock right now, as a UTC-flavored Time whose + # #hour and #wday read in the target's local frame. + def target_time_now + s = @target_time_sample + return nil unless s && s[:target_local_ts] && s[:sampled_at] + + elapsed = ::Time.now - s[:sampled_at] + ::Time.at(s[:target_local_ts] + elapsed).utc + end + + # Target UTC offset in seconds (east positive), or 0 when unknown. + def target_utc_offset + @target_time_sample&.dig(:utc_offset) || 0 + end + + def target_time_sample_age + sampled_at = @target_time_sample&.dig(:sampled_at) + sampled_at ? ::Time.now - sampled_at : nil + end + + # Seconds until the target's next allowed poll window opens, given the + # current async work_start/work_end/work_days config. See + # {Rex::Post::Meterpreter::AsyncWindow.seconds_until_next_window}. + def async_seconds_until_next_window(now = target_time_now) + return nil unless now + + cfg = async_config + AsyncWindow.seconds_until_next_window(now, cfg[:work_start], cfg[:work_end], cfg[:work_days]) + end + + # Clamp a Meterpreter request timeout when async mode is enabled. Job-lease + # requests only need to cover rapid polling. Scheduled requests must safely + # survive the longest valid work-window gap, including clock/DST drift. + def async_response_timeout(timeout) + return timeout unless async_mode_enabled? && timeout + + minimum = if async_lease_enabled? + ASYNC_LEASE_POLL_INTERVAL * 3 + 10 + else + cfg = async_config + poll = cfg[:poll_interval].to_i + worst_case = poll + (poll * cfg[:jitter].to_i / 100) + AsyncWindow::MAXIMUM_GAP + ASYNC_SCHEDULE_SAFETY_MARGIN + worst_case * 3 + 10 + end + [timeout.to_i, minimum].max + end + + # Run a block while maintaining an explicit target-side rapid-poll lease. + # The target drops the lease automatically when renewals stop. + def with_async_lease + ttl = async_config[:lease_ttl].to_i + enabled = core.async_lease(enabled: true, ttl: ttl) + raise Rex::Post::Meterpreter::RequestError.new(COMMAND_ID_CORE_ASYNC_LEASE, 'The target refused the async job lease') unless enabled + + start_async_lease_renewer(ttl) + yield + ensure + stop_async_lease_renewer + unless ::Thread.current[:msf_async_forced_stop] + begin + core.async_lease(enabled: false, ttl: ttl, wait: false) + rescue ::StandardError => e + dlog("Unable to release async job lease: #{e.message}", 'meterpreter/async') + end + end + self.async_lease_enabled = false + end + + def start_async_lease_renewer(ttl) + @async_lease_mutex ||= ::Mutex.new + @async_lease_condition ||= ::ConditionVariable.new + @async_lease_mutex.synchronize { @async_lease_stopping = false } + renew_interval = [ttl / 3, 10].max + + @async_lease_thread = Rex::ThreadFactory.spawn('AsyncLeaseRenewer', false) do + loop do + stopping = @async_lease_mutex.synchronize do + @async_lease_condition.wait(@async_lease_mutex, renew_interval) + @async_lease_stopping + end + break if stopping + + begin + core.async_lease(enabled: true, ttl: ttl, timeout: ttl) + rescue ::StandardError => e + dlog("Async job lease renewal failed: #{e.message}", 'meterpreter/async') + end + end + end + end + + def stop_async_lease_renewer + return unless @async_lease_thread + + @async_lease_mutex.synchronize do + @async_lease_stopping = true + @async_lease_condition.broadcast + end + @async_lease_thread.join(5) + if @async_lease_thread.alive? + @async_lease_thread.kill + @async_lease_thread.join + end + @async_lease_thread = nil + 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 @@ -586,7 +726,7 @@ def async_shell(main_shell) 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_worker, true) shell.instance_variable_set(:@async_pipe, pipe) # Re-load each extension on the async shell using the same code path as diff --git a/lib/rex/post/meterpreter/client_core.rb b/lib/rex/post/meterpreter/client_core.rb index 356f5adf317f5..087a70e5a5aa8 100644 --- a/lib/rex/post/meterpreter/client_core.rb +++ b/lib/rex/post/meterpreter/client_core.rb @@ -545,6 +545,32 @@ def transport_sleep(seconds) return true end + # + # Sample the target's wall clock. Returns a hash suitable for + # {Rex::Post::Meterpreter::Client#target_time_sample=}, or nil if the + # payload doesn't support the command. + # + # @return [Hash{Symbol=>Numeric}, nil] + # * :target_unix_ts - target's UTC seconds since epoch + # * :target_local_ts - target's wall-clock seconds since epoch (as if local were UTC) + # * :utc_offset - target_local_ts - target_unix_ts + # * :sampled_at - operator-local Time when the sample was taken + # + def get_target_time + request = Packet.create_request(COMMAND_ID_CORE_GET_TARGET_TIME) + response = client.send_request(request) + utc_ts = response.get_tlv_value(TLV_TYPE_TARGET_UNIX_TS) + local_ts = response.get_tlv_value(TLV_TYPE_TARGET_LOCAL_UNIX_TS) + return nil unless utc_ts && local_ts + + { + target_unix_ts: utc_ts, + target_local_ts: local_ts, + utc_offset: local_ts - utc_ts, + sampled_at: ::Time.now + } + end + # # Enable or disable async mode on the implant. # When enabled, the implant polls at the configured interval @@ -555,14 +581,25 @@ def transport_sleep(seconds) # @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_end business hours end (0-24) # @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 = {}) + if opts[:enabled] + poll_interval = opts.fetch(:poll_interval, client.async_config[:poll_interval]).to_i + jitter = opts.fetch(:jitter, client.async_config[:jitter]).to_i + work_start = opts.fetch(:work_start, client.async_config[:work_start]).to_i + work_end = opts.fetch(:work_end, client.async_config[:work_end]).to_i + work_days = opts.fetch(:work_days, client.async_config[:work_days]).to_i + raise ArgumentError, 'Poll interval must be between 10 and 86400 seconds' unless poll_interval.between?(10, 86400) + raise ArgumentError, 'Jitter must be between 0 and 99 percent' unless jitter.between?(0, 99) + raise ArgumentError, 'Work start must be between 0 and 23' unless work_start.between?(0, 23) + raise ArgumentError, 'Work end must be between 0 and 24' unless work_end.between?(0, 24) + raise ArgumentError, 'Work days must select at least one day' unless work_days.between?(1, 0x7F) + opts = opts.merge(poll_interval: poll_interval, jitter: jitter, work_start: work_start, work_end: work_end, work_days: work_days) + end + 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] @@ -570,49 +607,35 @@ def async_mode(opts = {}) 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 + client.async_config.merge!(opts.select { |key, value| !value.nil? && key != :enabled }) + else + client.async_lease_enabled = false end response end + # Acquire, renew, or release a rapid-poll lease for one async job. + def async_lease(enabled:, ttl:, timeout: client.response_timeout, wait: true) + request = Packet.create_request(COMMAND_ID_CORE_ASYNC_LEASE) + request.add_tlv(TLV_TYPE_ASYNC_LEASE_ENABLED, enabled) + request.add_tlv(TLV_TYPE_ASYNC_LEASE_TTL, ttl) + unless wait + client.send_packet(request) + client.async_lease_enabled = false + return false + end + + response = client.send_request(request, timeout) + lease_enabled = response.get_tlv_value(TLV_TYPE_ASYNC_LEASE_ENABLED) + lease_enabled ? client.async_lease_renewed(ttl) : client.async_lease_enabled = false + lease_enabled + end + # # Change the active transport to the next one in the transport list. # @@ -823,20 +846,7 @@ def shutdown # kill the handler. This could be improved by the server side # sending a reply to shutdown first. # - # 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. @@ -1100,4 +1110,3 @@ def generate_migrate_payload(target_process) end end; end; end - diff --git a/lib/rex/post/meterpreter/core_ids.rb b/lib/rex/post/meterpreter/core_ids.rb index d44ef011bf4c2..8c8cb9e447df2 100644 --- a/lib/rex/post/meterpreter/core_ids.rb +++ b/lib/rex/post/meterpreter/core_ids.rb @@ -47,6 +47,8 @@ module Meterpreter 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 +COMMAND_ID_CORE_GET_TARGET_TIME = EXTENSION_ID_CORE + 36 +COMMAND_ID_CORE_ASYNC_LEASE = EXTENSION_ID_CORE + 37 end end diff --git a/lib/rex/post/meterpreter/packet.rb b/lib/rex/post/meterpreter/packet.rb index fbd9ede66e414..5c368abc1f6f3 100644 --- a/lib/rex/post/meterpreter/packet.rb +++ b/lib/rex/post/meterpreter/packet.rb @@ -117,7 +117,16 @@ module Meterpreter 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 + +# +# Target-side wall clock. Sampled at bootstrap and, when async is active, +# piggybacked onto check-in responses so the framework can compute the +# target's local time without a fresh roundtrip. +# +TLV_TYPE_TARGET_UNIX_TS = TLV_META_TYPE_QWORD | 707 +TLV_TYPE_TARGET_LOCAL_UNIX_TS = TLV_META_TYPE_QWORD | 708 +TLV_TYPE_ASYNC_LEASE_ENABLED = TLV_META_TYPE_BOOL | 709 +TLV_TYPE_ASYNC_LEASE_TTL = TLV_META_TYPE_UINT | 710 # Configuration & C2 options # diff --git a/lib/rex/post/meterpreter/packet_dispatcher.rb b/lib/rex/post/meterpreter/packet_dispatcher.rb index 2b09cc70d98cf..1845b44b2300e 100644 --- a/lib/rex/post/meterpreter/packet_dispatcher.rb +++ b/lib/rex/post/meterpreter/packet_dispatcher.rb @@ -208,6 +208,8 @@ def send_request(packet, timeout = self.response_timeout) # @param timeout [Integer,nil] number of seconds to wait, or nil to wait # forever def send_packet_wait_response(packet, timeout) + timeout = async_response_timeout(timeout) if respond_to?(:async_response_timeout) + if packet.type == PACKET_TYPE_REQUEST && commands.present? # XXX: Remove this condition once the payloads gem has had another major version bump from 2.x to 3.x and # rapid7/metasploit-payloads#451 has been landed to correct the `enumextcmd` behavior on Windows. Until then, skip @@ -241,11 +243,12 @@ def send_packet_wait_response(packet, timeout) end # Wait for the supplied time interval - response = waiter.wait(timeout) - - # Remove the waiter from the list of waiters in case it wasn't - # removed. This happens if the waiter timed out above. - remove_response_waiter(waiter) + begin + response = waiter.wait(timeout) + ensure + # Also remove the waiter if an async worker is cancelled while waiting. + remove_response_waiter(waiter) + end # wire in the UUID for this, as it should be part of every response # packet @@ -595,6 +598,28 @@ def dispatch_inbound_packet(packet) self.last_checkin = ::Time.now pivot_session = self.find_pivot_session(packet.session_guid) + target_client = pivot_session ? pivot_session.pivoted_session : self + + # Refresh the target-time sample opportunistically. The implant tacks + # TLV_TYPE_TARGET_UNIX_TS + TARGET_LOCAL_UNIX_TS onto async check-in + # responses so short-term work-window status remains useful without a + # dedicated roundtrip. Request timeouts use a separate conservative bound. + if target_client.respond_to?(:target_time_sample=) + utc_ts = packet.get_tlv_value(TLV_TYPE_TARGET_UNIX_TS) + local_ts = packet.get_tlv_value(TLV_TYPE_TARGET_LOCAL_UNIX_TS) + if utc_ts && local_ts + target_client.target_time_sample = { + target_unix_ts: utc_ts, + target_local_ts: local_ts, + utc_offset: local_ts - utc_ts, + sampled_at: self.last_checkin + } + end + end + + lease_enabled = packet.get_tlv_value(TLV_TYPE_ASYNC_LEASE_ENABLED) + target_client.async_lease_enabled = false if lease_enabled == false && target_client.respond_to?(:async_lease_enabled=) + pivot_session.pivoted_session.last_checkin = self.last_checkin if pivot_session # If the packet is a response, try to notify any potential diff --git a/lib/rex/post/meterpreter/ui/console.rb b/lib/rex/post/meterpreter/ui/console.rb index 216ef4d75cb28..7ac53420ca5a7 100644 --- a/lib/rex/post/meterpreter/ui/console.rb +++ b/lib/rex/post/meterpreter/ui/console.rb @@ -86,6 +86,10 @@ def interact_with_channel(channel, raw: false) channel.reset_ui end + def async_worker? + !!@async_worker + end + # # Queues a command to be run when the interactive loop is entered. # @@ -101,23 +105,31 @@ def queue_cmd(cmd) async ].freeze + ASYNC_WORKER_BLOCKED_COMMANDS = %w[ + async background bg exit quit irb pry sessions + channel shell interact portfwd rportfwd powershell_shell + detach sleep transport migrate pivot secure + ].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 + if client.async_mode_enabled? + if @async_worker && ASYNC_WORKER_BLOCKED_COMMANDS.include?(method) + log_error("Cannot run '#{method}' inside an async job.") + return + elsif !@async_worker && !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 end begin super rescue Exception => e - is_error_handled = self.client.on_run_command_error_proc && self.client.on_run_command_error_proc.call(e) == :handled + is_error_handled = !@async_worker && client.on_run_command_error_proc && client.on_run_command_error_proc.call(e) == :handled return if is_error_handled case e when Rex::TimeoutError, Rex::InvalidDestination @@ -132,6 +144,7 @@ def run_command(dispatcher, method, arguments) log_error("Error running command #{method}: #{e.class} #{e}") elog(e) end + raise if @async_worker end end @@ -159,4 +172,3 @@ def log_error(msg) end end end - 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 22e36930fefca..cfdcb2a821efc 100644 --- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb +++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/core.rb @@ -756,16 +756,15 @@ def cmd_async_help -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) + -e Work hours end 0-24 (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) + -l Async job lease TTL, renewed while a job runs (default: 300) + -x Maximum runtime for one async job (default: 86400) Examples: async mode on async config -i 300 -j 20 -s 8 -e 17 -d mon-fri - async config -i 600 -y 30 + async config -i 600 -l 300 -x 7200 async run ls async run execute -f cmd.exe -a "/c whoami" -H async queue @@ -802,6 +801,7 @@ def parse_work_days(spec) # Format a work days bitmask into a human-readable string. # def format_work_days(mask) + return 'all' if mask == 0 return 'all' if mask == 0x7F return 'mon-fri' if mask == 0x3E @@ -868,17 +868,22 @@ def async_subcmd_mode(args) return end cfg = client.async_config + validation_error = async_config_validation_error(cfg) + if validation_error + print_error(validation_error) + return + end 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.') + print_warning('Use "async run " for extension commands and Post modules; transport lifecycle and persistent interactive commands are unavailable.') 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 + client.core.async_mode(enabled: false) print_good('Async mode disabled. Session is now interactive.') else print_error("Unknown mode: #{subcmd}. Use 'on' or 'off'.") @@ -900,7 +905,8 @@ def async_subcmd_config(args) 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'} + Lease TTL : #{cfg[:lease_ttl]}s + Job timeout : #{cfg[:job_timeout]}s Mode : #{client.async_mode_enabled? ? 'enabled' : 'disabled'} CONFIG @@ -912,29 +918,39 @@ def async_subcmd_config(args) '-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)'], + '-e' => [true, 'Work end hour (0-24)'], '-d' => [true, 'Work days'], - '-y' => [true, 'Smart-sync burst window (seconds, 0 disables)'] + '-l' => [true, 'Async job lease TTL (seconds)'], + '-x' => [true, 'Maximum async job runtime (seconds)'] ) + updated = cfg.dup opts.parse(args) do |opt, _idx, val| case opt when '-i' - cfg[:poll_interval] = val.to_i + updated[:poll_interval] = val.to_i when '-j' - cfg[:jitter] = val.to_i + updated[:jitter] = val.to_i when '-s' - cfg[:work_start] = val.to_i + updated[:work_start] = val.to_i when '-e' - cfg[:work_end] = val.to_i + updated[:work_end] = val.to_i when '-d' - cfg[:work_days] = parse_work_days(val) - when '-y' - cfg[:smart_sync] = val.to_i + updated[:work_days] = parse_work_days(val) + when '-l' + updated[:lease_ttl] = val.to_i + when '-x' + updated[:job_timeout] = 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}).") + validation_error = async_config_validation_error(updated) + if validation_error + print_error(validation_error) + return + end + + cfg.replace(updated) + print_good("Async configuration updated (poll #{cfg[:poll_interval]}s, jitter #{cfg[:jitter]}%, hours #{cfg[:work_start]}:00-#{cfg[:work_end]}:00, lease #{cfg[:lease_ttl]}s).") if client.async_mode_enabled? print_status('Async mode is active. Sending updated config to target...') client.core.async_mode(enabled: true, **cfg) @@ -947,10 +963,6 @@ def async_subcmd_config(args) # Commands that cannot be executed via 'async run' because they require # an interactive channel or persistent socket that the async poll model # cannot service. - ASYNC_RUN_BLOCKED_COMMANDS = %w[ - shell interact channel portfwd rportfwd - ].freeze - # # async run # @@ -960,9 +972,14 @@ def async_subcmd_run(args) return end - blocked = ASYNC_RUN_BLOCKED_COMMANDS.include?(args.first) + unless client.async_mode_enabled? + print_error('Enable async mode before queuing async commands.') + return + end + + blocked = Console::ASYNC_WORKER_BLOCKED_COMMANDS.include?(args.first) if blocked - print_error("Command '#{args.first}' cannot be run in async mode; it requires an interactive channel.") + print_error("Command '#{args.first}' cannot run as an async job because it changes session lifecycle or requires persistent interaction.") return end @@ -1012,7 +1029,11 @@ def async_subcmd_run(args) # 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) + client.with_async_lease do + ::Timeout.timeout(client.async_config[:job_timeout]) do + async_shell.run_single(cmd_line, propagate_errors: true) + end + end captured = async_pipe ? async_pipe.read_subscriber('async') : '' client.async_store.complete(work_rid, nil, captured.empty? ? '(no output)' : captured) ensure @@ -1040,6 +1061,33 @@ def async_subcmd_run(args) end print_status("Queued: #{cmd_line} (rid: #{rid[0..7]})") + + gap = client.async_seconds_until_next_window + sample_age = client.target_time_sample_age + if sample_age.nil? || sample_age > 3600 + print_warning('Target clock sample is unavailable or stale; delivery waits for the next configured target-local window.') + elsif gap && gap > 0 + human = if gap < 3600 + "#{(gap / 60).round}m" + elsif gap < 86400 + "#{(gap / 3600.0).round(1)}h" + else + "#{(gap / 86400.0).round(1)}d" + end + print_warning("Target outside work window; delivery resumes in approximately #{human} based on the last target-clock sample.") + end + end + + def async_config_validation_error(cfg) + return 'Poll interval must be between 10 and 86400 seconds.' unless cfg[:poll_interval].between?(10, 86400) + return 'Jitter must be between 0 and 99 percent.' unless cfg[:jitter].between?(0, 99) + return 'Work start must be between 0 and 23.' unless cfg[:work_start].between?(0, 23) + return 'Work end must be between 0 and 24.' unless cfg[:work_end].between?(0, 24) + return 'Work days must select at least one day.' unless cfg[:work_days].between?(1, 0x7F) + return 'Lease TTL must be between 30 and 3600 seconds.' unless cfg[:lease_ttl].between?(30, 3600) + return 'Job timeout must be at least the lease TTL.' unless cfg[:job_timeout] >= cfg[:lease_ttl] + + nil end # @@ -1757,7 +1805,8 @@ def cmd_run(*args) # First try it as a Post module if we have access to the Metasploit # Framework instance. If we don't, or if no such module exists, # fall back to using the scripting interface. - if msf_loaded? && mod = client.framework.modules.create(script_name) + mod = client.framework.modules.create(script_name) if msf_loaded? + if mod original_mod = mod reloaded_mod = client.framework.modules.reload_module(original_mod) @@ -1783,6 +1832,8 @@ def cmd_run(*args) }) print_status("Session #{result.sid} created in the background.") if result.is_a?(Msf::Session) + elsif shell.async_worker? + print_error('Meterpreter scripts cannot run inside an async job; use a Post module instead.') else # the rest of the arguments get passed in through the binding client.execute_script(script_name, args) diff --git a/spec/lib/rex/post/meterpreter/async_result_store_spec.rb b/spec/lib/rex/post/meterpreter/async_result_store_spec.rb new file mode 100644 index 0000000000000..601de96f0663e --- /dev/null +++ b/spec/lib/rex/post/meterpreter/async_result_store_spec.rb @@ -0,0 +1,61 @@ +# frozen_string_literal: true + +require 'rex/post/meterpreter/async_result_store' + +RSpec.describe Rex::Post::Meterpreter::AsyncResultStore do + let(:store) { described_class.new } + + before do + allow(store).to receive(:dlog) + allow(store).to receive(:elog) + end + + it 'preserves pending and running entries when clearing terminal results' do + store.queue('pending', 'pending') + store.queue('running', 'running') + store.mark_running('running') + store.queue('complete', 'complete') + store.complete('complete', nil) + store.queue('cancelled', 'cancelled') + store.cancel('cancelled') + + expect(store.clear_completed).to eq(2) + expect(store.all.transform_values { |entry| entry[:status] }).to eq('pending' => :pending, 'running' => :running) + end + + it 'cancels work queued behind the current command when stopped' do + started = ::Queue.new + store.enqueue_work('running', 'running') do |rid| + started << true + sleep 0.1 + store.complete(rid, nil) + end + store.enqueue_work('pending', 'pending') { |rid| store.complete(rid, nil) } + started.pop + + store.stop_worker + + expect(store.fetch('running')[:status]).to eq(:cancelled) + expect(store.fetch('pending')[:status]).to eq(:cancelled) + end + + it 'starts a fresh worker after stopping' do + first_done = ::Queue.new + store.enqueue_work('first', 'first') do |rid| + store.complete(rid, nil) + first_done << true + end + first_done.pop + store.stop_worker + + second_done = ::Queue.new + store.enqueue_work('second', 'second') do |rid| + store.complete(rid, nil) + second_done << true + end + second_done.pop + store.stop_worker + + expect(store.fetch('second')[:status]).to eq(:complete) + end +end diff --git a/spec/lib/rex/post/meterpreter/async_window_spec.rb b/spec/lib/rex/post/meterpreter/async_window_spec.rb new file mode 100644 index 0000000000000..a8683d23992ff --- /dev/null +++ b/spec/lib/rex/post/meterpreter/async_window_spec.rb @@ -0,0 +1,65 @@ +# frozen_string_literal: true + +require 'time' +require 'rex/post/meterpreter/async_window' + +RSpec.describe Rex::Post::Meterpreter::AsyncWindow do + describe '.seconds_until_next_window' do + # Target-local wall clock — a Time whose #hour and #wday read in the + # target's local frame. Client#target_time_now produces the same shape: + # a UTC-flavored Time offset by the target's UTC offset. + def at(str) + ::Time.parse("#{str} UTC") + end + + it 'returns 0 with the default fully-permissive config' do + # Sat 22:00 — outside any narrow window, but the config is 24/7. + expect(described_class.seconds_until_next_window(at('2026-08-08 22:00:00'), 0, 24, 0x7F)).to eq(0) + end + + it 'returns 0 when inside window on a work day' do + # Wed 10:30 with mon-fri 09-17. + expect(described_class.seconds_until_next_window(at('2026-08-05 10:30:00'), 9, 17, 0x3E)).to eq(0) + end + + it 'covers weekend + Monday morning gap from Sat 22:00 with mon-fri 09-17' do + now = at('2026-08-08 22:00:00') # Sat + expected = at('2026-08-10 09:00:00') - now + gap = described_class.seconds_until_next_window(now, 9, 17, 0x3E) + expect(gap).to be_within(3600).of(expected) + end + + it 'covers after-hours gap from Fri 17:30 with mon-fri 09-17' do + now = at('2026-08-07 17:30:00') # Fri + expected = at('2026-08-10 09:00:00') - now + gap = described_class.seconds_until_next_window(now, 9, 17, 0x3E) + expect(gap).to be_within(3600).of(expected) + end + + it 'covers same-day pre-hours gap' do + # Wed 07:30 with 09-17 all days → 1.5h. + gap = described_class.seconds_until_next_window(at('2026-08-05 07:30:00'), 9, 17, 0x7F) + expect(gap).to be_within(120).of(5400) + end + + it 'treats a zero work-day mask as all days for target compatibility' do + gap = described_class.seconds_until_next_window(at('2026-08-05 10:30:00'), 9, 17, 0) + expect(gap).to eq(0) + end + + it 'supports overnight windows' do + expect(described_class.seconds_until_next_window(at('2026-08-05 23:30:00'), 22, 6, 0x7F)).to eq(0) + expect(described_class.seconds_until_next_window(at('2026-08-05 12:30:00'), 22, 6, 0x7F)).to be_within(1).of(9.5 * 3600) + end + + it 'assigns the after-midnight portion to the previous work day' do + monday_only = 0x02 + expect(described_class.seconds_until_next_window(at('2026-08-11 02:00:00'), 22, 6, monday_only)).to eq(0) + expect(described_class.seconds_until_next_window(at('2026-08-11 22:00:00'), 22, 6, monday_only)).to be > 0 + end + + it 'treats equal start and end as a full-day window on active days' do + expect(described_class.seconds_until_next_window(at('2026-08-05 12:30:00'), 9, 9, 0x7F)).to eq(0) + end + end +end