views.py

# Standard library
import re
from collections import defaultdict
from datetime import datetime
from math import ceil

# Packages
import flask
import dateutil
import talisker.requests
from feedgen.entry import FeedEntry
from feedgen.feed import FeedGenerator
from marshmallow import EXCLUDE
from marshmallow.exceptions import ValidationError
from mistune import Markdown
from sortedcontainers import SortedDict
from sqlalchemy import asc, desc, or_, and_, func, case
from sqlalchemy.exc import IntegrityError, DataError
from sqlalchemy.orm import contains_eager

# Local
from webapp.security.database import db_session
from webapp.security.models import CVE, Notice, Package, Status, Release
from webapp.security.schemas import CVESchema, NoticeSchema, ReleaseSchema
from webapp.security.auth import authorization_required
from webapp.security.api import SecurityAPI

markdown_parser = Markdown(
    hard_wrap=True, parse_block_html=True, parse_inline_html=True
)
session = talisker.requests.get_session()


def notice(notice_id):
    notice = db_session.query(Notice).get(notice_id)

    if not notice:
        flask.abort(404)

    package_descriptions = {}
    package_versions = {}
    release_packages = SortedDict()

    releases = {
        release.codename: release.version
        for release in db_session.query(Release).all()
    }

    if notice.release_packages:
        for codename, pkgs in notice.release_packages.items():
            release_version = releases[codename]
            release_packages[release_version] = {}
            for package in pkgs:
                if not package.get("is_visible", True):
                    continue

                name = package["name"]
                if not package["is_source"]:
                    release_packages[release_version][name] = package
                    continue

                if notice.get_type == "LSN":
                    if name not in package_descriptions:
                        package_versions[name] = []

                    package_versions[name].append(package["version"])
                    versions = ", >= ".join(package_versions[name])
                    description = (
                        f"{package['description']} - " f"(>= {versions})"
                    )
                    package_descriptions[name] = description

                    continue

                if name not in package_descriptions:
                    package_descriptions[name] = package["description"]

        package_descriptions = {
            key: package_descriptions[key]
            for key in sorted(package_descriptions.keys())
        }

    notice_cve_ids = [cve.id for cve in notice.cves]
    related_notices = (
        db_session.query(Notice)
        .filter(Notice.cves.any(CVE.id.in_(notice_cve_ids)))
        .filter(Notice.id != notice.id)
        .all()
    )

    if notice.get_type == "LSN":
        template = "security/notices/lsn.html"
    else:
        template = "security/notices/usn.html"

    notice = {
        "id": notice.id,
        "title": notice.title,
        "published": notice.published,
        "summary": notice.summary,
        "details": markdown_parser(notice.get_processed_details),
        "instructions": markdown_parser(notice.instructions),
        "package_descriptions": package_descriptions,
        "release_packages": release_packages,
        "releases": notice.releases,
        "cves": notice.cves,
        "references": notice.references,
        "related_notices": [
            {
                "id": related_notice.id,
                "package_list": ", ".join(related_notice.package_list),
            }
            for related_notice in related_notices
        ],
    }

    return flask.render_template(template, notice=notice)


def notices():
    page = flask.request.args.get("page", default=1, type=int)
    details = flask.request.args.get("details", type=str)
    release = flask.request.args.get("release", type=str)
    order_by = flask.request.args.get("order", type=str)

    releases = (
        db_session.query(Release)
        .order_by(desc(Release.release_date))
        .filter(Release.version.isnot(None))
        .all()
    )
    notices_query = db_session.query(Notice)

    if release:
        notices_query = notices_query.join(Release, Notice.releases).filter(
            Release.codename == release
        )

    if details:
        notices_query = notices_query.filter(
            or_(
                Notice.id.ilike(f"%{details}%"),
                Notice.details.ilike(f"%{details}%"),
                Notice.title.ilike(f"%{details}%"),
                Notice.cves.any(CVE.id.ilike(f"%{details}%")),
            )
        )

    # Snapshot total results for search
    page_size = 10
    total_results = notices_query.count()
    total_pages = ceil(total_results / page_size)
    offset = page * page_size - page_size

    if page < 1 or 1 < page > total_pages:
        flask.abort(404)

    sort = asc if order_by == "oldest" else desc
    notices = (
        notices_query.order_by(sort(Notice.published))
        .offset(offset)
        .limit(page_size)
        .all()
    )

    return flask.render_template(
        "security/notices.html",
        notices=notices,
        releases=releases,
        pagination=dict(
            current_page=page,
            total_pages=total_pages,
            total_results=total_results,
            page_first_result=offset + 1,
            page_last_result=offset + len(notices),
        ),
    )


# USN Feeds
# ===


def notices_feed(feed_type):
    if feed_type not in ["atom", "rss"]:
        flask.abort(404)

    url_root = flask.request.url_root
    base_url = flask.request.base_url

    feed = FeedGenerator()
    feed.generator("Feedgen")

    feed.id(url_root)
    feed.copyright(
        f"{datetime.now().year} Canonical Ltd. "
        "Ubuntu and Canonical are registered trademarks of Canonical Ltd."
    )
    feed.title("Ubuntu security notices")
    feed.description("Recent content on Ubuntu security notices")
    feed.link(href=base_url, rel="self")

    def feed_entry(notice, url_root):
        _id = notice.id
        title = f"{_id}: {notice.title}"
        description = notice.details
        published = notice.published
        notice_path = flask.url_for(".notice", notice_id=notice.id).lstrip("/")
        link = f"{url_root}{notice_path}"

        entry = FeedEntry()
        entry.id(link)
        entry.title(title)
        entry.description(description)
        entry.link(href=link)
        entry.published(f"{published} UTC")
        entry.author({"name": "Ubuntu Security Team"})

        return entry

    notices = (
        db_session.query(Notice)
        .order_by(desc(Notice.published))
        .limit(10)
        .all()
    )

    for notice in notices:
        feed.add_entry(feed_entry(notice, url_root), order="append")

    payload = feed.atom_str() if feed_type == "atom" else feed.rss_str()
    return flask.Response(payload, mimetype="text/xml")


# USN API
# ===
def _update_notice_object(notice, data):
    """
    Set fields on a Notice model object
    """

    notice.title = data["title"]
    notice.summary = data["summary"]
    notice.details = data["description"]
    notice.release_packages = data["release_packages"]
    notice.published = data["published"]
    notice.references = data["references"]
    notice.instructions = data["instructions"]
    notice.is_hidden = data.get("is_hidden", False)

    notice.releases = [
        db_session.query(Release).get(codename)
        for codename in data["release_packages"].keys()
    ]

    notice.cves.clear()
    for cve_id in set(data["cves"]):
        notice.cves.append(db_session.query(CVE).get(cve_id) or CVE(id=cve_id))

    return notice


def single_notices_sitemap(offset):
    notices = (
        db_session.query(Notice)
        .order_by(Notice.published)
        .offset(offset)
        .limit(10000)
        .all()
    )

    xml_sitemap = flask.render_template(
        "sitemap.xml",
        links=[
            {
                "url": f"https://ubuntu.com/security/notices/{notice.id}",
                "last_updated": notice.published.strftime("%Y-%m-%d"),
            }
            for notice in notices
        ],
    )

    response = flask.make_response(xml_sitemap)
    response.headers["Content-Type"] = "application/xml"
    response.headers["Cache-Control"] = "public, max-age=43200"

    return response


def notices_sitemap():
    notices_count = db_session.query(Notice).order_by(Notice.published).count()

    base_url = "https://ubuntu.com/security/notices"

    xml_sitemap = flask.render_template(
        "sitemap_index_template.xml",
        base_url=base_url,
        links=[
            {
                "url": f"{base_url}/sitemap-{link * 10000}.xml",
            }
            for link in range(ceil(notices_count / 10000))
        ],
    )

    response = flask.make_response(xml_sitemap)
    response.headers["Content-Type"] = "application/xml"
    response.headers["Cache-Control"] = "public, max-age=43200"

    return response


@authorization_required
def create_notice():
    """
    POST method to create a new notice
    """

    notice_schema = NoticeSchema()
    notice_schema.context["release_codenames"] = [
        rel.codename for rel in db_session.query(Release).all()
    ]

    try:
        notice_data = notice_schema.load(flask.request.json)
    except ValidationError as error:
        return (
            flask.jsonify(
                {"message": "Invalid payload", "errors": error.messages}
            ),
            400,
        )

    db_session.add(
        _update_notice_object(Notice(id=notice_data["id"]), notice_data)
    )

    try:
        db_session.commit()
    except IntegrityError:
        return (
            flask.jsonify(
                {"message": f"Notice {notice_data['id']} already exists"}
            ),
            400,
        )

    return flask.jsonify({"message": "Notice created"}), 201


@authorization_required
def update_notice(notice_id):
    """
    PUT method to update a single notice
    """
    notice = db_session.query(Notice).without_default_filters().get(notice_id)

    if not notice:
        return (
            flask.jsonify({"message": f"Notice {notice_id} doesn't exist"}),
            404,
        )

    notice_schema = NoticeSchema()
    notice_schema.context["release_codenames"] = [
        rel.codename for rel in db_session.query(Release).all()
    ]

    try:
        notice_data = notice_schema.load(flask.request.json, unknown=EXCLUDE)
    except ValidationError as error:
        return (
            flask.jsonify(
                {"message": "Invalid payload", "errors": error.messages}
            ),
            400,
        )

    notice = _update_notice_object(notice, notice_data)

    db_session.add(notice)
    db_session.commit()

    return flask.jsonify({"message": "Notice updated"}), 200


@authorization_required
def delete_notice(notice_id):
    """
    DELETE method to delete a single notice
    """
    notice = db_session.query(Notice).without_default_filters().get(notice_id)

    if not notice:
        return (
            flask.jsonify({"message": f"Notice {notice_id} doesn't exist"}),
            404,
        )

    db_session.delete(notice)
    db_session.commit()

    return flask.jsonify({"message": f"Notice {notice_id} deleted"}), 200


@authorization_required
def create_release():
    """
    POST method to create a new release
    """

    release_schema = ReleaseSchema()

    try:
        release_data = release_schema.load(flask.request.json)
    except ValidationError as error:
        return (
            flask.jsonify(
                {"message": "Invalid payload", "errors": error.messages}
            ),
            400,
        )

    release = Release(
        codename=release_data["codename"],
        version=release_data["version"],
        name=release_data["name"],
        development=release_data["development"],
        lts=release_data["lts"],
        release_date=release_data["release_date"],
        esm_expires=release_data["esm_expires"],
        support_expires=release_data["support_expires"],
    )

    db_session.add(release)

    try:
        db_session.commit()
    except IntegrityError:
        return (
            flask.jsonify(
                {
                    "message": (
                        f"Release with [codename:'{release_data['codename']}']"
                        f" or [version:'{release_data['version']}'] or "
                        f"[name:'{release_data['name']}'] already exists"
                    )
                }
            ),
            400,
        )

    return flask.jsonify({"message": "Release created"}), 200


@authorization_required
def delete_release(codename):
    """
    DELETE method to delete a single release
    """
    release = db_session.query(Release).get(codename)

    if not release:
        return (
            flask.jsonify({"message": f"Release {codename} doesn't exist"}),
            404,
        )

    if len(release.statuses) > 0:
        return (
            flask.jsonify(
                {
                    "message": (
                        f"Cannot delete '{codename}' release. "
                        f"Release already in use"
                    )
                }
            ),
            400,
        )

    db_session.delete(release)
    db_session.commit()

    return flask.jsonify({"message": f"Release {codename} deleted"}), 200


# CVE views
# ===
def cve_index():
    """
    Display the list of CVEs, with pagination.
    Also accepts the following filtering query parameters:
    - order-by - "oldest" or "newest"
    - query - search query for the description field
    - priority
    - limit - default 20
    - offset - default 0
    """
    # Query parameters
    query = flask.request.args.get("q", "").strip()
    priority = flask.request.args.get("priority")
    package = flask.request.args.get("package")
    limit = flask.request.args.get("limit", default=20, type=int)
    offset = flask.request.args.get("offset", default=0, type=int)
    component = flask.request.args.get("component")
    versions = flask.request.args.getlist("version")
    statuses = flask.request.args.getlist("status")

    is_cve_id = re.match(r"^CVE-\d{4}-\d{4,7}$", query.upper())

    if is_cve_id and db_session.query(CVE).get(query.upper()):
        return flask.redirect(f"/security/{query.lower()}")

    all_releases = (
        db_session.query(Release)
        .order_by(desc(Release.release_date))
        .filter(Release.codename != "upstream")
        .all()
    )

    releases_query = db_session.query(Release).order_by(Release.release_date)

    if versions and not any(a in ["", "current"] for a in versions):
        releases_query = releases_query.filter(Release.codename.in_(versions))
    else:
        releases_query = releases_query.filter(
            or_(
                Release.support_expires > datetime.now(),
                Release.esm_expires > datetime.now(),
            )
        ).filter(Release.codename != "upstream")

    releases = releases_query.all()

    should_filter_by_version_and_status = (
        versions and statuses and len(versions) == len(statuses)
    )

    clean_versions = []
    clean_statuses = []
    if should_filter_by_version_and_status:
        raw_all_statuses = db_session.execute(
            "SELECT unnest(enum_range(NULL::statuses));"
        ).fetchall()
        all_statuses = ["".join(s) for s in raw_all_statuses]

        clean_versions = [
            (
                [version]
                if version not in ["", "current"]
                else [r.codename for r in releases]
            )
            for version in versions
        ]

        clean_statuses = [
            ([status] if status != "" else all_statuses) for status in statuses
        ]

    # query cves by filters
    cves_query = db_session.query(
        CVE, func.count("*").over().label("total")
    ).filter(CVE.status == "active")

    if priority:
        cves_query = cves_query.filter(CVE.priority == priority)

    if query:
        cves_query = cves_query.filter(
            or_(
                CVE.description.ilike(f"%{query}%"),
                CVE.ubuntu_description.ilike(f"%{query}%"),
            )
        )

    parameters = []
    if package:
        parameters.append(Status.package_name == package)

    if component:
        parameters.append(Status.component == component)

    if should_filter_by_version_and_status:
        conditions = []
        for key, version in enumerate(clean_versions):
            conditions.append(
                and_(
                    Status.release_codename.in_(version),
                    Status.status.in_(clean_statuses[key]),
                )
            )

        parameters.append(or_(*[c for c in conditions]))

        conditions = []
        for key, version in enumerate(clean_versions):
            sub_conditions = [
                Status.release_codename.in_(version),
                Status.status.in_(clean_statuses[key]),
                CVE.id == Status.cve_id,
            ]

            if package:
                sub_conditions.append(Status.package_name == package)

            if component:
                sub_conditions.append(Status.component == component)

            condition = Package.statuses.any(
                and_(*[sc for sc in sub_conditions])
            )

            conditions.append(condition)

        parameters.append(Status.package.has(and_(*[c for c in conditions])))
    else:
        parameters.append(Status.status.in_(Status.active_statuses))

    if len(parameters) > 0:
        cves_query = cves_query.filter(
            CVE.statuses.any(and_(*[p for p in parameters]))
        )

    cves_query = (
        cves_query.group_by(CVE.id)
        .order_by(
            case(
                [(CVE.published.is_(None), 1)],
                else_=0,
            ),
            desc(CVE.published),
        )
        .limit(limit)
        .offset(offset)
        .from_self()
        .join(CVE.statuses)
        .options(contains_eager(CVE.statuses))
    )

    raw_cves = cves_query.all()

    # Pagination
    total_results = raw_cves[0][1] if raw_cves else 0

    cves = []
    for raw_cve in raw_cves:
        packages = raw_cve[0].packages

        # filter by package name
        if package:
            packages = {
                package_name: package_statuses
                for package_name, package_statuses in packages.items()
                if package_name == package
            }

        # filter by component
        if component:
            packages = {
                package_name: package_statuses
                for package_name, package_statuses in packages.items()
                if any(
                    status.component == component
                    for status in package_statuses.values()
                )
            }

        if should_filter_by_version_and_status:
            packages = {
                package_name: package_statuses
                for package_name, package_statuses in packages.items()
                if all(
                    any(
                        package_status.release_codename in version
                        and package_status.status in clean_statuses[key]
                        for package_status in package_statuses.values()
                    )
                    for key, version in enumerate(clean_versions)
                )
            }

        # do not return cve if it has no packages left
        if not packages:
            continue

        cve = {
            "id": raw_cve[0].id,
            "priority": raw_cve[0].priority,
            "packages": packages,
        }

        cves.append(cve)

    return flask.render_template(
        "security/cve/index.html",
        releases=releases,
        all_releases=all_releases,
        cves=cves,
        total_results=total_results,
        total_pages=ceil(total_results / limit),
        offset=offset,
        limit=limit,
        priority=priority,
        query=query,
        package=package,
        component=component,
        versions=versions,
        statuses=statuses,
    )


def cve(cve_id):
    """
    Retrieve and display an individual CVE details page
    """

    security_api = SecurityAPI(
        session=session,
        base_url="https://ubuntu.com/security/",
    )

    cve = security_api.get_cve(cve_id)

    if not cve:
        flask.abort(404)

    if cve.get("published"):
        cve["published"] = dateutil.parser.parse(cve["published"]).strftime(
            "%-d %B %Y"
        )

    releases = security_api.get_releases()

    return flask.render_template(
        "security/cve/cve.html", cve=cve, releases=releases
    )


# CVE API
# ===
def update_statuses(cve, data, packages):
    statuses = cve.packages

    statuses_query = db_session.query(Status).filter(Status.cve_id == cve.id)
    statuses_to_delete = {
        f"{v.package_name}||{v.release_codename}": v
        for v in statuses_query.all()
    }

    for package_data in data.get("packages", []):
        name = package_data["name"]

        if packages.get(name) is None:
            package = Package(name=name)
            package.source = package_data["source"]
            package.ubuntu = package_data["ubuntu"]
            package.debian = package_data["debian"]
            packages[name] = package

            db_session.add(package)

        for status_data in package_data["statuses"]:
            update_status = False
            codename = status_data["release_codename"]

            status = statuses[name].get(codename)
            if status is None:
                update_status = True
                status = Status(
                    cve_id=cve.id, package_name=name, release_codename=codename
                )
            elif f"{name}||{codename}" in statuses_to_delete:
                del statuses_to_delete[f"{name}||{codename}"]

            if status.status != status_data["status"]:
                update_status = True
                status.status = status_data["status"]

            if status.description != status_data["description"]:
                update_status = True
                status.description = status_data["description"]

            if status.component != status_data.get("component"):
                update_status = True
                status.component = status_data.get("component")

            if status.pocket != status_data.get("pocket"):
                update_status = True
                status.pocket = status_data.get("pocket")

            if update_status:
                statuses[name][codename] = status
                db_session.add(status)

    for key in statuses_to_delete:
        db_session.delete(statuses_to_delete[key])


@authorization_required
def delete_cve(cve_id):
    """
    Delete a CVE from db
    @params string: query string with the CVE id
    """
    cve_query = db_session.query(CVE)
    cve = cve_query.filter(CVE.id == cve_id).first()

    try:
        db_session.delete(cve)
        db_session.commit()

    except IntegrityError as error:
        return flask.jsonify({"message": error.orig.args[0]}), 400

    return flask.jsonify({"message": "CVE deleted successfully"}), 200


@authorization_required
def bulk_upsert_cve():
    """
    Receives a PUT request from load_cve.py
    Parses the object and bulk inserts or updates
    @returns 3 lists of CVEs, created CVEs, updated CVEs and failed CVEs
    """

    cves_schema = CVESchema(many=True)
    cves_schema.context["release_codenames"] = [
        rel.codename for rel in db_session.query(Release).all()
    ]

    try:
        cves_data = cves_schema.load(flask.request.json)
    except ValidationError as error:
        return (
            flask.jsonify(
                {"message": "Invalid payload", "errors": error.messages}
            ),
            400,
        )

    if len(cves_data) > 50:
        return (
            flask.jsonify(
                {
                    "message": (
                        "Please only submit up to 50 CVEs at a time. "
                        f"({len(cves_data)} submitted)"
                    )
                }
            ),
            413,
        )

    packages = {}
    for package in db_session.query(Package).all():
        packages[package.name] = package

    for data in cves_data:
        update_cve = False
        cve = db_session.query(CVE).get(data["id"])

        if cve is None:
            update_cve = True
            cve = CVE(id=data["id"])

        if cve.status != data.get("status"):
            update_cve = True
            cve.status = data.get("status")

        published_date = (
            cve.published.strftime("%Y-%B-%d") if cve.published else None
        )
        data_published_date = (
            data.get("published").strftime("%Y-%B-%d")
            if data.get("published")
            else None
        )
        if published_date != data_published_date:
            update_cve = True
            cve.published = data.get("published")

        if cve.priority != data.get("priority"):
            update_cve = True
            cve.priority = data.get("priority")

        if cve.cvss3 != data.get("cvss3"):
            update_cve = True
            cve.cvss3 = data.get("cvss3")

        if cve.description != data.get("description"):
            update_cve = True
            cve.description = data.get("description")

        if cve.ubuntu_description != data.get("ubuntu_description"):
            update_cve = True
            cve.ubuntu_description = data.get("ubuntu_description")

        if cve.notes != data.get("notes"):
            update_cve = True
            cve.notes = data.get("notes")

        if cve.references != data.get("references"):
            update_cve = True
            cve.references = data.get("references")

        if cve.bugs != data.get("bugs"):
            update_cve = True
            cve.bugs = data.get("bugs")

        if cve.patches != data.get("patches"):
            update_cve = True
            cve.patches = data.get("patches")

        if cve.tags != data.get("tags"):
            update_cve = True
            cve.tags = data.get("tags")

        if cve.mitigation != data.get("mitigation"):
            update_cve = True
            cve.mitigation = data.get("mitigation")

        if update_cve:
            db_session.add(cve)

        update_statuses(cve, data, packages)

    created = defaultdict(lambda: 0)
    updated = defaultdict(lambda: 0)
    deleted = defaultdict(lambda: 0)

    for item in db_session.new:
        created[type(item).__name__] += 1

    for item in db_session.dirty:
        updated[type(item).__name__] += 1

    for item in db_session.deleted:
        deleted[type(item).__name__] += 1

    try:
        db_session.commit()
    except DataError as error:
        return (
            flask.jsonify(
                {
                    "message": "Failed bulk upserting session",
                    "error": error.orig.args[0],
                }
            ),
            400,
        )

    return (
        flask.jsonify(
            {"created": created, "updated": updated, "deleted": deleted}
        ),
        200,
    )


def single_cves_sitemap(offset):
    cves = (
        db_session.query(CVE)
        .order_by(CVE.published)
        .offset(offset)
        .limit(10000)
        .all()
    )

    xml_sitemap = flask.render_template(
        "sitemap.xml",
        links=[
            {
                "url": f"https://ubuntu.com/security/{cve.id}",
                "last_updated": (
                    cve.published.strftime("%Y-%m-%d") if cve.published else ""
                ),
            }
            for cve in cves
        ],
    )

    response = flask.make_response(xml_sitemap)
    response.headers["Content-Type"] = "application/xml"
    response.headers["Cache-Control"] = "public, max-age=43200"

    return response


def cves_sitemap():
    cves_count = db_session.query(CVE).order_by(CVE.published).count()

    base_url = "https://ubuntu.com/security/cve"

    xml_sitemap = flask.render_template(
        "sitemap_index_template.xml",
        base_url=base_url,
        links=[
            {
                "url": f"{base_url}/sitemap-{link * 10000}.xml",
            }
            for link in range(ceil(cves_count / 10000))
        ],
    )

    response = flask.make_response(xml_sitemap)
    response.headers["Content-Type"] = "application/xml"
    response.headers["Cache-Control"] = "public, max-age=43200"

    return response