Fix/refactor to use clouscraper

agent/definitions.py

@@ -80,14 +80,21 @@
         return super().init_poolmanager(*args, **kwargs)  # type:ignore[no-untyped-call]
 
 
+class HttpSession(cloudscraper.CloudScraper):  # type:ignore[no-any-unimported,misc]
+    """Wrapper for the requests session class."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.max_redirects = MAX_REDIRECTS
+        self.verify = False
+        self.mount("https://", SSLAdapter())
+
+
 class Exploit(abc.ABC):
     """Base Exploit"""
 
     def __init__(self) -> None:
-        self.session = cloudscraper.create_scraper()
-        self.session.max_redirects = MAX_REDIRECTS
-        self.session.verify = False
-        self.session.mount("https://", SSLAdapter())
+        self.session = HttpSession()
 
     @abc.abstractmethod
     def accept(self, target: Target) -> bool:
@@ -99,7 +106,7 @@
         Returns:
             True if the target is valid; otherwise False.
         """
         raise NotImplementedError()
 
     @abc.abstractmethod
     def check(self, target: Target) -> list[Vulnerability]:
@@ -111,7 +118,7 @@
         Returns:
             List of identified vulnerabilities.
         """
         raise NotImplementedError()
 
     @property
     def __key__(self) -> str:

agent/exploits/cve_2023_27997.py

@@ -83,6 +83,7 @@
                 target.origin + "/remote/info", timeout=DEFAULT_TIMEOUT
             )
         except requests_exceptions.RequestException:
+            breakpoint()
             return []
         reg = re.compile("salt='([0-9a-f]{8})'")
         matches = reg.findall(r.text)

agent/exploits/cve_2024_20419.py

@@ -5,7 +5,6 @@
 import logging
 import urllib3
 
-import cloudscraper
 from requests import exceptions as requests_exceptions
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
@@ -71,7 +70,7 @@
     )
 
 
-def _is_endpoint_reachable(session: cloudscraper.CloudScraper, url: str) -> bool:  # type: ignore[no-any-unimported]
+def _is_endpoint_reachable(session: definitions.HttpSession, url: str) -> bool:
     try:
         response = session.get(url, timeout=10)
         response.raise_for_status()
@@ -82,8 +81,8 @@
         return False
 
 
-def _get_auth_token(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper, url: str, username: str
+def _get_auth_token(
+    session: definitions.HttpSession, url: str, username: str
 ) -> Any | None:
     try:
         get_url = f"{url}/backend/settings/oauth_adfs?hostname=polar"
@@ -135,7 +134,7 @@
                 "POST Request failed with status code: %d", post_response.status_code
             )
             return None
     except requests_exceptions.RequestException as e:
         logger.error("HTTP Request failed during auth token retrieval: %s", e)
         return None
 

agent/exploits/cve_2024_23113.py

@@ -5,7 +5,6 @@
 import re
 import socket
 
-import cloudscraper
 from requests import exceptions as requests_exceptions
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
@@ -139,8 +138,8 @@ def _get_fortinet_version_snmp(host: str) -> tuple[str, tuple[int, int, int]] |
     return None
 
 
-def _get_fortinet_version_http(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper, host: str
+def _get_fortinet_version_http(
+    session: definitions.HttpSession, host: str
 ) -> tuple[str, tuple[int, int, int]] | None:
     try:
         response = session.get(f"https://{host}", timeout=DEFAULT_TIMEOUT.seconds)
@@ -184,8 +183,8 @@ def _get_fortinet_version_tcp(host: str) -> tuple[str, tuple[int, int, int]] | N
     return None
 
 
-def _get_fortinet_version(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper, host: str
+def _get_fortinet_version(
+    session: definitions.HttpSession, host: str
 ) -> tuple[str, tuple[int, int, int]] | None:
     # Try HTTP First
     result = _get_fortinet_version_http(session, host)

agent/exploits/cve_2024_4040.py

@@ -2,7 +2,6 @@
 
 from typing import Any
 
-import cloudscraper
 from requests import exceptions as requests_exceptions
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
 from ostorlab.agent.kb import kb
@@ -32,7 +31,7 @@ class CVE20244040Exploit(definitions.Exploit):
 
     def __init__(self) -> None:
         self.token = ""
-        self.session = cloudscraper.create_scraper()
+        self.session = definitions.HttpSession()
 
     def accept(self, target: definitions.Target) -> bool:
         try:

agent/exploits/cve_2024_40725.py

@@ -5,7 +5,6 @@
 import logging
 
 import pathlib
-import cloudscraper
 import semantic_version
 from pygments import highlight
 from pygments import lexers
@@ -35,7 +34,7 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_server_version(session: cloudscraper.CloudScraper, url: str) -> str | None:  # type: ignore[no-any-unimported]
+def _get_server_version(session: definitions.HttpSession, url: str) -> str | None:
     """Gets the server version from the given URL.
 
     Args:
@@ -115,8 +114,8 @@ def _contains_code_content(content: str) -> bool:
         return False
 
 
-def _check_files(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper, url: str, filenames: list[str]
+def _check_files(
+    session: definitions.HttpSession, url: str, filenames: list[str]
 ) -> bool:
     """Checks the given files for source code disclosure.
 

agent/exploits/cve_2024_7120.py

@@ -2,7 +2,6 @@
 
 import logging
 
-import cloudscraper
 from requests import exceptions as requests_exceptions
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
@@ -60,7 +59,7 @@ def _create_vulnerability(target: definitions.Target) -> definitions.Vulnerabili
     )
 
 
-def _is_endpoint_reachable(session: cloudscraper.CloudScraper, url: str) -> bool:  # type: ignore[no-any-unimported]
+def _is_endpoint_reachable(session: definitions.HttpSession, url: str) -> bool:
     """Checks if the endpoint is reachable and contains specific content."""
     try:
         response = session.get(url, timeout=10)
@@ -75,7 +74,7 @@ def _is_endpoint_reachable(session: cloudscraper.CloudScraper, url: str) -> bool
     return False
 
 
-def _detect_command_injection(session: cloudscraper.CloudScraper, url: str) -> bool:  # type: ignore[no-any-unimported]
+def _detect_command_injection(session: definitions.HttpSession, url: str) -> bool:
     """Checks if the command injection is successful."""
     try:
         injection_url = f"{url}/vpn/list_base_config.php?type=mod&parts=base_config&template={INJECTION_PAYLOAD}"

agent/exploits/cve_2024_9634.py

@@ -7,7 +7,6 @@
 import requests
 from requests import exceptions as requests_exceptions
 from packaging import version
-import cloudscraper
 
 from agent import definitions
 from agent import exploits_registry
@@ -30,8 +29,8 @@ def _is_vulnerable_version(version_str: str) -> bool:
     return version.parse(version_str.strip()) <= MAX_SAFE_VERSION
 
 
-def _send_request(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper,
+def _send_request(
+    session: definitions.HttpSession,
     target: definitions.Target,
     request: definitions.Request,
 ) -> Any | None:

agent/exploits/exposed_docker_resgistry_api.py

@@ -3,7 +3,6 @@
 import logging
 import json
 
-import cloudscraper
 from requests import exceptions as requests_exceptions
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
@@ -66,8 +65,8 @@ def _create_vulnerability(
     return vulnerability
 
 
-def _get_repositories_info(  # type: ignore[no-any-unimported]
-    session: cloudscraper.CloudScraper, target: definitions.Target, repo_list: list[str]
+def _get_repositories_info(
+    session: definitions.HttpSession, target: definitions.Target, repo_list: list[str]
 ) -> dict[str, list[str]]:
     """
     Retrieves repository information from the Docker Registry API.

tests/exploits/cve_2014_7169_test.py

@@ -23,7 +23,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.get", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.get", side_effect=side_effect)
 
     target = definitions.Target(
         scheme="http", host="127.0.0.1", port=8081, path="/cgi-bin/vulnerable.cgi"
@@ -54,7 +54,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.get", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.get", side_effect=side_effect)
     target = definitions.Target(scheme="https", host="127.0.0.1", port=8081)
     exploit_instance = cve_2014_7169.CVE20147169Exploit()
 

tests/exploits/cve_2018_14558_test.py

@@ -21,7 +21,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.get", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.get", side_effect=side_effect)
     target = definitions.Target(scheme="https", host="127.0.0.1", port=443)
     exploit_instance = cve_2018_14558.CVE201814558Exploit()
 
@@ -49,7 +49,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.post", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.post", side_effect=side_effect)
     target = definitions.Target(scheme="https", host="127.0.0.1", port=443)
     exploit_instance = cve_2018_14558.CVE201814558Exploit()
 

tests/exploits/cve_2018_7841_test.py

@@ -22,7 +22,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.post", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.post", side_effect=side_effect)
 
     target = definitions.Target(scheme="https", host="127.0.0.1", port=443)
     exploit_instance = cve_2018_7841.CVE20187841Exploit()
@@ -50,7 +50,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         mock_response.elapsed = elapsed
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.post", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.post", side_effect=side_effect)
     target = definitions.Target(scheme="https", host="127.0.0.1", port=443)
     exploit_instance = cve_2018_7841.CVE20187841Exploit()
 

tests/exploits/cve_2023_27997_test.py

@@ -28,7 +28,7 @@ def side_effect(*args, **kwargs):  # type: ignore[no-untyped-def]
         seed += 1
         return mock_response
 
-    mocker.patch("cloudscraper.CloudScraper.post", side_effect=side_effect)
+    mocker.patch("agent.definitions.HttpSession.post", side_effect=side_effect)
 
     target = definitions.Target(scheme="https", host="91.135.170.42", port=8443)
     vulnerabilities = exploit_instance.check(target)

tests/exploits/cve_2024_23113_test.py

@@ -173,7 +173,7 @@ def testCVE202423113_withHTTPDetection_returnVulnerabilities(
     """Test HTTP-based detection for various Fortinet products."""
     mock_response = mock.MagicMock()
     mock_response.text = content
-    mocker.patch("cloudscraper.CloudScraper.get", return_value=mock_response)
+    mocker.patch("agent.definitions.HttpSession.get", return_value=mock_response)
 
     exploit_instance = cve_2024_23113.CVE202423113Exploit()
     target = definitions.Target("tcp", "192.168.1.1", 443)

tests/exploits/cve_2024_40725_test.py

@@ -94,7 +94,7 @@ def testCVE202440725_whenNoFilenamesFound_reportNothing(
     mocker: mock.Mock, caplog: pytest.LogCaptureFixture
 ) -> None:
     """Test check method when an IOError occurs reading filenames."""
-    mocker.patch("cloudscraper.create_scraper")
+    mocker.patch("agent.definitions.HttpSession")
     mocker.patch("builtins.open", side_effect=IOError("File not found"))
 
     exploit_instance = cve_2024_40725.ApacheSourceCodeDisclosureExploit()
@@ -111,7 +111,7 @@ def testCVE202440725_whenRequestException_reportLoggedErrors() -> None:
     with mock.patch(
         "agent.exploits.cve_2024_40725._read_filenames", return_value=["index.php"]
     ), mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.RequestException("Request error"),
     ), mock.patch("agent.exploits.cve_2024_40725.logger") as mock_logger:
         exploit_instance = cve_2024_40725.ApacheSourceCodeDisclosureExploit()
@@ -128,7 +128,7 @@ def testCVE202440725_whenRequestException_reportLoggedErrors() -> None:
 def testCVE202440725_whenHTTPError_reportLoggedErrors() -> None:
     """Test the check method when requests.Get raises an HTTPError."""
     with mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.exceptions.HTTPError(
             "HTTP error", response=requests.models.Response()
         ),
@@ -153,7 +153,7 @@ def testCVE202440725_whenHTTPError_reportLoggedErrors() -> None:
 def testCVE202440725_whenConnectionError_reportLoggedErrors() -> None:
     """Test the check method when requests.Get raises a ConnectionError."""
     with mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.exceptions.ConnectionError("Connection error"),
     ), mock.patch(
         "agent.exploits.cve_2024_40725._contains_code_content", return_value=True
@@ -176,7 +176,7 @@ def testCVE202440725_whenConnectionError_reportLoggedErrors() -> None:
 def testCVE202440725_whenTimeoutError_reportLoggedErrors() -> None:
     """Test the check method when requests.Get raises a Timeout."""
     with mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.exceptions.Timeout("Timeout error"),
     ), mock.patch(
         "agent.exploits.cve_2024_40725._contains_code_content", return_value=True

tests/exploits/cve_2024_8522_test.py

@@ -77,7 +77,7 @@ def testCheck_whenVulnerable_reportFinding(
     )
 
     # Patch the 'get' request to return the mocked response
-    with mock.patch("cloudscraper.CloudScraper.get", return_value=mock_response):
+    with mock.patch("agent.definitions.HttpSession.get", return_value=mock_response):
         exploit_instance = cve_2024_8522.LearnPressSQLInjectionExploit()
         target = definitions.Target("http", "example.com", 80)
 
@@ -116,7 +116,7 @@ def testCheck_whenRequestException_reportNoVulnerabilities(
 ) -> None:
     """Test the check method when a RequestException occurs."""
     with mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.RequestException("Request error"),
     ):
         exploit_instance = cve_2024_8522.LearnPressSQLInjectionExploit()
@@ -133,7 +133,7 @@ def testCheck_whenTimeoutError_reportLoggedErrors(
 ) -> None:
     """Test the check method when a Timeout error occurs."""
     with mock.patch(
-        "cloudscraper.CloudScraper.get",
+        "agent.definitions.HttpSession.get",
         side_effect=requests.exceptions.Timeout("Timeout error"),
     ):
         exploit_instance = cve_2024_8522.LearnPressSQLInjectionExploit()