Merge pull request #51 from MislavReversingLabs/main

2.6.2

Author: MislavReversingLabs

CHANGELOG.md

@@ -310,5 +310,12 @@ v2.5.1 (2024-04-02)
 2.6.1 (2024-07-03)
 -------------------
 
+#### Improvements
+- Added more unit tests for all currently available modules.
+
+
+2.6.2 (2024-07-10)
+-------------------
+
 #### Improvements
 - Added more unit tests for all currently available modules.

ReversingLabs/SDK/__init__.py

@@ -5,4 +5,4 @@
 A Python SDK for communicating with ReversingLabs services.
 """
 
-__version__ = "2.6.1"
+__version__ = "2.6.2"

ReversingLabs/SDK/ticloud.py

@@ -5879,47 +5879,45 @@ def feed_query(self, category, filter, weeks=0, all_time=False):
         filtered by category. The service can return a list of malware family names
         newly added to each category, the number of unieque new samples added for each
         malware family in a category, and a list of top 20 malware families per category
-            param: category: Corresponds to the verticals feed category the user is requesting to access.
+            :param category: Corresponds to the verticals feed category the user is requesting to access.
             Only one category can be requested in each query. Note that the response for the 'exploit'
             category contains addional 'scanner_coverage' data not found in other categories.
             Enum: 'financial', 'retail', 'ransomware', 'apt', 'exploit', 'configuration'
-            type: category: str
-            param: filter: applied to filter data to request. Enum: 'first_seen', 'counts', 'top_list'
-            type: filter: str
-            param: weeks: specifies the number of weeks for which the data will be returned in response
-            type: weeks: int
-            param: all_time: Instructs the service to return all available data for the requested category
-            type all_time: boolean
+            :type category: str
+            :param filter: applied to filter data to request. Enum: 'first_seen', 'counts', 'top_list'
+            :type filter: str
+            :param weeks: specifies the number of weeks for which the data will be returned in response
+            :type weeks: int
+            :param all_time: Instructs the service to return all available data for the requested category
+            :type all_time: boolean
         """
         if category not in VERTICAL_FEEDS_CATEGORIES:
             raise WrongInputError("Only the following categories are allowed: {category}".format(
                 category=VERTICAL_FEEDS_CATEGORIES))
 
-        if filter.lower() not in ("counts", "top_list", "first_seen"):
+        if filter not in ("counts", "top_list", "first_seen"):
             raise WrongInputError("Only the following filters are allowed: 'counts', 'top_list' and 'first_seen'")
 
+        query_params = {
+            "format": "json"
+        }
+
         if isinstance(weeks, int):
 
             if weeks not in range(0, 30):
                 raise WrongInputError("The value for weeks can be a number between 0 and 30")
 
-            query_params = {
-                "weeks": weeks,
-                "format": "json"
-            }
-
-            raise WrongInputError("Weeks needs to be provided as integer")
+            query_params["weeks"] = weeks
 
         if all_time:
+            query_params["all_time"] = "true"
 
-            query_params = {
-                "all_time": "true",
-                "format": "json"
-            }
+            if "weeks" in query_params:
+                raise WrongInputError("The all_time parameter can not be used together with weeks.")
 
         endpoint = self.__API_FEED_ENDPOINT.format(
-            category = category,
-            filter = filter
+            category=category,
+            filter=filter
         )
 
         url = self._url.format(endpoint=endpoint)

tests/test_a1000.py

@@ -89,6 +89,60 @@ def test_sample_from_url(self, requests_mock):
 			files=None
 		)
 
+	def test_sample_from_file(self):
+		with pytest.raises(WrongInputError, match=r"file_source parameter must be a file open in 'rb' mode."):
+			self.a1000.upload_sample_from_file(file_source="/path/to/file")
+
+	def test_file_status(self, requests_mock):
+		self.a1000.file_analysis_status(sample_hashes=[SHA1, SHA1], sample_status="MALICIOUS")
+
+		expected_url = f"{self.host}/api/samples/status/"
+
+		requests_mock.post.assert_called_with(
+			url=expected_url,
+			verify=True,
+			proxies=None,
+			headers={"User-Agent": DEFAULT_USER_AGENT, "Authorization": f"Token {self.token}"},
+			params={"status": "MALICIOUS"},
+			json=None,
+			data={"hash_values": [SHA1, SHA1]},
+			files=None
+		)
+
+	def test_url_status(self, requests_mock):
+		self.a1000.check_submitted_url_status(task_id="aaa")
+
+		expected_url = f"{self.host}/api/uploads/v2/url-samples/aaa"
+
+		requests_mock.get.assert_called_with(
+			url=expected_url,
+			verify=True,
+			proxies=None,
+			headers={"User-Agent": DEFAULT_USER_AGENT, "Authorization": f"Token {self.token}"},
+			params=None
+		)
+
+	def test_url_report(self, requests_mock):
+		with pytest.raises(WrongInputError, match=r"task_id parameter must be a string."):
+			self.a1000.get_submitted_url_report(task_id=123, retry=False)
+
+	def test_detailed_report(self, requests_mock):
+		with pytest.raises(WrongInputError, match=r"fields parameter must be a list of strings."):
+			self.a1000.get_detailed_report_v2(sample_hashes=SHA1, fields="field1,field2")
+
+	def test_download_sample(self, requests_mock):
+		self.a1000.download_sample(sample_hash=SHA1)
+
+		expected_url = f"{self.host}/api/samples/{SHA1}/download/"
+
+		requests_mock.get.assert_called_with(
+			url=expected_url,
+			verify=True,
+			proxies=None,
+			headers={"User-Agent": DEFAULT_USER_AGENT, "Authorization": f"Token {self.token}"},
+			params=None
+		)
+
 	def test_wrong_id(self, requests_mock):
 		with pytest.raises(WrongInputError, match=r"task_id parameter must be a string."):
 			self.a1000.get_submitted_url_report(task_id=123, retry=False)
@@ -304,6 +358,9 @@ def test_start_yara_retro(self, requests_mock):
 			files=None
 		)
 
+	def test_yara_retro_status(self, requests_mock):
+		pass
+
 	def test_wrong_operation(self, requests_mock):
 		with pytest.raises(WrongInputError, match=r"operation parameter must be either 'START' or 'STOP'"):
 			self.a1000.start_or_stop_yara_local_retro_scan("BEGIN")
@@ -343,7 +400,7 @@ def test_list_containers(self, requests_mock):
 			files=None
 		)
 
-	def test_network_report(self, requests_mock):
+	def test_domain_report(self, requests_mock):
 		domain = "some.test.domain"
 
 		self.a1000.network_domain_report(domain)
@@ -357,3 +414,25 @@ def test_network_report(self, requests_mock):
 			headers={"User-Agent": DEFAULT_USER_AGENT, "Authorization": f"Token {self.token}"},
 			params=None
 		)
+
+	def test_ip_to_domain(self, requests_mock):
+		self.a1000.network_ip_to_domain("1.2.3.4")
+
+		params = {
+			"page": None,
+			"page_size": 500
+		}
+
+		expected_url = f"{self.host}/api/network-threat-intel/ip/1.2.3.4/resolutions/"
+
+		requests_mock.get.assert_called_with(
+			url=expected_url,
+			verify=True,
+			proxies=None,
+			headers={"User-Agent": DEFAULT_USER_AGENT, "Authorization": f"Token {self.token}"},
+			params=params
+		)
+
+	def test_files_from_ip(self, requests_mock):
+		with pytest.raises(WrongInputError, match=r"is allowed as the classification input"):
+			self.a1000.network_files_from_ip(ip_addr="1.2.3.4", classification="KNOWN")