1
|
|
|
"""The module describes the VirusTotalAPIUrls 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/url_class.html |
7
|
|
|
""" |
8
|
|
|
|
9
|
|
|
import base64 |
10
|
|
|
import hashlib |
11
|
|
|
import errno |
12
|
|
|
import requests |
13
|
|
|
|
14
|
|
|
from .vtapi3base import VirusTotalAPI |
15
|
|
|
from .vtapi3error import VirusTotalAPIError |
16
|
|
|
|
17
|
|
|
class VirusTotalAPIUrls(VirusTotalAPI): |
18
|
|
|
"""The analysis new URLs and retrieving information about any URLs from the VirusTotal database |
19
|
|
|
methods are defined in the class. |
20
|
|
|
|
21
|
|
|
Methods: |
22
|
|
|
get_url_id_base64(): Get base64 encoded URL identifier. |
23
|
|
|
get_url_id_sha256(): Get the URL identifier as a SHA256 hash. |
24
|
|
|
upload(): Upload URL for analysis. |
25
|
|
|
get_report(): Retrieve information about an URL. |
26
|
|
|
analyse(): Analyse an URL. |
27
|
|
|
get_comments(): Retrieve comments for an URL. |
28
|
|
|
put_comments(): Add a comment to a URL. |
29
|
|
|
get_votes(): Retrieve votes for an URL. |
30
|
|
|
put_votes(): Add a votes to a URL. |
31
|
|
|
get_network_location(): Get the domain or IP address for a URL. |
32
|
|
|
""" |
33
|
|
|
|
34
|
|
|
@staticmethod |
35
|
|
|
def get_url_id_base64(url): |
36
|
|
|
"""Get base64 encoded URL identifier. |
37
|
|
|
|
38
|
|
|
Args: |
39
|
|
|
url: The URL for which you want to get the identifier (str). |
40
|
|
|
|
41
|
|
|
Return: |
42
|
|
|
The identifier of the url, base64 encoded (str). |
43
|
|
|
""" |
44
|
|
|
return base64.urlsafe_b64encode(url.encode('utf-8')).decode('utf-8').rstrip('=') |
45
|
|
|
|
46
|
|
|
@staticmethod |
47
|
|
|
def get_url_id_sha256(url): |
48
|
|
|
"""Get the URL identifier as a SHA256 hash. |
49
|
|
|
|
50
|
|
|
Args: |
51
|
|
|
url: The URL for which you want to get the identifier (str). |
52
|
|
|
|
53
|
|
|
Return: |
54
|
|
|
The identifier of the url, SHA256 encoded (str). |
55
|
|
|
""" |
56
|
|
|
return hashlib.sha256(url.encode()).hexdigest() |
57
|
|
|
|
58
|
|
|
def upload(self, url): |
59
|
|
|
"""Upload URL for analysis. |
60
|
|
|
|
61
|
|
|
Args: |
62
|
|
|
url: URL to be analyzed (str). |
63
|
|
|
|
64
|
|
|
Return: |
65
|
|
|
The response from the server as a byte sequence. |
66
|
|
|
|
67
|
|
|
Exception |
68
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
69
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
70
|
|
|
""" |
71
|
|
|
self._last_http_error = None |
72
|
|
|
self._last_result = None |
73
|
|
|
data = {'url': url} |
74
|
|
|
api_url = self.base_url + '/urls' |
75
|
|
|
try: |
76
|
|
|
response = requests.post(api_url, headers=self.headers, data=data, |
77
|
|
|
timeout=self.timeout, proxies=self.proxies) |
78
|
|
|
except requests.exceptions.Timeout: |
79
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
80
|
|
|
except requests.exceptions.ConnectionError: |
81
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
82
|
|
|
else: |
83
|
|
|
self._last_http_error = response.status_code |
84
|
|
|
self._last_result = response.content |
85
|
|
|
return response.content |
86
|
|
|
|
87
|
|
|
def get_report(self, url_id): |
88
|
|
|
"""Retrieve information about an URL. |
89
|
|
|
|
90
|
|
|
Args: |
91
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
92
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
93
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
94
|
|
|
|
95
|
|
|
Return: |
96
|
|
|
The response from the server as a byte sequence. |
97
|
|
|
|
98
|
|
|
Exception |
99
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
100
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
101
|
|
|
""" |
102
|
|
|
self._last_http_error = None |
103
|
|
|
self._last_result = None |
104
|
|
|
api_url = self.base_url + '/urls/' + url_id |
105
|
|
|
try: |
106
|
|
|
response = requests.get(api_url, headers=self.headers, |
107
|
|
|
timeout=self.timeout, proxies=self.proxies) |
108
|
|
|
except requests.exceptions.Timeout: |
109
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
110
|
|
|
except requests.exceptions.ConnectionError: |
111
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
112
|
|
|
else: |
113
|
|
|
self._last_http_error = response.status_code |
114
|
|
|
self._last_result = response.content |
115
|
|
|
return response.content |
116
|
|
|
|
117
|
|
|
def analyse(self, url_id): |
118
|
|
|
"""Analyse an URL. |
119
|
|
|
|
120
|
|
|
Args: |
121
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
122
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
123
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
124
|
|
|
|
125
|
|
|
Return: |
126
|
|
|
The response from the server as a byte sequence. |
127
|
|
|
|
128
|
|
|
Exception |
129
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
130
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
131
|
|
|
""" |
132
|
|
|
self._last_http_error = None |
133
|
|
|
self._last_result = None |
134
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/analyse' |
135
|
|
|
try: |
136
|
|
|
response = requests.post(api_url, headers=self.headers, |
137
|
|
|
timeout=self.timeout, proxies=self.proxies) |
138
|
|
|
except requests.exceptions.Timeout: |
139
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
140
|
|
|
except requests.exceptions.ConnectionError: |
141
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
142
|
|
|
else: |
143
|
|
|
self._last_http_error = response.status_code |
144
|
|
|
self._last_result = response.content |
145
|
|
|
return response.content |
146
|
|
|
|
147
|
|
|
def get_comments(self, url_id, limit=10, cursor='""'): |
148
|
|
|
"""Retrieve comments for an URL. |
149
|
|
|
|
150
|
|
|
Args: |
151
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
152
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
153
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
154
|
|
|
limit: Maximum number of comments to retrieve (int). The default value is 10. |
155
|
|
|
cursor: Continuation cursor (str). The default value is ''. |
156
|
|
|
|
157
|
|
|
Return: |
158
|
|
|
The response from the server as a byte sequence. |
159
|
|
|
|
160
|
|
|
Exception |
161
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
162
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
163
|
|
|
""" |
164
|
|
|
self._last_http_error = None |
165
|
|
|
self._last_result = None |
166
|
|
|
query_string = {'limit': str(limit), 'cursor': cursor} |
167
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/comments' |
168
|
|
|
try: |
169
|
|
|
response = requests.get(api_url, headers=self.headers, params=query_string, |
170
|
|
|
timeout=self.timeout, proxies=self.proxies) |
171
|
|
|
except requests.exceptions.Timeout: |
172
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
173
|
|
|
except requests.exceptions.ConnectionError: |
174
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
175
|
|
|
else: |
176
|
|
|
self._last_http_error = response.status_code |
177
|
|
|
self._last_result = response.content |
178
|
|
|
return response.content |
179
|
|
|
|
180
|
|
|
def put_comments(self, url_id, text): |
181
|
|
|
"""Add a comment to a URL. |
182
|
|
|
|
183
|
|
|
Args: |
184
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
185
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
186
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
187
|
|
|
text: Text of the comment (str). |
188
|
|
|
|
189
|
|
|
Return: |
190
|
|
|
The response from the server as a byte sequence. |
191
|
|
|
|
192
|
|
|
Exception |
193
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
194
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
195
|
|
|
""" |
196
|
|
|
self._last_http_error = None |
197
|
|
|
self._last_result = None |
198
|
|
|
comments = {"data": {'type': 'comment', 'attributes': {'text': text}}} |
199
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/comments' |
200
|
|
|
try: |
201
|
|
|
response = requests.post(api_url, headers=self.headers, json=comments, |
202
|
|
|
timeout=self.timeout, proxies=self.proxies) |
203
|
|
|
except requests.exceptions.Timeout: |
204
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
205
|
|
|
except requests.exceptions.ConnectionError: |
206
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
207
|
|
|
else: |
208
|
|
|
self._last_http_error = response.status_code |
209
|
|
|
self._last_result = response.content |
210
|
|
|
return response.content |
211
|
|
|
|
212
|
|
|
def get_votes(self, url_id, limit=10, cursor='""'): |
213
|
|
|
"""Retrieve votes for a URL. |
214
|
|
|
|
215
|
|
|
Args: |
216
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
217
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
218
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
219
|
|
|
limit: Maximum number of comments to retrieve (int). The default value is 10. |
220
|
|
|
cursor: Continuation cursor (str). The default value is ''. |
221
|
|
|
|
222
|
|
|
Return: |
223
|
|
|
The response from the server as a byte sequence. |
224
|
|
|
|
225
|
|
|
Exception |
226
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
227
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
228
|
|
|
""" |
229
|
|
|
self._last_http_error = None |
230
|
|
|
self._last_result = None |
231
|
|
|
query_string = {'limit': str(limit), 'cursor': cursor} |
232
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/votes' |
233
|
|
|
try: |
234
|
|
|
response = requests.get(api_url, headers=self.headers, params=query_string, |
235
|
|
|
timeout=self.timeout, proxies=self.proxies) |
236
|
|
|
except requests.exceptions.Timeout: |
237
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
238
|
|
|
except requests.exceptions.ConnectionError: |
239
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
240
|
|
|
else: |
241
|
|
|
self._last_http_error = response.status_code |
242
|
|
|
self._last_result = response.content |
243
|
|
|
return response.content |
244
|
|
|
|
245
|
|
|
def put_votes(self, url_id, malicious=False): |
246
|
|
|
"""Add a vote for a URL. |
247
|
|
|
|
248
|
|
|
Args: |
249
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
250
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
251
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
252
|
|
|
malicious: Determines a malicious (True) or harmless (False) URL (bool). |
253
|
|
|
|
254
|
|
|
Return: |
255
|
|
|
The response from the server as a byte sequence. |
256
|
|
|
|
257
|
|
|
Exception |
258
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
259
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
260
|
|
|
""" |
261
|
|
|
self._last_http_error = None |
262
|
|
|
self._last_result = None |
263
|
|
|
if malicious: |
264
|
|
|
verdict = 'malicious' |
265
|
|
|
else: |
266
|
|
|
verdict = 'harmless' |
267
|
|
|
votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}} |
268
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/votes' |
269
|
|
|
try: |
270
|
|
|
response = requests.post(api_url, headers=self.headers, json=votes, |
271
|
|
|
timeout=self.timeout, proxies=self.proxies) |
272
|
|
|
except requests.exceptions.Timeout: |
273
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
274
|
|
|
except requests.exceptions.ConnectionError: |
275
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
276
|
|
|
else: |
277
|
|
|
self._last_http_error = response.status_code |
278
|
|
|
self._last_result = response.content |
279
|
|
|
return response.content |
280
|
|
|
|
281
|
|
|
def get_network_location(self, url_id): |
282
|
|
|
"""Get the domain or IP address for a URL. |
283
|
|
|
|
284
|
|
|
Args: |
285
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
286
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
287
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
288
|
|
|
|
289
|
|
|
Return: |
290
|
|
|
The response from the server as a byte sequence. |
291
|
|
|
|
292
|
|
|
Exception |
293
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
294
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
295
|
|
|
""" |
296
|
|
|
self._last_http_error = None |
297
|
|
|
self._last_result = None |
298
|
|
|
api_url = self.base_url + '/urls/' + url_id + '/network_location' |
299
|
|
|
try: |
300
|
|
|
response = requests.get(api_url, headers=self.headers, |
301
|
|
|
timeout=self.timeout, proxies=self.proxies) |
302
|
|
|
except requests.exceptions.Timeout: |
303
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
304
|
|
|
except requests.exceptions.ConnectionError: |
305
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
306
|
|
|
else: |
307
|
|
|
self._last_http_error = response.status_code |
308
|
|
|
self._last_result = response.content |
309
|
|
|
return response.content |
310
|
|
|
|
311
|
|
|
def get_relationship(self, url_id, relationship='/last_serving_ip_address', |
312
|
|
|
limit=10, cursor='""'): |
313
|
|
|
"""Retrieve objects related to an URL |
314
|
|
|
|
315
|
|
|
Args: |
316
|
|
|
url_id: URL identifier (str). This identifier can adopt two forms: the SHA-256 of the |
317
|
|
|
canonized URL (method 'get_url_id_sha256()'), the string resulting from encoding the |
318
|
|
|
URL in base64 without the "=" padding (method 'get_url_id_base64()'). |
319
|
|
|
relationship: Relationship name (str). The default value is '/last_serving_ip_address'. |
320
|
|
|
limit: Maximum number of comments to retrieve (int). The default value is 10. |
321
|
|
|
cursor: Continuation cursor (str). The default value is ''. |
322
|
|
|
|
323
|
|
|
Return: |
324
|
|
|
The response from the server as a byte sequence. |
325
|
|
|
|
326
|
|
|
Exception |
327
|
|
|
VirusTotalAPIError(Connection error): In case of server connection errors. |
328
|
|
|
VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded. |
329
|
|
|
""" |
330
|
|
|
self._last_http_error = None |
331
|
|
|
self._last_result = None |
332
|
|
|
query_string = {'limit': str(limit), 'cursor': cursor} |
333
|
|
|
api_url = self.base_url + '/urls/' + url_id + relationship |
334
|
|
|
try: |
335
|
|
|
response = requests.get(api_url, headers=self.headers, params=query_string, |
336
|
|
|
timeout=self.timeout, proxies=self.proxies) |
337
|
|
|
except requests.exceptions.Timeout: |
338
|
|
|
raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT) |
339
|
|
|
except requests.exceptions.ConnectionError: |
340
|
|
|
raise VirusTotalAPIError('Connection error', errno.ECONNABORTED) |
341
|
|
|
else: |
342
|
|
|
self._last_http_error = response.status_code |
343
|
|
|
self._last_result = response.content |
344
|
|
|
return response.content |
345
|
|
|
|