Validate Input URL With Parser (#194)

CHANGELOG.md

@@ -2,6 +2,14 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [3.21.0] - 2024-11-28
+### Fixed
+- Fixed `[NA-SPOT-SDK-03] URL Path Traversal in Python SDK` from pentesting report
+
+## [3.20.0] - 2024-11-20
+### Fixed
+- Fixed static analysis issues reported by semgrep tool
+
 ## [3.19.0] - 2024-11-06
 ### Added
 - Added `SpotSizeAttributes` model for Azure Stateful Nodes

requirements.txt

@@ -2,4 +2,4 @@ PyYaml==6.0.2
 requests==2.32.3
 pydoc-markdown==4.8.2
 mock==5.1.0
-pytest==8.3.2
+pytest==8.3.3

spotinst_sdk2/client.py

@@ -4,6 +4,9 @@
 
 import logging
 import requests
+
+import urllib3.util
+
 from .version import __version__
 
 VAR_SPOTINST_LOG_LEVEL = 'SPOTINST_LOG_LEVEL'
@@ -41,6 +44,15 @@ def __init__(self, session=None,
 
         self.timeout = timeout
 
+    def validate_url(self, url):
+        self.print_output("Input Url - " + self.base_url + url)
+        self.print_output("Parsed Url - " +
+                          urllib3.util.parse_url(self.base_url + url).url)
+        if urllib3.util.parse_url(self.base_url + url).url == self.base_url + url:
+            return self.base_url + url
+        else:
+            raise Exception("UNSAFE_URL")
+
     def send_get(self, url, entity_name, query_params=None):
         agent = self.resolve_user_agent()
 
@@ -61,9 +73,9 @@ def send_get(self, url, entity_name, query_params=None):
         self.print_output("Request Query Params - " + str(query_params))
 
         result = requests.get(
-            self.base_url + url, params=query_params, headers=headers, timeout=self.timeout)
+            self.validate_url(url), params=query_params, headers=headers, timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             data = json.loads(result.content.decode('utf-8'))
@@ -86,9 +98,9 @@ def send_delete(self, url, entity_name):
         self.print_output("Request Query Params - " + str(query_params))
 
         result = requests.delete(
-            self.base_url + url, params=query_params, headers=headers, timeout=self.timeout)
+            self.validate_url(url), params=query_params, headers=headers, timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             return True
@@ -111,13 +123,13 @@ def send_delete_with_body(self, body, url, entity_name):
         self.print_output("Request Body - " + str(body))
 
         result = requests.delete(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             headers=headers,
             data=body,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             return True
@@ -140,12 +152,12 @@ def send_delete_with_params(self, url, entity_name, user_query_params):
         self.print_output("Request Query Params - " + str(query_params))
 
         result = requests.delete(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             headers=headers,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             return True
@@ -173,13 +185,13 @@ def send_post(self, url, entity_name, body=None, query_params=None):
         self.print_output("Request Body - " + str(body))
 
         result = requests.post(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             data=body,
             headers=headers,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             data = json.loads(result.content.decode('utf-8'))
@@ -205,13 +217,13 @@ def send_post_with_params(self, url, entity_name, body, user_query_params):
         self.print_output("Request Body - " + str(body))
 
         result = requests.post(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             data=body,
             headers=headers,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             data = json.loads(result.content.decode('utf-8'))
@@ -240,13 +252,13 @@ def send_put(self, url, entity_name, query_params=None, body=None):
         self.print_output("Request Body - " + str(body))
 
         result = requests.put(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             data=body,
             headers=headers,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             data = json.loads(result.content.decode('utf-8'))
@@ -271,13 +283,13 @@ def send_put_with_params(self, body, url, entity_name, user_query_params):
         self.print_output("Request Body - " + str(body))
 
         result = requests.put(
-            self.base_url + url,
+            self.validate_url(url),
             params=query_params,
             data=body,
             headers=headers,
             timeout=self.timeout)
 
-        if result.status_code == requests.codes.ok:
+        if result.status_code == requests.codes['ok']:
             self.print_output("Success")
             self.print_output("Response - " + str(result.json()))
             data = json.loads(result.content.decode('utf-8'))

spotinst_sdk2/version.py

@@ -1 +1 @@
-__version__ = '3.20.0'
+__version__ = '3.21.0'