api.py

"""
Exploits API endpoints

NOTE: All exploit APIs are currently limited to CISA source as only those are handled by
      Product Security.
"""
from datetime import datetime

from django.db.models import BooleanField, ExpressionWrapper, Max, Q
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import OpenApiParameter, extend_schema
from rest_framework.exceptions import ParseError
from rest_framework.generics import ListAPIView
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView

from osidb.api_views import RudimentaryUserPathLoggingMixin
from osidb.models import Flaw, PsModule

from .constants import REPORT_EXPLOIT_SOURCES
from .helpers import ExploitResult
from .models import EPSS, Exploit
from .serializers import (
    AffectExploitReportSerializer,
    EPSSSerializer,
    ExploitOnlyReportDataSerializer,
    ExploitsSimpleSerializer,
    FlawReportDataSerializer,
    SupportedProductsSerializer,
)


class ExploitsCollect(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for re-collecting exploit data.

    **NOTE:** Currently for CISA data only, which is very small and collection is fast.
    """

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "result_cisa": {"type": "string"},
                },
            }
        }
    )
    def put(self, request):
        # Import in runtime to avoid problems with migrations-check and schema-check
        from collectors.exploits_cisa.tasks import cisa_collector_main

        try:
            cisa_collector_main()
            result_cisa = "Success"
        except Exception:
            result_cisa = "Failed"

        # NOTE: Exploit data collector function calls elevate ACL privileges, do not return any data
        #       from this endpoint.
        return Response({"result_cisa": result_cisa})


class ExploitsStatus(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for getting basic information about exploits in the database.

    **NOTE:** Everyone is allowed to see basic information.
    """

    permission_classes = [AllowAny]

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "exploits_count": {"type": "integer"},
                    "exploits_count_relevant": {"type": "integer"},
                    "last_exploit": {"type": "integer"},
                },
            }
        }
    )
    def get(self, request):
        count = Exploit.objects.filter(source="CISA").count()
        count_relevant = (
            Exploit.objects.filter(source="CISA")
            .filter(flaw__isnull=False)
            .distinct()
            .count()
        )
        max_date = str(
            Exploit.objects.filter(source="CISA").aggregate(Max("date"))["date__max"]
        )

        return Response(
            {
                "exploits_count": count,
                "exploits_count_relevant": count_relevant,
                "last_exploit": max_date,
            }
        )


class ExploitsCVEMap(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for getting simple exploits information mapped to impacted CVEs.

    The Insights Vulnerability application needs this format.

    Format of results:
    ```
    {
      "page_size": <Number of CVEs on the page>,
      "cves": {
        "CVE-2222-0001": [<List of exploits>],
        "CVE-2222-0002": [<List of exploits>],
        ...
      },
    }
    ```
    """

    def __init__(self):
        self.paginator = LimitOffsetPagination()
        super().__init__()

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "page_size": {"type": "integer"},
                    "cves": {"type": "object"},
                },
            }
        }
    )
    def get(self, request):
        flaws_with_exploit = (
            Flaw.objects.filter(exploit__source="CISA").order_by("cve_id").distinct()
        )
        flaws_page = self.paginator.paginate_queryset(
            flaws_with_exploit, request, view=self
        )

        cves = {}
        for flaw in flaws_page:
            exploits = flaw.exploit_set.filter(source="CISA")
            cves[flaw.cve_id] = ExploitsSimpleSerializer(exploits, many=True).data

        return self.paginator.get_paginated_response(
            {"page_size": len(cves), "cves": cves}
        )


class ExploitsReportDate(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for getting date based report for Incident Response.

    Format of results:
    ```
    {
      "cutoff_date": <Date>,
      "evaluated_cves": <Number of new CVEs with exploits>,
      "action_required": [<List of affects requiring action>],
      "no_action": [<List of CVEs not requiring action with reason>],
      "not_relevant": [<List of CVEs which are not in the database with reason>],
    }
    ```

    **NOTE:** No pagination is performed on this endpoint as data is limited by date and is expected
              to be fairly small. Also, because data is broken into three categories it is not
              exactly obvious how to create pages.
    """

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "cutoff_date": {"type": "string"},
                    "evaluated_cves": {"type": "integer"},
                    "action_required": {"type": "array", "items": {"type": "object"}},
                    "no_action": {"type": "array", "items": {"type": "object"}},
                    "not_relevant": {"type": "array", "items": {"type": "object"}},
                },
            }
        },
        parameters=[
            OpenApiParameter(
                "date",
                OpenApiTypes.DATE,
                OpenApiParameter.PATH,
                description="Date format: YYYY-MM-DD",
            ),
        ],
    )
    def get(self, request, date=None):
        try:
            parsed_date = datetime.strptime(date, "%Y-%m-%d").date()
        except ValueError:
            raise ParseError("Invalid date.")

        new_exploits = (
            Exploit.objects.filter(source="CISA")
            .filter(date__gte=parsed_date)
            .order_by("cve")
        )
        new_exploit_results = [ExploitResult(e) for e in new_exploits]

        result = {"cutoff_date": date, "evaluated_cves": new_exploits.count()}

        # Action required section
        new_exploit_results_action = [e for e in new_exploit_results if e.included]
        actions = []
        for exploit_result in new_exploit_results_action:
            action_affects = exploit_result.included_affects
            actions.extend(
                AffectExploitReportSerializer(action_affects, many=True).data
            )
        result["action_required"] = actions

        # No action section
        new_exploit_results_no_action = [
            e for e in new_exploit_results if e.in_database and not e.included
        ]
        no_actions = []
        for exploit_result in new_exploit_results_no_action:
            no_actions.append(
                {"cve": exploit_result.cve, "reason": exploit_result.message}
            )
        result["no_action"] = no_actions

        # Not relevant section
        new_exploit_results_not_relevant = [
            e for e in new_exploit_results if not e.in_database
        ]
        not_relevants = []
        for exploit_result in new_exploit_results_not_relevant:
            not_relevants.append(
                {"cve": exploit_result.cve, "reason": exploit_result.message}
            )
        result["not_relevant"] = not_relevants

        return Response(result)


class ExploitsReportPending(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for getting a report of pending actions for Incident Response.

    Format of results:
    ```
    {
      "pending_actions": [<List of affects requiring action>],
      "pending_actions_count": <Number of affects requiring action>,
    }
    ```

    **NOTE:** No pagination is performed on this endpoint as it is expected that the size of
              the list of pending actions will be mostly stable. Also, the paging cannot be done on
              the query level, as additional analysis of every exploit is required before a decision
              to include it in this report is done.
    """

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "pending_actions": {"type": "array", "items": {"type": "object"}},
                    "pending_actions_count": {"type": "integer"},
                },
            }
        }
    )
    def get(self, request):
        exploits = (
            Exploit.objects.filter(source="CISA")
            .filter(flaw__isnull=False)
            .order_by("cve")
        )
        exploit_results = [ExploitResult(e) for e in exploits]
        exploit_results_action = [e for e in exploit_results if e.included]

        actions = []
        for exploit_result in exploit_results_action:
            action_affects = exploit_result.included_affects
            actions.extend(
                AffectExploitReportSerializer(action_affects, many=True).data
            )

        return Response(
            {"pending_actions": actions, "pending_actions_count": len(actions)}
        )


class ExploitsReportExplanations(RudimentaryUserPathLoggingMixin, APIView):
    """
    API endpoint for getting a report of all CVEs with exploit and their status
    for Incident Response.

    Format of results:
    ```
    {
       "page_size": <Number of CVEs on the page>,
       "explanations": [<List of CVEs with exploit together with current status explanation>],
    }
    ```
    """

    def __init__(self):
        self.paginator = LimitOffsetPagination()
        super().__init__()

    @extend_schema(
        responses={
            200: {
                "type": "object",
                "properties": {
                    "page_size": {"type": "integer"},
                    "explanations": {"type": "array", "items": {"type": "object"}},
                },
            }
        }
    )
    def get(self, request):
        exploits = (
            Exploit.objects.filter(source="CISA")
            .filter(flaw__isnull=False)
            .order_by("cve")
        )
        exploit_page = self.paginator.paginate_queryset(exploits, request, view=self)
        exploit_results_page = [ExploitResult(e) for e in exploit_page]

        explanations = []
        for exploit_result in exploit_results_page:
            explanations.append(
                {"cve": exploit_result.cve, "explanation": exploit_result.message}
            )

        return self.paginator.get_paginated_response(
            {"page_size": len(explanations), "explanations": explanations}
        )


class ExploitsReportData(RudimentaryUserPathLoggingMixin, ListAPIView):
    """Export only the data required to generate the exploits report"""

    serializer_class = ExploitOnlyReportDataSerializer
    queryset = (
        Exploit.objects.filter(source__in=REPORT_EXPLOIT_SOURCES)
        .order_by("cve")
        .annotate(
            flaw_exists=ExpressionWrapper(
                Q(flaw__isnull=False), output_field=BooleanField()
            )
        )
    )


class ExploitsFlawData(RudimentaryUserPathLoggingMixin, ListAPIView):
    """Flaw, affect, and tracker data for Exploits"""

    serializer_class = FlawReportDataSerializer
    queryset = (
        Flaw.objects.filter(exploit__source__in=REPORT_EXPLOIT_SOURCES)
        .prefetch_related("affects__trackers")
        .distinct()
        .order_by("cve_id")
    )


class EPSSRelevant(RudimentaryUserPathLoggingMixin, ListAPIView):
    """
    API endpoint for getting list of Red Hat relevant CVEs with their EPSS score.
    """

    serializer_class = EPSSSerializer
    queryset = EPSS.objects.exclude(flaw=None)


class SupportedProducts(RudimentaryUserPathLoggingMixin, ListAPIView):
    """
    API endpoint for getting a list of all supported products.
    """

    serializer_class = SupportedProductsSerializer
    queryset = PsModule.objects.filter(active_ps_update_streams__isnull=False).distinct(
        "name"
    )