Add Jetpack version based detection

agent/exploits/jetpack_version_detection.py

@@ -0,0 +1,207 @@
+import datetime
+import logging
+import re
+from urllib import parse as urlparse
+
+import requests
+import semver
+from requests import exceptions as requests_exceptions
+
+from agent import definitions
+from agent import exploits_registry
+from agent.exploits import webexploit
+
+logging.basicConfig(level=logging.ERROR)
+logger = logging.getLogger(__name__)
+
+MAX_REDIRECTS = 2
+DEFAULT_TIMEOUT = datetime.timedelta(seconds=90)
+VULNERABILITY_TITLE = (
+    "UNAUTHORIZED FORM DATA EXPOSURE VULNERABILITY IN JETPACK'S CONTACT FORM FEATURE"
+)
+VULNERABILITY_REFERENCE = "jetpack data exposure"
+VULNERABILITY_DESCRIPTION = """The vulnerability resides in the Contact Form feature in Jetpack, and could be used by any logged in users on a site to read forms submitted by visitors on the site."""
+RISK_RATING = "CRITICAL"
+
+# Define vulnerable ranges with actual values
+VULNERABLE_RANGES = [
+    ("13.9.0", "13.9.0"),
+    ("13.8.0", "13.8.1"),
+    ("13.7.0", "13.7.0"),
+    ("13.6.0", "13.6.0"),
+    ("13.5.0", "13.5.0"),
+    ("13.4.0", "13.4.3"),
+    ("13.3.0", "13.3.1"),
+    ("13.2.0", "13.2.2"),
+    ("13.1.0", "13.1.3"),
+    ("13.0.0", "13.0.0"),
+    ("12.9.0", "12.9.3"),
+    ("12.8.0", "12.8.1"),
+    ("12.7.0", "12.7.1"),
+    ("12.6.0", "12.6.2"),
+    ("12.5.0", "12.5.0"),
+    ("12.4.0", "12.4.0"),
+    ("12.3.0", "12.3.0"),
+    ("12.2.0", "12.2.1"),
+    ("12.1.0", "12.1.1"),
+    ("12.0.0", "12.0.1"),
+    ("11.9.0", "11.9.2"),
+    ("11.8.0", "11.8.5"),
+    ("11.7.0", "11.7.2"),
+    ("11.6.0", "11.6.1"),
+    ("11.5.0", "11.5.2"),
+    ("11.4.0", "11.4.1"),
+    ("11.3.0", "11.3.3"),
+    ("11.2.0", "11.2.1"),
+    ("11.1.0", "11.1.3"),
+    ("11.0.0", "11.0.1"),
+    ("10.9.0", "10.9.2"),
+    ("10.8.0", "10.8.1"),
+    ("10.7.0", "10.7.1"),
+    ("10.6.0", "10.6.1"),
+    ("10.5.0", "10.5.2"),
+    ("10.4.0", "10.4.1"),
+    ("10.3.0", "10.3.1"),
+    ("10.2.0", "10.2.2"),
+    ("10.1.0", "10.1.1"),
+    ("10.0.0", "10.0.1"),
+    ("9.9.0", "9.9.2"),
+    ("9.8.0", "9.8.2"),
+    ("9.7.0", "9.7.2"),
+    ("9.6.0", "9.6.3"),
+    ("9.5.0", "9.5.4"),
+    ("9.4.0", "9.4.3"),
+    ("9.3.0", "9.3.4"),
+    ("9.2.0", "9.2.3"),
+    ("9.1.0", "9.1.2"),
+    ("9.0.0", "9.0.4"),
+    ("8.9.0", "8.9.3"),
+    ("8.8.0", "8.8.4"),
+    ("8.7.0", "8.7.3"),
+    ("8.6.0", "8.6.3"),
+    ("8.5.0", "8.5.2"),
+    ("8.4.0", "8.4.4"),
+    ("8.3.0", "8.3.2"),
+    ("8.2.0", "8.2.5"),
+    ("8.1.0", "8.1.3"),
+    ("8.0.0", "8.0.2"),
+    ("7.9.0", "7.9.3"),
+    ("7.8.0", "7.8.3"),
+    ("7.7.0", "7.7.5"),
+    ("7.6.0", "7.6.3"),
+    ("7.5.0", "7.5.6"),
+    ("7.4.0", "7.4.4"),
+    ("7.3.0", "7.3.4"),
+    ("7.2.0", "7.2.4"),
+    ("7.1.0", "7.1.4"),
+    ("7.0.0", "7.0.4"),
+    ("6.9.0", "6.9.3"),
+    ("6.8.0", "6.8.4"),
+    ("6.7.0", "6.7.3"),
+    ("6.6.0", "6.6.4"),
+    ("6.5.0", "6.5.3"),
+    ("6.4.0", "6.4.5"),
+    ("6.3.0", "6.3.6"),
+    ("6.2.0", "6.2.4"),
+    ("6.1.0", "6.1.4"),
+    ("6.0.0", "6.0.3"),
+    ("5.9.0", "5.9.3"),
+    ("5.8.0", "5.8.3"),
+    ("5.7.0", "5.7.4"),
+    ("5.6.0", "5.6.4"),
+    ("5.5.0", "5.5.4"),
+    ("5.4.0", "5.4.3"),
+    ("5.3.0", "5.3.3"),
+    ("5.2.0", "5.2.4"),
+    ("5.1.0", "5.1.3"),
+    ("5.0.0", "5.0.2"),
+    ("4.9.0", "4.9.2"),
+    ("4.8.0", "4.8.4"),
+    ("4.7.0", "4.7.3"),
+    ("4.6.0", "4.6.2"),
+    ("4.5.0", "4.5.2"),
+    ("4.4.0", "4.4.4"),
+    ("4.3.0", "4.3.4"),
+    ("4.2.0", "4.2.4"),
+    ("4.1.0", "4.1.3"),
+    ("4.0.0", "4.0.6"),
+    (None, "3.9.9"),
+]
+
+
+@exploits_registry.register
+class JetpackExploit(webexploit.WebExploit):
+    accept_request = definitions.Request(
+        method="GET", path="/wp-content/plugins/jetpack/readme.txt"
+    )
+    check_request = definitions.Request(
+        method="GET", path="/wp-content/plugins/jetpack/readme.txt"
+    )
+    accept_pattern = [
+        re.compile("=== Jetpack - WP Security, Backup, Speed, & Growth ===")
+    ]
+    version_pattern = re.compile("Stable\stag:\s(\d+\.\d+\.\d+)")
+
+    metadata = definitions.VulnerabilityMetadata(
+        title=VULNERABILITY_TITLE,
+        description=VULNERABILITY_DESCRIPTION,
+        reference=VULNERABILITY_REFERENCE,
+        risk_rating=RISK_RATING,
+    )
+
+    def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
+        """Rule to detect specific vulnerability on a specific target.
+
+        Args:
+            target: Target to scan
+
+        Returns:
+            List of identified vulnerabilities.
+        """
+        session = requests.Session()
+        session.max_redirects = MAX_REDIRECTS
+        session.verify = False
+
+        vulnerabilities: list[definitions.Vulnerability] = []
+
+        target_endpoint = urlparse.urljoin(target.origin, self.check_request.path)
+
+        try:
+            req = requests.Request(
+                method=self.check_request.method,
+                url=target_endpoint,
+                data=self.check_request.data,
+            ).prepare()
+            resp = session.send(req, timeout=DEFAULT_TIMEOUT.seconds)
+        except requests_exceptions.RequestException as e:
+            logger.error(f"Network error occurred: {e}")
+            return vulnerabilities
+
+        if (matched := self.version_pattern.findall(resp.text)) != []:
+            extracted_version = matched[0]
+
+            if _is_vulnerable(extracted_version) is True:
+                vulnerability = self._create_vulnerability(target)
+                vulnerabilities.append(vulnerability)
+        return vulnerabilities
+
+
+def _is_vulnerable(extracted_version: str) -> bool:
+    """Check if the extracted version is in the list of vulnerable ranges.
+
+    Args:
+        extracted_version: Version of Jetpack
+
+    Returns:
+        True if the version is vulnerable, False otherwise.
+    """
+    extracted_ver = semver.Version.parse(extracted_version)
+
+    for min_version, max_version in VULNERABLE_RANGES:
+        min_ver = semver.Version.parse(min_version) if min_version else None
+        max_ver = semver.Version.parse(max_version)
+
+        if (min_ver is None or extracted_ver >= min_ver) and extracted_ver <= max_ver:
+            return True
+
+    return False

tests/exploits/jetpack_version_detection_test.py

@@ -0,0 +1,76 @@
+"""Unit tests for Agent Asteroid: Jetpack Exploit"""
+
+import pytest
+import requests_mock as req_mock
+
+from agent import definitions
+from agent.exploits import jetpack_version_detection
+
+
+def testCVE20249487_whenVulnerable_reportFinding(
+    requests_mock: req_mock.mocker.Mocker,
+) -> None:
+    """Jetpack Exploit unit test: case when target is vulnerable."""
+    requests_mock.get(
+        "http://localhost:80/wp-content/plugins/jetpack/readme.txt",
+        text="""
+        === Jetpack - WP Security, Backup, Speed, & Growth ===
+        Tags: Security, backup, malware, scan, performance
+        Stable tag: 4.9.2
+        Requires at least: 6.5
+        """,
+        status_code=200,
+    )
+    exploit_instance = jetpack_version_detection.JetpackExploit()
+    target = definitions.Target("http", "localhost", 80)
+
+    accept = exploit_instance.accept(target)
+    vulnerabilities = exploit_instance.check(target)
+
+    assert accept is True
+    vulnerability = vulnerabilities[0]
+    assert (
+        vulnerability.entry.title
+        == "UNAUTHORIZED FORM DATA EXPOSURE VULNERABILITY IN JETPACK'S CONTACT FORM FEATURE"
+    )
+    assert vulnerability.technical_detail == (
+        "http://localhost:80 is vulnerable to jetpack data exposure, "
+        "UNAUTHORIZED FORM DATA EXPOSURE VULNERABILITY IN JETPACK'S CONTACT FORM FEATURE"
+    )
+
+
+def testCVE20249487_whenNotVulnerable_reportNothing(
+    requests_mock: req_mock.mocker.Mocker,
+) -> None:
+    """jetpack_version_detection unit test: case when target is vulnerable."""
+    requests_mock.get(
+        "http://localhost:80/wp-content/plugins/jetpack/readme.txt",
+        text="""
+        === Jetpack - WP Security, Backup, Speed, & Growth ===
+        Tags: Security, backup, malware, scan, performance
+        Stable tag: 13.9.1
+        Requires at least: 6.5
+        """,
+        status_code=200,
+    )
+    exploit_instance = jetpack_version_detection.JetpackExploit()
+    target = definitions.Target("http", "localhost", 80)
+
+    accept = exploit_instance.accept(target)
+    vulnerabilities = exploit_instance.check(target)
+
+    assert accept is True
+    assert len(vulnerabilities) == 0
+
+
+def testCVE20249487_whenNetworkErrorOccurs_reportNothing(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Jetpack Exploit unit test: case when network error occurs."""
+
+    exploit_instance = jetpack_version_detection.JetpackExploit()
+    target = definitions.Target("http", "nonexesit", 80)
+
+    vulnerabilities = exploit_instance.check(target)
+    assert len(vulnerabilities) == 0
+    assert "Network error occurred" in caplog.text