airbrake.py
142 lines
| 5.5 KiB
| text/x-python
|
PythonLexer
r2 | # -*- coding: utf-8 -*- | |||
r112 | # Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors | |||
r2 | # | |||
r112 | # Licensed under the Apache License, Version 2.0 (the "License"); | |||
# you may not use this file except in compliance with the License. | ||||
# You may obtain a copy of the License at | ||||
r2 | # | |||
r112 | # http://www.apache.org/licenses/LICENSE-2.0 | |||
r2 | # | |||
r112 | # Unless required by applicable law or agreed to in writing, software | |||
# distributed under the License is distributed on an "AS IS" BASIS, | ||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||||
# See the License for the specific language governing permissions and | ||||
# limitations under the License. | ||||
r2 | ||||
import logging | ||||
import uuid | ||||
from datetime import datetime | ||||
log = logging.getLogger(__name__) | ||||
def parse_airbrake_xml(request): | ||||
root = request.context.airbrake_xml_etree | ||||
error = root.find('error') | ||||
notifier = root.find('notifier') | ||||
server_env = root.find('server-environment') | ||||
request_data = root.find('request') | ||||
user = root.find('current-user') | ||||
if request_data is not None: | ||||
cgi_data = request_data.find('cgi-data') | ||||
if cgi_data is None: | ||||
cgi_data = [] | ||||
error_dict = { | ||||
'class_name': error.findtext('class') or '', | ||||
'error': error.findtext('message') or '', | ||||
"occurences": 1, | ||||
"http_status": 500, | ||||
"priority": 5, | ||||
"server": 'unknown', | ||||
'url': 'unknown', 'request': {} | ||||
} | ||||
if user is not None: | ||||
error_dict['username'] = user.findtext('username') or \ | ||||
user.findtext('id') | ||||
if notifier is not None: | ||||
error_dict['client'] = notifier.findtext('name') | ||||
if server_env is not None: | ||||
error_dict["server"] = server_env.findtext('hostname', 'unknown') | ||||
whitelist_environ = ['REMOTE_USER', 'REMOTE_ADDR', 'SERVER_NAME', | ||||
'CONTENT_TYPE', 'HTTP_REFERER'] | ||||
if request_data is not None: | ||||
error_dict['url'] = request_data.findtext('url', 'unknown') | ||||
component = request_data.findtext('component') | ||||
action = request_data.findtext('action') | ||||
if component and action: | ||||
error_dict['view_name'] = '%s:%s' % (component, action) | ||||
for node in cgi_data: | ||||
key = node.get('key') | ||||
if key.startswith('HTTP') or key in whitelist_environ: | ||||
error_dict['request'][key] = node.text | ||||
elif 'query_parameters' in key: | ||||
error_dict['request']['GET'] = {} | ||||
for x in node: | ||||
error_dict['request']['GET'][x.get('key')] = x.text | ||||
elif 'request_parameters' in key: | ||||
error_dict['request']['POST'] = {} | ||||
for x in node: | ||||
error_dict['request']['POST'][x.get('key')] = x.text | ||||
elif key.endswith('cookie'): | ||||
error_dict['request']['COOKIE'] = {} | ||||
for x in node: | ||||
error_dict['request']['COOKIE'][x.get('key')] = x.text | ||||
elif key.endswith('request_id'): | ||||
error_dict['request_id'] = node.text | ||||
elif key.endswith('session'): | ||||
error_dict['request']['SESSION'] = {} | ||||
for x in node: | ||||
error_dict['request']['SESSION'][x.get('key')] = x.text | ||||
else: | ||||
if key in ['rack.session.options']: | ||||
# skip secret configs | ||||
continue | ||||
try: | ||||
if len(node): | ||||
error_dict['request'][key] = dict( | ||||
[(x.get('key'), x.text,) for x in node]) | ||||
else: | ||||
error_dict['request'][key] = node.text | ||||
except Exception as e: | ||||
log.warning('Airbrake integration exception: %s' % e) | ||||
error_dict['request'].pop('HTTP_COOKIE', '') | ||||
error_dict['ip'] = error_dict.pop('REMOTE_ADDR', '') | ||||
error_dict['user_agent'] = error_dict.pop('HTTP_USER_AGENT', '') | ||||
if 'request_id' not in error_dict: | ||||
error_dict['request_id'] = str(uuid.uuid4()) | ||||
if request.context.possibly_public: | ||||
# set ip for reports that come from airbrake js client | ||||
error_dict["timestamp"] = datetime.utcnow() | ||||
if request.environ.get("HTTP_X_FORWARDED_FOR"): | ||||
ip = request.environ.get("HTTP_X_FORWARDED_FOR", '') | ||||
first_ip = ip.split(',')[0] | ||||
remote_addr = first_ip.strip() | ||||
else: | ||||
remote_addr = (request.environ.get("HTTP_X_REAL_IP") or | ||||
request.environ.get('REMOTE_ADDR')) | ||||
error_dict["ip"] = remote_addr | ||||
blacklist = ['password', 'passwd', 'pwd', 'auth_tkt', 'secret', 'csrf', | ||||
'session', 'test'] | ||||
lines = [] | ||||
for l in error.find('backtrace'): | ||||
lines.append({'file': l.get("file", ""), | ||||
'line': l.get("number", ""), | ||||
'fn': l.get("method", ""), | ||||
'module': l.get("module", ""), | ||||
'cline': l.get("method", ""), | ||||
'vars': {}}) | ||||
error_dict['traceback'] = list(reversed(lines)) | ||||
# filtering is not provided by airbrake | ||||
keys_to_check = ( | ||||
error_dict['request'].get('COOKIE'), | ||||
error_dict['request'].get('COOKIES'), | ||||
error_dict['request'].get('POST'), | ||||
error_dict['request'].get('SESSION'), | ||||
) | ||||
for source in [_f for _f in keys_to_check if _f]: | ||||
for k in source.keys(): | ||||
for bad_key in blacklist: | ||||
if bad_key in k.lower(): | ||||
source[k] = '***' | ||||
return error_dict | ||||