vtapi3.vtapi3files.VirusTotalAPIFiles.get_file_id() - Code Metrics - drobotun/virustotalapi3 - Measure and Improve Code Quality continuously with Scrutinizer

VirusTotalAPIFiles.get_file_id() B
last analyzed 2023-03-02 09:52 UTC

↳ Parent: vtapi3.vtapi3files

Complexity

Conditions

Size

Total Lines	33
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	17
nop	2
dl	0
loc	33
rs	8
c	0
b	0
f	0

"""The module describes the VirusTotalAPIFiles class

   Author: Evgeny Drobotun (c) 2019
   License: MIT (https://github.com/drobotun/virustotalapi3/blob/master/LICENSE)

   More information: https://virustotalapi3.readthedocs.io/en/latest/file_class.html
"""

import hashlib
import errno
import requests

from .vtapi3base import VirusTotalAPI
from .vtapi3error import VirusTotalAPIError

class VirusTotalAPIFiles(VirusTotalAPI):
    """The analysis new files and retrieving information about any file from the VirusTotal database
       methods are defined in the class.

       Methods:
          get_file_id(): Get SHA256, SHA1 or MD5 file identifier.
          upload(): Upload and analyse a file.
          get_upload_url(): Get a URL for uploading files larger than 32MB.
          get_report(): Retrieve information about a file.
          reanalyse(): Reanalyse a file already in VirusTotal.
          get_comments(): Retrieve comments for a file.
          put_comments(): Add a comment to a file.
          get_votes(): Retrieve votes for a file.
          put_votes(): Add a votes to a file.
          get_relationship(): Retrieve objects related to a file.
          get_behaviours(): Get the PCAP for the sandbox.
          get_download_url(): Get a download URL for a file (special privileges required).
          get_download(): Download a file (special privileges required).
    """

    @staticmethod
    def get_file_id(file_path, hash_alg='sha256'):
        """Get SHA256, SHA1 or MD5 file identifier.

           Args:
              file_path: Path to the file to be scanned (str).
              hash: Necessary identifier ('sha256', 'sha1' or 'md5'). The default value is
                 'sha256'.

           Return:
              The SHA256, SHA1 or MD5 identifier of the file.

           Exception
              VirusTotalAPIError(File not found): In case the file is not found.
              VirusTotalAPIError(Permission error): In case do not have access rights to the file.
              VirusTotalAPIError(IO Error): If an IO error occurs during file operations.
        """
        buffer_size = 65536
        hasher = hashlib.new(hash_alg)
        try:
            with open(file_path, 'rb') as file:
                buffer = file.read(buffer_size)
                while len(buffer) > 0:
                    hasher.update(buffer)
                    buffer = file.read(buffer_size)
        except FileNotFoundError:
            raise VirusTotalAPIError('File not found', errno.ENOENT)
        except PermissionError:
            raise VirusTotalAPIError('Permission error', errno.EPERM)
        except OSError:
            raise VirusTotalAPIError('IO Error', errno.EIO)
        else:
            return hasher.hexdigest()

    def upload(self, file_path):
        """Upload and analyse a file.

        Args:
           file_path: Path to the file to be scanned (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
           VirusTotalAPIError(File not found): In case the file you want to upload to the server is
              not found.
           VirusTotalAPIError(Permission error): In case do not have access rights to the file.
           VirusTotalAPIError(IO Error): If an IO error occurs during file operations.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files'
        try:
            with open(file_path, 'rb') as file:
                files = {'file': (file_path, file)}
                response = requests.post(api_url, headers=self.headers, files=files,
                                         timeout=self.timeout, proxies=self.proxies)
        except FileNotFoundError:
            raise VirusTotalAPIError('File not found', errno.ENOENT)
        except PermissionError:
            raise VirusTotalAPIError('Permission error', errno.EPERM)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        except OSError:
            raise VirusTotalAPIError('IO Error', errno.EIO)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_upload_url(self):
        """Get a URL for uploading files larger than 32MB.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files/upload_url'
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_report(self, file_id):
        """Retrieve information about a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).

       Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files/' + file_id
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def analyse(self, file_id):
        """Reanalyse a file already in VirusTotal.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files/' + file_id + '/analyse'
        try:
            response = requests.post(api_url, headers=self.headers,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_comments(self, file_id, limit=10, cursor='""'):
        """Retrieve comments for a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/files/' + file_id + '/comments'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_comments(self, file_id, text):
        """Add a comment to a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
           text: Text of the comment (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        comments = {"data": {'type': 'comment', 'attributes': {'text': text}}}
        api_url = self.base_url + '/files/' + file_id + '/comments'
        try:
            response = requests.post(api_url, headers=self.headers, json=comments,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_votes(self, file_id, limit=10, cursor='""'):
        """Retrieve votes for a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/files/' + file_id + '/votes'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_votes(self, file_id, malicious=False):
        """Add a votes to a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
           malicious: Determines a malicious (True) or harmless (False) file (bool).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        if malicious:
            verdict = 'malicious'
        else:
            verdict = 'harmless'
        votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}}
        api_url = self.base_url + '/files/' + file_id + '/votes'
        try:
            response = requests.post(api_url, headers=self.headers, json=votes,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_relationship(self, file_id, relationship='/behaviours', limit=10, cursor='""'):
        """Retrieve objects related to a file.

        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
           relationship: Relationship name (str). The default value is "/behaviours".
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/files/' + file_id + relationship
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_behaviours(self, sandbox_id):
        """Get the PCAP for the sandbox.

        Args:
           sandbox_id: Identifier obtained using the 'get_relationship' method with
              the value of the 'relationship' argument equal to 'behaviours' (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/file_behaviours/' + sandbox_id + '/pcap'
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_download_url(self, file_id):
        """Get a download URL for a file. This function requires special privileges
           (you need a private key to access the VirusTotal API).
        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files/' + file_id + '/download_url'
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_download(self, file_id):
        """Download a file. This function requires special privileges (you need a private
           key to access the VirusTotal API).
        Args:
           file_id: SHA-256, SHA-1 or MD5 identifying the file (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/files/' + file_id + '/download'
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content


1			"""The module describes the VirusTotalAPIFiles class
2
3			Author: Evgeny Drobotun (c) 2019
4			License: MIT (https://github.com/drobotun/virustotalapi3/blob/master/LICENSE)
5
6			More information: https://virustotalapi3.readthedocs.io/en/latest/file_class.html
7			"""
8
9			import hashlib
10			import errno
11			import requests
12
13			from .vtapi3base import VirusTotalAPI
14			from .vtapi3error import VirusTotalAPIError
15
16			class VirusTotalAPIFiles(VirusTotalAPI):
17			"""The analysis new files and retrieving information about any file from the VirusTotal database
18			methods are defined in the class.
19
20			Methods:
21			get_file_id(): Get SHA256, SHA1 or MD5 file identifier.
22			upload(): Upload and analyse a file.
23			get_upload_url(): Get a URL for uploading files larger than 32MB.
24			get_report(): Retrieve information about a file.
25			reanalyse(): Reanalyse a file already in VirusTotal.
26			get_comments(): Retrieve comments for a file.
27			put_comments(): Add a comment to a file.
28			get_votes(): Retrieve votes for a file.
29			put_votes(): Add a votes to a file.
30			get_relationship(): Retrieve objects related to a file.
31			get_behaviours(): Get the PCAP for the sandbox.
32			get_download_url(): Get a download URL for a file (special privileges required).
33			get_download(): Download a file (special privileges required).
34			"""
35
36			@staticmethod
37			def get_file_id(file_path, hash_alg='sha256'):
38			"""Get SHA256, SHA1 or MD5 file identifier.
39
40			Args:
41			file_path: Path to the file to be scanned (str).
42			hash: Necessary identifier ('sha256', 'sha1' or 'md5'). The default value is
43			'sha256'.
44
45			Return:
46			The SHA256, SHA1 or MD5 identifier of the file.
47
48			Exception
49			VirusTotalAPIError(File not found): In case the file is not found.
50			VirusTotalAPIError(Permission error): In case do not have access rights to the file.
51			VirusTotalAPIError(IO Error): If an IO error occurs during file operations.
52			"""
53			buffer_size = 65536
54			hasher = hashlib.new(hash_alg)
55			try:
56			with open(file_path, 'rb') as file:
57			buffer = file.read(buffer_size)
58			while len(buffer) > 0:
59			hasher.update(buffer)
60			buffer = file.read(buffer_size)
61			except FileNotFoundError:
62			raise VirusTotalAPIError('File not found', errno.ENOENT)
63			except PermissionError:
64			raise VirusTotalAPIError('Permission error', errno.EPERM)
65			except OSError:
66			raise VirusTotalAPIError('IO Error', errno.EIO)
67			else:
68			return hasher.hexdigest()
69
70			def upload(self, file_path):
71			"""Upload and analyse a file.
72
73			Args:
74			file_path: Path to the file to be scanned (str).
75
76			Return:
77			The response from the server as a byte sequence.
78
79			Exception
80			VirusTotalAPIError(Connection error): In case of server connection errors.
81			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
82			VirusTotalAPIError(File not found): In case the file you want to upload to the server is
83			not found.
84			VirusTotalAPIError(Permission error): In case do not have access rights to the file.
85			VirusTotalAPIError(IO Error): If an IO error occurs during file operations.
86			"""
87			self._last_http_error = None
88			self._last_result = None
89			api_url = self.base_url + '/files'
90			try:
91			with open(file_path, 'rb') as file:
92			files = {'file': (file_path, file)}
93			response = requests.post(api_url, headers=self.headers, files=files,
94			timeout=self.timeout, proxies=self.proxies)
95			except FileNotFoundError:
96			raise VirusTotalAPIError('File not found', errno.ENOENT)
97			except PermissionError:
98			raise VirusTotalAPIError('Permission error', errno.EPERM)
99			except requests.exceptions.Timeout:
100			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
101			except requests.exceptions.ConnectionError:
102			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
103			except OSError:
104			raise VirusTotalAPIError('IO Error', errno.EIO)
105			else:
106			self._last_http_error = response.status_code
107			self._last_result = response.content
108			return response.content
109
110			def get_upload_url(self):
111			"""Get a URL for uploading files larger than 32MB.
112
113			Return:
114			The response from the server as a byte sequence.
115
116			Exception
117			VirusTotalAPIError(Connection error): In case of server connection errors.
118			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
119			"""
120			self._last_http_error = None
121			self._last_result = None
122			api_url = self.base_url + '/files/upload_url'
123			try:
124			response = requests.get(api_url, headers=self.headers,
125			timeout=self.timeout, proxies=self.proxies)
126			except requests.exceptions.Timeout:
127			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
128			except requests.exceptions.ConnectionError:
129			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
130			else:
131			self._last_http_error = response.status_code
132			self._last_result = response.content
133			return response.content
134
135			def get_report(self, file_id):
136			"""Retrieve information about a file.
137
138			Args:
139			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
140
141			Return:
142			The response from the server as a byte sequence.
143
144			Exception
145			VirusTotalAPIError(Connection error): In case of server connection errors.
146			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
147			"""
148			self._last_http_error = None
149			self._last_result = None
150			api_url = self.base_url + '/files/' + file_id
151			try:
152			response = requests.get(api_url, headers=self.headers,
153			timeout=self.timeout, proxies=self.proxies)
154			except requests.exceptions.Timeout:
155			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
156			except requests.exceptions.ConnectionError:
157			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
158			else:
159			self._last_http_error = response.status_code
160			self._last_result = response.content
161			return response.content
162
163			def analyse(self, file_id):
164			"""Reanalyse a file already in VirusTotal.
165
166			Args:
167			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
168
169			Return:
170			The response from the server as a byte sequence.
171
172			Exception
173			VirusTotalAPIError(Connection error): In case of server connection errors.
174			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
175			"""
176			self._last_http_error = None
177			self._last_result = None
178			api_url = self.base_url + '/files/' + file_id + '/analyse'
179			try:
180			response = requests.post(api_url, headers=self.headers,
181			timeout=self.timeout, proxies=self.proxies)
182			except requests.exceptions.Timeout:
183			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
184			except requests.exceptions.ConnectionError:
185			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
186			else:
187			self._last_http_error = response.status_code
188			self._last_result = response.content
189			return response.content
190
191			def get_comments(self, file_id, limit=10, cursor='""'):
192			"""Retrieve comments for a file.
193
194			Args:
195			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
196			limit: Maximum number of comments to retrieve (int). The default value is 10.
197			cursor: Continuation cursor (str). The default value is ''.
198
199			Return:
200			The response from the server as a byte sequence.
201
202			Exception
203			VirusTotalAPIError(Connection error): In case of server connection errors.
204			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
205			"""
206			self._last_http_error = None
207			self._last_result = None
208			query_string = {'limit': str(limit), 'cursor': cursor}
209			api_url = self.base_url + '/files/' + file_id + '/comments'
210			try:
211			response = requests.get(api_url, headers=self.headers, params=query_string,
212			timeout=self.timeout, proxies=self.proxies)
213			except requests.exceptions.Timeout:
214			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
215			except requests.exceptions.ConnectionError:
216			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
217			else:
218			self._last_http_error = response.status_code
219			self._last_result = response.content
220			return response.content
221
222			def put_comments(self, file_id, text):
223			"""Add a comment to a file.
224
225			Args:
226			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
227			text: Text of the comment (str).
228
229			Return:
230			The response from the server as a byte sequence.
231
232			Exception
233			VirusTotalAPIError(Connection error): In case of server connection errors.
234			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
235			"""
236			self._last_http_error = None
237			self._last_result = None
238			comments = {"data": {'type': 'comment', 'attributes': {'text': text}}}
239			api_url = self.base_url + '/files/' + file_id + '/comments'
240			try:
241			response = requests.post(api_url, headers=self.headers, json=comments,
242			timeout=self.timeout, proxies=self.proxies)
243			except requests.exceptions.Timeout:
244			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
245			except requests.exceptions.ConnectionError:
246			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
247			else:
248			self._last_http_error = response.status_code
249			self._last_result = response.content
250			return response.content
251
252			def get_votes(self, file_id, limit=10, cursor='""'):
253			"""Retrieve votes for a file.
254
255			Args:
256			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
257			limit: Maximum number of comments to retrieve (int). The default value is 10.
258			cursor: Continuation cursor (str). The default value is ''.
259
260			Return:
261			The response from the server as a byte sequence.
262
263			Exception
264			VirusTotalAPIError(Connection error): In case of server connection errors.
265			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
266			"""
267			self._last_http_error = None
268			self._last_result = None
269			query_string = {'limit': str(limit), 'cursor': cursor}
270			api_url = self.base_url + '/files/' + file_id + '/votes'
271			try:
272			response = requests.get(api_url, headers=self.headers, params=query_string,
273			timeout=self.timeout, proxies=self.proxies)
274			except requests.exceptions.Timeout:
275			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
276			except requests.exceptions.ConnectionError:
277			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
278			else:
279			self._last_http_error = response.status_code
280			self._last_result = response.content
281			return response.content
282
283			def put_votes(self, file_id, malicious=False):
284			"""Add a votes to a file.
285
286			Args:
287			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
288			malicious: Determines a malicious (True) or harmless (False) file (bool).
289
290			Return:
291			The response from the server as a byte sequence.
292
293			Exception
294			VirusTotalAPIError(Connection error): In case of server connection errors.
295			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
296			"""
297			self._last_http_error = None
298			self._last_result = None
299			if malicious:
300			verdict = 'malicious'
301			else:
302			verdict = 'harmless'
303			votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}}
304			api_url = self.base_url + '/files/' + file_id + '/votes'
305			try:
306			response = requests.post(api_url, headers=self.headers, json=votes,
307			timeout=self.timeout, proxies=self.proxies)
308			except requests.exceptions.Timeout:
309			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
310			except requests.exceptions.ConnectionError:
311			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
312			else:
313			self._last_http_error = response.status_code
314			self._last_result = response.content
315			return response.content
316
317			def get_relationship(self, file_id, relationship='/behaviours', limit=10, cursor='""'):
318			"""Retrieve objects related to a file.
319
320			Args:
321			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
322			relationship: Relationship name (str). The default value is "/behaviours".
323			limit: Maximum number of comments to retrieve (int). The default value is 10.
324			cursor: Continuation cursor (str). The default value is ''.
325
326			Return:
327			The response from the server as a byte sequence.
328
329			Exception
330			VirusTotalAPIError(Connection error): In case of server connection errors.
331			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
332			"""
333			self._last_http_error = None
334			self._last_result = None
335			query_string = {'limit': str(limit), 'cursor': cursor}
336			api_url = self.base_url + '/files/' + file_id + relationship
337			try:
338			response = requests.get(api_url, headers=self.headers, params=query_string,
339			timeout=self.timeout, proxies=self.proxies)
340			except requests.exceptions.Timeout:
341			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
342			except requests.exceptions.ConnectionError:
343			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
344			else:
345			self._last_http_error = response.status_code
346			self._last_result = response.content
347			return response.content
348
349			def get_behaviours(self, sandbox_id):
350			"""Get the PCAP for the sandbox.
351
352			Args:
353			sandbox_id: Identifier obtained using the 'get_relationship' method with
354			the value of the 'relationship' argument equal to 'behaviours' (str).
355
356			Return:
357			The response from the server as a byte sequence.
358
359			Exception
360			VirusTotalAPIError(Connection error): In case of server connection errors.
361			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
362			"""
363			self._last_http_error = None
364			self._last_result = None
365			api_url = self.base_url + '/file_behaviours/' + sandbox_id + '/pcap'
366			try:
367			response = requests.get(api_url, headers=self.headers,
368			timeout=self.timeout, proxies=self.proxies)
369			except requests.exceptions.Timeout:
370			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
371			except requests.exceptions.ConnectionError:
372			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
373			else:
374			self._last_http_error = response.status_code
375			self._last_result = response.content
376			return response.content
377
378			def get_download_url(self, file_id):
379			"""Get a download URL for a file. This function requires special privileges
380			(you need a private key to access the VirusTotal API).
381			Args:
382			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
383
384			Return:
385			The response from the server as a byte sequence.
386
387			Exception
388			VirusTotalAPIError(Connection error): In case of server connection errors.
389			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
390			"""
391			self._last_http_error = None
392			self._last_result = None
393			api_url = self.base_url + '/files/' + file_id + '/download_url'
394			try:
395			response = requests.get(api_url, headers=self.headers,
396			timeout=self.timeout, proxies=self.proxies)
397			except requests.exceptions.Timeout:
398			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
399			except requests.exceptions.ConnectionError:
400			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
401			else:
402			self._last_http_error = response.status_code
403			self._last_result = response.content
404			return response.content
405
406			def get_download(self, file_id):
407			"""Download a file. This function requires special privileges (you need a private
408			key to access the VirusTotal API).
409			Args:
410			file_id: SHA-256, SHA-1 or MD5 identifying the file (str).
411
412			Return:
413			The response from the server as a byte sequence.
414
415			Exception
416			VirusTotalAPIError(Connection error): In case of server connection errors.
417			VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
418			"""
419			self._last_http_error = None
420			self._last_result = None
421			api_url = self.base_url + '/files/' + file_id + '/download'
422			try:
423			response = requests.get(api_url, headers=self.headers,
424			timeout=self.timeout, proxies=self.proxies)
425			except requests.exceptions.Timeout:
426			raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
427			except requests.exceptions.ConnectionError:
428			raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
429			else:
430			self._last_http_error = response.status_code
431			self._last_result = response.content
432			return response.content
433

drobotun / virustotalapi3

VirusTotalAPIFiles.get_file_id() B last analyzed 2023-03-02 09:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

VirusTotalAPIFiles.get_file_id() B
last analyzed 2023-03-02 09:52 UTC