Merge pull request #46 from citizenjosh/pixeebot/drip-2024-01-25-pixe…

…e-python/add-requests-timeouts Add timeout to `requests` calls
DelineaXPM · Jul 3, 2024 · 51e4005 · 51e4005
2 parents 2a06f46 + 319db5b
commit 51e4005
Showing 1 changed file with 8 additions and 8 deletions.
diff --git a/delinea/secrets/server.py b/delinea/secrets/server.py
@@ -167,7 +167,7 @@ def get_access_grant(token_url, grant_request):
                 other than a valid Access Grant
         """
 
-        response = requests.post(token_url, grant_request)
+        response = requests.post(token_url, grant_request, timeout=60)
 
         try:  # TSS returns a 200 (OK) containing HTML for some error conditions
             return json.loads(SecretServer.process(response).content)
@@ -301,14 +301,14 @@ def get_secret_json(self, id, query_params=None):
         endpoint_url = f"{self.api_url}/secrets/{id}"
 
         if query_params is None:
-            return self.process(requests.get(endpoint_url, headers=self.headers())).text
+            return self.process(requests.get(endpoint_url, headers=self.headers(), timeout=60)).text
         else:
             return self.process(
                 requests.get(
                     endpoint_url,
                     params=query_params,
                     headers=self.headers(),
-                )
+                timeout=60)
             ).text
 
     def get_secret(self, id, fetch_file_attachments=True, query_params=None):
@@ -343,15 +343,15 @@ def get_secret(self, id, fetch_file_attachments=True, query_params=None):
                     endpoint_url = f"{self.api_url}/secrets/{id}/fields/{item['slug']}"
                     if query_params is None:
                         item["itemValue"] = self.process(
-                            requests.get(endpoint_url, headers=self.headers())
+                            requests.get(endpoint_url, headers=self.headers(), timeout=60)
                         )
                     else:
                         item["itemValue"] = self.process(
                             requests.get(
                                 endpoint_url,
                                 params=query_params,
                                 headers=self.headers(),
-                            )
+                            timeout=60)
                         )
         return secret
 
@@ -391,14 +391,14 @@ def search_secrets(self, query_params=None):
         endpoint_url = f"{self.api_url}/secrets"
 
         if query_params is None:
-            return self.process(requests.get(endpoint_url, headers=self.headers())).text
+            return self.process(requests.get(endpoint_url, headers=self.headers(), timeout=60)).text
         else:
             return self.process(
                 requests.get(
                     endpoint_url,
                     params=query_params,
                     headers=self.headers(),
-                )
+                timeout=60)
             ).text
 
     def get_secret_ids_by_folderid(self, folder_id):
@@ -417,7 +417,7 @@ def get_secret_ids_by_folderid(self, folder_id):
         params = {"filter.folderId": folder_id}
         endpoint_url = f"{self.api_url}/secrets/search-total"
         params["take"] = self.process(
-            requests.get(endpoint_url, params=params, headers=self.headers())
+            requests.get(endpoint_url, params=params, headers=self.headers(), timeout=60)
         ).text
         response = self.search_secrets(query_params=params)