__init__.py

#!/usr/bin/env python3

from __future__ import annotations

import logging
import logging.config
from importlib.metadata import version
from typing import Any

from feedgen.feed import FeedGenerator  # type: ignore[import-untyped]
from flask import Flask, Response, flash, make_response, redirect, render_template, request, url_for
from flask_bootstrap import Bootstrap5  # type: ignore[import-untyped]
from flask_restx import Api, Resource  # type: ignore[import-untyped]
from werkzeug import Response as WerkzeugResponse

from vulnerabilitylookup.default import get_config
from vulnerabilitylookup.vulnerabilitylookup import VulnerabilityLookup

from .helpers import get_secret_key, sri_load
from .proxied import ReverseProxied

logging.config.dictConfig(get_config("logging"))

app: Flask = Flask(__name__)

app.wsgi_app = ReverseProxied(app.wsgi_app)  # type: ignore

app.config["SECRET_KEY"] = get_secret_key()
Bootstrap5(app)
app.config["BOOTSTRAP_SERVE_LOCAL"] = True
app.config["SESSION_COOKIE_NAME"] = "vulnerability-lookup"
app.config["SESSION_COOKIE_SAMESITE"] = "Strict"
app.debug = False

vulnerabilitylookup: VulnerabilityLookup = VulnerabilityLookup()


# ##### Global methods passed to jinja


def get_sri(directory: str, filename: str) -> str:
    sha512 = sri_load()[directory][filename]
    return f"sha512-{sha512}"


app.jinja_env.globals.update(get_sri=get_sri)

# #### Web interface ####


@app.route("/", methods=["GET", "POST"])
@app.route("/search", methods=["GET", "POST"])
def search() -> str | WerkzeugResponse:
    if request.method == "HEAD":
        # Just returns ack if the webserver is running
        return "Ack"
    if request.method == "GET":
        vulnerability_id = request.args.get("vulnerability_id")
        vendor = request.args.get("vendor")
        product = request.args.get("product")
    if request.method == "POST":
        vulnerability_id = None
        product = None
        vendor = None
        # trigger search
        if search_query := request.form.get("freetext_search"):  # Normal search
            vulnerability_id = search_query
        elif search_query := request.form.get("vendor_vuln"):  # Vuln id search from vendor
            vulnerability_id = search_query
            vendor = request.form.get("vendor")
        elif search_query := request.form.get("product"):  # Search for all vulns related to this vendor and product
            product = search_query
            vendor = request.form.get("vendor")

    if vulnerability_id:
        if vulnerabilitylookup.get_vulnerability(vulnerability_id, with_meta=True):
            # The search query was a vulnerability ID, just display that.
            return redirect(url_for("vulnerability_view", vulnerability_id=vulnerability_id))

        elif linked_vulns := vulnerabilitylookup.get_linked_vulnerabilities(vulnerability_id):
            # unable to find a vulnerability with that id. Attempt to find linked entries.
            return render_template("search.html", vulnerability_id=vulnerability_id, linked_vulns=linked_vulns)
        elif vulnerability_id in vulnerabilitylookup.get_vendors():
            # FIXME: this is dirty.
            vendor = vulnerability_id
            # Got a vendor, show the proper template.
            vendor_products = vulnerabilitylookup.get_vendor_products(vendor)
            vendor_vulns = vulnerabilitylookup.get_vendor_vulnerabilities(vendor)
            return render_template("search.html", vendor=vendor, vendor_products=vendor_products, vendor_vulns=vendor_vulns)
        flash(f"Nothing found in the database for {vulnerability_id}.", "warning")
    elif vendor and not product:
        if vendor in vulnerabilitylookup.get_vendors():
            # Got a vendor, show the proper template.
            vendor_products = vulnerabilitylookup.get_vendor_products(vendor)
            vendor_vulns = vulnerabilitylookup.get_vendor_vulnerabilities(vendor)
            return render_template("search.html", vendor=vendor, vendor_products=vendor_products, vendor_vulns=vendor_vulns)
        else:
            flash(f"Nothing found in the database for {vendor}.", "warning")
    elif vendor and product:
        if vulnerabilities := vulnerabilitylookup.get_vendor_product_vulnerabilities(vendor, product):
            return render_template("search.html", vendor=vendor, product=product, vp_vulnerabilities=vulnerabilities)
        else:
            flash(f"Nothing found in the database for {vendor} / {product}.", "warning")

    # render search page
    return render_template("search.html")


@app.route("/recent", methods=["GET"])
@app.route("/recent/<int:page>", methods=["GET"])
def recent(page: int = 1) -> str:
    # For the webinterface, we want the most recent entries by source
    source_to_show = [
        "github",
        "cvelistv5",
        "pysec",
        "gsd",
        "ossf_malicious_packages",
        "csaf_certbund",
        "csaf_siemens",
        "csaf_redhat",
        "csaf_cisa",
        "csaf_cisco",
        "csaf_sick",
        "csaf_nozominetworks",
        "csaf_ox",
    ]
    default_source = "cvelistv5"
    recent = {source: vulnerabilitylookup.get_last(source, page=page) for source in source_to_show}
    try:
        p = int(page)
        if p < 1:
            page = 1
    except ValueError:
        page = 1
    return render_template("recent.html", recent=recent, default_source=default_source, current_page=page)


@app.route("/cve/<string:vulnerability_id>", methods=["GET"])
@app.route("/vuln/<string:vulnerability_id>", methods=["GET"])
def vulnerability_view(vulnerability_id: str) -> str:
    if vulnerability := vulnerabilitylookup.get_vulnerability(vulnerability_id, with_meta=True):
        # The search query was a vulnerability ID, just display that.
        source = vulnerabilitylookup.get_vulnerability_source(vulnerability_id)
        linked_vulns = vulnerabilitylookup.get_linked_vulnerabilities(vulnerability_id)
        return render_template(
            "vuln.html", source=source, vulnerability_id=vulnerability_id, vulnerability_data=vulnerability, linked_vulns=linked_vulns
        )

    flash(f"Nothing found in the database for {vulnerability_id}.", "warning")
    return render_template("search.html")


@app.route("/rss/recent/<string:source>", methods=["GET"])
@app.route("/rss/recent/<string:source>/<int:length>", methods=["GET"])
def rss_recent(source: str, length: int = 15) -> Response:
    public_domain = get_config("generic", "public_domain")
    fg = FeedGenerator()
    fg.id(f"https://{public_domain}/rss/recent/{source}/{length}")
    fg.title(f"Most recent entries from {source}")
    fg.author({"name": "Vulnerability Lookup"})
    fg.link(href=f"https://{public_domain}", rel="alternate")
    # fg.logo('http://ex.com/logo.jpg')
    fg.subtitle(f"Contains only the most {length} recent entries.")
    fg.language("en")
    for vuln_id, entry in vulnerabilitylookup.get_last(source, page=1, number=length):
        fe = fg.add_entry()
        fe.id(f"https://{public_domain}/vuln/{vuln_id}")
        fe.title(vuln_id)
        fe.link(href=f"https://{public_domain}/vuln/{vuln_id}")
        if source in ["github", "pysec", "ossf_malicious_packages"]:
            fe.content(entry["details"])
        elif source == "cvelistv5" and "title" in entry["containers"]["cna"]:
            fe.content(entry["containers"]["cna"]["title"])
        else:
            pass
        fe.description()
    r = make_response(fg.rss_str(pretty=True))
    r.mimetype = "application/rss+xml"
    r.headers["Content-Type"] = "application/rss+xml"
    return r


# #### API ####

api = Api(
    app,
    title="VulnerabilityLookup API",
    description="API to query VulnerabilityLookup.",
    version=version("vulnerabilitylookup"),
    doc="/doc",
)


@api.route("/redis_up")
@api.doc(description="Check if redis is up and running")
class RedisUp(Resource):  # type: ignore[misc]
    def get(self) -> bool:
        return vulnerabilitylookup.check_redis_up()


@api.route("/api/cve/<string:vulnerability_id>")
@api.route("/vulnerability/<string:vulnerability_id>")
@api.doc(description="Get a vulnerability")
class Vulnerability(Resource):  # type: ignore[misc]
    def get(self, vulnerability_id: str) -> dict[str, Any] | None:
        with_meta = True if request.args.get("with_meta") else False
        return vulnerabilitylookup.get_vulnerability(vulnerability_id, with_meta=with_meta)


@api.route("/api/dbInfo")
@api.route("/info")
@api.doc(description="Get more information about the current databases in use and when it was updated")
class Info(Resource):  # type: ignore[misc]
    def get(self) -> dict[str, Any]:
        return vulnerabilitylookup.get_info()


@api.route("/api/last")
@api.route("/last")
@api.route("/last/<int:number>")
@api.route("/last/<string:source>")
@api.route("/last/<string:source>/<int:number>")
@api.doc(description="Get the last CVEs")
class Last(Resource):  # type: ignore[misc]
    def get(self, source: str | None = None, number: int | None = 30) -> list[dict[str, Any]]:
        return [entry for v_id, entry in vulnerabilitylookup.get_last(source, number)]


@api.route("/api/browse")
@api.doc(description="Get the known vendors")
class Vendors(Resource):  # type: ignore[misc]
    def get(self) -> list[str]:
        return list(vulnerabilitylookup.get_vendors())


@api.route("/api/browse/<string:vendor>")
@api.doc(description="Get the known products for a vendor")
class VendorProducts(Resource):  # type: ignore[misc]
    def get(self, vendor: str) -> list[str]:
        return list(vulnerabilitylookup.get_vendor_products(vendor))


@api.route("/api/search/<string:vendor>/<string:product>")
@api.doc(description="Get the the vulnerabilities per vendor and a specific product")
class VendorProductVulnerabilities(Resource):  # type: ignore[misc]
    def get(self, vendor: str, product: str) -> dict[str, list[tuple[str, dict[str, Any]]]]:
        return vulnerabilitylookup.get_vendor_product_vulnerabilities(vendor, product)