Merge ~lgp171188/launchpad:security-tracker-cve-changes into launchpad:master

[Guruprasad]

Add new fields and methods to ICve and Cve

Add the 'date_made_public', 'discoverer', and the 'cvss' fields. Also
add the 'setCVSSVectorForAuthority()' method.

[Guruprasad (lgp171188)]

Self-approving this MP as its dependency DB changes have been merged and this just reinstates a previously approved, merged, and reverted change.

review: Approve

diff --git a/lib/lp/bugs/interfaces/cve.py b/lib/lp/bugs/interfaces/cve.py
index 2ba5a53..3a860da 100644
--- a/lib/lp/bugs/interfaces/cve.py
+++ b/lib/lp/bugs/interfaces/cve.py
@@ -30,12 +30,15 @@ from zope.interface import (
 from zope.schema import (
     Choice,
     Datetime,
+    Dict,
     Int,
+    Text,
     TextLine,
     )
 
 from lp import _
 from lp.app.validators.validation import valid_cve_sequence
+from lp.services.fields import PersonChoice
 
 
 class CveStatus(DBEnumeratedType):
@@ -121,12 +124,45 @@ class ICve(Interface):
                               description=_("A title for the CVE")))
     references = Attribute("The set of CVE References for this CVE.")
 
+    date_made_public = exported(
+        Datetime(title=_('Date Made Public'), required=False, readonly=True),
+        as_of='devel'
+    )
+
+    discoverer = exported(
+        PersonChoice(
+            title=_('Discoverer'),
+            required=False,
+            readonly=True,
+            vocabulary='ValidPerson'
+        ),
+        as_of='devel'
+    )
+
+    cvss = exported(
+        Dict(
+            title=_('CVSS'),
+            description=_(
+                'The CVSS vector strings from various authorities '
+                'that publish it.'
+            ),
+            key_type=Text(title=_('The authority that published the score.')),
+            value_type=Text(title=_('The CVSS vector string.')),
+            required=False,
+            readonly=True,
+        ),
+        as_of='devel'
+    )
+
     def createReference(source, content, url=None):
         """Create a new CveReference for this CVE."""
 
     def removeReference(ref):
         """Remove a CveReference."""
 
+    def setCVSSVectorForAuthority(authority, vector_string):
+        """Set the CVSS vector string from an authority."""
+
 
 @exported_as_webservice_collection(ICve)
 class ICveSet(Interface):
@@ -140,7 +176,8 @@ class ICveSet(Interface):
     def __iter__():
         """Iterate through all the Cve records."""
 
-    def new(sequence, description, cvestate=CveStatus.CANDIDATE):
+    def new(sequence, description, cvestate=CveStatus.CANDIDATE,
+            date_made_public=None, discoverer=None, cvss=None):
         """Create a new ICve."""
 
     @collection_default_content()
diff --git a/lib/lp/bugs/model/cve.py b/lib/lp/bugs/model/cve.py
index 27a40b5..f49ce00 100644
--- a/lib/lp/bugs/model/cve.py
+++ b/lib/lp/bugs/model/cve.py
@@ -9,10 +9,12 @@ __all__ = [
 import operator
 
 import pytz
+from storm.databases.postgres import JSON
 from storm.locals import (
     DateTime,
     Desc,
     Int,
+    Reference,
     ReferenceSet,
     Store,
     Unicode,
@@ -60,11 +62,29 @@ class Cve(StormBase, BugLinkTargetMixin):
     references = ReferenceSet(
         id, 'CveReference.cve_id', order_by='CveReference.id')
 
-    def __init__(self, sequence, status, description):
+    date_made_public = DateTime(tzinfo=pytz.UTC, allow_none=True)
+    discoverer_id = Int(name='discoverer', allow_none=True)
+    discoverer = Reference(discoverer_id, 'Person.id')
+    _cvss = JSON(name='cvss', allow_none=True)
+
+    @property
+    def cvss(self):
+        return self._cvss or {}
+
+    @cvss.setter
+    def cvss(self, value):
+        assert value is None or isinstance(value, dict)
+        self._cvss = value
+
+    def __init__(self, sequence, status, description,
+                 date_made_public=None, discoverer=None, cvss=None):
         super().__init__()
         self.sequence = sequence
         self.status = status
         self.description = description
+        self.date_made_public = date_made_public
+        self.discoverer = discoverer
+        self._cvss = cvss
 
     @property
     def url(self):
@@ -111,6 +131,12 @@ class Cve(StormBase, BugLinkTargetMixin):
         getUtility(IXRefSet).delete(
             {('cve', self.sequence): [('bug', str(bug.id))]})
 
+    def setCVSSVectorForAuthority(self, authority, vector_string):
+        """See ICveReference."""
+        if self._cvss is None:
+            self._cvss = {}
+        self._cvss[authority] = vector_string
+
 
 @implementer(ICveSet)
 class CveSet:
@@ -136,10 +162,18 @@ class CveSet:
         """See ICveSet."""
         return iter(IStore(Cve).find(Cve))
 
-    def new(self, sequence, description, status=CveStatus.CANDIDATE):
+    def new(self, sequence, description, status=CveStatus.CANDIDATE,
+            date_made_public=None, discoverer=None, cvss=None):
         """See ICveSet."""
-        cve = Cve(sequence=sequence, status=status,
-            description=description)
+        cve = Cve(
+            sequence=sequence,
+            status=status,
+            description=description,
+            date_made_public=date_made_public,
+            discoverer=discoverer,
+            cvss=cvss
+        )
+
         IStore(Cve).add(cve)
         return cve
 
diff --git a/lib/lp/bugs/tests/test_cve.py b/lib/lp/bugs/tests/test_cve.py
index 90bffc7..6882b34 100644
--- a/lib/lp/bugs/tests/test_cve.py
+++ b/lib/lp/bugs/tests/test_cve.py
@@ -3,10 +3,19 @@
 
 """CVE related tests."""
 
+from datetime import datetime
+
+import pytz
+from testtools.matchers import MatchesStructure
+from testtools.testcase import ExpectedException
 from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
 
 from lp.bugs.interfaces.bugtasksearch import BugTaskSearchParams
-from lp.bugs.interfaces.cve import ICveSet
+from lp.bugs.interfaces.cve import (
+    CveStatus,
+    ICveSet,
+    )
 from lp.testing import (
     login_person,
     person_logged_in,
@@ -133,3 +142,161 @@ class TestBugLinks(TestCaseWithFactory):
         self.assertContentEqual([bug1], cve2.bugs)
         self.assertContentEqual([cve2], bug1.cves)
         self.assertContentEqual([], bug2.cves)
+
+
+class TestCve(TestCaseWithFactory):
+    """Tests for Cve fields and methods."""
+
+    layer = DatabaseFunctionalLayer
+
+    def test_cveset_new_method_optional_parameters(self):
+        cve = getUtility(ICveSet).new(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            status=CveStatus.CANDIDATE
+        )
+        self.assertThat(cve, MatchesStructure.byEquality(
+            sequence='2099-1234',
+            status=CveStatus.CANDIDATE,
+            description='A critical vulnerability',
+            date_made_public=None,
+            discoverer=None,
+            cvss={}
+        ))
+
+    def test_cveset_new_method_parameters(self):
+        person = self.factory.makePerson()
+        today = datetime.now(tz=pytz.UTC)
+        cvss = {
+            'nvd': 'CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H'
+        }
+        cve = getUtility(ICveSet).new(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            status=CveStatus.CANDIDATE,
+            date_made_public=today,
+            discoverer=person,
+            cvss=cvss
+        )
+        self.assertThat(cve, MatchesStructure.byEquality(
+            sequence='2099-1234',
+            status=CveStatus.CANDIDATE,
+            description='A critical vulnerability',
+            date_made_public=today,
+            discoverer=person,
+            cvss=cvss
+        ))
+
+    def test_cve_date_made_public_invalid_values(self):
+        invalid_values = ['', 'abcd', {'a': 1},
+                          [1, 'a', '2', 'b'], '2022-01-01']
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+        )
+        for invalid_value in invalid_values:
+            with ExpectedException(TypeError, 'Expected datetime,.*'):
+                removeSecurityProxy(cve).date_made_public = invalid_value
+
+    def test_cve_discoverer_id_invalid_values(self):
+        invalid_values = ['', 'abcd', '2022-01-01', datetime.now()]
+
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+        )
+        for invalid_value in invalid_values:
+            with ExpectedException(TypeError, 'Expected int,.*'):
+                removeSecurityProxy(cve).discoverer_id = invalid_value
+
+    def test_cve_cvss_invalid_values(self):
+        invalid_values = ['', 'abcd', '2022-01-01', datetime.now()]
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+        )
+        for invalid_value in invalid_values:
+            with ExpectedException(AssertionError):
+                removeSecurityProxy(cve).cvss = invalid_value
+
+    def test_cvss_value_returned_when_null(self):
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+        )
+        cve = removeSecurityProxy(cve)
+        self.assertIsNone(cve._cvss)
+        self.assertEqual({}, cve.cvss)
+
+    def test_setCVSSVectorForAuthority_initially_unset(self):
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+        )
+        unproxied_cve = removeSecurityProxy(cve)
+        self.assertIsNone(unproxied_cve._cvss)
+        self.assertEqual({}, unproxied_cve.cvss)
+
+        cve.setCVSSVectorForAuthority(
+            authority="nvd",
+            vector_string="CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
+        )
+
+        self.assertEqual(
+            {"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"},
+            unproxied_cve.cvss
+        )
+
+    def test_setCVSSVectorForAuthority_overwrite_existing_key_value(self):
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+            cvss={"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"}
+        )
+        unproxied_cve = removeSecurityProxy(cve)
+        self.assertEqual(
+            {"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"},
+            unproxied_cve.cvss
+        )
+
+        cve.setCVSSVectorForAuthority(
+            authority="nvd",
+            vector_string="CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N"
+        )
+
+        self.assertEqual(
+            {"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N"},
+            unproxied_cve.cvss
+        )
+
+    def test_setCVSSVectorForAuthority_add_new_when_initial_value_set(self):
+        cve = self.factory.makeCVE(
+            sequence='2099-1234',
+            description='A critical vulnerability',
+            cvestate=CveStatus.CANDIDATE,
+            cvss={"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"}
+        )
+        unproxied_cve = removeSecurityProxy(cve)
+        self.assertEqual(
+            {"nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"},
+            unproxied_cve.cvss
+        )
+
+        cve.setCVSSVectorForAuthority(
+            authority="nist",
+            vector_string="CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N"
+        )
+
+        self.assertEqual(
+            {
+                "nvd": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H",
+                "nist": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N"
+            },
+            unproxied_cve.cvss
+        )
diff --git a/lib/lp/testing/factory.py b/lib/lp/testing/factory.py
index 041cb89..9e37780 100644
--- a/lib/lp/testing/factory.py
+++ b/lib/lp/testing/factory.py
@@ -4582,11 +4582,21 @@ class BareLaunchpadObjectFactory(ObjectFactory):
         return secret, token
 
     def makeCVE(self, sequence, description=None,
-                cvestate=CveStatus.CANDIDATE):
+                cvestate=CveStatus.CANDIDATE,
+                date_made_public=None, discoverer=None,
+                cvss=None):
         """Create a new CVE record."""
         if description is None:
             description = self.getUniqueUnicode()
-        return getUtility(ICveSet).new(sequence, description, cvestate)
+
+        return getUtility(ICveSet).new(
+            sequence,
+            description,
+            cvestate,
+            date_made_public,
+            discoverer,
+            cvss
+        )
 
     def makePublisherConfig(self, distribution=None, root_dir=None,
                             base_url=None, copy_base_url=None):