Enrique Sánchez has proposed merging
~enriqueesanchz/launchpad:add-vulnerabilityjob-model into launchpad:master.
Commit message:
Add `ImportVulnerabilityJob`
It is the `VulnerabilityJob` that imports data from cve trackers. Only SOSS is
supported now.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~enriqueesanchz/launchpad/+git/launchpad/+merge/491368
--
Your team Launchpad code reviewers is requested to review the proposed merge of
~enriqueesanchz/launchpad:add-vulnerabilityjob-model into launchpad:master.
diff --git a/lib/lp/bugs/configure.zcml b/lib/lp/bugs/configure.zcml
index b0bec51..687a347 100644
--- a/lib/lp/bugs/configure.zcml
+++ b/lib/lp/bugs/configure.zcml
@@ -1121,6 +1121,22 @@
interface="lp.bugs.interfaces.apportjob.IProcessApportBlobJobSource"/>
</lp:securedutility>
+ <!-- VulnerabilityJob -->
+ <class class="lp.bugs.model.vulnerabilityjob.VulnerabilityJob">
+ <allow interface="lp.bugs.interfaces.vulnerabilityjob.IVulnerabilityJob" />
+ </class>
+
+ <!-- ImportVulnerabilityJobSource -->
+ <lp:securedutility
+ component="lp.bugs.model.importvulnerabilityjob.ImportVulnerabilityJob"
+ provides="lp.bugs.interfaces.vulnerabilityjob.IImportVulnerabilityJobSource">
+ <allow interface="lp.bugs.interfaces.vulnerabilityjob.IImportVulnerabilityJobSource"/>
+ </lp:securedutility>
+ <class class="lp.bugs.model.importvulnerabilityjob.ImportVulnerabilityJob">
+ <allow interface="lp.bugs.interfaces.vulnerabilityjob.IImportVulnerabilityJob" />
+ <allow interface="lp.bugs.interfaces.vulnerabilityjob.IVulnerabilityJob" />
+ </class>
+
<!-- FileBugData -->
<class class="lp.bugs.model.bug.FileBugData">
<allow interface="lp.bugs.interfaces.bug.IFileBugData" />
diff --git a/lib/lp/bugs/enums.py b/lib/lp/bugs/enums.py
index 109ec8f..be1a968 100644
--- a/lib/lp/bugs/enums.py
+++ b/lib/lp/bugs/enums.py
@@ -9,6 +9,7 @@ __all__ = [
"BugNotificationLevel",
"BugNotificationStatus",
"VulnerabilityStatus",
+ "VulnerabilityHandlerEnum",
]
from lazr.enum import DBEnumeratedType, DBItem, use_template
@@ -168,3 +169,15 @@ class VulnerabilityStatus(DBEnumeratedType):
This vulnerability is now retired.
""",
)
+
+
+class VulnerabilityHandlerEnum(DBEnumeratedType):
+ SOSS = DBItem(
+ 1,
+ """
+ SOSS Handler
+
+ Specific handler to use for SOSS vulnerability data imports and
+ exports.
+ """,
+ )
diff --git a/lib/lp/bugs/interfaces/vulnerabilityjob.py b/lib/lp/bugs/interfaces/vulnerabilityjob.py
new file mode 100644
index 0000000..f057825
--- /dev/null
+++ b/lib/lp/bugs/interfaces/vulnerabilityjob.py
@@ -0,0 +1,152 @@
+# Copyright 2025 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__all__ = [
+ "VulnerabilityJobType",
+ "IVulnerabilityJob",
+ "VulnerabilityJobInProgress",
+ "VulnerabilityJobException",
+ "IImportVulnerabilityJobSource",
+ "IImportVulnerabilityJob",
+ "IExportVulnerabilityJobSource",
+ "IExportVulnerabilityJob",
+]
+
+from lazr.enum import DBEnumeratedType, DBItem
+from zope.interface import Attribute, Interface
+from zope.schema import Choice, Int, Object, Text
+
+from lp import _
+from lp.bugs.enums import VulnerabilityHandlerEnum
+from lp.services.job.interfaces.job import IJob, IJobSource, IRunnableJob
+
+
+class VulnerabilityJobType(DBEnumeratedType):
+ IMPORT_DATA = DBItem(
+ 1,
+ """
+ Import Vulnerability data
+
+ This job imports Vulnerability data from the format given by the
+ handler.
+ """,
+ )
+
+ EXPORT_DATA = DBItem(
+ 2,
+ """
+ Export Vulnerability data
+
+ This job exports Vulnerability data to the format given by the hanlder.
+ """,
+ )
+
+
+class IVulnerabilityJob(Interface):
+ """A Job that acts on vulnerabilities."""
+
+ id = Int(
+ title=_("DB ID"),
+ required=True,
+ readonly=True,
+ description=_("The tracking number for this job."),
+ )
+
+ handler = Choice(
+ title=_("The handler for this job."),
+ vocabulary=VulnerabilityHandlerEnum,
+ required=True,
+ readonly=True,
+ )
+
+ job_type = Choice(
+ title=_("Job type"),
+ vocabulary=VulnerabilityJobType,
+ required=True,
+ readonly=True,
+ )
+
+ job = Object(
+ title=_("The common Job attributes"), schema=IJob, required=True
+ )
+
+ metadata = Attribute("A dict of data about the job.")
+
+ def destroySelf():
+ """Destroy this object."""
+
+
+class VulnerabilityJobInProgress(Exception):
+ """The VulnerabilityJob for the handler is already in progress."""
+
+ def __init__(self, job):
+ super().__init__()
+ self.job = job
+
+
+class VulnerabilityJobException(Exception):
+ """There was an error during VulnerabilityJob creation."""
+
+
+class IImportVulnerabilityJobSource(IJobSource):
+ """An interface for acquiring IImportVulnerabilityJobs."""
+
+ def create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ ):
+ """Create a new import job for a handler."""
+
+ def get(handler):
+ """Retrieve the import job for a handler, if any.
+
+ :return: `None` or an `IImportVulnerabilityJobSource`.
+ """
+
+
+class IImportVulnerabilityJob(IRunnableJob):
+ """A Job that imports SVT data."""
+
+ error_description = Text(
+ title=_("Error description"),
+ description=_(
+ "A short description of the last error this "
+ "job encountered, if any."
+ ),
+ readonly=True,
+ required=False,
+ )
+
+
+class IExportVulnerabilityJobSource(IJobSource):
+ """An interface for acquiring IExportVulnerabilityJob."""
+
+ def create(
+ handler,
+ sources,
+ ):
+ """Create a new export job for sources using a handler."""
+
+ def get(handler):
+ """Retrieve the export job for a handler, if any.
+
+ :return: `None` or an `IExportVulnerabilityJobSource`.
+ """
+
+
+class IExportVulnerabilityJob(IRunnableJob):
+ """A Job that exports SVT data."""
+
+ error_description = Text(
+ title=_("Error description"),
+ description=_(
+ "A short description of the last error this "
+ "job encountered, if any."
+ ),
+ readonly=True,
+ required=False,
+ )
diff --git a/lib/lp/bugs/model/importvulnerabilityjob.py b/lib/lp/bugs/model/importvulnerabilityjob.py
new file mode 100644
index 0000000..3753a7f
--- /dev/null
+++ b/lib/lp/bugs/model/importvulnerabilityjob.py
@@ -0,0 +1,280 @@
+# Copyright 2025 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__all__ = [
+ "ImportVulnerabilityJob",
+]
+
+import logging
+import re
+
+from zope.component import getUtility
+from zope.interface import implementer, provider
+
+from lp.app.enums import InformationType
+from lp.app.interfaces.launchpad import ILaunchpadCelebrities
+from lp.bugs.enums import VulnerabilityHandlerEnum
+from lp.bugs.interfaces.vulnerabilityjob import (
+ IImportVulnerabilityJob,
+ IImportVulnerabilityJobSource,
+ VulnerabilityJobException,
+ VulnerabilityJobInProgress,
+ VulnerabilityJobType,
+)
+from lp.bugs.model.vulnerabilityjob import (
+ VulnerabilityJob,
+ VulnerabilityJobDerived,
+)
+from lp.bugs.scripts.soss.models import SOSSRecord
+from lp.bugs.scripts.soss.sossimport import SOSSImporter
+from lp.code.interfaces.githosting import IGitHostingClient
+from lp.code.interfaces.gitlookup import IGitLookup
+from lp.services.config import config
+from lp.services.database.interfaces import IPrimaryStore, IStore
+from lp.services.job.interfaces.job import JobStatus
+from lp.services.job.model.job import Job
+from lp.testing import person_logged_in
+
+CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d+$")
+logger = logging.getLogger(__name__)
+
+
+@implementer(IImportVulnerabilityJob)
+@provider(IImportVulnerabilityJobSource)
+class ImportVulnerabilityJob(VulnerabilityJobDerived):
+ class_job_type = VulnerabilityJobType.IMPORT_DATA
+
+ user_error_types = (VulnerabilityJobException,)
+
+ config = config.IImportVulnerabilityJobSource
+
+ @property
+ def git_repository(self):
+ return self.metadata.get("git_repository")
+
+ @property
+ def git_ref(self):
+ return self.metadata.get("git_ref")
+
+ @property
+ def git_paths(self):
+ return self.metadata.get("git_paths")
+
+ @property
+ def information_type(self):
+ return self.metadata.get("information_type")
+
+ @property
+ def import_since_commit_sha1(self):
+ return self.metadata.get("import_since_commit_sha1")
+
+ @property
+ def error_description(self):
+ return self.metadata.get("error_description")
+
+ @classmethod
+ def create(
+ cls,
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1=False,
+ ):
+ """Create a new `ImportVulnerabilityJob`.
+
+ :param handler: What handler to use for importing the data. Can be one
+ of a group of predefined classes (SOSS, UCT, ...).
+ :param git_repository: Git repository to import from.
+ :param git_ref: Git branch/tag to get data from.
+ :param git_paths: List of relative directories within the repository to
+ get data from.
+ :param information_type: Whether imported data (bugs) should be private
+ or public. Can be one of a group of predefined options (PUBLIC,
+ PRIVATE_SECURITY...). If the source git repository is private, then
+ the information_type needs to be private also.
+ :param import_since_commit_sha1: Import data from files that were
+ altered since the given commit_sha1
+ """
+ store = IPrimaryStore(VulnerabilityJob)
+
+ vulnerability_job = store.find(
+ VulnerabilityJob,
+ VulnerabilityJob.job_id == Job.id,
+ VulnerabilityJob.job_type == cls.class_job_type,
+ VulnerabilityJob.handler == handler,
+ ).one()
+
+ if vulnerability_job is not None and (
+ vulnerability_job.job.status == JobStatus.WAITING
+ or vulnerability_job.job.status == JobStatus.RUNNING
+ ):
+ raise VulnerabilityJobInProgress(cls(vulnerability_job))
+
+ # Schedule the initialization.
+ metadata = {
+ "git_repository": git_repository,
+ "git_ref": git_ref,
+ "git_paths": git_paths,
+ "information_type": information_type,
+ "import_since_commit_sha1": import_since_commit_sha1,
+ }
+
+ vulnerability_job = VulnerabilityJob(
+ handler, cls.class_job_type, metadata
+ )
+ store.add(vulnerability_job)
+ derived_job = cls(vulnerability_job)
+ derived_job.celeryRunOnCommit()
+ IStore(VulnerabilityJob).flush()
+ return derived_job
+
+ @classmethod
+ def get(cls, handler):
+ """See `IImportVulnerabilityJob`."""
+ vulnerability_job = (
+ IStore(VulnerabilityJob)
+ .find(
+ VulnerabilityJob,
+ VulnerabilityJob.job_id == Job.id,
+ VulnerabilityJob.job_type == cls.class_job_type,
+ VulnerabilityJob.handler == handler,
+ )
+ .one()
+ )
+ return None if vulnerability_job is None else cls(vulnerability_job)
+
+ def __repr__(self):
+ """Returns an informative representation of the job."""
+ parts = "%s for" % self.__class__.__name__
+ parts += " handler: %s" % self.handler
+ parts += ", metadata: %s" % self.metadata
+ return "<%s>" % parts
+
+ def _get_parser_importer(
+ self,
+ handler: VulnerabilityHandlerEnum,
+ information_type: InformationType,
+ ):
+ """Decide which parser and importer to use
+
+ :return: a tuple of (parser, importer) where parser is the function
+ that gets a blob and returns a record and importer is the function that
+ gets a record and imports it.
+ """
+
+ if handler == VulnerabilityHandlerEnum.SOSS:
+ parser = SOSSRecord.from_yaml
+ importer = SOSSImporter(
+ information_type=information_type
+ ).import_cve
+ else:
+ exception = VulnerabilityJobException("Handler not found")
+ self.notifyUserError(exception)
+ raise exception
+
+ return parser, importer
+
+ def run(self):
+ """See `IRunnableJob`."""
+ self.metadata["result"] = {"succeeded": [], "failed": []}
+ admin = getUtility(ILaunchpadCelebrities).admin
+
+ # InformationType is passed as a value as DBItem is not serializable
+ information_type = InformationType.items[self.information_type]
+ parser, importer = self._get_parser_importer(
+ self.context.handler, information_type
+ )
+
+ # Get git repository
+ git_lookup = getUtility(IGitLookup)
+ repository = git_lookup.getByUrl(self.git_repository)
+ if not repository:
+ exception = VulnerabilityJobException("Git repository not found")
+ self.notifyUserError(exception)
+ raise exception
+
+ # Get git reference
+ ref = repository.getRefByPath(self.git_ref)
+ if not ref:
+ exception = VulnerabilityJobException("Git ref not found")
+ self.notifyUserError(exception)
+ raise exception
+
+ # turnip API call to get added/modified files
+ stats = getUtility(IGitHostingClient).getDiffStats(
+ path=self.git_repository,
+ old=self.import_since_commit_sha1,
+ new=ref.commit_sha1,
+ logger=logger,
+ )
+
+ files = [*stats.get("added", ()), *stats.get("modified", ())]
+ for file in files:
+ # Check if files that changed are in the desired path
+ found_path = False
+ for path in self.git_paths:
+ if file.startswith(path):
+ found_path = True
+ break
+
+ if not found_path:
+ logger.debug(
+ f"[ImportVulnerabilityJob] {file} is not in git_paths"
+ )
+ continue
+
+ cve_sequence = file.rsplit("/", maxsplit=1)[-1]
+ if not CVE_PATTERN.match(cve_sequence):
+ logger.debug(
+ f"[ImportVulnerabilityJob] {cve_sequence} is not a CVE "
+ "sequence"
+ )
+ continue
+
+ try:
+ logger.debug(f"[ImportVulnerabilityJob] Getting {file}")
+ blob = ref.getBlob(file)
+
+ logger.debug(
+ f"[ImportVulnerabilityJob] Parsing {cve_sequence}"
+ )
+ record = parser(blob)
+
+ # Logged as admin
+ with person_logged_in(admin):
+ bug, vulnerability = importer(record, cve_sequence)
+
+ if bug and vulnerability:
+ self.metadata["result"]["succeeded"].append(cve_sequence)
+ else:
+ self.metadata["result"]["failed"].append(cve_sequence)
+ except Exception as e:
+ self.notifyUserError(e)
+
+ def notifyUserError(self, error):
+ """Calls up and also saves the error text in this job's metadata.
+
+ See `BaseRunnableJob`.
+ """
+ # This method is called when error is an instance of
+ # self.user_error_types.
+ super().notifyUserError(error)
+ error_description = self.metadata.get("error_description", [])
+ error_description.append(str(error))
+ self.metadata = dict(
+ self.metadata, error_description=error_description
+ )
+
+ def getOopsVars(self):
+ """See `IRunnableJob`."""
+ vars = super().getOopsVars()
+ vars.extend(
+ [
+ ("vulnerabilityjob_job_id", self.context.id),
+ ("vulnerability_job_type", self.context.job_type.title),
+ ("handler", self.context.handler),
+ ]
+ )
+ return vars
diff --git a/lib/lp/bugs/model/vulnerabilityjob.py b/lib/lp/bugs/model/vulnerabilityjob.py
new file mode 100644
index 0000000..80a0753
--- /dev/null
+++ b/lib/lp/bugs/model/vulnerabilityjob.py
@@ -0,0 +1,102 @@
+# Copyright 2025 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__all__ = [
+ "VulnerabilityJob",
+ "VulnerabilityJobDerived",
+]
+
+from lazr.delegates import delegate_to
+from storm.databases.postgres import JSON
+from storm.locals import And, Int, Reference
+from zope.interface import implementer
+
+from lp.app.errors import NotFoundError
+from lp.bugs.enums import VulnerabilityHandlerEnum
+from lp.bugs.interfaces.vulnerabilityjob import (
+ IVulnerabilityJob,
+ VulnerabilityJobType,
+)
+from lp.services.database.enumcol import DBEnum
+from lp.services.database.interfaces import IStore
+from lp.services.database.stormbase import StormBase
+from lp.services.job.model.job import EnumeratedSubclass, Job
+from lp.services.job.runner import BaseRunnableJob
+
+
+@implementer(IVulnerabilityJob)
+class VulnerabilityJob(StormBase):
+ """Base class for jobs related to Vulnerabilities."""
+
+ __storm_table__ = "VulnerabilityJob"
+
+ id = Int(primary=True)
+
+ handler = DBEnum(enum=VulnerabilityHandlerEnum, allow_none=False)
+
+ job_type = DBEnum(enum=VulnerabilityJobType, allow_none=False)
+
+ job_id = Int(name="job")
+ job = Reference(job_id, Job.id)
+
+ metadata = JSON("json_data", allow_none=False)
+
+ def __init__(self, handler, job_type, metadata):
+ super().__init__()
+ self.job = Job()
+ self.handler = handler
+ self.job_type = job_type
+ self.metadata = metadata
+
+ def makeDerived(self):
+ return VulnerabilityJobDerived.makeSubclass(self)
+
+
+@delegate_to(IVulnerabilityJob)
+class VulnerabilityJobDerived(BaseRunnableJob, metaclass=EnumeratedSubclass):
+ """Abstract class for deriving from VulnerabilityJob."""
+
+ def __init__(self, job):
+ self.context = job
+
+ @classmethod
+ def get(cls, job_id):
+ """Get a job by id.
+
+ :return: the VulnerabilityJob with the specified id, as
+ the current VulnerabilityJobDerived subclass.
+ :raises: NotFoundError if there is no job with the specified id,
+ or its job_type does not match the desired subclass.
+ """
+ job = VulnerabilityJob.get(job_id)
+ if job.job_type != cls.class_job_type:
+ raise NotFoundError(
+ "No object found with id %d and type %s"
+ % (job_id, cls.class_job_type.title)
+ )
+ return cls(job)
+
+ @classmethod
+ def iterReady(cls):
+ """Iterate through all ready VulnerabilityJob."""
+ jobs = IStore(VulnerabilityJob).find(
+ VulnerabilityJob,
+ And(
+ VulnerabilityJob.job_type == cls.class_job_type,
+ VulnerabilityJob.job == Job.id,
+ Job.id.is_in(Job.ready_jobs),
+ ),
+ )
+ return (cls(job) for job in jobs)
+
+ def getOopsVars(self):
+ """See `IRunnableJob`."""
+ vars = super().getOopsVars()
+ vars.extend(
+ [
+ ("vulnerabilityjob_job_id", self.context.id),
+ ("vulnerability_job_type", self.context.job_type.title),
+ ("handler", self.context.handler),
+ ]
+ )
+ return vars
diff --git a/lib/lp/bugs/scripts/soss/sossimport.py b/lib/lp/bugs/scripts/soss/sossimport.py
index 1c9ad0a..c96ec23 100644
--- a/lib/lp/bugs/scripts/soss/sossimport.py
+++ b/lib/lp/bugs/scripts/soss/sossimport.py
@@ -85,13 +85,17 @@ class SOSSImporter:
self.information_type = information_type
self.dry_run = dry_run
self.bug_importer = getUtility(ILaunchpadCelebrities).bug_importer
- self.soss = getUtility(IDistributionSet).getByName(DISTRIBUTION_NAME)
self.person_set = getUtility(IPersonSet)
self.source_package_name_set = getUtility(ISourcePackageNameSet)
self.bugtask_set = getUtility(IBugTaskSet)
self.vulnerability_set = getUtility(IVulnerabilitySet)
self.bug_set = getUtility(IBugSet)
self.cve_set = getUtility(ICveSet)
+ self.soss = getUtility(IDistributionSet).getByName(DISTRIBUTION_NAME)
+
+ if self.soss is None:
+ logger.error("[SOSSImporter] SOSS distribution not found")
+ raise Exception("SOSS distribution not found")
def import_cve_from_file(
self, cve_path: str
diff --git a/lib/lp/bugs/tests/test_importvulnerabilityjob.py b/lib/lp/bugs/tests/test_importvulnerabilityjob.py
new file mode 100644
index 0000000..80aec80
--- /dev/null
+++ b/lib/lp/bugs/tests/test_importvulnerabilityjob.py
@@ -0,0 +1,595 @@
+# Copyright 2025 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from pathlib import Path
+
+import transaction
+from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
+
+from lp.app.enums import InformationType
+from lp.app.interfaces.launchpad import ILaunchpadCelebrities
+from lp.bugs.enums import VulnerabilityHandlerEnum
+from lp.bugs.interfaces.vulnerabilityjob import (
+ IImportVulnerabilityJob,
+ IImportVulnerabilityJobSource,
+ VulnerabilityJobException,
+ VulnerabilityJobInProgress,
+)
+from lp.bugs.model.importvulnerabilityjob import ImportVulnerabilityJob
+from lp.code.tests.helpers import GitHostingFixture
+from lp.services.features.testing import FeatureFixture
+from lp.services.job.interfaces.job import JobStatus
+from lp.services.job.tests import block_on_job
+from lp.testing import TestCaseWithFactory, person_logged_in
+from lp.testing.layers import CeleryJobLayer, DatabaseFunctionalLayer
+
+
+class ImportVulnerabilityJobTests(TestCaseWithFactory):
+ """Test case for ImportVulnerabilityJob."""
+
+ layer = DatabaseFunctionalLayer
+
+ def setUp(self):
+ super().setUp()
+ self.repository = self.factory.makeGitRepository()
+ self.refs = self.factory.makeGitRefs(
+ repository=self.repository,
+ paths=("ref/heads/main", "ref/tags/v1.0"),
+ )
+ self.cve_path = (
+ Path(__file__).parent
+ / ".."
+ / ".."
+ / "bugs"
+ / "scripts"
+ / "soss"
+ / "tests"
+ / "sampledata"
+ / "CVE-2025-1979"
+ )
+
+ @property
+ def job_source(self):
+ return getUtility(IImportVulnerabilityJobSource)
+
+ def test_getOopsVars(self):
+ """Test getOopsVars method."""
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ vars = job.getOopsVars()
+ naked_job = removeSecurityProxy(job)
+ self.assertIn(("vulnerabilityjob_job_id", naked_job.id), vars)
+ self.assertIn(
+ ("vulnerability_job_type", naked_job.job_type.title), vars
+ )
+ self.assertIn(("handler", naked_job.handler), vars)
+
+ def _getJobs(self):
+ """Return the pending IImportVulnerabilityJob as a list."""
+ return list(IImportVulnerabilityJob.iterReady())
+
+ def _getJobCount(self):
+ """Return the number of IImportVulnerabilityJob in the
+ queue."""
+ return len(self._getJobs())
+
+ def test___repr__(self):
+ """Test __repr__ method."""
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+ metadata = {
+ "git_repository": git_repository,
+ "git_ref": git_ref,
+ "git_paths": git_paths,
+ "information_type": information_type,
+ "import_since_commit_sha1": import_since_commit_sha1,
+ }
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+
+ expected = (
+ "<ImportVulnerabilityJob for "
+ f"handler: {handler}, "
+ f"metadata: {metadata}"
+ ">"
+ )
+ self.assertEqual(expected, repr(job))
+
+ def test_create_with_existing_in_progress_job(self):
+ """If there's already a waiting/running ImportVulnerabilityJob for the
+ handler ImportVulnerabilityJob.create() raises an exception.
+ """
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ # Job waiting status
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ waiting_exception = self.assertRaises(
+ VulnerabilityJobInProgress,
+ self.job_source.create,
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ self.assertEqual(job, waiting_exception.job)
+
+ # Job status from WAITING to RUNNING
+ job.start()
+ running_exception = self.assertRaises(
+ VulnerabilityJobInProgress,
+ self.job_source.create,
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ self.assertEqual(job, running_exception.job)
+
+ def test_create_with_existing_completed_job(self):
+ """If there's already a completed ImportVulnerabilityJob for the
+ handler the job can be runned again.
+ """
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ job.start()
+ job.complete()
+ self.assertEqual(job.status, JobStatus.COMPLETED)
+
+ job_duplicated = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ job_duplicated.start()
+ job_duplicated.complete()
+ self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
+
+ def test_create_with_existing_failed_job(self):
+ """If there's a failed ImportVulnerabilityJob for the handler the job
+ can be runned again.
+ """
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ job.start()
+ job.fail()
+ self.assertEqual(job.status, JobStatus.FAILED)
+
+ job_duplicated = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ job_duplicated.start()
+ job_duplicated.complete()
+ self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
+
+ def test_arguments(self):
+ """Test that ImportVulnerabilityJob specified with arguments can
+ be gotten out again."""
+ git_repository = self.factory.makeGitRepository()
+ self.useFixture(GitHostingFixture(blob=b"Some text"))
+
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+ metadata = {
+ "git_repository": git_repository,
+ "git_ref": git_ref,
+ "git_paths": git_paths,
+ "information_type": information_type,
+ "import_since_commit_sha1": import_since_commit_sha1,
+ }
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+
+ naked_job = removeSecurityProxy(job)
+ self.assertEqual(naked_job.handler, handler)
+ self.assertEqual(naked_job.git_repository, git_repository)
+ self.assertEqual(naked_job.git_ref, git_ref)
+ self.assertEqual(naked_job.git_paths, git_paths)
+ self.assertEqual(naked_job.information_type, information_type)
+ self.assertEqual(
+ naked_job.import_since_commit_sha1, import_since_commit_sha1
+ )
+ self.assertEqual(naked_job.metadata, metadata)
+
+ def test_run_import(self):
+ """Run ImportVulnerabilityJob."""
+ with open(self.cve_path, encoding="utf-8") as file:
+ self.useFixture(
+ GitHostingFixture(
+ blob=file.read(),
+ refs=self.refs,
+ diff_stats={"added": ["cves/CVE-2025-1979"]},
+ )
+ )
+
+ cve = self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository=self.repository.git_https_url,
+ git_ref="ref/tags/v1.0",
+ git_paths=["cves"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1=None,
+ )
+ job.run()
+
+ # Check that it created the bug and vulnerability
+ self.assertEqual(len(cve.bugs), 1)
+
+ admin = getUtility(ILaunchpadCelebrities).admin
+ with person_logged_in(admin):
+ self.assertEqual(len(list(cve.vulnerabilities)), 1)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": ["CVE-2025-1979"], "failed": []},
+ )
+
+ def test_run_import_with_wrong_git_paths(self):
+ """Run ImportVulnerabilityJob with wrong git_paths."""
+ with open(self.cve_path, encoding="utf-8") as file:
+ self.useFixture(
+ GitHostingFixture(
+ blob=file.read(),
+ refs=self.refs,
+ diff_stats={"added": ["cves/CVE-2025-1979"]},
+ )
+ )
+
+ cve = self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository=self.repository.git_https_url,
+ git_ref="ref/tags/v1.0",
+ git_paths=["wrong_path"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1=None,
+ )
+ job.run()
+
+ # Check that it did not create the bug and vulnerability
+ self.assertEqual(len(cve.bugs), 0)
+
+ admin = getUtility(ILaunchpadCelebrities).admin
+ with person_logged_in(admin):
+ self.assertEqual(len(list(cve.vulnerabilities)), 0)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": [], "failed": []},
+ )
+
+ def test_run_import_with_wrong_git_repository(self):
+ """Run ImportVulnerabilityJob with wrong git_repository."""
+ self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository="wrong_url",
+ git_ref="ref/heads/main",
+ git_paths=["cves"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1=None,
+ )
+
+ self.assertRaises(VulnerabilityJobException, job.run)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": [], "failed": []},
+ )
+
+ def test_run_import_with_wrong_git_ref(self):
+ """Run ImportVulnerabilityJob with wrong git_ref."""
+ with open(self.cve_path, encoding="utf-8") as file:
+ self.useFixture(
+ GitHostingFixture(
+ blob=file.read(),
+ refs=self.refs,
+ diff_stats={"added": ["cves/CVE-2025-1979"]},
+ )
+ )
+
+ self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository=self.repository.git_https_url,
+ git_ref="ref/heads/wrong-ref",
+ git_paths=["cves"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1=None,
+ )
+
+ self.assertRaises(VulnerabilityJobException, job.run)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": [], "failed": []},
+ )
+
+ def test_run_import_with_wrong_blob(self):
+ """Run ImportVulnerabilityJob with a blob that is not a cve record.
+ This will not raise an exception, it will only not be imported.
+ """
+ self.useFixture(
+ GitHostingFixture(
+ blob=b"Bad blob",
+ refs=self.refs,
+ diff_stats={"added": ["cves/CVE-2025-1979"]},
+ )
+ )
+
+ cve = self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository=self.repository.git_https_url,
+ git_ref="ref/tags/v1.0",
+ git_paths=["cves"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1=None,
+ )
+ job.run()
+
+ # Check that it did not create the bug and vulnerability
+ self.assertEqual(len(cve.bugs), 0)
+
+ admin = getUtility(ILaunchpadCelebrities).admin
+ with person_logged_in(admin):
+ self.assertEqual(len(list(cve.vulnerabilities)), 0)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": [], "failed": []},
+ )
+
+ def test_run_import_with_import_since_commit_sha1(self):
+ """Run ImportVulnerabilityJob using import_since_commit_sha1"""
+ with open(self.cve_path, encoding="utf-8") as file:
+ self.useFixture(
+ GitHostingFixture(
+ blob=file.read(),
+ refs=self.refs,
+ diff_stats={"added": ["cves/CVE-2025-1979"]},
+ )
+ )
+
+ cve = self.factory.makeCVE("2025-1979")
+ self.factory.makeDistribution(name="soss")
+
+ job = self.job_source.create(
+ handler=VulnerabilityHandlerEnum.SOSS,
+ git_repository=self.repository.git_https_url,
+ git_ref="ref/tags/v1.0",
+ git_paths=["cves"],
+ information_type=InformationType.PRIVATESECURITY.value,
+ import_since_commit_sha1="1" * 40,
+ )
+ job.run()
+
+ # Check that it created the bug and vulnerability
+ self.assertEqual(len(cve.bugs), 1)
+
+ admin = getUtility(ILaunchpadCelebrities).admin
+ with person_logged_in(admin):
+ self.assertEqual(len(list(cve.vulnerabilities)), 1)
+
+ self.assertEqual(
+ job.metadata["result"],
+ {"succeeded": ["CVE-2025-1979"], "failed": []},
+ )
+
+ def test_get(self):
+ """ImportVulnerabilityJob.get() returns the import job for the given
+ handler.
+ """
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ # There is no job before creating it
+ self.assertIs(None, self.job_source.get(handler))
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ job_gotten = self.job_source.get(handler)
+
+ self.assertIsInstance(job, ImportVulnerabilityJob)
+ self.assertEqual(job, job_gotten)
+
+ def test_error_description_when_no_error(self):
+ """The ImportVulnerabilityJob.error_description property returns
+ None when no error description is recorded."""
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ self.assertIs(None, removeSecurityProxy(job).error_description)
+
+ def test_error_description_set_when_notifying_about_user_errors(self):
+ """Test that error_description is set by notifyUserError()."""
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+
+ job = self.job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ message = "This is an example message."
+ job.notifyUserError(VulnerabilityJobException(message))
+ self.assertEqual([message], removeSecurityProxy(job).error_description)
+
+
+class TestViaCelery(TestCaseWithFactory):
+ layer = CeleryJobLayer
+
+ def setUp(self):
+ super().setUp()
+ self.repository = self.factory.makeGitRepository()
+ self.refs = self.factory.makeGitRefs(
+ repository=self.repository,
+ paths=("ref/heads/main", "ref/tags/v1.0"),
+ )
+
+ def test_job(self):
+ """Job runs via Celery."""
+ self.factory.makeCVE("2025-1979")
+ transaction.commit()
+
+ fixture = FeatureFixture(
+ {
+ "jobs.celery.enabled_classes": "ImportVulnerabilityJob",
+ }
+ )
+ self.useFixture(fixture)
+ job_source = getUtility(IImportVulnerabilityJobSource)
+
+ handler = VulnerabilityHandlerEnum.SOSS
+ git_repository = self.repository.git_https_url
+ git_ref = "ref/heads/main"
+ git_paths = ["cves"]
+ information_type = InformationType.PRIVATESECURITY.value
+ import_since_commit_sha1 = None
+ metadata = {
+ "git_repository": git_repository,
+ "git_ref": git_ref,
+ "git_paths": git_paths,
+ "information_type": information_type,
+ "import_since_commit_sha1": import_since_commit_sha1,
+ }
+
+ with block_on_job():
+ job_source.create(
+ handler,
+ git_repository,
+ git_ref,
+ git_paths,
+ information_type,
+ import_since_commit_sha1,
+ )
+ transaction.commit()
+
+ job = job_source.get(handler)
+ self.assertEqual(job.handler, handler)
+ self.assertEqual(job.metadata, metadata)
diff --git a/lib/lp/services/config/schema-lazr.conf b/lib/lp/services/config/schema-lazr.conf
index 0d5f38b..5dbcf5a 100644
--- a/lib/lp/services/config/schema-lazr.conf
+++ b/lib/lp/services/config/schema-lazr.conf
@@ -2250,6 +2250,10 @@ crontab_group: MAIN
[IUpdatePreviewDiffJobSource]
link: IBranchMergeProposalJobSource
+[IImportVulnerabilityJobSource]
+module: lp.bugs.interfaces.vulnerabilityjob
+dbuser: launchpad_main
+
[IWebhookDeliveryJobSource]
module: lp.services.webhooks.interfaces
dbuser: webhookrunner
_______________________________________________
Mailing list: https://launchpad.net/~launchpad-reviewers
Post to : launchpad-reviewers@lists.launchpad.net
Unsubscribe : https://launchpad.net/~launchpad-reviewers
More help : https://help.launchpad.net/ListHelp
