backend/src/appenlight/lib/utils/airbrake.py

# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# 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
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# 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.

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