Merge branch '2.7.0-branch' into 'develop'

v2.7.1

See merge request integrations/sdk/reversinglabs-sdk-py3!7

Author: MislavReversingLabs

.gitlab-ci.yml

@@ -4,3 +4,4 @@ include:
   - project: 'integrations/ops/release-pipeline'
     ref: master
     file: '/reversinglabs-sdk-py3/gitlab-ci.yml'
+

CHANGELOG.md

@@ -332,7 +332,7 @@ v2.5.1 (2024-04-02)
   - Implemented the default user agent string in embedded `FileAnalysis` calls.
 
 
-2.7.0 (2024-07-24)
+2.7.0 (2024-09-25)
 -------------------
 
 #### Improvements
@@ -346,6 +346,13 @@ v2.5.1 (2024-04-02)
   - Parameters `internet_simulation` and `sample_name` of the `DynamicAnalysis.detonate_sample` method are now deprecated. Use `**optional_parameters` instead.
 
 
+2.7.1 (2024-10-09)
+-------------------
+
+#### Improvements
+- **fie** module:
+  - Introduced a new module called **fie** which corresponds to the ReversingLabs **File Investigation Engine (FIE)** service. 
+  - The module currently has one class with four methods for sending files to FIE for analysis and fetching the short classification or more detailed analysis reports.
 
 
 ### Scheduled removals

README.md

@@ -928,7 +928,37 @@ class TitaniumScale(object):
 - `get_yara_id`
   - Retrieves the identifier of the current set of YARA rules on the TitaniumScale Worker instance.
 
+***
+
+## Module: fie
+A Python module representing the ReversingLabs File Inspection Engine platform.
+#### Class:
+```python
+class FileInspectionEngine(object):
+    def __init__(self, host, verify, proxies, user_agent)
+```
+#### Parameters:
+`host` - File Inspection Engine address
+`verify` - verify SSL certificate  
+`proxies` - optional proxies in use  
+`user_agent` - optional user agent string  
 
+#### Methods:
+- `test_connection`
+    - Creates a lightweight request towards the FIE scan API to test the connection.
+- `scan_using_file_path`
+    - Sends a file to the FIE for inspection and returns a simple verdict in the submit response.
+    - Uses a file path string as input.
+- `scan_using_open_file`
+    - Sends a file to the FIE for inspection and returns a simple verdict in the submit response.
+    - Uses an open file handle as input.
+- `report_using_file_path`
+    - Sends a file to the FIE for inspection and returns a more complex analysis report in the submit response.
+    - Uses a file path string as input.
+- `report_using_open_file`
+    - Sends a file to the FIE for inspection and returns a more complex analysis report in the submit response.
+    - Uses an open file handle as input.
+  
 ***
 
 ## Examples
@@ -1065,6 +1095,23 @@ results = titanium_scale.upload_sample_and_get_results(
 )
 ```
 
+#### File Inspection Engine
+```python
+from ReversingLabs.SDK.fie import FileInspectionEngine
+
+
+fie = FileInspectionEngine(
+    host="http://fie.address",
+    verify=True
+)
+
+results = fie.scan_using_file_path(
+    file_path="/local/path/to/file.exe"
+)
+
+print(results.json())
+```
+
 #### Error handling
 Each module raises corresponding custom exceptions according to the error status code returned in the response. 
 Custom exception classes that correspond to error status codes also carry the original response object in its entirety. 

ReversingLabs/SDK/__init__.py

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

ReversingLabs/SDK/fie.py

@@ -0,0 +1,175 @@
+"""
+author: Mislav Sever
+
+File Inspection Engine (FIE)
+A Python module for the ReversingLabs File Inspection Engine REST API.
+"""
+
+import requests
+from io import BytesIO
+
+from ReversingLabs.SDK.helper import DEFAULT_USER_AGENT, RESPONSE_CODE_ERROR_MAP, WrongInputError
+
+
+class FileInspectionEngine(object):
+
+	__SCAN_ENDPOINT = "/scan"
+	__REPORT_ENDPOINT = "/report"
+
+	def __init__(self, host, verify=True, proxies=None, user_agent=DEFAULT_USER_AGENT):
+		self._host = self.__validate_host(host)
+		self._url = "{host}{{endpoint}}".format(host=self._host)
+		self._verify = verify
+
+		self._headers = {"User-Agent": user_agent}
+
+		if proxies:
+			if not isinstance(proxies, dict):
+				raise WrongInputError("proxies parameter must be a dictionary.")
+			if len(proxies) == 0:
+				raise WrongInputError("proxies parameter can not be an empty dictionary.")
+		self._proxies = proxies
+
+	@staticmethod
+	def __validate_host(host):
+		"""Returns a formatted host URL including the protocol prefix.
+			:param host: URL string
+			:type host: str
+			:returns: formatted URL string
+			:rtype: str
+		"""
+		if not isinstance(host, str):
+			raise WrongInputError("host parameter must be string.")
+
+		if not host.startswith(("http://", "https://")):
+			raise WrongInputError("host parameter must contain a protocol definition at the beginning.")
+
+		host = host.rstrip("/")
+
+		return host
+
+	def test_connection(self):
+		"""Creates a lightweight request towards the FIE scan API to test the connection.
+		"""
+		fake_file = BytesIO(b'this is a sample text')
+
+		response = self.scan_using_open_file(
+			file_source=fake_file
+		)
+
+		self.__raise_on_error(response)
+
+		return
+
+	def scan_using_file_path(self, file_path):
+		"""Sends a file to the FIE for inspection and returns a simple verdict in the submit response.
+		Uses a file path string as input.
+			:param file_path: local path to the file
+			:type file_path: str
+			:return: response
+			:rtype: requests.Response
+		"""
+		if not isinstance(file_path, str):
+			raise WrongInputError("file_path must be a string.")
+
+		try:
+			file_handle = open(file_path, "rb")
+		except IOError as error:
+			raise WrongInputError("Error while opening file in 'rb' mode - {error}".format(error=str(error)))
+
+		response = self.__upload_file(
+			file_source=file_handle,
+			endpoint=self.__SCAN_ENDPOINT
+		)
+
+		return response
+
+	def scan_using_open_file(self, file_source):
+		"""Sends a file to the FIE for inspection and returns a simple verdict in the submit response.
+		Uses an open file handle as input.
+			:param file_source: open file in rb mode
+			:type file_source: file or BinaryIO
+			:return: response
+			:rtype: requests.Response
+		"""
+		response = self.__upload_file(
+			file_source=file_source,
+			endpoint=self.__SCAN_ENDPOINT
+		)
+
+		return response
+
+	def report_using_file_path(self, file_path):
+		"""Sends a file to the FIE for inspection and returns a more complex analysis report in the submit response.
+		Uses a file path string as input.
+			:param file_path: local path to the file
+			:type file_path: str
+			:return: response
+			:rtype: requests.Response
+		"""
+		if not isinstance(file_path, str):
+			raise WrongInputError("file_path must be a string.")
+
+		try:
+			file_handle = open(file_path, "rb")
+		except IOError as error:
+			raise WrongInputError("Error while opening file in 'rb' mode - {error}".format(error=str(error)))
+
+		response = self.__upload_file(
+			file_source=file_handle,
+			endpoint=self.__REPORT_ENDPOINT
+		)
+
+		return response
+
+	def report_using_open_file(self, file_source):
+		"""Sends a file to the FIE for inspection and returns a more complex analysis report in the submit response.
+		Uses an open file handle as input.
+			:param file_source: open file in rb mode
+			:type file_source: file or BinaryIO
+			:return: response
+			:rtype: requests.Response
+		"""
+		response = self.__upload_file(
+			file_source=file_source,
+			endpoint=self.__REPORT_ENDPOINT
+		)
+
+		return response
+
+	def __upload_file(self, file_source, endpoint):
+		"""Internal method for utilizing the FIE endpoints.
+			:param file_source: open file in rb mode
+			:type file_source: file or BinaryIO
+			:param endpoint: endpoint string
+			:type endpoint: str
+			:return: response
+			:rtype: requests.Response
+		"""
+		if not hasattr(file_source, "read"):
+			raise WrongInputError("file_source parameter must be a file open in 'rb' mode.")
+
+		url = self._url.format(endpoint=endpoint)
+
+		response = requests.post(
+			url=url,
+			data=file_source,
+			verify=self._verify,
+			proxies=self._proxies,
+			headers=self._headers
+		)
+
+		self.__raise_on_error(response)
+
+		return response
+
+	@staticmethod
+	def __raise_on_error(response):
+		"""Accepts a response object for validation and raises an exception if an error status code is received.
+			:return: response
+			:rtype: requests.Response
+		"""
+		exception = RESPONSE_CODE_ERROR_MAP.get(response.status_code, None)
+		if not exception:
+			return
+		raise exception(response_object=response)

setup.py

@@ -43,6 +43,7 @@
     ],
     project_urls={
         "Documentation": "https://github.com/reversinglabs/reversinglabs-sdk-py3/blob/main/README.md",
-        "Source": "https://github.com/reversinglabs/reversinglabs-sdk-py3"
+        "Source": "https://github.com/reversinglabs/reversinglabs-sdk-py3",
+        "Changes": "https://github.com/reversinglabs/reversinglabs-sdk-py3/blob/main/CHANGELOG.md"
     },
 )

tests/test_fie.py

@@ -0,0 +1,53 @@
+import pytest
+from unittest import mock
+from ReversingLabs.SDK import __version__
+from ReversingLabs.SDK.fie import FileInspectionEngine
+from ReversingLabs.SDK.helper import WrongInputError, DEFAULT_USER_AGENT
+
+
+def test_fie_object():
+	invalid_host = "my.host"
+	valid_host = f"https://{invalid_host}"
+
+	fie = FileInspectionEngine(
+		host=valid_host,
+		verify=True
+	)
+
+	assert fie._url == valid_host + "{endpoint}"
+
+	with pytest.raises(WrongInputError, match=r"host parameter must contain a protocol definition at the beginning."):
+		FileInspectionEngine(host=invalid_host)
+
+	user_agent = fie._headers.get("User-Agent")
+	assert __version__ in user_agent
+
+
+@pytest.fixture
+def requests_mock():
+	with mock.patch('ReversingLabs.SDK.fie.requests', autospec=True) as requests_mock:
+		yield requests_mock
+
+
+class TestFIE:
+	host = "http://my.host"
+
+	@classmethod
+	def setup_class(cls):
+		cls.fie = FileInspectionEngine(cls.host)
+
+	def test_scan_using_path(self):
+		with pytest.raises(WrongInputError, match=r"file_path must be a string."):
+			self.fie.scan_using_file_path(file_path=123)
+
+	def test_scan_using_file(self):
+		with pytest.raises(WrongInputError, match=r"file_source parameter must be a file open in 'rb' mode."):
+			self.fie.scan_using_open_file(file_source="/path/to/file")
+
+	def test_report_using_path(self):
+		with pytest.raises(WrongInputError, match=r"file_path must be a string."):
+			self.fie.report_using_file_path(file_path=123)
+
+	def test_report_using_file(self):
+		with pytest.raises(WrongInputError, match=r"file_source parameter must be a file open in 'rb' mode."):
+			self.fie.report_using_open_file(file_source="/path/to/file")