forked from conda/conda
-
Notifications
You must be signed in to change notification settings - Fork 0
/
exception_handler.py
389 lines (334 loc) · 14 KB
/
exception_handler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
# Copyright (C) 2012 Anaconda, Inc
# SPDX-License-Identifier: BSD-3-Clause
"""Error handling and error reporting."""
import os
import sys
from functools import lru_cache, partial
from logging import getLogger
from .common.compat import ensure_text_type, on_win
log = getLogger(__name__)
class ExceptionHandler:
def __call__(self, func, *args, **kwargs):
try:
return func(*args, **kwargs)
except:
_, exc_val, exc_tb = sys.exc_info()
return self.handle_exception(exc_val, exc_tb)
def write_out(self, *content):
from logging import getLogger
from .cli.main import init_loggers
init_loggers()
getLogger("conda.stderr").info("\n".join(content))
@property
def http_timeout(self):
from .base.context import context
return context.remote_connect_timeout_secs, context.remote_read_timeout_secs
@property
def user_agent(self):
from .base.context import context
return context.user_agent
@property
def error_upload_url(self):
from .base.context import context
return context.error_upload_url
def handle_exception(self, exc_val, exc_tb):
from errno import ENOSPC
from .exceptions import (
CondaError,
CondaMemoryError,
NoSpaceLeftError,
)
if isinstance(exc_val, CondaError):
if exc_val.reportable:
return self.handle_reportable_application_exception(exc_val, exc_tb)
else:
return self.handle_application_exception(exc_val, exc_tb)
if isinstance(exc_val, EnvironmentError):
if getattr(exc_val, "errno", None) == ENOSPC:
return self.handle_application_exception(
NoSpaceLeftError(exc_val), exc_tb
)
if isinstance(exc_val, MemoryError):
return self.handle_application_exception(CondaMemoryError(exc_val), exc_tb)
if isinstance(exc_val, KeyboardInterrupt):
self._print_conda_exception(CondaError("KeyboardInterrupt"), exc_tb)
return 1
if isinstance(exc_val, SystemExit):
return exc_val.code
return self.handle_unexpected_exception(exc_val, exc_tb)
def handle_application_exception(self, exc_val, exc_tb):
self._print_conda_exception(exc_val, exc_tb)
return exc_val.return_code
def _print_conda_exception(self, exc_val, exc_tb):
from .exceptions import print_conda_exception
print_conda_exception(exc_val, exc_tb)
def handle_unexpected_exception(self, exc_val, exc_tb):
error_report = self.get_error_report(exc_val, exc_tb)
self.print_unexpected_error_report(error_report)
self._upload(error_report)
rc = getattr(exc_val, "return_code", None)
return rc if rc is not None else 1
def handle_reportable_application_exception(self, exc_val, exc_tb):
error_report = self.get_error_report(exc_val, exc_tb)
from .base.context import context
if context.json:
error_report.update(exc_val.dump_map())
self.print_expected_error_report(error_report)
self._upload(error_report)
return exc_val.return_code
def get_error_report(self, exc_val, exc_tb):
from .exceptions import CondaError, _format_exc
command = " ".join(ensure_text_type(s) for s in sys.argv)
info_dict = {}
if " info" not in command:
# get info_dict, but if we get an exception here too, record it without trampling
# the original exception
try:
from .cli.main_info import get_info_dict
info_dict = get_info_dict()
except Exception as info_e:
info_traceback = _format_exc()
info_dict = {
"error": repr(info_e),
"exception_name": info_e.__class__.__name__,
"exception_type": str(exc_val.__class__),
"traceback": info_traceback,
}
error_report = {
"error": repr(exc_val),
"exception_name": exc_val.__class__.__name__,
"exception_type": str(exc_val.__class__),
"command": command,
"traceback": _format_exc(exc_val, exc_tb),
"conda_info": info_dict,
}
if isinstance(exc_val, CondaError):
error_report["conda_error_components"] = exc_val.dump_map()
return error_report
def print_unexpected_error_report(self, error_report):
from .base.context import context
if context.json:
from .cli.common import stdout_json
stdout_json(error_report)
else:
message_builder = []
message_builder.append("")
message_builder.append(
"# >>>>>>>>>>>>>>>>>>>>>> ERROR REPORT <<<<<<<<<<<<<<<<<<<<<<"
)
message_builder.append("")
message_builder.extend(
" " + line for line in error_report["traceback"].splitlines()
)
message_builder.append("")
message_builder.append("`$ %s`" % error_report["command"])
message_builder.append("")
if error_report["conda_info"]:
from .cli.main_info import get_env_vars_str, get_main_info_str
try:
# TODO: Sanitize env vars to remove secrets (e.g credentials for PROXY)
message_builder.append(get_env_vars_str(error_report["conda_info"]))
message_builder.append(
get_main_info_str(error_report["conda_info"])
)
except Exception as e:
log.warn("%r", e, exc_info=True)
message_builder.append("conda info could not be constructed.")
message_builder.append("%r" % e)
message_builder.extend(
[
"",
"An unexpected error has occurred. Conda has prepared the above report."
"",
"If you suspect this error is being caused by a malfunctioning plugin,",
"consider using the --no-plugins option to turn off plugins.",
"",
"Example: conda --no-plugins install <package>",
"",
"Alternatively, you can set the CONDA_NO_PLUGINS environment variable on",
"the command line to run the command without plugins enabled.",
"",
"Example: CONDA_NO_PLUGINS=true conda install <package>",
"",
]
)
self.write_out(*message_builder)
def print_expected_error_report(self, error_report):
from .base.context import context
if context.json:
from .cli.common import stdout_json
stdout_json(error_report)
else:
message_builder = []
message_builder.append("")
message_builder.append(
"# >>>>>>>>>>>>>>>>>>>>>> ERROR REPORT <<<<<<<<<<<<<<<<<<<<<<"
)
message_builder.append("")
message_builder.append("`$ %s`" % error_report["command"])
message_builder.append("")
if error_report["conda_info"]:
from .cli.main_info import get_env_vars_str, get_main_info_str
try:
# TODO: Sanitize env vars to remove secrets (e.g credentials for PROXY)
message_builder.append(get_env_vars_str(error_report["conda_info"]))
message_builder.append(
get_main_info_str(error_report["conda_info"])
)
except Exception as e:
log.warn("%r", e, exc_info=True)
message_builder.append("conda info could not be constructed.")
message_builder.append("%r" % e)
message_builder.append("")
message_builder.append(
"V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V"
)
message_builder.append("")
message_builder.extend(error_report["error"].splitlines())
message_builder.append("")
message_builder.append(
"A reportable application error has occurred. Conda has prepared the above report."
)
message_builder.append("")
self.write_out(*message_builder)
# FUTURE: Python 3.8+, replace with functools.cached_property
@property
@lru_cache(maxsize=None)
def _isatty(self):
try:
return os.isatty(0) or on_win
except Exception as e:
log.debug("%r", e)
return True
def _upload(self, error_report) -> None:
"""Determine whether or not to upload the error report."""
from .base.context import context
post_upload = False
if context.report_errors is False:
# no prompt and no submission
do_upload = False
elif context.report_errors is True or context.always_yes:
# no prompt and submit
do_upload = True
elif context.json or context.quiet or not self._isatty:
# never prompt under these conditions, submit iff always_yes
do_upload = bool(not context.offline and context.always_yes)
else:
# prompt whether to submit
do_upload = self._ask_upload()
post_upload = True
# the upload state is one of the following:
# - True: upload error report
# - False: do not upload error report
# - None: while prompting a timeout occurred
if do_upload:
# user wants report to be submitted
self._execute_upload(error_report)
if post_upload:
# post submission text
self._post_upload(do_upload)
def _ask_upload(self):
from .auxlib.type_coercion import boolify
from .common.io import timeout
try:
do_upload = timeout(
40,
partial(
input,
"If submitted, this report will be used by core maintainers to improve\n"
"future releases of conda.\n"
"Would you like conda to send this report to the core maintainers? "
"[y/N]: ",
),
)
return do_upload and boolify(do_upload)
except Exception as e:
log.debug("%r", e)
return False
def _execute_upload(self, error_report):
import getpass
import json
from .auxlib.entity import EntityEncoder
headers = {
"User-Agent": self.user_agent,
}
_timeout = self.http_timeout
username = getpass.getuser()
error_report["is_ascii"] = (
True if all(ord(c) < 128 for c in username) else False
)
error_report["has_spaces"] = True if " " in str(username) else False
data = json.dumps(error_report, sort_keys=True, cls=EntityEncoder) + "\n"
data = data.replace(str(username), "USERNAME_REMOVED")
response = None
try:
# requests does not follow HTTP standards for redirects of non-GET methods
# That is, when following a 301 or 302, it turns a POST into a GET.
# And no way to disable. WTF
import requests
redirect_counter = 0
url = self.error_upload_url
response = requests.post(
url, headers=headers, timeout=_timeout, data=data, allow_redirects=False
)
response.raise_for_status()
while response.status_code in (301, 302) and response.headers.get(
"Location"
):
url = response.headers["Location"]
response = requests.post(
url,
headers=headers,
timeout=_timeout,
data=data,
allow_redirects=False,
)
response.raise_for_status()
redirect_counter += 1
if redirect_counter > 15:
from . import CondaError
raise CondaError("Redirect limit exceeded")
log.debug("upload response status: %s", response and response.status_code)
except Exception as e: # pragma: no cover
log.info("%r", e)
try:
if response and response.ok:
self.write_out("Upload successful.")
else:
self.write_out("Upload did not complete.")
if response and response.status_code:
self.write_out(" HTTP %s" % response.status_code)
except Exception as e:
log.debug("%r" % e)
def _post_upload(self, do_upload):
if do_upload is True:
# report was submitted
self.write_out(
"",
"Thank you for helping to improve conda.",
"Opt-in to always sending reports (and not see this message again)",
"by running",
"",
" $ conda config --set report_errors true",
"",
)
elif do_upload is None:
# timeout was reached while prompting user
self.write_out(
"",
"Timeout reached. No report sent.",
"",
)
else:
# no report submitted
self.write_out(
"",
"No report sent. To permanently opt-out, use",
"",
" $ conda config --set report_errors false",
"",
)
def conda_exception_handler(func, *args, **kwargs):
exception_handler = ExceptionHandler()
return_value = exception_handler(func, *args, **kwargs)
return return_value