views.py

#
# Copyright (c) nexB Inc. and others. All rights reserved.
# VulnerableCode is a trademark of nexB Inc.
# SPDX-License-Identifier: Apache-2.0
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
# See https://github.com/aboutcode-org/vulnerablecode for support or download.
# See https://aboutcode.org for more information about nexB OSS projects.
#

import json
import logging
import time
from collections import defaultdict
from typing import List

from cvss.exceptions import CVSS2MalformedError
from cvss.exceptions import CVSS3MalformedError
from cvss.exceptions import CVSS4MalformedError
from django.contrib import messages
from django.contrib.auth.views import LoginView
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.mail import send_mail
from django.db.models import Exists
from django.db.models import OuterRef
from django.db.models import Prefetch
from django.db.models import Q
from django.http import HttpResponse
from django.http.response import Http404
from django.shortcuts import get_object_or_404
from django.shortcuts import render
from django.urls import reverse_lazy
from django.views import View
from django.views import generic
from django.views.generic.detail import DetailView
from django.views.generic.edit import FormMixin
from django.views.generic.edit import FormView
from django.views.generic.list import ListView

from vulnerabilities import models
from vulnerabilities.forms import AdminLoginForm
from vulnerabilities.forms import AdvisorySearchForm
from vulnerabilities.forms import AdvisoryToDoForm
from vulnerabilities.forms import AltchaForm
from vulnerabilities.forms import ApiUserCreationForm
from vulnerabilities.forms import PackageSearchForm
from vulnerabilities.forms import PipelineSchedulePackageForm
from vulnerabilities.forms import VulnerabilitySearchForm
from vulnerabilities.models import ISSUE_TYPE_CHOICES
from vulnerabilities.models import AdvisorySetMember
from vulnerabilities.models import AdvisoryToDoV2
from vulnerabilities.models import AdvisoryV2
from vulnerabilities.models import Group
from vulnerabilities.models import GroupedAdvisory
from vulnerabilities.models import ImpactedPackageAffecting
from vulnerabilities.models import ImpactedPackageFixedBy
from vulnerabilities.models import PipelineRun
from vulnerabilities.models import PipelineSchedule
from vulnerabilities.pipelines.v2_importers.epss_importer_v2 import EPSSImporterPipeline
from vulnerabilities.severity_systems import EPSS
from vulnerabilities.severity_systems import SCORING_SYSTEMS
from vulnerabilities.tasks import compute_queue_load_factor
from vulnerabilities.throttling import AnonUserUIThrottle
from vulnerabilities.utils import TYPES_WITH_MULTIPLE_IMPORTERS
from vulnerabilities.utils import get_advisories_from_groups
from vulnerabilities.utils import safe_altcha_redirect
from vulnerablecode import __version__ as VULNERABLECODE_VERSION
from vulnerablecode.settings import VULNERABLECODE_ALTCHA_SESSION_TIMEOUT
from vulnerablecode.settings import VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN
from vulnerablecode.settings import VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS
from vulnerablecode.settings import env

PAGE_SIZE = 10


class VulnerableCodeView(View):
    """
    Base ListView for VulnerableCode views that includes throttling.
    """

    throttle_classes = [AnonUserUIThrottle]

    def dispatch(self, request, *args, **kwargs):
        throttle = AnonUserUIThrottle()

        if not throttle.allow_request(request, self):
            return HttpResponse("Rate limit exceeded", status=429)

        return super().dispatch(request, *args, **kwargs)


class VulnerableCodeDetailView(DetailView, VulnerableCodeView):
    """
    Base DetailView for VulnerableCode views that includes throttling.
    """

    pass


class VulnerableCodeListView(ListView, VulnerableCodeView):
    """
    Base ListView for VulnerableCode views that includes throttling.
    """

    pass


class VulnerableCodeCreateView(generic.CreateView, VulnerableCodeView):
    """
    Base CreateView for VulnerableCode views that includes throttling.
    """

    pass


class PackageSearch(VulnerableCodeListView):
    model = models.Package
    template_name = "packages.html"
    ordering = ["type", "namespace", "name", "version"]
    paginate_by = PAGE_SIZE

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        request_query = self.request.GET
        context["package_search_form"] = PackageSearchForm(request_query)
        context["search"] = request_query.get("search")
        return context

    def get_queryset(self, query=None):
        """
        Return a Package queryset for the ``query``.
        Make a best effort approach to find matching packages either based
        on exact purl, partial purl or just name and namespace.
        """
        query = query or self.request.GET.get("search") or ""
        return (
            self.model.objects.search(query)
            .with_vulnerability_counts()
            .prefetch_related()
            .order_by("package_url")
        )


class VulnerabilitySearch(VulnerableCodeListView):
    model = models.Vulnerability
    template_name = "vulnerabilities.html"
    ordering = ["vulnerability_id"]
    paginate_by = PAGE_SIZE

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        request_query = self.request.GET
        context["vulnerability_search_form"] = VulnerabilitySearchForm(request_query)
        context["search"] = request_query.get("search")
        return context

    def get_queryset(self, query=None):
        query = query or self.request.GET.get("search") or ""
        return self.model.objects.search(query=query).with_package_counts()


class PackageDetails(VulnerableCodeDetailView):
    model = models.Package
    template_name = "package_details.html"
    slug_url_kwarg = "purl"
    slug_field = "purl"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        package = self.object
        context["package"] = package
        context["affected_by_vulnerabilities"] = package.affected_by.order_by("vulnerability_id")
        # Ghost package should not fix any vulnerability.
        context["fixing_vulnerabilities"] = (
            None if package.is_ghost else package.fixing.order_by("vulnerability_id")
        )
        context["package_search_form"] = PackageSearchForm(self.request.GET)
        context["fixed_package_details"] = package.fixed_package_details

        context["history"] = list(package.history)
        return context

    def get_object(self, queryset=None):
        if queryset is None:
            queryset = self.get_queryset()

        purl = self.kwargs.get(self.slug_url_kwarg)
        if purl:
            queryset = queryset.for_purl(purl)
        else:
            cls = self.__class__.__name__
            raise AttributeError(
                f"Package details view {cls} must be called with a purl, " f"but got: {purl!r}"
            )

        try:
            package = queryset.get()
        except queryset.model.DoesNotExist:
            raise Http404(f"No Package found for purl: {purl}")
        return package


class PackageSearchV2(VulnerableCodeListView):
    model = models.PackageV2
    template_name = "packages_v2.html"
    ordering = ["type", "namespace", "name", "version"]
    paginate_by = PAGE_SIZE

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        request_query = self.request.GET
        context["package_search_form"] = PackageSearchForm(request_query)
        context["search"] = request_query.get("search")
        return context

    def get_queryset(self, query=None):
        """
        Return a Package queryset for the ``query``.
        """

        query = query or self.request.GET.get("search") or ""

        qs = self.model.objects.search(query)

        affecting_exists = ImpactedPackageAffecting.objects.filter(
            package_id=OuterRef("pk"),
            impacted_package__advisory___all_impacts_unfurled_at__isnull=False,
            impacted_package__advisory__is_latest=True,
        )

        fixed_by_exists = ImpactedPackageFixedBy.objects.filter(
            package_id=OuterRef("pk"),
            impacted_package__advisory___all_impacts_unfurled_at__isnull=False,
            impacted_package__advisory__is_latest=True,
        )

        return qs.annotate(
            is_vulnerable=Exists(affecting_exists),
            is_fixing=Exists(fixed_by_exists),
        ).filter(Q(is_vulnerable=True) | Q(is_fixing=True))


class AffectedByAdvisoriesListView(VulnerableCodeListView):
    model = models.AdvisoryV2
    template_name = "affected_by_advisories.html"
    paginate_by = PAGE_SIZE

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        purl = self.kwargs.get("purl")
        package = models.PackageV2.objects.for_purl(purl).first()
        context["fixed_package_details"] = get_fixed_package_details(package)
        return context

    def get_queryset(self):
        purl = self.kwargs.get("purl")
        return (
            models.AdvisoryV2.objects.latest_affecting_advisories_for_purl(purl)
            .only("advisory_id", "summary", "url", "date_published")
            .prefetch_related("aliases")
        )


class FixingAdvisoriesListView(VulnerableCodeListView):
    model = models.AdvisoryV2
    template_name = "fixing_advisories.html"
    paginate_by = PAGE_SIZE

    def get_queryset(self):
        purl = self.kwargs.get("purl")
        return (
            models.AdvisoryV2.objects.latest_fixed_by_advisories_for_purl(purl)
            .only("advisory_id", "summary", "url", "date_published")
            .prefetch_related("aliases")
        )


class PackageV2Details(VulnerableCodeDetailView):
    model = models.PackageV2
    template_name = "package_details_v2.html"
    slug_url_kwarg = "purl"
    slug_field = "purl"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        package = self.object

        next_non_vulnerable, latest_non_vulnerable = package.get_non_vulnerable_versions

        context["package"] = package
        context["next_non_vulnerable"] = next_non_vulnerable
        context["latest_non_vulnerable"] = latest_non_vulnerable
        context["package_search_form"] = PackageSearchForm(self.request.GET)

        if not package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
            affecting_advisories = AdvisoryV2.objects.latest_affecting_advisories_for_purl(
                purl=package.purl
            )

            fixed_by_advisories = AdvisoryV2.objects.latest_fixed_by_advisories_for_purl(
                purl=package.purl
            )

            context["grouped"] = False

            affected_by_advisories_url = None
            fixing_advisories_url = None

            affected_by_advisories_qs_ids = affecting_advisories.only("id")
            fixing_advisories_qs_ids = fixed_by_advisories.only("id")

            affected_by_advisories = list(affected_by_advisories_qs_ids[:101])
            if len(affected_by_advisories) > 100:
                affected_by_advisories_url = reverse_lazy(
                    "affected_by_advisories_v2", kwargs={"purl": package.package_url}
                )
                context["affected_by_advisories_v2_url"] = affected_by_advisories_url

            else:
                fixed_pkg_details = get_fixed_package_details(package)
                context["fixed_package_details"] = fixed_pkg_details
                affecting_advs = []
                for adv in affecting_advisories:
                    affecting_advs.append(
                        {
                            "advisory_id": adv.advisory_id.split("/")[-1],
                            "advisory": adv,
                        }
                    )
                context["affected_by_advisories_v2"] = affecting_advs
                context["affected_by_advisories_v2_url"] = None

            fixing_advisories = list(fixing_advisories_qs_ids[:101])
            if len(fixing_advisories) > 100:
                fixing_advisories_url = reverse_lazy(
                    "fixing_advisories_v2", kwargs={"purl": package.package_url}
                )
                context["fixing_advisories_v2_url"] = fixing_advisories_url
                context["fixing_advisories_v2"] = []

            else:
                fixed_by_advisories = fixed_by_advisories.prefetch_related(
                    "aliases",
                )
                fixed_by_advisories = list(fixed_by_advisories)
                fix_advs = []
                for fixed_by_adv in fixed_by_advisories:
                    fix_advs.append(
                        {
                            "advisory_id": fixed_by_adv.advisory_id.split("/")[-1],
                            "advisory": fixed_by_adv,
                        }
                    )

                context["fixing_advisories_v2"] = fix_advs

            return context

        context["grouped"] = True
        fixed_pkg_details = get_fixed_package_details(package)
        context["fixed_package_details"] = fixed_pkg_details

        affected_by_advisories_qs = (
            models.AdvisorySet.objects.filter(package=package, relation_type="affecting")
            .select_related("primary_advisory")
            .prefetch_related(
                Prefetch(
                    "members",
                    queryset=AdvisorySetMember.objects.filter(is_primary=False).select_related(
                        "advisory"
                    ),
                    to_attr="secondary_members",
                )
            )
        )

        fixing_advisories_qs = (
            models.AdvisorySet.objects.filter(package=package, relation_type="fixing")
            .select_related("primary_advisory")
            .prefetch_related(
                Prefetch(
                    "members",
                    queryset=AdvisorySetMember.objects.filter(is_primary=False).select_related(
                        "advisory"
                    ),
                    to_attr="secondary_members",
                )
            )
        )

        affected_groups = [
            (
                Group(
                    aliases=list(adv.aliases.all()),
                    primary=adv.primary_advisory,
                    secondaries=[a.advisory for a in adv.secondary_members],
                )
            )
            for adv in affected_by_advisories_qs
        ]
        fixing_groups = [
            (
                Group(
                    aliases=list(adv.aliases.all()),
                    primary=adv.primary_advisory,
                    secondaries=[a.advisory for a in adv.secondary_members],
                )
            )
            for adv in fixing_advisories_qs
        ]

        affected_advisories: List[GroupedAdvisory] = get_advisories_from_groups(affected_groups)
        fixing_advisories: List[GroupedAdvisory] = get_advisories_from_groups(fixing_groups)

        context["affected_by_advisories_v2"] = affected_advisories
        context["fixing_advisories_v2"] = fixing_advisories

        return context

    def get_object(self, queryset=None):
        if queryset is None:
            queryset = self.get_queryset()

        purl = self.kwargs.get(self.slug_url_kwarg)
        if purl:
            queryset = queryset.for_purl(purl)
        else:
            cls = self.__class__.__name__
            raise AttributeError(
                f"Package details view {cls} must be called with a purl, " f"but got: {purl!r}"
            )

        try:
            package = queryset.get()
        except queryset.model.DoesNotExist:
            raise Http404(f"No Package found for purl: {purl}")
        return package


def get_fixed_package_details(package):
    rows = package.affected_in_impacts.values_list(
        "advisory__avid",
        "fixed_by_packages",
    )

    pkg_ids = {pkg_id for _, pkg_id in rows if pkg_id}

    pkg_map = {
        p.id: p
        for p in models.PackageV2.objects.filter(id__in=pkg_ids, is_ghost=False).annotate(
            is_vulnerable=Exists(
                models.ImpactedPackage.objects.filter(
                    affecting_packages=OuterRef("pk"),
                    advisory__is_latest=True,
                    advisory___all_impacts_unfurled_at__isnull=False,
                )
            )
        )
    }

    fixed_pkg_details = defaultdict(list)

    for avid, pkg_id in rows:
        if not pkg_id:
            continue

        pkg = pkg_map.get(pkg_id)
        if not pkg:
            continue

        fixed_pkg_details[avid].append(
            {
                "pkg": pkg,
                "is_vulnerable": pkg.is_vulnerable,
            }
        )

    return fixed_pkg_details


class VulnerabilityDetails(VulnerableCodeDetailView):
    model = models.Vulnerability
    template_name = "vulnerability_details.html"
    slug_url_kwarg = "vulnerability_id"
    slug_field = "vulnerability_id"

    def get_queryset(self):
        return (
            super()
            .get_queryset()
            .select_related()
            .prefetch_related(
                Prefetch(
                    "references",
                    queryset=models.VulnerabilityReference.objects.only(
                        "reference_id", "reference_type", "url"
                    ),
                ),
                Prefetch(
                    "aliases",
                    queryset=models.Alias.objects.only("alias"),
                ),
                Prefetch(
                    "weaknesses",
                    queryset=models.Weakness.objects.only("cwe_id"),
                ),
                Prefetch(
                    "severities",
                    queryset=models.VulnerabilitySeverity.objects.only(
                        "scoring_system", "value", "url", "scoring_elements", "published_at"
                    ),
                ),
                Prefetch(
                    "exploits",
                    queryset=models.Exploit.objects.only(
                        "data_source", "description", "required_action", "due_date", "notes"
                    ),
                ),
            )
        )

    def get_context_data(self, **kwargs):
        """
        Build context with preloaded QuerySets and minimize redundant queries.
        """
        context = super().get_context_data(**kwargs)
        vulnerability = self.object

        # Pre-fetch and process data in Python instead of the template
        weaknesses_present_in_db = [
            weakness_object
            for weakness_object in vulnerability.weaknesses.all()
            if weakness_object.weakness
        ]

        valid_severities = self.object.severities.exclude(scoring_system=EPSS.identifier).filter(
            scoring_elements__isnull=False, scoring_system__in=SCORING_SYSTEMS.keys()
        )

        severity_vectors = []

        for severity in valid_severities:
            try:
                vector_values_system = SCORING_SYSTEMS[severity.scoring_system]
                if not vector_values_system:
                    logging.error(f"Unknown scoring system: {severity.scoring_system}")
                    continue
                vector_values = vector_values_system.get(severity.scoring_elements)
                if vector_values:
                    severity_vectors.append({"vector": vector_values, "origin": severity.url})
            except (
                CVSS2MalformedError,
                CVSS3MalformedError,
                CVSS4MalformedError,
                NotImplementedError,
            ):
                logging.error(f"CVSSMalformedError for {severity.scoring_elements}")

        epss_severity = vulnerability.severities.filter(scoring_system="epss").first()
        epss_data = None
        if epss_severity:
            epss_data = {
                "percentile": epss_severity.scoring_elements,
                "score": epss_severity.value,
                "published_at": epss_severity.published_at,
            }

        context.update(
            {
                "vulnerability": vulnerability,
                "vulnerability_search_form": VulnerabilitySearchForm(self.request.GET),
                "severities": list(vulnerability.severities.all()),
                "severity_vectors": severity_vectors,
                "references": list(vulnerability.references.all()),
                "aliases": list(vulnerability.aliases.all()),
                "weaknesses": weaknesses_present_in_db,
                "status": vulnerability.get_status_label,
                "history": vulnerability.history,
                "epss_data": epss_data,
            }
        )
        return context


class AdvisoryDetails(VulnerableCodeDetailView):
    model = models.AdvisoryV2
    template_name = "advisory_detail.html"
    slug_url_kwarg = "avid"
    slug_field = "avid"

    def get_object(self, queryset=None):
        avid = self.kwargs.get(self.slug_url_kwarg)
        obj = models.AdvisoryV2.objects.latest_for_avid(avid)

        if not obj:
            raise Http404("Advisory not found")

        return obj

    def get_queryset(self):
        return (
            super()
            .get_queryset()
            .select_related()
            .prefetch_related(
                Prefetch(
                    "references",
                    queryset=models.AdvisoryReference.objects.only(
                        "reference_id", "reference_type", "url"
                    ),
                ),
                Prefetch(
                    "aliases",
                    queryset=models.AdvisoryAlias.objects.only("alias"),
                ),
                Prefetch(
                    "weaknesses",
                    queryset=models.AdvisoryWeakness.objects.only("cwe_id"),
                ),
                Prefetch(
                    "severities",
                    queryset=models.AdvisorySeverity.objects.only(
                        "scoring_system", "value", "url", "scoring_elements", "published_at"
                    ),
                ),
                Prefetch(
                    "exploits",
                    queryset=models.AdvisoryExploit.objects.only(
                        "data_source", "description", "required_action", "due_date", "notes"
                    ),
                ),
                Prefetch(
                    "related_ssvcs",
                    queryset=models.SSVC.objects.select_related("source_advisory").only(
                        "vector",
                        "options",
                        "decision",
                        "source_advisory__id",
                        "source_advisory__url",
                    ),
                ),
                Prefetch(
                    "source_ssvcs",
                    queryset=models.SSVC.objects.select_related("source_advisory").only(
                        "vector",
                        "options",
                        "decision",
                        "source_advisory__id",
                        "source_advisory__url",
                    ),
                ),
            )
        )

    def get_context_data(self, **kwargs):
        """
        Build context with preloaded QuerySets and minimize redundant queries.
        """
        context = super().get_context_data(**kwargs)
        advisory = self.object

        # Pre-fetch and process data in Python instead of the template
        weaknesses_present_in_db = [
            weakness_object
            for weakness_object in advisory.weaknesses.all()
            if weakness_object.weakness
        ]

        valid_severities = (
            self.object.severities.exclude(scoring_system=EPSS.identifier)
            .filter(scoring_elements__isnull=False, scoring_system__in=SCORING_SYSTEMS.values())
            .exclude(scoring_elements="")
        )

        epss_severity = advisory.severities.filter(scoring_system="epss").first()
        epss_data = None
        epss_advisory = None
        if not epss_severity:
            epss_advisory = (
                advisory.related_advisory_severities.filter(
                    datasource_id=EPSSImporterPipeline.pipeline_id
                )
                .latest_per_avid()
                .first()
            )
            if epss_advisory:
                epss_severity = epss_advisory.severities.filter(scoring_system="epss").first()
        if epss_severity:
            # If the advisory itself does not have EPSS severity, but has a related advisory with EPSS severity, we use the related advisory's EPSS severity and URL as the source of EPSS data.
            epss_data = {
                "percentile": epss_severity.scoring_elements,
                "score": epss_severity.value,
                "published_at": epss_severity.published_at,
                "source": epss_advisory.url if epss_advisory else advisory.url,
                "advisory": epss_advisory if epss_advisory else advisory,
            }

        ssvc_entries = []
        seen = set()

        severity_vectors = []

        for severity in valid_severities:
            try:
                vector_values_system = SCORING_SYSTEMS.get(severity.scoring_system)
                if not vector_values_system:
                    logging.error(f"Unknown scoring system: {severity.scoring_system}")
                    continue
                if vector_values_system.identifier in ["cvssv3.1_qr"]:
                    continue
                vector_values = vector_values_system.get(severity.scoring_elements)
                if vector_values:
                    severity_vectors.append({"vector": vector_values, "origin": severity.url})
            except (
                CVSS2MalformedError,
                CVSS3MalformedError,
                CVSS4MalformedError,
                NotImplementedError,
            ):
                logging.error(f"CVSSMalformedError for {severity.scoring_elements}")

        def add_ssvc(ssvc):
            key = (ssvc.vector, ssvc.source_advisory_id)
            if key in seen:
                return
            seen.add(key)
            ssvc_entries.append(
                {
                    "vector": ssvc.vector,
                    "decision": ssvc.decision,
                    "options": ssvc.options,
                    "advisory_url": ssvc.source_advisory.url,
                    "advisory": ssvc.source_advisory,
                }
            )

        for ssvc in advisory.source_ssvcs.all():
            add_ssvc(ssvc)

        for ssvc in advisory.related_ssvcs.all():
            add_ssvc(ssvc)

        context["ssvcs"] = ssvc_entries
        context.update(
            {
                "advisory": advisory,
                "severities": list(advisory.severities.all()),
                "references": list(advisory.references.all()),
                "patches": list(advisory.patches.all()),
                "aliases": list(advisory.aliases.all()),
                "severity_vectors": severity_vectors,
                "weaknesses": weaknesses_present_in_db,
                "status": advisory.get_status_label,
                "epss_data": epss_data,
            }
        )
        return context


class HomePage(VulnerableCodeView):
    template_name = "index.html"

    def get(self, request):
        request_query = request.GET
        context = {
            "vulnerability_search_form": VulnerabilitySearchForm(request_query),
            "package_search_form": PackageSearchForm(request_query),
            "release_url": f"https://github.com/aboutcode-org/vulnerablecode/releases/tag/v{VULNERABLECODE_VERSION}",
        }
        return render(request=request, template_name=self.template_name, context=context)


class HomePageV2(VulnerableCodeView):
    template_name = "index.html"

    def get(self, request):
        request_query = request.GET
        context = {
            "vulnerability_search_form": AdvisorySearchForm(request_query),
            "package_search_form": PackageSearchForm(request_query),
            "release_url": f"https://github.com/aboutcode-org/vulnerablecode/releases/tag/v{VULNERABLECODE_VERSION}",
        }
        return render(request=request, template_name=self.template_name, context=context)


email_template = """
Dear VulnerableCode.io user:

We have received a request to send a VulnerableCode.io API key to this email address.
Here is your API key:

   Token {auth_token}

If you did NOT request this API key, you can either ignore this email or contact us at support@nexb.com and let us know in the forward that you did not request an API key.

The API root is at https://public.vulnerablecode.io/api
To learn more about using the VulnerableCode.io API, please refer to the live API documentation at https://public.vulnerablecode.io/api/docs
To learn about VulnerableCode, refer to the general documentation at https://vulnerablecode.readthedocs.io

--
Sincerely,
The nexB support Team.

VulnerableCode is a free and open database of software package vulnerabilities
and the tools to aggregate and correlate these vulnerabilities.

Chat at https://gitter.im/aboutcode-org/vulnerablecode
Docs at https://vulnerablecode.readthedocs.org/
Source code and issues at https://github.com/nexB/vulnerablecode
"""


class ApiUserCreateView(VulnerableCodeCreateView):
    model = models.ApiUser
    form_class = ApiUserCreationForm
    template_name = "api_user_creation_form.html"

    def form_valid(self, form):
        try:
            response = super().form_valid(form)
        except ValidationError:
            messages.error(self.request, "Email is invalid or already taken")
            return redirect(self.get_success_url())

        send_mail(
            subject="VulnerableCode.io API key request",
            message=email_template.format(auth_token=self.object.auth_token),
            from_email=env.str("FROM_EMAIL", default=""),
            recipient_list=[self.object.email],
            fail_silently=True,
        )

        messages.success(
            self.request, f"Your API key token has been sent to your email: {self.object.email}."
        )

        return response

    def get_success_url(self):
        return reverse_lazy("api_user_request")


class VulnerabilityPackagesDetails(VulnerableCodeDetailView):
    """
    View to display all packages affected by or fixing a specific vulnerability.
    URL: /vulnerabilities/{vulnerability_id}/packages
    """

    model = models.Vulnerability
    template_name = "vulnerability_package_details.html"
    slug_url_kwarg = "vulnerability_id"
    slug_field = "vulnerability_id"

    def get_queryset(self):
        """
        Prefetch and optimize related data to minimize database hits.
        """
        return (
            super()
            .get_queryset()
            .prefetch_related(
                Prefetch(
                    "affecting_packages",
                    queryset=models.Package.objects.only("type", "namespace", "name", "version"),
                ),
                Prefetch(
                    "fixed_by_packages",
                    queryset=models.Package.objects.only("type", "namespace", "name", "version"),
                ),
            )
        )

    def get_context_data(self, **kwargs):
        """
        Build context with preloaded QuerySets and minimize redundant queries.
        """
        context = super().get_context_data(**kwargs)
        vulnerability = self.object
        (
            sorted_fixed_by_packages,
            sorted_affected_packages,
            all_affected_fixed_by_matches,
        ) = vulnerability.aggregate_fixed_and_affected_packages()
        context.update(
            {
                "affected_packages": sorted_affected_packages,
                "fixed_by_packages": sorted_fixed_by_packages,
                "all_affected_fixed_by_matches": all_affected_fixed_by_matches,
            }
        )
        return context


class AdvisoryPackagesDetails(VulnerableCodeDetailView):
    """
    View to display all packages affected by or fixing a specific vulnerability.
    URL: /advisories/{id}/packages
    """

    model = models.AdvisoryV2
    template_name = "advisory_package_details.html"
    slug_url_kwarg = "avid"

    def get_object(self, queryset=None):
        avid = self.kwargs.get(self.slug_url_kwarg)
        if not avid:
            raise Http404("Missing advisory identifier")

        advisory = models.AdvisoryV2.objects.latest_for_avid(avid)

        if not advisory:
            raise Http404(f"No advisory found for avid: {avid}")

        return advisory

    def get_queryset(self):
        """
        Prefetch and optimize related data to minimize database hits.
        """
        return (
            super()
            .get_queryset()
            .prefetch_related(
                Prefetch(
                    "impacted_packages",
                    queryset=models.ImpactedPackage.objects.filter(
                        advisory__is_latest=True, advisory___all_impacts_unfurled_at__isnull=False
                    ).prefetch_related(
                        Prefetch(
                            "affecting_packages",
                            queryset=models.PackageV2.objects.only(
                                "type", "namespace", "name", "version"
                            ),
                        ),
                        Prefetch(
                            "fixed_by_packages",
                            queryset=models.PackageV2.objects.only(
                                "type", "namespace", "name", "version"
                            ),
                        ),
                    ),
                )
            )
        )


class AdvisoryPackageCommitPatchDetails(DetailView):
    """
    View to display all packages introduce by or fixing a specific vulnerability.
    URL: /advisories/commits/{id}
    """

    model = models.AdvisoryV2
    template_name = "advisory_package_commit_details.html"
    slug_url_kwarg = "avid"

    def get_object(self, queryset=None):
        avid = self.kwargs.get(self.slug_url_kwarg)
        if not avid:
            raise Http404("Missing advisory identifier")

        advisory = models.AdvisoryV2.objects.latest_for_avid(avid)

        if not advisory:
            raise Http404(f"No advisory found for avid: {avid}")

        return advisory

    def get_queryset(self):
        """
        Prefetch and optimize related data to minimize database hits.
        """
        return (
            super()
            .get_queryset()
            .prefetch_related(
                Prefetch(
                    "impacted_packages",
                    queryset=models.ImpactedPackage.objects.order_by("base_purl").prefetch_related(
                        Prefetch(
                            "introduced_by_package_commit_patches",
                            queryset=models.PackageCommitPatch.objects.only(
                                "id", "commit_hash", "vcs_url"
                            ),
                        ),
                        Prefetch(
                            "fixed_by_package_commit_patches",
                            queryset=models.PackageCommitPatch.objects.only(
                                "id", "commit_hash", "vcs_url"
                            ),
                        ),
                    ),
                )
            )
        )


class PipelineScheduleListView(VulnerableCodeListView, FormMixin):
    model = PipelineSchedule
    context_object_name = "schedule_list"
    template_name = "pipeline_dashboard.html"
    paginate_by = 20
    form_class = PipelineSchedulePackageForm

    def get_form_kwargs(self):
        kwargs = super().get_form_kwargs()
        kwargs["data"] = self.request.GET
        return kwargs

    def get_queryset(self):
        form = self.form_class(self.request.GET)
        if form.is_valid():
            return PipelineSchedule.objects.filter(
                pipeline_id__icontains=form.cleaned_data.get("search")
            )
        return PipelineSchedule.objects.all()

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        cache_timeout = 60 * 10
        load_per_queue = cache.get("load_per_queue")

        if load_per_queue is None:
            load_per_queue = compute_queue_load_factor()
            cache.set("load_per_queue", load_per_queue, cache_timeout)

        context["load_per_queue"] = load_per_queue
        context["active_pipeline_count"] = PipelineSchedule.objects.filter(is_active=True).count()
        context["disabled_pipeline_count"] = PipelineSchedule.objects.filter(
            is_active=False
        ).count()
        return context


class PipelineRunListView(VulnerableCodeListView):
    model = PipelineRun
    context_object_name = "run_list"
    template_name = "pipeline_run_list.html"
    paginate_by = 20
    slug_url_kwarg = "pipeline_id"
    slug_field = "pipeline_id"

    def get_queryset(self):
        pipeline = get_object_or_404(
            PipelineSchedule,
            pipeline_id=self.kwargs["pipeline_id"],
        )
        return pipeline.pipelineruns.defer("log")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        pipeline = get_object_or_404(
            PipelineSchedule,
            pipeline_id=self.kwargs["pipeline_id"],
        )
        context["pipeline_name"] = pipeline.pipeline_class.__name__
        context["pipeline_description"] = pipeline.description
        context["pipeline_run_retention_days"] = VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS
        context["minimum_pipeline_runs_retain"] = VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN
        return context


class PipelineRunDetailView(VulnerableCodeDetailView):
    model = PipelineRun
    template_name = "pipeline_run_details.html"
    context_object_name = "run"

    def get_object(self):
        pipeline_id = self.kwargs["pipeline_id"]
        run_id = self.kwargs["run_id"]
        return get_object_or_404(
            PipelineRun,
            pipeline__pipeline_id=pipeline_id,
            run_id=run_id,
        )

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        pipeline_id = self.kwargs["pipeline_id"]
        run_id = self.kwargs["run_id"]
        run = get_object_or_404(
            PipelineRun,
            pipeline__pipeline_id=pipeline_id,
            run_id=run_id,
        )
        context["pipeline_name"] = run.pipeline_class.__name__
        return context


class AdminLoginView(LoginView):
    template_name = "admin_login.html"
    authentication_form = AdminLoginForm

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["site_title"] = "VulnerableCode site admin"
        context["site_header"] = "VulnerableCode Administration"
        return context


class AdvisoryToDoListView(ListView, FormMixin):
    model = AdvisoryToDoV2
    context_object_name = "todo_list"
    template_name = "advisory_todos.html"
    paginate_by = 20
    form_class = AdvisoryToDoForm

    def get_form_kwargs(self):
        kwargs = super().get_form_kwargs()
        kwargs["data"] = self.request.GET
        return kwargs

    def get_queryset(self):
        form = self.form_class(self.request.GET)
        resolved = self.request.GET.get("resolved")
        issue_type = self.request.GET.get("issue_type")

        qs = (
            super()
            .get_queryset()
            .filter(is_todo_stale=False)
            .order_by("-advisories_count", "-oldest_advisory_date")
        )
        if resolved in ["True", "False"]:
            qs = qs.filter(is_resolved=(resolved == "True"))

        if issue_type:
            qs = qs.filter(issue_type=issue_type)

        qs.prefetch_related("advisories__aliases")

        if form.is_valid() and (avid := form.cleaned_data.get("avid")):
            return qs.filter(advisories__avid=avid).distinct()
        elif form.is_valid() and (search := form.cleaned_data.get("search")):
            return qs.filter(advisories__aliases__alias__icontains=search).distinct()

        return qs

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["issue_choices"] = ISSUE_TYPE_CHOICES

        return context


class AdvisoryPackageCurationView(DetailView):
    model = AdvisoryToDoV2
    template_name = "package_curation.html"
    slug_url_kwarg = "todo_id"
    slug_field = "todo_id"

    def get_queryset(self):
        package_curation_issue_types = [
            "CONFLICTING_FIXED_BY_PACKAGES",
            "CONFLICTING_AFFECTED_PACKAGES",
            "CONFLICTING_AFFECTED_AND_FIXED_BY_PACKAGES",
        ]
        return super().get_queryset().filter(issue_type__in=package_curation_issue_types)

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        todo = self.object

        context["advisory_summaries"] = {
            adv.avid: adv.summary for adv in todo.advisories.all() if adv.summary.strip()
        }
        context["vulnerability_id"] = todo.alias
        context["curation_items"] = json.dumps(todo.issue_detail["curation_items"])
        return context


class AltchaView(FormView):
    template_name = "altcha.html"
    form_class = AltchaForm

    def dispatch(self, request, *args, **kwargs):
        """Do not show Altcha challenge to already validated user."""
        verified_at = request.session.get("altcha_verified_at")

        if verified_at:
            if time.time() - verified_at < VULNERABLECODE_ALTCHA_SESSION_TIMEOUT:
                next_url = request.GET.get("next", "/")
                return safe_altcha_redirect(next_url)

        return super().dispatch(request, *args, **kwargs)

    def form_valid(self, form):
        self.request.session["altcha_verified_at"] = time.time()

        next_url = self.request.GET.get("next", "/")
        return safe_altcha_redirect(next_url)


class AdvisorySeverityCurationView(DetailView):
    model = AdvisoryToDoV2
    template_name = "severity_curation.html"
    slug_url_kwarg = "todo_id"
    slug_field = "todo_id"

    def get_queryset(self):
        return super().get_queryset().filter(issue_type="CONFLICTING_SEVERITY_SCORES")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        todo = self.object

        context["advisory_summaries"] = {
            adv.avid: adv.summary for adv in todo.advisories.all() if adv.summary.strip()
        }
        context["vulnerability_id"] = todo.alias
        context["curation_items"] = json.dumps(todo.issue_detail["curation_items"])
        return context


class AdvisoryWeaknessCurationView(DetailView):
    model = AdvisoryToDoV2
    template_name = "weakness_curation.html"
    slug_url_kwarg = "todo_id"
    slug_field = "todo_id"

    def get_queryset(self):
        return super().get_queryset().filter(issue_type="CONFLICTING_WEAKNESSES")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        todo = self.object

        context["advisory_summaries"] = {
            adv.avid: adv.summary for adv in todo.advisories.all() if adv.summary.strip()
        }
        context["vulnerability_id"] = todo.alias
        context["curation_items"] = json.dumps(todo.issue_detail["curation_items"])
        return context


class AdvisoryMitigationCurationView(DetailView):
    model = AdvisoryToDoV2
    template_name = "mitigation_curation.html"
    slug_url_kwarg = "todo_id"
    slug_field = "todo_id"

    def get_queryset(self):
        return super().get_queryset().filter(issue_type="MISSING_FIXED_BY_PACKAGE")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        todo = self.object

        context["advisory_summaries"] = {
            adv.avid: adv.summary for adv in todo.advisories.all() if adv.summary.strip()
        }
        context["avid"] = next(iter(context["advisory_summaries"]))
        context["curation_items"] = [
            a.base_purl
            for a in todo.advisories.first().impacted_packages.filter(
                fixed_by_packages__isnull=True
            )
        ]
        return context