show.py

from collections import OrderedDict
from collections import defaultdict

from flask import flash
from flask import redirect
from flask import render_template
from flask import url_for
from flask_login import current_user
from markupsafe import escape
from sqlalchemy import and_
from sqlalchemy_continuum import version_class
from sqlalchemy_continuum import versioning_manager

from config import TRACKER_ADVISORY_URL
from config import TRACKER_BUGTRACKER_URL
from config import TRACKER_GROUP_URL
from config import TRACKER_ISSUE_URL
from config import TRACKER_LOG_ENTRIES_PER_PAGE
from config import TRACKER_SUMMARY_LENGTH_MAX
from tracker import db
from tracker import tracker
from tracker.advisory import advisory_escape_html
from tracker.advisory import advisory_extend_html
from tracker.advisory import advisory_format_issue_listing
from tracker.advisory import generate_advisory
from tracker.advisory import render_html_advisory
from tracker.form.advisory import AdvisoryForm
from tracker.model import CVE
from tracker.model import Advisory
from tracker.model import CVEGroup
from tracker.model import CVEGroupEntry
from tracker.model import CVEGroupPackage
from tracker.model import Package
from tracker.model.advisory import advisory_regex
from tracker.model.cve import cve_id_regex
from tracker.model.cvegroup import pkgname_regex
from tracker.model.cvegroup import vulnerability_group_regex
from tracker.model.enum import Publication
from tracker.model.enum import Remote
from tracker.model.enum import Status
from tracker.model.package import filter_duplicate_packages
from tracker.model.package import sort_packages
from tracker.user import reporter_required
from tracker.user import user_can_delete_group
from tracker.user import user_can_delete_issue
from tracker.user import user_can_edit_group
from tracker.user import user_can_edit_issue
from tracker.user import user_can_handle_advisory
from tracker.user import user_can_watch_log
from tracker.user import user_can_watch_user_log
from tracker.util import add_params_to_uri
from tracker.util import json_response
from tracker.util import multiline_to_list
from tracker.view.error import not_found


def get_bug_data(cves, pkgs, versions, group):
    references = []
    references = [ref for ref in multiline_to_list(group.reference)
                  if ref not in references]
    list(map(lambda issue: references.extend(
        [ref for ref in multiline_to_list(issue.reference) if ref not in references]), cves))

    severity_sorted_issues = sorted(cves, key=lambda issue: issue.issue_type)
    severity_sorted_issues = sorted(severity_sorted_issues, key=lambda issue: issue.severity)
    unique_issue_types = []
    for issue in severity_sorted_issues:
        if issue.issue_type not in unique_issue_types:
            unique_issue_types.append(issue.issue_type)

    bug_desc = render_template('bug.txt', cves=cves, group=group, references=references,
                               pkgs=pkgs, unique_issue_types=unique_issue_types,
                               TRACKER_ISSUE_URL=TRACKER_ISSUE_URL,
                               TRACKER_GROUP_URL=TRACKER_GROUP_URL)
    pkg_str = ' '.join((pkg.pkgname for pkg in pkgs))
    group_type = 'multiple issues' if len(unique_issue_types) > 1 else unique_issue_types[0]
    summary = '[{}] [Security] {} ({})'.format(pkg_str, group_type, ' '.join([cve.id for cve in cves]))

    if TRACKER_SUMMARY_LENGTH_MAX != 0 and len(summary) > TRACKER_SUMMARY_LENGTH_MAX:
        summary = "[{}] [Security] {} (Multiple CVE's)".format(pkg_str, group_type)

    return {
        'item_summary': summary,
        'detailed_desc': bug_desc
    }


def get_cve_data(cve):
    cve_model = CVE.query.get(cve)
    if not cve_model:
        return None

    entries = (db.session.query(CVEGroupEntry, CVEGroup, CVEGroupPackage, Advisory)
               .filter_by(cve=cve_model)
               .join(CVEGroup, CVEGroupEntry.group)
               .join(CVEGroupPackage, CVEGroup.packages)
               .outerjoin(Advisory, Advisory.group_package_id == CVEGroupPackage.id)
               .order_by(CVEGroup.created.desc()).order_by(CVEGroupPackage.pkgname)).all()

    group_packages = defaultdict(set)
    advisories = set()
    groups = set()
    for cve, group, pkg, advisory in entries:
        group_packages[group].add(pkg.pkgname)
        groups.add(group)
        if advisory:
            advisories.add(advisory)

    groups = sorted(groups, key=lambda item: item.created, reverse=True)
    groups = sorted(groups, key=lambda item: item.status)
    advisories = sorted(advisories, key=lambda item: item.id, reverse=True)
    group_packages = dict(map(lambda item: (item[0], sorted(item[1])), group_packages.items()))

    return {'issue': cve_model,
            'groups': groups,
            'group_packages': group_packages,
            'advisories': advisories}


@tracker.route('/<regex("{}"):cve>.json'.format(cve_id_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):cve>/json'.format(cve_id_regex[1:-1]), methods=['GET'])
@json_response
def show_cve_json(cve):
    data = get_cve_data(cve)
    if not data:
        return not_found(json=True)

    cve = data['issue']
    references = cve.reference.replace('\r', '').split('\n') if cve.reference else []
    packages = list(set(sorted([item for sublist in data['group_packages'].values() for item in sublist])))
    advisories = data['advisories']
    if not current_user.role.is_reporter:
        advisories = list(filter(lambda advisory: advisory.publication == Publication.published, advisories))

    json_data = OrderedDict()
    json_data['name'] = cve.id
    json_data['type'] = cve.issue_type
    json_data['severity'] = cve.severity.label
    json_data['vector'] = cve.remote.label
    json_data['description'] = cve.description
    json_data['groups'] = [str(group) for group in data['groups']]
    json_data['packages'] = packages
    json_data['advisories'] = [advisory.id for advisory in advisories]
    json_data['references'] = references
    json_data['notes'] = cve.notes if cve.notes else None
    return json_data


@tracker.route('/<regex("{}"):cve>'.format(cve_id_regex[1:]), methods=['GET'])
def show_cve(cve):

    data = get_cve_data(cve)

    if not data:
        return not_found()

    packages = list(set(sorted([item for sublist in data['group_packages'].values() for item in sublist])))
    title = '{} - {}'.format(data['issue'].id, ' '.join(packages)) \
            if len(packages) else \
            '{}'.format(data['issue'].id)

    advisories = data['advisories']
    if not current_user.role.is_reporter:
        advisories = list(filter(lambda advisory: advisory.publication == Publication.published, advisories))

    return render_template('cve.html',
                           title=title,
                           issue=data['issue'],
                           groups=data['groups'],
                           group_packages=data['group_packages'],
                           advisories=advisories,
                           can_watch_log=user_can_watch_log(),
                           can_edit=user_can_edit_issue(advisories),
                           can_delete=user_can_delete_issue(advisories))


@tracker.route('/<regex("{}"):cve>/log'.format(cve_id_regex[1:-1]), methods=['GET'])
def show_cve_log(cve):
    data = get_cve_data(cve)
    if not data:
        return not_found()

    title = '{} - log'.format(data['issue'].id)

    return render_template('log/cve_log.html',
                           title=title,
                           issue=data['issue'],
                           can_watch_user_log=user_can_watch_user_log())


def get_group_data(avg):
    avg_id = int(avg.replace('AVG-', ''))
    entries = (db.session.query(CVEGroup, CVE, CVEGroupPackage, Advisory, Package)
               .filter(CVEGroup.id == avg_id)
               .join(CVEGroupEntry, CVEGroup.issues)
               .join(CVE, CVEGroupEntry.cve)
               .join(CVEGroupPackage, CVEGroup.packages)
               .outerjoin(Package, Package.name == CVEGroupPackage.pkgname)
               .outerjoin(Advisory, Advisory.group_package_id == CVEGroupPackage.id)).all()
    if not entries:
        return None

    group = None
    issues = set()
    packages = set()
    advisories = set()
    issue_types = set()
    versions = set()
    for group_entry, cve, pkg, advisory, package in entries:
        group = group_entry
        issues.add(cve)
        issue_types.add(cve.issue_type)
        packages.add(pkg)
        if package:
            versions.add(package)
        if advisory:
            advisories.add(advisory)

    advisories = sorted(advisories, key=lambda item: item.id, reverse=True)
    issue_types = list(issue_types)
    issues = sorted(issues, key=lambda item: item, reverse=True)
    packages = sorted(packages, key=lambda item: item.pkgname)
    versions = filter_duplicate_packages(sort_packages(list(versions)), True)
    advisories_pending = group.status == Status.fixed and group.advisory_qualified and len(advisories) <= 0

    return {
        'group': group,
        'packages': packages,
        'versions': versions,
        'issues': issues,
        'issue_types': issue_types,
        'advisories': advisories,
        'advisories_pending': advisories_pending
    }


@tracker.route('/group/<regex("{}"):avg>.json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/group/<regex("{}"):avg>/json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/avg/<regex("{}"):avg>.json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/avg/<regex("{}"):avg>/json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):avg>.json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):avg>/json'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@json_response
def show_group_json(avg):
    data = get_group_data(avg)
    if not data:
        return not_found(json=True)

    group = data['group']
    advisories = data['advisories']
    if not current_user.role.is_reporter:
        advisories = list(filter(lambda advisory: advisory.publication == Publication.published, advisories))
    issues = data['issues']
    packages = data['packages']
    issue_types = data['issue_types']
    references = group.reference.replace('\r', '').split('\n') if group.reference else []

    json_data = OrderedDict()
    json_data['name'] = group.name
    json_data['packages'] = [package.pkgname for package in packages]
    json_data['status'] = group.status.label
    json_data['severity'] = group.severity.label
    json_data['type'] = 'multiple issues' if len(issue_types) > 1 else issue_types[0]
    json_data['affected'] = group.affected
    json_data['fixed'] = group.fixed if group.fixed else None
    json_data['ticket'] = group.bug_ticket if group.bug_ticket else None
    json_data['issues'] = [str(cve) for cve in issues]
    json_data['advisories'] = [advisory.id for advisory in advisories]
    json_data['references'] = references
    json_data['notes'] = group.notes if group.notes else None

    return json_data


@tracker.route('/group/<regex("{}"):avg>'.format(vulnerability_group_regex[1:]), methods=['GET'])
@tracker.route('/avg/<regex("{}"):avg>'.format(vulnerability_group_regex[1:]), methods=['GET'])
@tracker.route('/<regex("{}"):avg>'.format(vulnerability_group_regex[1:]), methods=['GET'])
def show_group(avg):
    data = get_group_data(avg)
    if not data:
        return not_found()

    group = data['group']
    advisories = data['advisories']
    if not current_user.role.is_reporter:
        advisories = list(filter(lambda advisory: advisory.publication == Publication.published, advisories))
    issues = data['issues']
    packages = data['packages']
    issue_types = data['issue_types']
    versions = data['versions']
    issue_type = 'multiple issues' if len(issue_types) > 1 else issue_types[0]
    pkgnames = list(set(sorted([pkg.pkgname for pkg in packages])))

    form = AdvisoryForm()
    form.advisory_type.data = issue_type

    return render_template('group.html',
                           title='{} - {}'.format(group, ' '.join(pkgnames)),
                           form=form,
                           group=group,
                           packages=packages,
                           issues=issues,
                           advisories=advisories,
                           versions=versions,
                           Status=Status,
                           issue_type=issue_type,
                           advisories_pending=data['advisories_pending'],
                           can_edit=user_can_edit_group(advisories),
                           can_delete=user_can_delete_group(advisories),
                           can_handle_advisory=user_can_handle_advisory(),
                           can_watch_log=user_can_watch_log())


def get_package_data(pkgname):
    entries = (db.session.query(Package, CVEGroup, CVE, Advisory)
               .filter(Package.name == pkgname)
               .outerjoin(CVEGroupPackage, CVEGroupPackage.pkgname == Package.name)
               .outerjoin(CVEGroup, CVEGroupPackage.group)
               .outerjoin(CVEGroupEntry, CVEGroup.issues)
               .outerjoin(CVE, CVEGroupEntry.cve)
               .outerjoin(Advisory, and_(Advisory.group_package_id == CVEGroupPackage.id,
                                         Advisory.publication == Publication.published))
               ).all()

    # fallback for dropped packages
    if not entries:
        entries = (db.session.query(CVEGroupPackage, CVEGroup, CVE, Advisory)
                   .filter(CVEGroupPackage.pkgname == pkgname)
                   .join(CVEGroup, CVEGroupPackage.group)
                   .join(CVEGroupEntry, CVEGroup.issues)
                   .join(CVE, CVEGroupEntry.cve)
                   .outerjoin(Advisory, and_(Advisory.group_package_id == CVEGroupPackage.id,
                                             Advisory.publication == Publication.published))
                   ).all()

    if not entries:
        return None

    groups = set()
    issues = set()
    advisories = set()
    versions = set()
    for package, group, cve, advisory in entries:
        if isinstance(package, Package):
            versions.add(package)
        if group:
            groups.add(group)
        if cve:
            issues.add((cve, group))
        if advisory:
            advisories.add(advisory)

    issues = [{'issue': e[0], 'group': e[1]} for e in issues]
    issues = sorted(issues, key=lambda item: item['issue'], reverse=True)
    issues = sorted(issues, key=lambda item: item['group'].status)
    groups = sorted(groups, key=lambda item: item.id, reverse=True)
    groups = sorted(groups, key=lambda item: item.status)
    advisories = sorted(advisories, key=lambda item: item.id, reverse=True)
    versions = filter_duplicate_packages(sort_packages(list(versions)), True)
    package = versions[0] if versions else None

    return {
        'package': package,
        'pkgname': pkgname,
        'versions': versions,
        'groups': groups,
        'issues': issues,
        'advisories': advisories
    }


@tracker.route('/group/<regex("{}"):avg>/log'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/avg/<regex("{}"):avg>/log'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):avg>/log'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
def show_group_log(avg):
    data = get_group_data(avg)
    if not data:
        return not_found(json=True)

    group = data['group']

    return render_template('log/group_log.html',
                           title='{} - log'.format(group),
                           group=group,
                           Status=Status,
                           advisories_pending=data['advisories_pending'],
                           can_watch_user_log=user_can_watch_user_log())


@tracker.route('/package/<regex("{}"):pkgname>.json'.format(pkgname_regex[1:-1]), methods=['GET'])
@tracker.route('/package/<regex("{}"):pkgname>/json'.format(pkgname_regex[1:-1]), methods=['GET'])
@json_response
def show_package_json(pkgname):
    data = get_package_data(pkgname)
    if not data:
        return not_found(json=True)

    advisories = data['advisories']
    versions = data['versions']
    groups = data['groups']
    issues = data['issues']

    json_advisory = []
    for advisory in advisories:
        entry = OrderedDict()
        entry['name'] = advisory.id
        entry['date'] = advisory.created.strftime('%Y-%m-%d')
        entry['severity'] = advisory.group_package.group.severity.label
        entry['type'] = advisory.advisory_type
        entry['reference'] = advisory.reference if advisory.reference else None
        json_advisory.append(entry)

    json_versions = []
    for version in versions:
        entry = OrderedDict()
        entry['version'] = version.version
        entry['database'] = version.database
        json_versions.append(entry)

    json_groups = []
    for group in groups:
        entry = OrderedDict()
        entry['name'] = group.name
        entry['status'] = group.status.label
        entry['severity'] = group.severity.label
        json_groups.append(entry)

    json_issues = []
    for issue in issues:
        group = issue['group']
        issue = issue['issue']
        entry = OrderedDict()
        entry['name'] = issue.id
        entry['severity'] = issue.severity.label
        entry['type'] = issue.issue_type
        entry['status'] = group.status.label
        json_issues.append(entry)

    json_data = OrderedDict()
    json_data['name'] = pkgname
    json_data['versions'] = json_versions
    json_data['advisories'] = json_advisory
    json_data['groups'] = json_groups
    json_data['issues'] = json_issues
    return json_data


@tracker.route('/package/<regex("{}"):pkgname>'.format(pkgname_regex[1:]), methods=['GET'])
def show_package(pkgname):
    data = get_package_data(pkgname)
    if not data:
        return not_found()

    groups = data['groups']
    data['groups'] = {'open': list(filter(lambda group: group.status.open(), groups)),
                      'resolved': list(filter(lambda group: group.status.resolved(), groups))}

    issues = data['issues']
    data['issues'] = {'open': list(filter(lambda issue: issue['group'].status.open(), issues)),
                      'resolved': list(filter(lambda issue: issue['group'].status.resolved(), issues))}

    return render_template('package.html',
                           title='{}'.format(pkgname),
                           package=data)


@tracker.route('/advisory/<regex("{}"):advisory_id>/raw'.format(advisory_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):advisory_id>/raw'.format(advisory_regex[1:-1]), methods=['GET'])
def show_advisory_raw(advisory_id):
    result = show_advisory(advisory_id, raw=True)
    if isinstance(result, tuple):
        return result
    if not isinstance(result, str):
        return result
    return result, 200, {'Content-Type': 'text/plain; charset=utf-8'}


@tracker.route('/advisory/<regex("{}"):advisory_id>/generate/raw'.format(advisory_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):advisory_id>/generate/raw'.format(advisory_regex[1:-1]), methods=['GET'])
def show_generated_advisory_raw(advisory_id):
    result = show_generated_advisory(advisory_id, raw=True)
    if isinstance(result, tuple):
        return result
    if not isinstance(result, str):
        return result
    return result, 200, {'Content-Type': 'text/plain; charset=utf-8'}


@tracker.route('/advisory/<regex("{}"):advisory_id>'.format(advisory_regex[1:]), methods=['GET'])
@tracker.route('/<regex("{}"):advisory_id>'.format(advisory_regex[1:]), methods=['GET'])
def show_advisory(advisory_id, raw=False):
    entries = (db.session.query(Advisory, CVEGroup, CVEGroupPackage, CVE)
               .filter(Advisory.id == advisory_id)
               .join(CVEGroupPackage, Advisory.group_package)
               .join(CVEGroup, CVEGroupPackage.group)
               .join(CVEGroupEntry, CVEGroup.issues)
               .join(CVE, CVEGroupEntry.cve)
               .order_by(CVE.id)
               ).all()
    if not entries:
        return not_found()

    advisory = entries[0][0]
    group = entries[0][1]
    package = entries[0][2]
    issues = [issue for (advisory, group, package, issue) in entries]

    if not advisory.content:
        if raw:
            return redirect('/{}/generate/raw'.format(advisory_id))
        return redirect('/{}/generate'.format(advisory_id))

    if raw:
        return advisory.content
    asa = advisory_extend_html(advisory_escape_html(advisory.content), issues, package)
    return render_html_advisory(advisory=advisory, package=package, group=group, raw_asa=asa, generated=False)


@tracker.route('/advisory/<regex("{}"):advisory_id>/generate'.format(advisory_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):advisory_id>/generate'.format(advisory_regex[1:-1]), methods=['GET'])
def show_generated_advisory(advisory_id, raw=False):
    advisory = generate_advisory(advisory_id, with_subject=True, raw=raw)
    if not advisory:
        return not_found()
    return advisory


@tracker.route('/advisory/<regex("{}"):advisory_id>/log'.format(advisory_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):advisory_id>/log'.format(advisory_regex[1:-1]), methods=['GET'])
def show_advisory_log(advisory_id):
    advisory = (db.session.query(Advisory)
                .filter(Advisory.id == advisory_id)
                ).first()
    if not advisory:
        return not_found()

    return render_template('log/advisory_log.html',
                           title='{} - log'.format(advisory_id),
                           advisory=advisory,
                           can_watch_user_log=user_can_watch_user_log())


# TODO: define permission to view this
@tracker.route('/log', defaults={'page': 1}, methods=['GET'])
@tracker.route('/log/page/<int:page>', methods=['GET'])
def show_log(page=1):
    Transaction = versioning_manager.transaction_cls
    VersionClassCVE = version_class(CVE)
    VersionClassGroup = version_class(CVEGroup)
    VersionClassAdvisory = version_class(Advisory)

    pagination = (db.session.query(Transaction, VersionClassCVE, VersionClassGroup, VersionClassAdvisory)
                  .outerjoin(VersionClassCVE, Transaction.id == VersionClassCVE.transaction_id)
                  .outerjoin(VersionClassGroup, Transaction.id == VersionClassGroup.transaction_id)
                  .outerjoin(VersionClassAdvisory, Transaction.id == VersionClassAdvisory.transaction_id)
                  .order_by(Transaction.issued_at.desc())
                  .filter((VersionClassCVE.transaction_id) |
                          (VersionClassGroup.transaction_id) |
                          (VersionClassAdvisory.transaction_id))
                  ).paginate(page, TRACKER_LOG_ENTRIES_PER_PAGE, True)

    return render_template('log/log.html',
                           title=f'Log',
                           can_watch_user_log=user_can_watch_user_log(),
                           pagination=pagination,
                           CVE=CVE,
                           CVEGroup=CVEGroup,
                           Advisory=Advisory)


@tracker.route('/group/<regex("{}"):avg>/create_bug'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/avg/<regex("{}"):avg>/create_bug'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@tracker.route('/<regex("{}"):avg>/create_bug'.format(vulnerability_group_regex[1:-1]), methods=['GET'])
@reporter_required
def create_bug_ticket_redirect(avg):
    data = get_group_data(avg)
    if not data:
        return not_found()

    group = data['group']
    issues = data['issues']
    packages = data['packages']
    versions = data['versions']

    if group.status != Status.vulnerable:
        flash('Bug ticket can only be created for vulnerable groups.', 'warning')
        return redirect(url_for('tracker.show_group', avg=avg))

    pkg_identifier = ''
    if not versions:
        pkg_identifier = packages[0].pkgname
    else:
        pkg_identifier = versions[0].base

    pkgname = packages[0].pkgname
    gitlab_url = f"https://gitlab.archlinux.org/archlinux/packaging/packages/{pkgname}/-/issues/new"

    bug_data = get_bug_data(issues, packages, versions, group)
    params = {
        'issue[title]': bug_data['item_summary'],
        'issue[description]': bug_data['detailed_desc']
    }

    final_url = add_params_to_uri(gitlab_url, params)
    return redirect(final_url)