Add the vulnerability report

closes: #6773

Author: git-hyagi

CHANGES/6773.feature

@@ -0,0 +1 @@
+Added the vulnerability report data model.

pulpcore/app/models/vulnerability_report.py

@@ -0,0 +1,16 @@
+from django.db import models
+
+from pulpcore.plugin.models import BaseModel
+from pulpcore.plugin.util import get_domain_pk
+
+
+class VulnerabilityReport(BaseModel):
+    """
+    Model used in vulnerability report.
+    """
+
+    vulns = models.JSONField()
+    pulp_domain = models.ForeignKey("core.Domain", default=get_domain_pk, on_delete=models.CASCADE)
+
+    class Meta:
+        default_related_name = "%(app_label)s_%(model_name)s"

pulpcore/app/serializers/vulnerability_report.py

@@ -0,0 +1,17 @@
+from rest_framework import serializers
+
+from pulpcore.app.models.vulnerability_report import VulnerabilityReport
+from pulpcore.plugin.serializers import IdentityField, ModelSerializer
+
+
+class VulnerabilityReportSerializer(ModelSerializer):
+    """
+    A serializer for the VulnerabilityReport Model.
+    """
+
+    vulns = serializers.JSONField()
+    pulp_href = IdentityField(view_name="vuln_report-detail")
+
+    class Meta:
+        model = VulnerabilityReport
+        fields = ModelSerializer.Meta.fields + ("vulns",)

pulpcore/app/tasks/vulnerability_report.py

@@ -0,0 +1,195 @@
+import aiohttp
+import json
+
+from asgiref.sync import sync_to_async
+from queue import Empty, Queue
+from threading import Thread
+from typing import Callable, Optional, Tuple, Dict, Any
+
+
+from pulpcore.plugin.util import get_domain
+from pulpcore.plugin.models import CreatedResource
+from pulpcore.constants import (
+    OSV_QUERY_URL,
+    VULNERABILITY_TASK_THREAD_TIMEOUT,
+)
+from pulpcore.app.models.vulnerability_report import VulnerabilityReport
+
+# Create a thread-safe queue to share Content units between threads
+content_queue = Queue()
+
+
+async def check_content(func: Callable, args: Optional[Tuple] = None) -> None:
+    """
+    Start the background_thread and make the API requests (scan) to osv.dev with the packages
+    from Queue.
+
+    Args:
+        func (callable | str): The function to populate content_queue Queue with the content_units.
+            Each item in the queue must follow the osv.dev request data format.
+        args (tuple): The positional arguments to pass on to the func.
+    """
+    if not args:
+        args = ()
+    background_thread = Thread(target=func, args=args)
+    await _scan_packages(background_thread)
+    background_thread.join()
+
+
+async def _scan_packages(background_thread: Thread) -> None:
+    """
+    Makes a request to the osv.dev API and store the results in VulnerabilityReport model.
+
+    Args:
+        background_thread (Thread): We need to pass the thread object used to populate the queue to
+            prevent deadlock issues.
+    """
+    scanned_packages = await _scan_packages_from_queue(
+        background_thread=background_thread,
+    )
+    await _save_vulnerability_report(scanned_packages)
+
+
+async def _scan_packages_from_queue(
+    background_thread: Thread, http_client: Optional[aiohttp.ClientSession] = None
+) -> Dict[str, Any]:
+    """
+    Scans packages from a queue by making HTTP requests to OSV API.
+
+    Args:
+        background_thread (Thread): Thread populating the queue
+        http_client (Optional[aiohttp.ClientSession]): HTTP client for making requests
+
+    Returns:
+        Dict[str, Any]: Dictionary mapping package names to vulnerability data
+    """
+
+    # Use provided client or create a new one
+    if http_client:
+        return await _process_queue_with_client(background_thread, http_client)
+
+    async with aiohttp.ClientSession() as session:
+        return await _process_queue_with_client(background_thread, session)
+
+
+async def _process_queue_with_client(
+    background_thread: Thread,
+    session: aiohttp.ClientSession,
+) -> Dict[str, Any]:
+    """
+    Process queue items using the provided HTTP client session.
+
+    Args:
+        background_thread (Thread): Thread populating the queue
+        session (aiohttp.ClientSession): HTTP client session
+
+    Returns:
+        Dict[str, Any]: Dictionary mapping package names to vulnerability data
+    """
+    scanned_packages = {}
+    try:
+        for osv_data in iter(
+            lambda: content_queue.get(timeout=VULNERABILITY_TASK_THREAD_TIMEOUT), None
+        ):
+            if isinstance(osv_data, Exception):
+                raise RuntimeError(f"Background vuln report task failed to execute: {osv_data}")
+
+            vulnerability_data = await _query_osv_api(session, osv_data)
+            package_name = _get_package_name(osv_data)
+
+            if vulnerability_data.get("vulns"):
+                scanned_packages[package_name] = vulnerability_data["vulns"]
+
+            # Handle pagination
+            if next_page_token := vulnerability_data.get("next_page_token"):
+                next_page_request = build_osv_data(
+                    osv_data["package"]["name"],
+                    osv_data["package"]["ecosystem"],
+                    osv_data.get("version"),
+                    next_page_token,
+                )
+                content_queue.put(next_page_request)
+
+    except Empty:
+        if not background_thread.is_alive():
+            raise RuntimeError("Vuln report task thread died unexpectedly.")
+        else:
+            raise RuntimeError("Background vuln report thread took too long.")
+
+    return scanned_packages
+
+
+async def _query_osv_api(
+    session: aiohttp.ClientSession, osv_data: Dict[str, Any]
+) -> Dict[str, Any]:
+    """
+    Make a single request to the OSV API.
+
+    Args:
+        session (aiohttp.ClientSession): HTTP client session
+        osv_data (Dict[str, Any]): OSV query data
+
+    Returns:
+        Dict[str, Any]: JSON response from OSV API
+    """
+    data = json.dumps(osv_data)
+    async with session.post(url=OSV_QUERY_URL, data=data) as response:
+        response_body = await response.text()
+        return json.loads(response_body)
+
+
+def _get_package_name(osv_data: Dict[str, Any]) -> str:
+    """
+    Extract package name from OSV data.
+
+    Args:
+        osv_data (Dict[str, Any]): OSV query data
+
+    Returns:
+        str: Formatted package name
+    """
+    osv_package_name = osv_data["package"]["name"]
+    osv_package_version = osv_data.get("version", "")
+    return "{package}-{version}".format(
+        package=osv_package_name,
+        version=osv_package_version,
+    )
+
+
+async def _save_vulnerability_report(scanned_packages: Dict[str, Any]) -> None:
+    """
+    Save vulnerability report to the database.
+
+    Args:
+        scanned_packages (Dict[str, Any]): Dictionary mapping package names to vulnerability data
+    """
+    vuln_report, created = await sync_to_async(VulnerabilityReport.objects.get_or_create)(
+        vulns=scanned_packages, pulp_domain=get_domain()
+    )
+    if created:
+        await CreatedResource.objects.acreate(content_object=vuln_report)
+
+
+def build_osv_data(
+    name: str, ecosystem: str, version: Optional[str] = None, next_page_token: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Helper function to build the osv.dev request data based on content object
+
+    Args:
+      name (str): Name of the package. Should match the name used in the package ecosystem.
+      ecosystem (str): The ecosystem for this package. Check osv.dev documentation for
+        the complete list of valid ecosystem names.
+      version (str): The version string to query for.
+      next_page_token (Optional[str]): If the previous query fetched a large number of results,
+        the response will be paginated.
+
+    Returns (Dict[str, Any]): The package which vulnerability will be checked. A dicitonary following the
+        expected payload format from osv.dev.
+    """
+    osv_data: Dict[str, Any] = {"package": {"name": name, "ecosystem": ecosystem}}
+    if version:
+        osv_data["version"] = version
+    if next_page_token:
+        osv_data["page_token"] = next_page_token
+    return osv_data

pulpcore/app/viewsets/vulnerability_report.py

@@ -0,0 +1,12 @@
+from rest_framework.mixins import DestroyModelMixin, ListModelMixin, RetrieveModelMixin
+
+from pulpcore.app.models.vulnerability_report import VulnerabilityReport as VulnReport
+from pulpcore.app.serializers.vulnerability_report import VulnerabilityReportSerializer
+from pulpcore.plugin.viewsets import NamedModelViewSet
+
+
+class VulnerabilityReport(NamedModelViewSet, ListModelMixin, RetrieveModelMixin, DestroyModelMixin):
+
+    endpoint_name = "vuln_report"
+    queryset = VulnReport.objects.all()
+    serializer_class = VulnerabilityReportSerializer

pulpcore/constants.py

@@ -127,3 +127,9 @@
 # The upper boundary represents an unsigned 32-bit integer and prevents overflow
 ORPHAN_PROTECTION_TIME_LOWER_BOUND = 0
 ORPHAN_PROTECTION_TIME_UPPER_BOUND = 4294967295  # (2^32)-1
+
+# OSV API URL
+OSV_QUERY_URL = "https://api.osv.dev/v1/query"
+
+# Timeout when waiting on tasks scan thread queue to avoid indefinite blocking.
+VULNERABILITY_TASK_THREAD_TIMEOUT = 60

pulpcore/tests/unit/test_vulnerability_report.py

@@ -0,0 +1,100 @@
+import pytest
+import json
+
+from queue import Queue
+from threading import Thread
+from unittest.mock import Mock, AsyncMock, patch
+
+from pulpcore.app.tasks.vulnerability_report import (
+    build_osv_data,
+    _query_osv_api,
+    _process_queue_with_client,
+)
+from pulpcore.constants import OSV_QUERY_URL
+
+
+@pytest.mark.asyncio
+async def test_process_queue_with_client():
+    """Test successful processing of queue items with vulnerabilities."""
+
+    osv_data = {"package": {"name": "django", "ecosystem": "PyPI"}, "version": "5.1"}
+    vulnerability_data = {"vulns": [{"id": "GHSA-test-1234", "summary": "Test vulnerability"}]}
+
+    # Mock background thread as alive
+    mock_thread = Mock(spec=Thread)
+    mock_thread.is_alive.return_value = True
+
+    # Mock session
+    mock_session = Mock()
+
+    # Create a test queue and patch the global content_queue
+    test_queue = Queue()
+    test_queue.put(osv_data)
+    test_queue.put(None)  # Sentinel value to notify no more items in queue
+
+    with patch("pulpcore.app.tasks.vulnerability_report.content_queue", test_queue), patch(
+        "pulpcore.app.tasks.vulnerability_report._query_osv_api", return_value=vulnerability_data
+    ) as mock_query, patch(
+        "pulpcore.app.tasks.vulnerability_report._get_package_name", return_value="django-5.1"
+    ) as mock_get_name:
+
+        result = await _process_queue_with_client(mock_thread, mock_session)
+        assert result == {"django-5.1": vulnerability_data["vulns"]}
+        mock_query.assert_called_once_with(mock_session, osv_data)
+        mock_get_name.assert_called_once_with(osv_data)
+
+
+@pytest.mark.asyncio
+async def test_query_osv_api():
+    """Test OSV API query with mocked HTTP response."""
+    # Arrange
+    osv_data = {"package": {"name": "django", "ecosystem": "PyPI"}, "version": "5.1"}
+    expected_response = {
+        "vulns": [
+            {
+                "id": "GHSA-abcd-1234",
+                "summary": "Test vulnerability",
+                "details": "This is a test vulnerability",
+            }
+        ]
+    }
+
+    # Mock the HTTP response
+    mock_response = AsyncMock()
+    mock_response.text.return_value = json.dumps(expected_response)
+
+    # Mock the session and its context manager
+    mock_session = Mock()
+    mock_session.post.return_value = AsyncMock()
+    mock_session.post.return_value.__aenter__.return_value = mock_response
+    mock_session.post.return_value.__aexit__.return_value = None
+
+    result = await _query_osv_api(mock_session, osv_data)
+
+    assert result == expected_response
+    mock_session.post.assert_called_once_with(url=OSV_QUERY_URL, data=json.dumps(osv_data))
+    mock_response.text.assert_called_once()
+
+
+def test_build_osv_data():
+    """Test if the osv_data is correctly built with the required fields."""
+    pkg_name = "test"
+    pkg_ecosystem = "npm"
+    osv_data = build_osv_data(pkg_name, pkg_ecosystem)
+    assert osv_data == {"package": {"name": pkg_name, "ecosystem": pkg_ecosystem}}
+
+    # verify if the osv_data is correctly built with the optional field version
+    version = "beta"
+    osv_data = build_osv_data(pkg_name, pkg_ecosystem, version)
+    assert osv_data == {
+        "package": {"name": pkg_name, "ecosystem": pkg_ecosystem},
+        "version": version,
+    }
+
+    # verify if the osv_data is correctly built with the optional field next_page_token
+    token = "1234"
+    osv_data = build_osv_data(pkg_name, pkg_ecosystem, next_page_token=token)
+    assert osv_data == {
+        "package": {"name": pkg_name, "ecosystem": pkg_ecosystem},
+        "page_token": token,
+    }