launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #28991
[Merge] ~andrey-fedoseev/launchpad:uct-export into launchpad:master
Andrey Fedoseev has proposed merging ~andrey-fedoseev/launchpad:uct-export into launchpad:master.
Commit message:
Enable exporting bugs to UCT CVE records
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~andrey-fedoseev/launchpad/+git/launchpad/+merge/428152
--
Your team Launchpad code reviewers is requested to review the proposed merge of ~andrey-fedoseev/launchpad:uct-export into launchpad:master.
diff --git a/lib/contrib/cve_lib.py b/lib/contrib/cve_lib.py
index 6672864..da211fd 100644
--- a/lib/contrib/cve_lib.py
+++ b/lib/contrib/cve_lib.py
@@ -11,6 +11,7 @@ import math
import os
import re
import sys
+from collections import OrderedDict
import yaml
@@ -1195,17 +1196,17 @@ def load_cve(cve, strict=False, srcmap=None):
]
extra_fields = ["CRD", "PublicDateAtUSN", "Mitigation"]
- data = dict()
+ data = OrderedDict()
# maps entries in data to their source line - if didn't supply one
# create a local one to simplify the code
if srcmap is None:
- srcmap = dict()
- srcmap.setdefault("pkgs", dict())
- srcmap.setdefault("tags", dict())
- data.setdefault("tags", dict())
- srcmap.setdefault("patches", dict())
- data.setdefault("patches", dict())
- affected = dict()
+ srcmap = OrderedDict()
+ srcmap.setdefault("pkgs", OrderedDict())
+ srcmap.setdefault("tags", OrderedDict())
+ data.setdefault("tags", OrderedDict())
+ srcmap.setdefault("patches", OrderedDict())
+ data.setdefault("patches", OrderedDict())
+ affected = OrderedDict()
lastfield = ""
fields_seen = []
if not os.path.exists(cve):
@@ -1254,7 +1255,7 @@ def load_cve(cve, strict=False, srcmap=None):
code = EXIT_FAIL
elif lastfield == "CVSS":
try:
- cvss = dict()
+ cvss = OrderedDict()
result = re.search(
r" (.+)\: (\S+)( \[(.*) (.*)\])?", line
)
@@ -1277,7 +1278,7 @@ def load_cve(cve, strict=False, srcmap=None):
# line where the CVSS block starts - so convert it
# to a dict first if needed
if type(srcmap["CVSS"]) is tuple:
- srcmap["CVSS"] = dict()
+ srcmap["CVSS"] = OrderedDict()
srcmap["CVSS"].setdefault(
cvss["source"], (cve, linenum)
)
@@ -1419,7 +1420,7 @@ def load_cve(cve, strict=False, srcmap=None):
msg += "%s: %d: unknown entry '%s'\n" % (cve, linenum, rel)
code = EXIT_FAIL
continue
- affected.setdefault(pkg, dict())
+ affected.setdefault(pkg, OrderedDict())
if rel in affected[pkg]:
msg += (
"%s: %d: duplicate entry for '%s': original at line %d\n"
@@ -1433,7 +1434,7 @@ def load_cve(cve, strict=False, srcmap=None):
code = EXIT_FAIL
continue
affected[pkg].setdefault(rel, [state, details])
- srcmap["pkgs"].setdefault(pkg, dict())
+ srcmap["pkgs"].setdefault(pkg, OrderedDict())
srcmap["pkgs"][pkg].setdefault(rel, (cve, linenum))
elif field not in required_fields + extra_fields:
msg += "%s: %d: unknown field '%s'\n" % (cve, linenum, field)
@@ -1539,9 +1540,9 @@ def amend_external_subproject_pkg(cve, data, srcmap, amendments, code, msg):
if not success:
return code, msg
- data.setdefault("pkgs", dict())
- data["pkgs"].setdefault(pkg, dict())
- srcmap["pkgs"].setdefault(pkg, dict())
+ data.setdefault("pkgs", OrderedDict())
+ data["pkgs"].setdefault(pkg, OrderedDict())
+ srcmap["pkgs"].setdefault(pkg, OrderedDict())
# override existing release info if it exists
data["pkgs"][pkg][release] = [state, details]
srcmap["pkgs"][pkg][release] = (cve, linenum)
diff --git a/lib/lp/bugs/scripts/tests/sampledata/CVE-2022-23222 b/lib/lp/bugs/scripts/tests/sampledata/CVE-2022-23222
index 6d665be..30574d6 100644
--- a/lib/lp/bugs/scripts/tests/sampledata/CVE-2022-23222
+++ b/lib/lp/bugs/scripts/tests/sampledata/CVE-2022-23222
@@ -14,7 +14,7 @@ Ubuntu-Description:
execute arbitrary code.
Notes:
sbeattie> Ubuntu 21.10 / 5.13+ kernels disable unprivileged BPF by default.
- sbeattie> kernels 5.8 and older are not affected, priority high is for
+ kernels 5.8 and older are not affected, priority high is for
5.10 and 5.11 based kernels only
Mitigation:
seth-arnold> set kernel.unprivileged_bpf_disabled to 1
diff --git a/lib/lp/bugs/scripts/tests/test_uctimport.py b/lib/lp/bugs/scripts/tests/test_uct.py
similarity index 73%
rename from lib/lp/bugs/scripts/tests/test_uctimport.py
rename to lib/lp/bugs/scripts/tests/test_uct.py
index d9aaa4b..47ff535 100644
--- a/lib/lp/bugs/scripts/tests/test_uctimport.py
+++ b/lib/lp/bugs/scripts/tests/test_uct.py
@@ -13,8 +13,11 @@ from lp.bugs.enums import VulnerabilityStatus
from lp.bugs.interfaces.bugtask import BugTaskImportance, BugTaskStatus
from lp.bugs.model.bug import Bug
from lp.bugs.model.bugtask import BugTask
-from lp.bugs.scripts.uctimport import (
+from lp.bugs.model.cve import Cve as CveModel
+from lp.bugs.scripts.uct import (
CVE,
+ CVSS,
+ UCTExporter,
UCTImporter,
UCTImportError,
UCTRecord,
@@ -27,12 +30,15 @@ from lp.testing.layers import ZopelessDatabaseLayer
class TestUCTRecord(TestCase):
- def test_load(self):
- cve_path = Path(__file__).parent / "sampledata" / "CVE-2022-23222"
- uct_record = UCTRecord.load(cve_path)
- self.assertEqual(
+
+ maxDiff = None
+
+ def test_load_save(self):
+ load_from = Path(__file__).parent / "sampledata" / "CVE-2022-23222"
+ uct_record = UCTRecord.load(load_from)
+ self.assertDictEqual(
UCTRecord(
- path=cve_path,
+ parent_dir="sampledata",
assigned_to="",
bugs=[
"https://github.com/mm2/Little-CMS/issues/29";,
@@ -40,17 +46,20 @@ class TestUCTRecord(TestCase):
"https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=745471";,
],
cvss=[
- {
- "source": "nvd",
- "vector": (
- "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H"
+ CVSS(
+ authority="nvd",
+ vector_string=(
+ "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H "
+ "[7.8 HIGH]"
),
- "baseScore": "7.8",
- "baseSeverity": "HIGH",
- }
+ ),
],
candidate="CVE-2022-23222",
- date_made_public=datetime.datetime(
+ crd=None,
+ public_date_at_USN=datetime.datetime(
+ 2022, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
+ ),
+ public_date=datetime.datetime(
2022, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
),
description=(
@@ -63,17 +72,12 @@ class TestUCTRecord(TestCase):
mitigation=(
"seth-arnold> set kernel.unprivileged_bpf_disabled to 1"
),
- notes=[
- UCTRecord.Note(
- author="sbeattie",
- text=(
- "Ubuntu 21.10 / 5.13+ kernels disable "
- "unprivileged BPF by default.\nkernels 5.8 and "
- "older are not affected, priority high is "
- "for\n5.10 and 5.11 based kernels only"
- ),
- ),
- ],
+ notes=(
+ "sbeattie> Ubuntu 21.10 / 5.13+ kernels disable "
+ "unprivileged BPF by default.\n kernels 5.8 and "
+ "older are not affected, priority high is "
+ "for\n 5.10 and 5.11 based kernels only"
+ ),
priority=UCTRecord.Priority.CRITICAL,
references=["https://ubuntu.com/security/notices/USN-5368-1";],
ubuntu_description=(
@@ -88,10 +92,10 @@ class TestUCTRecord(TestCase):
name="linux",
statuses=[
UCTRecord.DistroSeriesPackageStatus(
- distroseries="devel",
- status=UCTRecord.PackageStatus.NOT_AFFECTED,
- reason="5.15.0-25.25",
- priority=UCTRecord.Priority.MEDIUM,
+ distroseries="upstream",
+ status=UCTRecord.PackageStatus.RELEASED,
+ reason="5.17~rc1",
+ priority=None,
),
UCTRecord.DistroSeriesPackageStatus(
distroseries="impish",
@@ -100,10 +104,10 @@ class TestUCTRecord(TestCase):
priority=UCTRecord.Priority.MEDIUM,
),
UCTRecord.DistroSeriesPackageStatus(
- distroseries="upstream",
- status=UCTRecord.PackageStatus.RELEASED,
- reason="5.17~rc1",
- priority=None,
+ distroseries="devel",
+ status=UCTRecord.PackageStatus.NOT_AFFECTED,
+ reason="5.15.0-25.25",
+ priority=UCTRecord.Priority.MEDIUM,
),
],
priority=None,
@@ -123,9 +127,9 @@ class TestUCTRecord(TestCase):
name="linux-hwe",
statuses=[
UCTRecord.DistroSeriesPackageStatus(
- distroseries="devel",
- status=UCTRecord.PackageStatus.DOES_NOT_EXIST,
- reason="",
+ distroseries="upstream",
+ status=UCTRecord.PackageStatus.RELEASED,
+ reason="5.17~rc1",
priority=None,
),
UCTRecord.DistroSeriesPackageStatus(
@@ -135,9 +139,9 @@ class TestUCTRecord(TestCase):
priority=None,
),
UCTRecord.DistroSeriesPackageStatus(
- distroseries="upstream",
- status=UCTRecord.PackageStatus.RELEASED,
- reason="5.17~rc1",
+ distroseries="devel",
+ status=UCTRecord.PackageStatus.DOES_NOT_EXIST,
+ reason="",
priority=None,
),
],
@@ -146,49 +150,73 @@ class TestUCTRecord(TestCase):
patches=[],
),
],
- ),
- uct_record,
+ ).__dict__,
+ uct_record.__dict__,
+ )
+
+ output_dir = Path(self.makeTemporaryDirectory())
+ saved_to_path = uct_record.save(output_dir)
+ self.assertEqual(
+ output_dir / "sampledata" / "CVE-2022-23222", saved_to_path
)
+ self.assertEqual(load_from.read_text(), saved_to_path.read_text())
class TextCVE(TestCaseWithFactory):
layer = ZopelessDatabaseLayer
+ maxDiff = None
- def test_make_from_uct_record(self):
+ def setUp(self, *args, **kwargs):
+ super().setUp(*args, **kwargs)
celebrities = getUtility(ILaunchpadCelebrities)
ubuntu = celebrities.ubuntu
supported_series = self.factory.makeDistroSeries(
- distribution=ubuntu, status=SeriesStatus.SUPPORTED
+ distribution=ubuntu,
+ status=SeriesStatus.SUPPORTED,
+ name="focal",
)
current_series = self.factory.makeDistroSeries(
- distribution=ubuntu, status=SeriesStatus.CURRENT
+ distribution=ubuntu,
+ status=SeriesStatus.CURRENT,
+ name="jammy",
)
devel_series = self.factory.makeDistroSeries(
- distribution=ubuntu, status=SeriesStatus.DEVELOPMENT
+ distribution=ubuntu,
+ status=SeriesStatus.DEVELOPMENT,
+ name="kinetic",
)
dsp1 = self.factory.makeDistributionSourcePackage(distribution=ubuntu)
dsp2 = self.factory.makeDistributionSourcePackage(distribution=ubuntu)
assignee = self.factory.makePerson()
- uct_record = UCTRecord(
- path=Path("./active/CVE-2022-23222"),
+ self.uct_record = UCTRecord(
+ parent_dir="active",
assigned_to=assignee.name,
bugs=["https://github.com/mm2/Little-CMS/issues/29";],
- cvss=[],
+ cvss=[
+ CVSS(
+ authority="nvd",
+ vector_string=(
+ "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H "
+ "[7.8 HIGH]"
+ ),
+ ),
+ ],
candidate="CVE-2022-23222",
- date_made_public=datetime.datetime(
+ crd=datetime.datetime(
+ 2020, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
+ ),
+ public_date_at_USN=datetime.datetime(
+ 2021, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
+ ),
+ public_date=datetime.datetime(
2022, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
),
description="description",
discovered_by="tr3e wang",
mitigation="mitigation",
- notes=[
- UCTRecord.Note(
- author="author",
- text="text",
- ),
- ],
+ notes="author> text",
priority=UCTRecord.Priority.CRITICAL,
references=["https://ubuntu.com/security/notices/USN-5368-1";],
ubuntu_description="ubuntu-description",
@@ -216,17 +244,8 @@ class TextCVE(TestCaseWithFactory):
),
],
priority=None,
- tags={"not-ue"},
- patches=[
- UCTRecord.Patch(
- patch_type="break-fix",
- entry=(
- "457f44363a8894135c85b7a9afd2bd8196db24ab "
- "c25b2ae136039ffa820c26138ed4a5e5f3ab3841|"
- "local-CVE-2022-23222-fix"
- ),
- )
- ],
+ tags=set(),
+ patches=[],
),
UCTRecord.Package(
name=dsp2.sourcepackagename.name,
@@ -256,29 +275,29 @@ class TextCVE(TestCaseWithFactory):
),
],
)
- cve = CVE.make_from_uct_record(uct_record)
- self.assertEqual("CVE-2022-23222", cve.sequence)
- self.assertEqual(
- datetime.datetime(
+
+ self.cve = CVE(
+ sequence="CVE-2022-23222",
+ crd=datetime.datetime(
+ 2020, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
+ ),
+ public_date_at_USN=datetime.datetime(
+ 2021, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
+ ),
+ public_date=datetime.datetime(
2022, 1, 14, 8, 15, tzinfo=datetime.timezone.utc
),
- cve.date_made_public,
- )
- self.assertEqual(
- [
+ distro_packages=[
CVE.DistroPackage(
package=dsp1,
- importance=BugTaskImportance.CRITICAL,
+ importance=None,
),
CVE.DistroPackage(
package=dsp2,
importance=BugTaskImportance.HIGH,
),
],
- cve.distro_packages,
- )
- self.assertEqual(
- [
+ series_packages=[
CVE.SeriesPackage(
package=SourcePackage(
sourcepackagename=dsp1.sourcepackagename,
@@ -302,7 +321,7 @@ class TextCVE(TestCaseWithFactory):
sourcepackagename=dsp1.sourcepackagename,
distroseries=devel_series,
),
- importance=BugTaskImportance.CRITICAL,
+ importance=None,
status=BugTaskStatus.FIXRELEASED,
status_explanation="reason 3",
),
@@ -311,7 +330,7 @@ class TextCVE(TestCaseWithFactory):
sourcepackagename=dsp2.sourcepackagename,
distroseries=supported_series,
),
- importance=BugTaskImportance.HIGH,
+ importance=None,
status=BugTaskStatus.DOESNOTEXIST,
status_explanation="",
),
@@ -320,7 +339,7 @@ class TextCVE(TestCaseWithFactory):
sourcepackagename=dsp2.sourcepackagename,
distroseries=current_series,
),
- importance=BugTaskImportance.HIGH,
+ importance=None,
status=BugTaskStatus.DOESNOTEXIST,
status_explanation="",
),
@@ -329,30 +348,45 @@ class TextCVE(TestCaseWithFactory):
sourcepackagename=dsp2.sourcepackagename,
distroseries=devel_series,
),
- importance=BugTaskImportance.HIGH,
+ importance=None,
status=BugTaskStatus.FIXRELEASED,
status_explanation="",
),
],
- cve.series_packages,
- )
- self.assertEqual(BugTaskImportance.CRITICAL, cve.importance)
- self.assertEqual(VulnerabilityStatus.ACTIVE, cve.status)
- self.assertEqual(assignee, cve.assignee)
- self.assertEqual("description", cve.description)
- self.assertEqual("ubuntu-description", cve.ubuntu_description)
- self.assertEqual(
- ["https://github.com/mm2/Little-CMS/issues/29";], cve.bug_urls
- )
- self.assertEqual(
- ["https://ubuntu.com/security/notices/USN-5368-1";], cve.references
+ importance=BugTaskImportance.CRITICAL,
+ status=VulnerabilityStatus.ACTIVE,
+ assignee=assignee,
+ discovered_by="tr3e wang",
+ description="description",
+ ubuntu_description="ubuntu-description",
+ bug_urls=["https://github.com/mm2/Little-CMS/issues/29";],
+ references=["https://ubuntu.com/security/notices/USN-5368-1";],
+ notes="author> text",
+ mitigation="mitigation",
+ cvss=[
+ CVSS(
+ authority="nvd",
+ vector_string=(
+ "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H "
+ "[7.8 HIGH]"
+ ),
+ ),
+ ],
)
- self.assertEqual("author> text", cve.notes)
- self.assertEqual("mitigation", cve.mitigation)
+ def test_make_from_uct_record(self):
+ cve = CVE.make_from_uct_record(self.uct_record)
+ self.assertDictEqual(self.cve.__dict__, cve.__dict__)
+
+ def test_to_uct_record(self):
+ uct_record = self.cve.to_uct_record()
+ self.assertListEqual(self.uct_record.packages, uct_record.packages)
+ self.assertDictEqual(self.uct_record.__dict__, uct_record.__dict__)
-class TestUCTImporter(TestCaseWithFactory):
+class TestUCTImporterExporter(TestCaseWithFactory):
+
+ maxDiff = None
layer = ZopelessDatabaseLayer
def setUp(self, *args, **kwargs):
@@ -362,19 +396,27 @@ class TestUCTImporter(TestCaseWithFactory):
self.esm = self.factory.makeDistribution("esm")
self.bug_importer = celebrities.bug_importer
self.ubuntu_supported_series = self.factory.makeDistroSeries(
- distribution=self.ubuntu, status=SeriesStatus.SUPPORTED
+ distribution=self.ubuntu,
+ status=SeriesStatus.SUPPORTED,
+ name="focal",
)
self.ubuntu_current_series = self.factory.makeDistroSeries(
- distribution=self.ubuntu, status=SeriesStatus.CURRENT
+ distribution=self.ubuntu, status=SeriesStatus.CURRENT, name="jammy"
)
self.ubuntu_devel_series = self.factory.makeDistroSeries(
- distribution=self.ubuntu, status=SeriesStatus.DEVELOPMENT
+ distribution=self.ubuntu,
+ status=SeriesStatus.DEVELOPMENT,
+ name="kinetic",
)
self.esm_supported_series = self.factory.makeDistroSeries(
- distribution=self.esm, status=SeriesStatus.SUPPORTED
+ distribution=self.esm,
+ status=SeriesStatus.SUPPORTED,
+ name="precise",
)
self.esm_current_series = self.factory.makeDistroSeries(
- distribution=self.esm, status=SeriesStatus.CURRENT
+ distribution=self.esm,
+ status=SeriesStatus.CURRENT,
+ name="trusty",
)
self.ubuntu_package = self.factory.makeDistributionSourcePackage(
distribution=self.ubuntu
@@ -407,7 +449,9 @@ class TestUCTImporter(TestCaseWithFactory):
self.now = datetime.datetime.now(datetime.timezone.utc)
self.cve = CVE(
sequence="CVE-2022-23222",
- date_made_public=self.now,
+ crd=None,
+ public_date=self.now,
+ public_date_at_USN=None,
distro_packages=[
CVE.DistroPackage(
package=self.ubuntu_package,
@@ -415,7 +459,7 @@ class TestUCTImporter(TestCaseWithFactory):
),
CVE.DistroPackage(
package=self.esm_package,
- importance=BugTaskImportance.MEDIUM,
+ importance=None,
),
],
series_packages=[
@@ -433,7 +477,7 @@ class TestUCTImporter(TestCaseWithFactory):
sourcepackagename=self.ubuntu_package.sourcepackagename, # noqa: E501
distroseries=self.ubuntu_current_series,
),
- importance=BugTaskImportance.LOW,
+ importance=None,
status=BugTaskStatus.DOESNOTEXIST,
status_explanation="does not exist",
),
@@ -442,7 +486,7 @@ class TestUCTImporter(TestCaseWithFactory):
sourcepackagename=self.ubuntu_package.sourcepackagename, # noqa: E501
distroseries=self.ubuntu_devel_series,
),
- importance=BugTaskImportance.LOW,
+ importance=None,
status=BugTaskStatus.INVALID,
status_explanation="not affected",
),
@@ -451,7 +495,7 @@ class TestUCTImporter(TestCaseWithFactory):
sourcepackagename=self.esm_package.sourcepackagename,
distroseries=self.esm_supported_series,
),
- importance=BugTaskImportance.MEDIUM,
+ importance=None,
status=BugTaskStatus.WONTFIX,
status_explanation="ignored",
),
@@ -460,7 +504,7 @@ class TestUCTImporter(TestCaseWithFactory):
sourcepackagename=self.esm_package.sourcepackagename,
distroseries=self.esm_current_series,
),
- importance=BugTaskImportance.MEDIUM,
+ importance=None,
status=BugTaskStatus.UNKNOWN,
status_explanation="needs triage",
),
@@ -468,14 +512,25 @@ class TestUCTImporter(TestCaseWithFactory):
importance=BugTaskImportance.MEDIUM,
status=VulnerabilityStatus.ACTIVE,
assignee=self.factory.makePerson(),
+ discovered_by="",
description="description",
ubuntu_description="ubuntu-description",
bug_urls=["https://github.com/mm2/Little-CMS/issues/29";],
references=["https://ubuntu.com/security/notices/USN-5368-1";],
notes="author> text",
mitigation="mitigation",
+ cvss=[
+ CVSS(
+ authority="nvd",
+ vector_string=(
+ "CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H "
+ "[7.8 HIGH]"
+ ),
+ ),
+ ],
)
self.importer = UCTImporter()
+ self.exporter = UCTExporter()
def checkBug(self, bug: Bug, cve: CVE):
self.assertEqual(cve.sequence, bug.title)
@@ -483,10 +538,6 @@ class TestUCTImporter(TestCaseWithFactory):
self.assertEqual(InformationType.PUBLICSECURITY, bug.information_type)
expected_description = cve.description
- if cve.ubuntu_description:
- expected_description = "{}\n\nUbuntu-Description:\n{}".format(
- expected_description, cve.ubuntu_description
- )
if cve.references:
expected_description = "{}\n\nReferences:\n{}".format(
expected_description, "\n".join(cve.references)
@@ -505,15 +556,21 @@ class TestUCTImporter(TestCaseWithFactory):
)
bug_tasks_by_target = {t.target: t for t in bug_tasks}
+ package_importances = {}
+
for distro_package in cve.distro_packages:
self.assertIn(distro_package.package, bug_tasks_by_target)
t = bug_tasks_by_target[distro_package.package]
+ package_importance = distro_package.importance or cve.importance
+ package_importances[
+ distro_package.package.sourcepackagename
+ ] = package_importance
conjoined_primary = t.conjoined_primary
if conjoined_primary:
expected_importance = conjoined_primary.importance
expected_status = conjoined_primary.status
else:
- expected_importance = distro_package.importance
+ expected_importance = package_importance
expected_status = BugTaskStatus.NEW
self.assertEqual(expected_importance, t.importance)
self.assertEqual(expected_status, t.status)
@@ -522,7 +579,11 @@ class TestUCTImporter(TestCaseWithFactory):
for series_package in cve.series_packages:
self.assertIn(series_package.package, bug_tasks_by_target)
t = bug_tasks_by_target[series_package.package]
- self.assertEqual(series_package.importance, t.importance)
+ package_importance = package_importances[
+ series_package.package.sourcepackagename
+ ]
+ sp_importance = series_package.importance or package_importance
+ self.assertEqual(sp_importance, t.importance)
self.assertEqual(series_package.status, t.status)
self.assertEqual(
series_package.status_explanation, t.status_explanation
@@ -546,7 +607,7 @@ class TestUCTImporter(TestCaseWithFactory):
self.assertEqual(self.bug_importer, vulnerability.creator)
self.assertEqual(self.lp_cve, vulnerability.cve)
self.assertEqual(cve.status, vulnerability.status)
- self.assertEqual(cve.description, vulnerability.description)
+ self.assertEqual(cve.ubuntu_description, vulnerability.description)
self.assertEqual(cve.notes, vulnerability.notes)
self.assertEqual(cve.mitigation, vulnerability.mitigation)
self.assertEqual(cve.importance, vulnerability.importance)
@@ -558,6 +619,12 @@ class TestUCTImporter(TestCaseWithFactory):
)
self.assertEqual([bug], vulnerability.bugs)
+ def checkLaunchpadCve(self, lp_cve: CveModel, cve: CVE):
+ self.assertDictEqual(
+ {cvss.authority: cvss.vector_string for cvss in cve.cvss},
+ lp_cve.cvss,
+ )
+
def test_create_bug(self):
bug = self.importer.create_bug(self.cve, self.lp_cve)
@@ -578,11 +645,11 @@ class TestUCTImporter(TestCaseWithFactory):
def test_find_existing_bug(self):
self.assertIsNone(
- self.importer.find_existing_bug(self.cve, self.lp_cve)
+ self.importer._find_existing_bug(self.cve, self.lp_cve)
)
bug = self.importer.create_bug(self.cve, self.lp_cve)
self.assertEqual(
- self.importer.find_existing_bug(self.cve, self.lp_cve), bug
+ self.importer._find_existing_bug(self.cve, self.lp_cve), bug
)
def test_find_existing_bug_multiple_bugs(self):
@@ -594,7 +661,7 @@ class TestUCTImporter(TestCaseWithFactory):
vulnerability.linkBug(another_bug)
self.assertRaises(
UCTImportError,
- self.importer.find_existing_bug,
+ self.importer._find_existing_bug,
self.cve,
self.lp_cve,
)
@@ -802,19 +869,42 @@ class TestUCTImporter(TestCaseWithFactory):
def test_import_cve(self):
self.importer.import_cve(self.cve)
self.assertIsNotNone(
- self.importer.find_existing_bug(self.cve, self.lp_cve)
+ self.importer._find_existing_bug(self.cve, self.lp_cve)
)
+ self.checkLaunchpadCve(self.lp_cve, self.cve)
def test_import_cve_dry_run(self):
importer = UCTImporter(dry_run=True)
importer.import_cve(self.cve)
- self.assertIsNone(importer.find_existing_bug(self.cve, self.lp_cve))
+ self.assertIsNone(importer._find_existing_bug(self.cve, self.lp_cve))
def test_naive_date_made_public(self):
cve = self.cve
- cve.date_made_public = cve.date_made_public.replace(tzinfo=None)
+ cve.public_date = cve.public_date.replace(tzinfo=None)
bug = self.importer.create_bug(cve, self.lp_cve)
self.assertEqual(
UTC,
bug.vulnerabilities[0].date_made_public.tzinfo,
)
+
+ def test_make_cve_from_bug(self):
+ self.importer.import_cve(self.cve)
+ bug = self.importer._find_existing_bug(self.cve, self.lp_cve)
+ cve = self.exporter._make_cve_from_bug(bug)
+ self.assertEqual(self.cve.sequence, cve.sequence)
+ self.assertEqual(self.cve.crd, cve.crd)
+ self.assertEqual(self.cve.public_date, cve.public_date)
+ self.assertEqual(self.cve.public_date_at_USN, cve.public_date_at_USN)
+ self.assertListEqual(self.cve.distro_packages, cve.distro_packages)
+ self.assertListEqual(self.cve.series_packages, cve.series_packages)
+ self.assertEqual(self.cve.importance, cve.importance)
+ self.assertEqual(self.cve.status, cve.status)
+ self.assertEqual(self.cve.assignee, cve.assignee)
+ self.assertEqual(self.cve.discovered_by, cve.discovered_by)
+ self.assertEqual(self.cve.description, cve.description)
+ self.assertEqual(self.cve.ubuntu_description, cve.ubuntu_description)
+ self.assertListEqual(self.cve.bug_urls, cve.bug_urls)
+ self.assertListEqual(self.cve.references, cve.references)
+ self.assertEqual(self.cve.notes, cve.notes)
+ self.assertEqual(self.cve.mitigation, cve.mitigation)
+ self.assertListEqual(self.cve.cvss, cve.cvss)
diff --git a/lib/lp/bugs/scripts/uct/__init__.py b/lib/lp/bugs/scripts/uct/__init__.py
new file mode 100644
index 0000000..da0635b
--- /dev/null
+++ b/lib/lp/bugs/scripts/uct/__init__.py
@@ -0,0 +1,6 @@
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from .models import CVE, CVSS, UCTRecord
+from .uctexport import UCTExporter
+from .uctimport import UCTImporter, UCTImportError
diff --git a/lib/lp/bugs/scripts/uctimport.py b/lib/lp/bugs/scripts/uct/models.py
similarity index 54%
rename from lib/lp/bugs/scripts/uctimport.py
rename to lib/lp/bugs/scripts/uct/models.py
index 235ea0d..6c87f86 100644
--- a/lib/lp/bugs/scripts/uctimport.py
+++ b/lib/lp/bugs/scripts/uct/models.py
@@ -1,49 +1,21 @@
-# Copyright 2022 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
-"""A UCT (Ubuntu CVE Tracker) bug importer
-
-This code can import CVE summaries stored in UCT repository to bugs in
-Launchpad.
-
-For each entry in UCT we:
-
-1. Create a Bug instance
-2. Create a Vulnerability instance and link it to the bug (multiple
- Vulnerabilities may be created if the CVE entry covers multiple
- distributions)
-3. Create a Bug Task for each package/distro-series in the CVE entry
-4. Update the statuses of Bug Tasks based on the information in the CVE entry
-"""
import logging
-from datetime import datetime, timezone
+from collections import defaultdict
+from datetime import datetime
from enum import Enum
-from itertools import chain
from pathlib import Path
-from typing import Any, Dict, List, NamedTuple, Optional, Set, Tuple
+from typing import Any, Dict, List, NamedTuple, Optional, Set, Tuple, Union
+from typing.io import TextIO
import dateutil.parser
-import transaction
from contrib.cve_lib import load_cve
from zope.component import getUtility
-from lp.app.enums import InformationType
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.bugs.enums import VulnerabilityStatus
-from lp.bugs.interfaces.bug import CreateBugParams, IBugSet
-from lp.bugs.interfaces.bugactivity import IBugActivitySet
-from lp.bugs.interfaces.bugtask import (
- BugTaskImportance,
- BugTaskStatus,
- IBugTaskSet,
-)
-from lp.bugs.interfaces.bugwatch import IBugWatchSet
-from lp.bugs.interfaces.cve import ICveSet
-from lp.bugs.interfaces.vulnerability import IVulnerabilitySet
-from lp.bugs.model.bug import Bug as BugModel
-from lp.bugs.model.bugtask import BugTask
-from lp.bugs.model.cve import Cve as CveModel
-from lp.bugs.model.vulnerability import Vulnerability
+from lp.bugs.interfaces.bugtask import BugTaskImportance, BugTaskStatus
from lp.registry.interfaces.distroseries import IDistroSeriesSet
from lp.registry.interfaces.person import IPersonSet
from lp.registry.interfaces.series import SeriesStatus
@@ -55,23 +27,32 @@ from lp.registry.model.distributionsourcepackage import (
from lp.registry.model.distroseries import DistroSeries
from lp.registry.model.person import Person
from lp.registry.model.sourcepackage import SourcePackage
-from lp.services.database.constants import UTC_NOW
+from lp.registry.model.sourcepackagename import SourcePackageName
+from lp.services.propertycache import cachedproperty
__all__ = [
"CVE",
- "UCTImporter",
+ "CVSS",
"UCTRecord",
- "UCTImportError",
]
-from lp.services.propertycache import cachedproperty
+logger = logging.getLogger(__name__)
+
-logger = logging.getLogger("lp.bugs.scripts.import")
+CVSS = NamedTuple(
+ "CVSS",
+ (
+ ("authority", str),
+ ("vector_string", str),
+ ),
+)
class UCTRecord:
"""
- UCTRecord represents a single CVE record in the ubuntu-cve-tracker.
+ UCTRecord represents a single CVE record (file) in the ubuntu-cve-tracker.
+
+ It contains exactly the same information as a UCT CVE record.
"""
class Priority(Enum):
@@ -121,37 +102,33 @@ class UCTRecord:
),
)
- Note = NamedTuple(
- "Note",
- (
- ("author", str),
- ("text", str),
- ),
- )
-
def __init__(
self,
- path: Path,
+ parent_dir: str,
assigned_to: str,
bugs: List[str],
- cvss: List[Dict[str, Any]],
+ cvss: List[CVSS],
candidate: str,
- date_made_public: Optional[datetime],
+ crd: Optional[datetime],
+ public_date: Optional[datetime],
+ public_date_at_USN: Optional[datetime],
description: str,
discovered_by: str,
mitigation: Optional[str],
- notes: List[Note],
+ notes: str,
priority: Priority,
references: List[str],
ubuntu_description: str,
packages: List[Package],
):
- self.path = path
+ self.parent_dir = parent_dir
self.assigned_to = assigned_to
self.bugs = bugs
self.cvss = cvss
self.candidate = candidate
- self.date_made_public = date_made_public
+ self.crd = crd
+ self.public_date = public_date
+ self.public_date_at_USN = public_date_at_USN
self.description = description
self.discovered_by = discovered_by
self.mitigation = mitigation
@@ -167,18 +144,6 @@ class UCTRecord:
return self.__dict__ == other.__dict__
@classmethod
- def pop_cve_property(
- cls, cve_data: Dict[str, Any], field_name: str, required=True
- ) -> Optional[Any]:
- if required:
- value = cve_data.pop(field_name)
- else:
- value = cve_data.pop(field_name, None)
- if isinstance(value, str):
- return value.strip()
- return value
-
- @classmethod
def load(cls, cve_path: Path) -> "UCTRecord":
"""
Create a `UCTRecord` instance from a file located at `cve_path`.
@@ -192,20 +157,18 @@ class UCTRecord:
cve_data = load_cve(str(cve_path)) # type: Dict[str, Any]
packages = []
- tags = cls.pop_cve_property(
+ tags = cls._pop_cve_property(
cve_data, "tags"
) # type: Dict[str, Set[str]]
- patches = cls.pop_cve_property(
+ patches = cls._pop_cve_property(
cve_data, "patches"
) # type: Dict[str, List[Tuple[str, str]]]
- for package, statuses_dict in sorted(
- cls.pop_cve_property(cve_data, "pkgs").items()
- ):
+ for package, statuses_dict in cls._pop_cve_property(
+ cve_data, "pkgs"
+ ).items():
statuses = []
- for distroseries, (status, reason) in sorted(
- statuses_dict.items()
- ):
- distroseries_priority = cls.pop_cve_property(
+ for distroseries, (status, reason) in statuses_dict.items():
+ distroseries_priority = cls._pop_cve_property(
cve_data,
"Priority_{package}_{distroseries}".format(
package=package,
@@ -225,7 +188,7 @@ class UCTRecord:
),
)
)
- package_priority = cls.pop_cve_property(
+ package_priority = cls._pop_cve_property(
cve_data,
"Priority_{package}".format(package=package),
required=False,
@@ -247,47 +210,59 @@ class UCTRecord:
)
)
- crd = cls.pop_cve_property(cve_data, "CRD", required=False)
+ crd = cls._pop_cve_property(cve_data, "CRD", required=False)
if crd == "unknown":
crd = None
- public_date = cls.pop_cve_property(
+ public_date = cls._pop_cve_property(
cve_data, "PublicDate", required=False
)
if public_date == "unknown":
public_date = None
- public_date_at_USN = cls.pop_cve_property(
+ public_date_at_USN = cls._pop_cve_property(
cve_data, "PublicDateAtUSN", required=False
)
if public_date_at_USN == "unknown":
public_date_at_USN = None
- date_made_public = crd or public_date or public_date_at_USN
+ cvss = []
+ for cvss_dict in cls._pop_cve_property(cve_data, "CVSS"):
+ cvss.append(
+ CVSS(
+ authority=cvss_dict["source"],
+ vector_string="{} [{} {}]".format(
+ cvss_dict["vector"],
+ cvss_dict["baseScore"],
+ cvss_dict["baseSeverity"],
+ ),
+ )
+ )
entry = UCTRecord(
- path=cve_path,
- assigned_to=cls.pop_cve_property(cve_data, "Assigned-to"),
- bugs=cls.pop_cve_property(cve_data, "Bugs").split("\n"),
- cvss=cls.pop_cve_property(cve_data, "CVSS"),
- candidate=cls.pop_cve_property(cve_data, "Candidate"),
- date_made_public=(
- dateutil.parser.parse(date_made_public)
- if date_made_public
+ parent_dir=cve_path.absolute().parent.name,
+ assigned_to=cls._pop_cve_property(cve_data, "Assigned-to"),
+ bugs=cls._pop_cve_property(cve_data, "Bugs").split("\n"),
+ cvss=cvss,
+ candidate=cls._pop_cve_property(cve_data, "Candidate"),
+ crd=dateutil.parser.parse(crd) if crd else None,
+ public_date=(
+ dateutil.parser.parse(public_date) if public_date else None
+ ),
+ public_date_at_USN=(
+ dateutil.parser.parse(public_date_at_USN)
+ if public_date_at_USN
else None
),
- description=cls.pop_cve_property(cve_data, "Description"),
- discovered_by=cls.pop_cve_property(cve_data, "Discovered-by"),
- mitigation=cls.pop_cve_property(
+ description=cls._pop_cve_property(cve_data, "Description"),
+ discovered_by=cls._pop_cve_property(cve_data, "Discovered-by"),
+ mitigation=cls._pop_cve_property(
cve_data, "Mitigation", required=False
),
- notes=[
- cls.Note(author=author, text=text)
- for author, text in cls.pop_cve_property(cve_data, "Notes")
- ],
- priority=cls.Priority(cls.pop_cve_property(cve_data, "Priority")),
- references=cls.pop_cve_property(cve_data, "References").split(
+ notes=cls._format_notes(cls._pop_cve_property(cve_data, "Notes")),
+ priority=cls.Priority(cls._pop_cve_property(cve_data, "Priority")),
+ references=cls._pop_cve_property(cve_data, "References").split(
"\n"
),
- ubuntu_description=cls.pop_cve_property(
+ ubuntu_description=cls._pop_cve_property(
cve_data, "Ubuntu-Description"
),
packages=packages,
@@ -301,8 +276,147 @@ class UCTRecord:
return entry
+ def save(self, output_dir: Path) -> Path:
+ """
+ Save UCTRecord to a file in UCT format.
+ """
+ if not output_dir.is_dir():
+ raise ValueError(
+ "{} does not exist or is not a directory", output_dir
+ )
+ output_path = output_dir / self.parent_dir / self.candidate
+ output_path.parent.mkdir(exist_ok=True)
+ output = open(str(output_path), "w")
+ if self.public_date_at_USN:
+ self._write_field(
+ "PublicDateAtUSN",
+ self._format_datetime(self.public_date_at_USN),
+ output,
+ )
+ self._write_field("Candidate", self.candidate, output)
+ if self.crd:
+ self._write_field("CRD", self._format_datetime(self.crd), output)
+ if self.public_date:
+ self._write_field(
+ "PublicDate", self._format_datetime(self.public_date), output
+ )
+ self._write_field("References", self.references, output)
+ self._write_field("Description", self.description.split("\n"), output)
+ self._write_field(
+ "Ubuntu-Description", self.ubuntu_description.split("\n"), output
+ )
+ self._write_field("Notes", self.notes.split("\n"), output)
+ self._write_field(
+ "Mitigation",
+ self.mitigation.split("\n") if self.mitigation else "",
+ output,
+ )
+ self._write_field("Bugs", self.bugs, output)
+ self._write_field("Priority", self.priority.value, output)
+ self._write_field("Discovered-by", self.discovered_by, output)
+ self._write_field("Assigned-to", self.assigned_to, output)
+ self._write_field(
+ "CVSS",
+ [
+ "{authority}: {vector_string}".format(**c._asdict())
+ for c in self.cvss
+ ],
+ output,
+ )
+ for package in self.packages:
+ output.write("\n")
+ patches = [
+ "{}: {}".format(patch.patch_type, patch.entry)
+ for patch in package.patches
+ ]
+ self._write_field(
+ "Patches_{}".format(package.name), patches, output
+ )
+ for status in package.statuses:
+ self._write_field(
+ "{}_{}".format(status.distroseries, package.name),
+ (
+ "{} ({})".format(status.status.value, status.reason)
+ if status.reason
+ else status.status.value
+ ),
+ output,
+ )
+ if package.priority:
+ self._write_field(
+ "Priority_{}".format(package.name),
+ package.priority.value,
+ output,
+ )
+ for status in package.statuses:
+ if status.priority:
+ self._write_field(
+ "Priority_{}_{}".format(
+ package.name, status.distroseries
+ ),
+ status.priority.value,
+ output,
+ )
+
+ if package.tags:
+ self._write_field(
+ "Tags_{}".format(package.name),
+ " ".join(package.tags),
+ output,
+ )
+
+ output.close()
+ return output_path
+
+ @classmethod
+ def _pop_cve_property(
+ cls, cve_data: Dict[str, Any], field_name: str, required=True
+ ) -> Optional[Any]:
+ if required:
+ value = cve_data.pop(field_name)
+ else:
+ value = cve_data.pop(field_name, None)
+ if isinstance(value, str):
+ return value.strip()
+ return value
+
+ @classmethod
+ def _write_field(
+ cls, name: str, value: Union[str, List[str]], output: TextIO
+ ) -> None:
+ if isinstance(value, str):
+ if value:
+ output.write("{}: {}\n".format(name, value))
+ else:
+ output.write("{}:\n".format(name))
+ elif isinstance(value, list):
+ output.write("{}:\n".format(name))
+ for line in value:
+ output.write(" {}\n".format(line))
+ else:
+ raise AssertionError()
+
+ @classmethod
+ def _format_datetime(cls, dt: datetime) -> str:
+ return dt.strftime("%Y-%m-%d %H:%M:%S %Z")
+
+ @classmethod
+ def _format_notes(cls, notes: List[Tuple[str, str]]) -> str:
+ lines = []
+ for author, text in notes:
+ note_lines = text.split("\n")
+ lines.append("{}> {}".format(author, note_lines[0]))
+ for line in note_lines[1:]:
+ lines.append(" " + line)
+ return "\n".join(lines)
+
class CVE:
+ """
+ `CVE` represents UCT CVE information mapped to Launchpad data structures.
+
+ Do not confuse this with `Cve` database model.
+ """
DistroPackage = NamedTuple(
"DistroPackage",
@@ -330,6 +444,7 @@ class CVE:
UCTRecord.Priority.UNTRIAGED: BugTaskImportance.UNDECIDED,
UCTRecord.Priority.NEGLIGIBLE: BugTaskImportance.WISHLIST,
}
+ PRIORITY_MAP_REVERSE = {v: k for k, v in PRIORITY_MAP.items()}
BUG_TASK_STATUS_MAP = {
UCTRecord.PackageStatus.IGNORED: BugTaskStatus.WONTFIX,
@@ -342,52 +457,64 @@ class CVE:
UCTRecord.PackageStatus.NEEDED: BugTaskStatus.NEW,
UCTRecord.PackageStatus.PENDING: BugTaskStatus.FIXCOMMITTED,
}
+ BUG_TASK_STATUS_MAP_REVERSE = {
+ v: k for k, v in BUG_TASK_STATUS_MAP.items()
+ }
VULNERABILITY_STATUS_MAP = {
"active": VulnerabilityStatus.ACTIVE,
"ignored": VulnerabilityStatus.IGNORED,
"retired": VulnerabilityStatus.RETIRED,
}
+ VULNERABILITY_STATUS_MAP_REVERSE = {
+ v: k for k, v in VULNERABILITY_STATUS_MAP.items()
+ }
def __init__(
self,
sequence: str,
- date_made_public: Optional[datetime],
+ crd: Optional[datetime],
+ public_date: Optional[datetime],
+ public_date_at_USN: Optional[datetime],
distro_packages: List[DistroPackage],
series_packages: List[SeriesPackage],
importance: BugTaskImportance,
status: VulnerabilityStatus,
assignee: Optional[Person],
+ discovered_by: str,
description: str,
ubuntu_description: str,
bug_urls: List[str],
references: List[str],
notes: str,
mitigation: str,
+ cvss: List[CVSS],
):
self.sequence = sequence
- self.date_made_public = date_made_public
+ self.crd = crd
+ self.public_date = public_date
+ self.public_date_at_USN = public_date_at_USN
self.distro_packages = distro_packages
self.series_packages = series_packages
self.importance = importance
self.status = status
self.assignee = assignee
+ self.discovered_by = discovered_by
self.description = description
self.ubuntu_description = ubuntu_description
self.bug_urls = bug_urls
self.references = references
self.notes = notes
self.mitigation = mitigation
+ self.cvss = cvss
@classmethod
def make_from_uct_record(cls, uct_record: UCTRecord) -> "CVE":
- # Some `UCTRecord` fields are not being used at the moment:
- # - cve.discovered_by: This is supposed to be `Cve.discoverer` but
- # there may be a difficulty there since the `Cve` table should only
- # be managed by syncing data from MITRE and not from
- # ubuntu-cve-tracker
- # - cve.cvss: `Cve.cvss`, but may have a similar issue to
- # `Cve.discoverer` as above.
+ """
+ Create a `CVE` from a `UCTRecord`
+
+ This maps UCT CVE information to Launchpad data structures.
+ """
distro_packages = []
series_packages = []
@@ -396,10 +523,9 @@ class CVE:
for uct_package in uct_record.packages:
source_package_name = spn_set.getOrCreateByName(uct_package.name)
- package_priority = uct_package.priority or uct_record.priority
package_importance = (
- cls.PRIORITY_MAP[package_priority]
- if package_priority
+ cls.PRIORITY_MAP[uct_package.priority]
+ if uct_package.priority
else None
)
@@ -427,12 +553,9 @@ class CVE:
if distro_package not in distro_packages:
distro_packages.append(distro_package)
- series_package_priority = (
- uct_package_status.priority or package_priority
- )
series_package_importance = (
- cls.PRIORITY_MAP[series_package_priority]
- if series_package_priority
+ cls.PRIORITY_MAP[uct_package_status.priority]
+ if uct_package_status.priority
else None
)
@@ -461,20 +584,107 @@ class CVE:
return cls(
sequence=uct_record.candidate,
- date_made_public=uct_record.date_made_public,
+ crd=uct_record.crd,
+ public_date=uct_record.public_date,
+ public_date_at_USN=uct_record.public_date_at_USN,
distro_packages=distro_packages,
series_packages=series_packages,
importance=cls.PRIORITY_MAP[uct_record.priority],
status=cls.infer_vulnerability_status(uct_record),
assignee=assignee,
+ discovered_by=uct_record.discovered_by,
description=uct_record.description,
ubuntu_description=uct_record.ubuntu_description,
bug_urls=uct_record.bugs,
references=uct_record.references,
- notes=cls.format_cve_notes(uct_record.notes),
+ notes=uct_record.notes,
mitigation=uct_record.mitigation,
+ cvss=uct_record.cvss,
+ )
+
+ def to_uct_record(self) -> UCTRecord:
+ """
+ Convert a `CVE` to a `UCTRecord`.
+
+ This maps Launchpad data structures to the format that UCT understands.
+ """
+ series_packages_by_name = defaultdict(
+ list
+ ) # type: Dict[SourcePackageName, List[CVE.SeriesPackage]]
+ for series_package in self.series_packages:
+ series_packages_by_name[
+ series_package.package.sourcepackagename
+ ].append(series_package)
+
+ packages = [] # type: List[UCTRecord.Package]
+ processed_packages = set() # type: Set[SourcePackageName]
+ for distro_package in self.distro_packages:
+ spn = distro_package.package.sourcepackagename
+ if spn in processed_packages:
+ continue
+ processed_packages.add(spn)
+ statuses = [] # type: List[UCTRecord.DistroSeriesPackageStatus]
+ for series_package in series_packages_by_name[spn]:
+ series = series_package.package.distroseries
+ if series.status == SeriesStatus.DEVELOPMENT:
+ series_name = "devel"
+ else:
+ series_name = series.name
+ statuses.append(
+ UCTRecord.DistroSeriesPackageStatus(
+ distroseries=series_name,
+ status=self.BUG_TASK_STATUS_MAP_REVERSE[
+ series_package.status
+ ],
+ reason=series_package.status_explanation,
+ priority=(
+ self.PRIORITY_MAP_REVERSE[
+ series_package.importance
+ ]
+ if series_package.importance
+ else None
+ ),
+ )
+ )
+
+ packages.append(
+ UCTRecord.Package(
+ name=spn.name,
+ statuses=statuses,
+ priority=(
+ self.PRIORITY_MAP_REVERSE[distro_package.importance]
+ if distro_package.importance
+ else None
+ ),
+ tags=set(),
+ patches=[],
+ )
+ )
+ return UCTRecord(
+ parent_dir=self.VULNERABILITY_STATUS_MAP_REVERSE.get(
+ self.status, ""
+ ),
+ assigned_to=self.assignee.name if self.assignee else "",
+ bugs=self.bug_urls,
+ cvss=self.cvss,
+ candidate=self.sequence,
+ crd=self.crd,
+ public_date=self.public_date,
+ public_date_at_USN=self.public_date_at_USN,
+ description=self.description,
+ discovered_by=self.discovered_by,
+ mitigation=self.mitigation,
+ notes=self.notes,
+ priority=self.PRIORITY_MAP_REVERSE[self.importance],
+ references=self.references,
+ ubuntu_description=self.ubuntu_description,
+ packages=packages,
)
+ @property
+ def date_made_public(self):
+ return self.crd or self.public_date_at_USN or self.public_date
+
@cachedproperty
def affected_distributions(self) -> Set[Distribution]:
return {p.package.distribution for p in self.distro_packages}
@@ -490,16 +700,8 @@ class CVE:
"""
Infer vulnerability status based on the parent folder of the CVE file.
"""
- cve_folder_name = uct_record.path.absolute().parent.name
return cls.VULNERABILITY_STATUS_MAP.get(
- cve_folder_name, VulnerabilityStatus.NEEDS_TRIAGE
- )
-
- @classmethod
- def format_cve_notes(cls, notes: List[UCTRecord.Note]) -> str:
- return "\n".join(
- "{author}> {text}".format(author=note.author, text=note.text)
- for note in notes
+ uct_record.parent_dir, VulnerabilityStatus.NEEDS_TRIAGE
)
@classmethod
@@ -537,237 +739,3 @@ class CVE:
"Could not find the distro series: %s", distro_series_name
)
return distro_series
-
-
-class UCTImportError(Exception):
- pass
-
-
-class UCTImporter:
- def __init__(self, dry_run=False):
- self.dry_run = dry_run
- self.bug_importer = getUtility(ILaunchpadCelebrities).bug_importer
-
- def import_cve_from_file(self, cve_path: Path) -> None:
- uct_record = UCTRecord.load(cve_path)
- cve = CVE.make_from_uct_record(uct_record)
- self.import_cve(cve)
-
- def import_cve(self, cve: CVE) -> None:
- if cve.date_made_public is None:
- logger.warning(
- "The CVE does not have a publication date, is it embargoed?"
- )
- return
- lp_cve = getUtility(ICveSet)[cve.sequence] # type: CveModel
- if lp_cve is None:
- logger.warning("Could not find the CVE in LP: %s", cve.sequence)
- return
- bug = self.find_existing_bug(cve, lp_cve)
- try:
- if bug is None:
- self.create_bug(cve, lp_cve)
- else:
- self.update_bug(bug, cve, lp_cve)
- except Exception:
- transaction.abort()
- raise
-
- if self.dry_run:
- logger.info("Dry-run mode enabled, all changes are reverted.")
- transaction.abort()
- else:
- transaction.commit()
-
- def find_existing_bug(
- self, cve: CVE, lp_cve: CveModel
- ) -> Optional[BugModel]:
- bug = None
- for vulnerability in lp_cve.vulnerabilities:
- if vulnerability.distribution in cve.affected_distributions:
- bugs = vulnerability.bugs
- if bugs:
- if bug and bugs[0] != bug:
- raise UCTImportError(
- "Multiple existing bugs are found "
- "for CVE {}".format(cve.sequence)
- )
- else:
- bug = bugs[0]
- return bug
-
- def create_bug(self, cve: CVE, lp_cve: CveModel) -> Optional[BugModel]:
-
- logger.debug("creating bug...")
-
- if not cve.series_packages:
- logger.warning("Could not find any affected packages")
- return None
-
- distro_package = cve.distro_packages[0]
-
- # Create the bug
- bug = getUtility(IBugSet).createBug(
- CreateBugParams(
- comment=self.make_bug_description(cve),
- title=cve.sequence,
- information_type=InformationType.PUBLICSECURITY,
- owner=self.bug_importer,
- target=distro_package.package,
- importance=distro_package.importance,
- cve=lp_cve,
- )
- ) # type: BugModel
-
- self.update_external_bug_urls(bug, cve.bug_urls)
-
- logger.info("Created bug with ID: %s", bug.id)
-
- self.create_bug_tasks(
- bug, cve.distro_packages[1:], cve.series_packages
- )
- self.update_statuses_and_importances(
- bug, cve.distro_packages[1:], cve.series_packages
- )
- self.assign_bug_tasks(bug, cve.assignee)
-
- # Make a note of the import in the activity log:
- getUtility(IBugActivitySet).new(
- bug=bug.id,
- datechanged=UTC_NOW,
- person=self.bug_importer,
- whatchanged="bug",
- message="UCT CVE entry {}".format(cve.sequence),
- )
-
- # Create the Vulnerabilities
- for distribution in cve.affected_distributions:
- self.create_vulnerability(bug, cve, lp_cve, distribution)
-
- return bug
-
- def update_bug(self, bug: BugModel, cve: CVE, lp_cve: CveModel) -> None:
- bug.description = self.make_bug_description(cve)
-
- self.create_bug_tasks(bug, cve.distro_packages, cve.series_packages)
- self.update_statuses_and_importances(
- bug, cve.distro_packages, cve.series_packages
- )
- self.assign_bug_tasks(bug, cve.assignee)
- self.update_external_bug_urls(bug, cve.bug_urls)
-
- # Update or add new Vulnerabilities
- vulnerabilities_by_distro = {
- v.distribution: v for v in bug.vulnerabilities
- }
- for distro in cve.affected_distributions:
- vulnerability = vulnerabilities_by_distro.get(distro)
- if vulnerability is None:
- self.create_vulnerability(bug, cve, lp_cve, distro)
- else:
- self.update_vulnerability(vulnerability, cve)
-
- def create_bug_tasks(
- self,
- bug: BugModel,
- distro_packages: List[CVE.DistroPackage],
- series_packages: List[CVE.SeriesPackage],
- ) -> None:
- bug_tasks = bug.bugtasks # type: List[BugTask]
- bug_task_by_target = {t.target: t for t in bug_tasks}
- bug_task_set = getUtility(IBugTaskSet)
- for target in (
- p.package for p in chain(distro_packages, series_packages)
- ):
- if target not in bug_task_by_target:
- bug_task_set.createTask(bug, self.bug_importer, target)
-
- def create_vulnerability(
- self,
- bug: BugModel,
- cve: CVE,
- lp_cve: CveModel,
- distribution: Distribution,
- ) -> Vulnerability:
- date_made_public = cve.date_made_public
- if date_made_public.tzinfo is None:
- date_made_public = date_made_public.replace(tzinfo=timezone.utc)
- vulnerability = getUtility(IVulnerabilitySet).new(
- distribution=distribution,
- creator=bug.owner,
- cve=lp_cve,
- status=cve.status,
- description=cve.description,
- notes=cve.notes,
- mitigation=cve.mitigation,
- importance=cve.importance,
- information_type=InformationType.PUBLICSECURITY,
- date_made_public=date_made_public,
- ) # type: Vulnerability
-
- vulnerability.linkBug(bug, bug.owner)
-
- logger.info("Created vulnerability with ID: %s", vulnerability.id)
-
- return vulnerability
-
- def update_vulnerability(
- self, vulnerability: Vulnerability, cve: CVE
- ) -> None:
- vulnerability.status = cve.status
- vulnerability.description = cve.description
- vulnerability.notes = cve.notes
- vulnerability.mitigation = cve.mitigation
- vulnerability.importance = cve.importance
- vulnerability.date_made_public = cve.date_made_public
-
- def assign_bug_tasks(
- self, bug: BugModel, assignee: Optional[Person]
- ) -> None:
- for bug_task in bug.bugtasks:
- bug_task.transitionToAssignee(assignee, validate=False)
-
- def update_statuses_and_importances(
- self,
- bug: BugModel,
- distro_packages: List[CVE.DistroPackage],
- series_packages: List[CVE.SeriesPackage],
- ) -> None:
- bug_tasks = bug.bugtasks # type: List[BugTask]
- bug_task_by_target = {t.target: t for t in bug_tasks}
-
- for dp in distro_packages:
- task = bug_task_by_target[dp.package]
- if task.importance != dp.importance:
- task.transitionToImportance(dp.importance, self.bug_importer)
-
- for sp in series_packages:
- task = bug_task_by_target[sp.package]
- if task.importance != sp.importance:
- task.transitionToImportance(sp.importance, self.bug_importer)
- if task.status != sp.status:
- task.transitionToStatus(sp.status, self.bug_importer)
- if task.status_explanation != sp.status_explanation:
- task.status_explanation = sp.status_explanation
-
- def update_external_bug_urls(
- self, bug: BugModel, bug_urls: List[str]
- ) -> None:
- bug_urls = set(bug_urls)
- for watch in bug.watches:
- if watch.url in bug_urls:
- bug_urls.remove(watch.url)
- else:
- watch.destroySelf()
- bug_watch_set = getUtility(IBugWatchSet)
- for external_bug_url in bug_urls:
- bug_watch_set.fromText(external_bug_url, bug, self.bug_importer)
-
- def make_bug_description(self, cve: CVE) -> str:
- parts = [cve.description]
- if cve.ubuntu_description:
- parts.extend(["", "Ubuntu-Description:", cve.ubuntu_description])
- if cve.references:
- parts.extend(["", "References:"])
- parts.extend(cve.references)
- return "\n".join(parts)
diff --git a/lib/lp/bugs/scripts/uct/uctexport.py b/lib/lp/bugs/scripts/uct/uctexport.py
new file mode 100644
index 0000000..757e94f
--- /dev/null
+++ b/lib/lp/bugs/scripts/uct/uctexport.py
@@ -0,0 +1,200 @@
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import logging
+from collections import defaultdict
+from pathlib import Path
+from typing import List, NamedTuple, Optional
+
+from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
+
+from lp.bugs.interfaces.bug import IBugSet
+from lp.bugs.model.bug import Bug as BugModel
+from lp.bugs.model.bugtask import BugTask
+from lp.bugs.model.cve import Cve as CveModel
+from lp.bugs.model.vulnerability import Vulnerability
+from lp.registry.model.distributionsourcepackage import (
+ DistributionSourcePackage,
+)
+from lp.registry.model.sourcepackage import SourcePackage
+
+from .models import CVE, CVSS
+
+__all__ = [
+ "UCTExporter",
+]
+
+
+logger = logging.getLogger(__name__)
+
+
+class UCTExporter:
+ """
+ `UCTExporter` is used to export LP Bugs, Vulnerabilities and Cve's to
+ UCT CVE files.
+ """
+
+ ParsedDescription = NamedTuple(
+ "ParsedDescription",
+ (
+ ("description", str),
+ ("references", List[str]),
+ ),
+ )
+
+ def export_bug_to_uct_file(
+ self, bug_id: int, output_dir: Path
+ ) -> Optional[Path]:
+ """
+ Export a bug with the given bug_id as a
+ UCT CVE record file in the `output_dir`
+
+ :param bug_id: ID of the `Bug` model to be exported
+ :param output_dir: the directory where the exported file will be stored
+ :return: path to the exported file
+ """
+ bug = getUtility(IBugSet).get(bug_id)
+ if not bug:
+ logger.error("Could not find a bug with ID: %s", bug_id)
+ return
+ cve = self._make_cve_from_bug(bug)
+ uct_record = cve.to_uct_record()
+ save_to_path = uct_record.save(output_dir)
+ logger.info(
+ "Bug with ID: %s is exported to: %s", bug_id, str(save_to_path)
+ )
+ return save_to_path
+
+ def _make_cve_from_bug(self, bug: BugModel) -> CVE:
+ """
+ Create a `CVE` instances from a `Bug` model and the related
+ Vulnerabilities and `Cve`.
+
+ `BugTasks` are converted to `CVE.DistroPackage` and `CVE.SEriesPackage`
+ objects.
+
+ Other `CVE` fields are populated from the information contained in the
+ `Bug`, its related Vulnerabilities and LP `Cve` model.
+
+ :param bug: `Bug` model
+ :return: `CVE` instance
+ """
+ vulnerabilities = list(bug.vulnerabilities)
+ if not vulnerabilities:
+ raise ValueError(
+ "Bug with ID: {} does not have vulnerabilities".format(bug.id)
+ )
+ vulnerability = vulnerabilities[0] # type: Vulnerability
+ if not vulnerability.cve:
+ raise ValueError(
+ "Bug with ID: {} - vulnerability "
+ "is not linked to a CVE".format(bug.id)
+ )
+ lp_cve = vulnerability.cve # type: CveModel
+
+ parsed_description = self._parse_bug_description(bug.description)
+
+ bug_urls = []
+ for bug_watch in bug.watches:
+ bug_urls.append(bug_watch.url)
+
+ bug_tasks = list(bug.bugtasks) # type: List[BugTask]
+
+ cve_importance = vulnerability.importance
+ package_importances = {}
+
+ distro_packages = []
+ for bug_task in bug_tasks:
+ target = removeSecurityProxy(bug_task.target)
+ if not isinstance(target, DistributionSourcePackage):
+ continue
+ dp_importance = bug_task.importance
+ package_importances[target.sourcepackagename] = dp_importance
+ distro_packages.append(
+ CVE.DistroPackage(
+ package=target,
+ importance=(
+ dp_importance
+ if dp_importance != cve_importance
+ else None
+ ),
+ )
+ )
+
+ series_packages = []
+ for bug_task in bug_tasks:
+ target = removeSecurityProxy(bug_task.target)
+ if not isinstance(target, SourcePackage):
+ continue
+ sp_importance = bug_task.importance
+ package_importance = package_importances[target.sourcepackagename]
+ series_packages.append(
+ CVE.SeriesPackage(
+ package=target,
+ importance=(
+ sp_importance
+ if sp_importance != package_importance
+ else None
+ ),
+ status=bug_task.status,
+ status_explanation=bug_task.status_explanation,
+ )
+ )
+
+ return CVE(
+ sequence="CVE-{}".format(lp_cve.sequence),
+ crd=None, # TODO: fix this
+ public_date=vulnerability.date_made_public,
+ public_date_at_USN=None, # TODO: fix this
+ distro_packages=distro_packages,
+ series_packages=series_packages,
+ importance=cve_importance,
+ status=vulnerability.status,
+ assignee=bug_tasks[0].assignee,
+ discovered_by="", # TODO: fix this
+ description=parsed_description.description,
+ ubuntu_description=vulnerability.description,
+ bug_urls=bug_urls,
+ references=parsed_description.references,
+ notes=vulnerability.notes,
+ mitigation=vulnerability.mitigation,
+ cvss=[
+ CVSS(
+ authority=authority,
+ vector_string=vector_string,
+ )
+ for authority, vector_string in lp_cve.cvss.items()
+ ],
+ )
+
+ def _parse_bug_description(
+ self, bug_description: str
+ ) -> "ParsedDescription":
+ """
+ Some `CVE` fields can't be mapped to Launchpad models.
+ They are saved to bug description.
+
+ This method extract those fields from the bug description.
+
+ :param bug_description: bug description
+ :return: parsed description
+ """
+ field_values = defaultdict(list)
+ current_field = "description"
+ known_fields = {
+ "References:": "references",
+ }
+ lines = bug_description.split("\n")
+ for line in lines:
+ line = line.strip()
+ if not line:
+ continue
+ if line in known_fields:
+ current_field = known_fields[line]
+ continue
+ field_values[current_field].append(line)
+ return UCTExporter.ParsedDescription(
+ description="\n".join(field_values.get("description", [])),
+ references=field_values.get("references", []),
+ )
diff --git a/lib/lp/bugs/scripts/uct/uctimport.py b/lib/lp/bugs/scripts/uct/uctimport.py
new file mode 100644
index 0000000..4228a86
--- /dev/null
+++ b/lib/lp/bugs/scripts/uct/uctimport.py
@@ -0,0 +1,420 @@
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""A UCT (Ubuntu CVE Tracker) bug importer
+
+This code can import CVE summaries stored in UCT repository to bugs in
+Launchpad.
+
+For each entry in UCT we:
+
+1. Create a Bug instance
+2. Create a Vulnerability instance for each affected distribution and link it
+ to the bug
+3. Create a Bug Task for each distribution/series package in the CVE entry
+4. Update the statuses of Bug Tasks based on the information in the CVE entry
+5. Update the information the related Launchpad's `Cve` model, if necessary
+"""
+import logging
+from datetime import timezone
+from itertools import chain
+from pathlib import Path
+from typing import List, Optional
+
+import transaction
+from zope.component import getUtility
+
+from lp.app.enums import InformationType
+from lp.app.interfaces.launchpad import ILaunchpadCelebrities
+from lp.bugs.interfaces.bug import CreateBugParams, IBugSet
+from lp.bugs.interfaces.bugactivity import IBugActivitySet
+from lp.bugs.interfaces.bugtask import BugTaskImportance, IBugTaskSet
+from lp.bugs.interfaces.bugwatch import IBugWatchSet
+from lp.bugs.interfaces.cve import ICveSet
+from lp.bugs.interfaces.vulnerability import IVulnerabilitySet
+from lp.bugs.model.bug import Bug as BugModel
+from lp.bugs.model.bugtask import BugTask
+from lp.bugs.model.cve import Cve as CveModel
+from lp.bugs.model.vulnerability import Vulnerability
+from lp.registry.model.distribution import Distribution
+from lp.registry.model.person import Person
+from lp.services.database.constants import UTC_NOW
+
+from .models import CVE, UCTRecord
+
+__all__ = [
+ "UCTImporter",
+ "UCTImportError",
+]
+
+logger = logging.getLogger(__name__)
+
+
+class UCTImportError(Exception):
+ pass
+
+
+class UCTImporter:
+ """
+ `UCTImporter` is used to import UCT CVE files to Launchpad database.
+ """
+
+ def __init__(self, dry_run=False):
+ self.dry_run = dry_run
+ self.bug_importer = getUtility(ILaunchpadCelebrities).bug_importer
+
+ def import_cve_from_file(self, cve_path: Path) -> None:
+ """
+ Import a UCT CVE record from a file located at `cve_path`.
+
+ :param cve_path: path to the UCT CVE file
+ """
+ logger.info("Importing %s", cve_path)
+ uct_record = UCTRecord.load(cve_path)
+ cve = CVE.make_from_uct_record(uct_record)
+ self.import_cve(cve)
+ logger.info("%s was imported successfully", cve_path)
+
+ def import_cve(self, cve: CVE) -> None:
+ """
+ Import a `CVE` instance to Launchpad database.
+
+ :param cve: `CVE` with information from UCT
+ """
+ if cve.date_made_public is None:
+ logger.warning(
+ "%s does not have a publication date, "
+ "is it embargoed? Aborting.",
+ cve.sequence,
+ )
+ return
+ if not cve.series_packages:
+ logger.warning(
+ "%s: could not find any affected packages, aborting.",
+ cve.series_packages,
+ )
+ return
+ lp_cve = getUtility(ICveSet)[cve.sequence] # type: CveModel
+ if lp_cve is None:
+ logger.warning(
+ "%s: could not find the CVE in LP. Aborting.", cve.sequence
+ )
+ return
+ bug = self._find_existing_bug(cve, lp_cve)
+ try:
+ if bug is None:
+ bug = self.create_bug(cve, lp_cve)
+ logger.info(
+ "%s: created bug with ID: %s", cve.sequence, bug.id
+ )
+ else:
+ logging.info(
+ "%s: found existing bug with ID: %s",
+ cve.sequence,
+ bug.id,
+ )
+ self.update_bug(bug, cve, lp_cve)
+ logger.info(
+ "%s: updated bug with ID: %s", cve.sequence, bug.id
+ )
+ self._update_launchpad_cve(lp_cve, cve)
+ except Exception:
+ transaction.abort()
+ raise
+
+ if self.dry_run:
+ logger.info(
+ "%s: dry-run mode enabled, all changes are reverted.",
+ cve.sequence,
+ )
+ transaction.abort()
+ else:
+ transaction.commit()
+
+ def create_bug(self, cve: CVE, lp_cve: CveModel) -> BugModel:
+ """
+ Create a `Bug` model based on the information contained in a `CVE`.
+
+ :param cve: `CVE` with information from UCT
+ :param lp_cve: Launchpad `Cve` model
+ """
+
+ distro_package = cve.distro_packages[0]
+
+ # Create the bug
+ bug = getUtility(IBugSet).createBug(
+ CreateBugParams(
+ comment=self._make_bug_description(cve),
+ title=cve.sequence,
+ information_type=InformationType.PUBLICSECURITY,
+ owner=self.bug_importer,
+ target=distro_package.package,
+ importance=distro_package.importance,
+ cve=lp_cve,
+ )
+ ) # type: BugModel
+
+ self._update_external_bug_urls(bug, cve.bug_urls)
+
+ self._create_bug_tasks(
+ bug, cve.distro_packages[1:], cve.series_packages
+ )
+ self._update_statuses_and_importances(
+ bug, cve.importance, cve.distro_packages, cve.series_packages
+ )
+ self._assign_bug_tasks(bug, cve.assignee)
+
+ # Make a note of the import in the activity log:
+ getUtility(IBugActivitySet).new(
+ bug=bug.id,
+ datechanged=UTC_NOW,
+ person=self.bug_importer,
+ whatchanged="bug",
+ message="UCT CVE entry {}".format(cve.sequence),
+ )
+
+ # Create the Vulnerabilities
+ for distribution in cve.affected_distributions:
+ self._create_vulnerability(bug, cve, lp_cve, distribution)
+
+ return bug
+
+ def update_bug(self, bug: BugModel, cve: CVE, lp_cve: CveModel) -> None:
+ """
+ Update a `Bug` model with the information contained in a `CVE`.
+
+ :param bug: `Bug` model to be updated
+ :param cve: `CVE` with information from UCT
+ :param lp_cve: Launchpad `Cve` model
+ """
+ bug.description = self._make_bug_description(cve)
+
+ self._create_bug_tasks(bug, cve.distro_packages, cve.series_packages)
+ self._update_statuses_and_importances(
+ bug, cve.importance, cve.distro_packages, cve.series_packages
+ )
+ self._assign_bug_tasks(bug, cve.assignee)
+ self._update_external_bug_urls(bug, cve.bug_urls)
+
+ # Update or add new Vulnerabilities
+ vulnerabilities_by_distro = {
+ v.distribution: v for v in bug.vulnerabilities
+ }
+ for distro in cve.affected_distributions:
+ vulnerability = vulnerabilities_by_distro.get(distro)
+ if vulnerability is None:
+ self._create_vulnerability(bug, cve, lp_cve, distro)
+ else:
+ self._update_vulnerability(vulnerability, cve)
+
+ def _find_existing_bug(
+ self, cve: CVE, lp_cve: CveModel
+ ) -> Optional[BugModel]:
+ bug = None
+ for vulnerability in lp_cve.vulnerabilities:
+ if vulnerability.distribution in cve.affected_distributions:
+ bugs = vulnerability.bugs
+ if bugs:
+ if bug and bugs[0] != bug:
+ raise UCTImportError(
+ "Multiple existing bugs are found "
+ "for CVE {}".format(cve.sequence)
+ )
+ else:
+ bug = bugs[0]
+ return bug
+
+ def _create_bug_tasks(
+ self,
+ bug: BugModel,
+ distro_packages: List[CVE.DistroPackage],
+ series_packages: List[CVE.SeriesPackage],
+ ) -> None:
+ """
+ Add bug tasks to the given `Bug` model based on the information
+ from a `CVE`.
+
+ `distro_packages` and `series_packages` from the `CVE`
+ are used as bug task targets.
+
+ This may be called multiple times, only new targets will be processed.
+
+ :param bug: `Bug` model to be updated
+ :param distro_packages: list of `DistroPackage`s from a `CVE`
+ :param series_packages: list of `SeriesPackage`s from a `CVE`
+ """
+ bug_tasks = bug.bugtasks # type: List[BugTask]
+ bug_task_by_target = {t.target: t for t in bug_tasks}
+ bug_task_set = getUtility(IBugTaskSet)
+ for target in (
+ p.package for p in chain(distro_packages, series_packages)
+ ):
+ if target not in bug_task_by_target:
+ bug_task_set.createTask(bug, self.bug_importer, target)
+
+ def _create_vulnerability(
+ self,
+ bug: BugModel,
+ cve: CVE,
+ lp_cve: CveModel,
+ distribution: Distribution,
+ ) -> Vulnerability:
+ """
+ Create a Vulnerability instance based on the information from
+ the given `CVE` instance and link to the specified `Bug`
+ and LP's `Cve` model.
+
+ :param bug: `Bug` model associated with the vulnerability
+ :param cve: `CVE` with information from UCT
+ :param lp_cve: Launchpad `Cve` model
+ :param distribution: a `Distribution` affected by the vulnerability
+ :return: a Vulnerability
+ """
+ date_made_public = cve.date_made_public
+ if date_made_public.tzinfo is None:
+ date_made_public = date_made_public.replace(tzinfo=timezone.utc)
+ vulnerability = getUtility(IVulnerabilitySet).new(
+ distribution=distribution,
+ creator=bug.owner,
+ cve=lp_cve,
+ status=cve.status,
+ description=cve.ubuntu_description,
+ notes=cve.notes,
+ mitigation=cve.mitigation,
+ importance=cve.importance,
+ information_type=InformationType.PUBLICSECURITY,
+ date_made_public=date_made_public,
+ ) # type: Vulnerability
+
+ vulnerability.linkBug(bug, bug.owner)
+
+ logger.info(
+ "%s: created vulnerability with ID: %s for distribution: %s",
+ cve.sequence,
+ vulnerability.id,
+ distribution.name,
+ )
+
+ return vulnerability
+
+ def _update_vulnerability(
+ self, vulnerability: Vulnerability, cve: CVE
+ ) -> None:
+ """
+ Update a `Vulnerability` model with the information
+ contained in a `CVE`
+
+ :param vulnerability: `Vulnerability` model to be updated
+ :param cve: `CVE` with information from UCT
+ """
+ vulnerability.status = cve.status
+ vulnerability.description = cve.ubuntu_description
+ vulnerability.notes = cve.notes
+ vulnerability.mitigation = cve.mitigation
+ vulnerability.importance = cve.importance
+ vulnerability.date_made_public = cve.date_made_public
+
+ def _assign_bug_tasks(
+ self, bug: BugModel, assignee: Optional[Person]
+ ) -> None:
+ """
+ Assign all bug tasks from the given bug to the given assignee.
+
+ :param bug: `Bug` model to be updated
+ :param assignee: a person to be assigned (may be None)
+ """
+ for bug_task in bug.bugtasks:
+ bug_task.transitionToAssignee(assignee, validate=False)
+
+ def _update_statuses_and_importances(
+ self,
+ bug: BugModel,
+ cve_importance: BugTaskImportance,
+ distro_packages: List[CVE.DistroPackage],
+ series_packages: List[CVE.SeriesPackage],
+ ) -> None:
+ """
+ Update statuses and importances of bug tasks according to the
+ information contained in `CVE`.
+
+ If a distro package doesn't have importance information, the
+ `cve_importance` is used instead.
+
+ If a series package doesn't have importance information, the
+ importance of the corresponding distro package is used instead.
+
+ :param bug: `Bug` model to be updated
+ :param cve_importance: `CVE` importance
+ :param distro_packages: list of `DistroPackage`s from a `CVE`
+ :param series_packages: list of `SeriesPackage`s from a `CVE`
+ """
+ bug_tasks = bug.bugtasks # type: List[BugTask]
+ bug_task_by_target = {t.target: t for t in bug_tasks}
+
+ package_importances = {}
+
+ for dp in distro_packages:
+ task = bug_task_by_target[dp.package]
+ dp_importance = dp.importance or cve_importance
+ package_importances[dp.package.sourcepackagename] = dp_importance
+ if task.importance != dp_importance:
+ task.transitionToImportance(dp_importance, self.bug_importer)
+
+ for sp in series_packages:
+ task = bug_task_by_target[sp.package]
+ package_importance = package_importances[
+ sp.package.sourcepackagename
+ ]
+ sp_importance = sp.importance or package_importance
+ if task.importance != sp_importance:
+ task.transitionToImportance(sp_importance, self.bug_importer)
+ if task.status != sp.status:
+ task.transitionToStatus(sp.status, self.bug_importer)
+ if task.status_explanation != sp.status_explanation:
+ task.status_explanation = sp.status_explanation
+
+ def _update_external_bug_urls(
+ self, bug: BugModel, bug_urls: List[str]
+ ) -> None:
+ """
+ Save links to external bug trackers as bug watches.
+
+ :param bug: `Bug` model to be updated
+ :param bug_urls: links to external bug trackers
+ """
+ bug_urls = set(bug_urls)
+ for watch in bug.watches:
+ if watch.url in bug_urls:
+ bug_urls.remove(watch.url)
+ else:
+ watch.destroySelf()
+ bug_watch_set = getUtility(IBugWatchSet)
+ for external_bug_url in bug_urls:
+ bug_watch_set.fromText(external_bug_url, bug, self.bug_importer)
+
+ def _make_bug_description(self, cve: CVE) -> str:
+ """
+ Some `CVE` fields can't be mapped to Launchpad models.
+
+ They are saved to bug description.
+
+ :param cve: `CVE` with information from UCT
+ :return: bug description
+ """
+ parts = [cve.description]
+ if cve.references:
+ parts.extend(["", "References:"])
+ parts.extend(cve.references)
+ return "\n".join(parts)
+
+ def _update_launchpad_cve(self, lp_cve: CveModel, cve: CVE) -> None:
+ """
+ Update LP's `Cve` model based on the information contained in `CVE`.
+
+ :param lp_cve: LP's `CVE` model to be updated
+ :param cve: `CVE` with information from UCT
+ """
+ for cvss in cve.cvss:
+ lp_cve.setCVSSVectorForAuthority(
+ cvss.authority, cvss.vector_string
+ )
diff --git a/scripts/uct-export.py b/scripts/uct-export.py
new file mode 100755
index 0000000..d65d224
--- /dev/null
+++ b/scripts/uct-export.py
@@ -0,0 +1,34 @@
+#!/usr/bin/python3 -S
+#
+# Copyright 2022 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+import _pythonpath # noqa: F401
+
+import logging
+from pathlib import Path
+
+from lp.bugs.scripts.uct import UCTExporter
+from lp.services.scripts.base import LaunchpadScript
+
+
+class UCTExportScript(LaunchpadScript):
+
+ usage = "usage: %prog [options] BUG_ID OUTPUT_DIR"
+ description = "Export bugs from to CVE entries used in ubuntu-cve-tracker."
+ loglevel = logging.INFO
+
+ def main(self):
+ if len(self.args) != 2:
+ self.parser.error(
+ "Please specify the bug ID and the output directory."
+ )
+
+ bug_id, output_dir = self.args
+
+ exporter = UCTExporter()
+ exporter.export_bug_to_uct_file(int(bug_id), Path(output_dir))
+
+
+if __name__ == "__main__":
+ script = UCTExportScript("lp.services.scripts.uctexport")
+ script.run()
diff --git a/scripts/uct-import.py b/scripts/uct-import.py
index 549f8d4..92d62d3 100755
--- a/scripts/uct-import.py
+++ b/scripts/uct-import.py
@@ -7,12 +7,13 @@ import _pythonpath # noqa: F401
import logging
from pathlib import Path
-from lp.bugs.scripts.uctimport import UCTImporter
+from lp.bugs.scripts.uct import UCTImporter
from lp.services.scripts.base import LaunchpadScript
class UCTImportScript(LaunchpadScript):
+ usage = "usage: %prog [options] CVE_PATH"
description = (
"Import bugs into Launchpad from CVE entries in ubuntu-cve-tracker."
)
diff --git a/setup.cfg b/setup.cfg
index 806401a..c348ed5 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -211,6 +211,9 @@ ignore =
# operators, at least for now.
W503,
W504
+per-file-ignores =
+ # Ignore unused imports in `__init__` files
+ */__init__.py: F401
[isort]
# database/* have some implicit relative imports.
Follow ups
