Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 29 additions & 4 deletions product_portfolio/importers.py
Original file line number Diff line number Diff line change
Expand Up @@ -713,8 +713,11 @@ def update_scancode_project(self):

class ImportPackageFromScanCodeIO:
"""
Creates, and assign to a product, packages in Dejacode from a ScanCode.io project
discovered packages.
Import packages discovered by a ScanCode.io project and assign them to a product.

For each package, associated vulnerabilities are imported and linked, including
reachability data when available.
Dependencies can optionally be imported as well.
"""

unique_together_fields = [
Expand Down Expand Up @@ -810,21 +813,43 @@ def import_vulnerability(vulnerability_data, product_package):
if not vulnerabilities:
return

vulnerability = vulnerabilities[0]

if cdx_vulnerability := vulnerability_data.get("cdx_vulnerability_data"):
if analysis_data := cdx_vulnerability.get("analysis"):
# CycloneDX model uses "response" while the local model uses "response"
# CycloneDX model uses "response" while the local model uses "responses"
if response_value := analysis_data.pop("response", None):
analysis_data["responses"] = response_value

VulnerabilityAnalysis.create_from_data(
user=product_package.dataspace,
data={
"product_package": product_package,
"vulnerability": vulnerabilities[0],
"vulnerability": vulnerability,
**analysis_data,
},
)

# Import reachability from the "symbol reachability analysis" scan when available.
is_reachable_raw = vulnerability_data.get("is_reachable")
is_reachable = None
if is_reachable_raw == "yes":
is_reachable = True
elif is_reachable_raw == "no":
is_reachable = False

if is_reachable is not None:
analysis, created = VulnerabilityAnalysis.objects.get_or_create(
product_package=product_package,
vulnerability=vulnerability,
dataspace=product_package.dataspace,
defaults={"is_reachable": is_reachable},
)
if not created and analysis.is_reachable is None:
VulnerabilityAnalysis.objects.filter(pk=analysis.pk).update(
is_reachable=is_reachable
)

def import_package(self, package_data):
# Vulnerabilities are assigned after the package creation.
affected_by_vulnerabilities = package_data.pop("affected_by_vulnerabilities", [])
Expand Down
157 changes: 157 additions & 0 deletions product_portfolio/tests/test_importers.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
from product_portfolio.models import ProductPackage
from product_portfolio.models import ProductRelationStatus
from product_portfolio.models import ScanCodeProject
from vulnerabilities.models import VulnerabilityAnalysis


class ProductRelationImporterTestCase(TestCase):
Expand Down Expand Up @@ -1414,3 +1415,159 @@ def test_product_portfolio_import_packages_from_scio_importer_vex(
self.assertEqual("code_not_present", analysis.justification)
self.assertEqual("AAAA", analysis.detail)
self.assertEqual(["can_not_fix", "update"], analysis.responses)

@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
def test_product_portfolio_import_packages_from_scio_importer_is_reachable(
self, mock_fetch_packages, mock_fetch_dependencies
):
def make_vulnerability_entry(advisory_id, is_reachable):
return {
"advisory_uid": f"github_osv/{advisory_id}",
"summary": "A vulnerability",
"is_reachable": is_reachable,
"cdx_vulnerability_data": {
"analysis": {"state": "in_triage", "detail": "Under review"},
},
}

mock_fetch_packages.return_value = [
{
"purl": "pkg:maven/abc/abc@1.0",
"type": "maven",
"namespace": "abc",
"name": "abc",
"version": "1.0",
"affected_by_vulnerabilities": [
make_vulnerability_entry("GHSA-yes", "yes"),
make_vulnerability_entry("GHSA-no", "no"),
make_vulnerability_entry("GHSA-unknown", "unknown"),
],
}
]
mock_fetch_dependencies.return_value = []

importer = ImportPackageFromScanCodeIO(
user=self.super_user,
project_uuid=uuid.uuid4(),
product=self.product1,
)
importer.save()

yes_analysis = VulnerabilityAnalysis.objects.get(
vulnerability__advisory_uid="github_osv/GHSA-yes"
)
no_analysis = VulnerabilityAnalysis.objects.get(
vulnerability__advisory_uid="github_osv/GHSA-no"
)
unknown_analysis = VulnerabilityAnalysis.objects.get(
vulnerability__advisory_uid="github_osv/GHSA-unknown"
)

self.assertTrue(yes_analysis.is_reachable)
self.assertFalse(no_analysis.is_reachable)
self.assertIsNone(unknown_analysis.is_reachable)

@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
def test_product_portfolio_import_packages_from_scio_importer_is_reachable_not_overwritten(
self, mock_fetch_packages, mock_fetch_dependencies
):
mock_fetch_packages.return_value = [
{
"purl": "pkg:maven/abc/abc@1.0",
"type": "maven",
"namespace": "abc",
"name": "abc",
"version": "1.0",
"affected_by_vulnerabilities": [
{
"advisory_uid": "github_osv/GHSA-existing",
"summary": "A vulnerability",
"is_reachable": "no",
"cdx_vulnerability_data": {
"analysis": {"state": "in_triage", "detail": "Under review"},
},
}
],
}
]
mock_fetch_dependencies.return_value = []

importer = ImportPackageFromScanCodeIO(
user=self.super_user,
project_uuid=uuid.uuid4(),
product=self.product1,
)
importer.save()

analysis = VulnerabilityAnalysis.objects.get(
vulnerability__advisory_uid="github_osv/GHSA-existing"
)
self.assertFalse(analysis.is_reachable)

# A second import with a conflicting value must not overwrite the existing one.
# Reassign return_value because import_package pops "affected_by_vulnerabilities".
mock_fetch_packages.return_value = [
{
"purl": "pkg:maven/abc/abc@1.0",
"type": "maven",
"namespace": "abc",
"name": "abc",
"version": "1.0",
"affected_by_vulnerabilities": [
{
"advisory_uid": "github_osv/GHSA-existing",
"summary": "A vulnerability",
"is_reachable": "yes",
}
],
}
]
importer2 = ImportPackageFromScanCodeIO(
user=self.super_user,
project_uuid=uuid.uuid4(),
product=self.product1,
)
importer2.save()

analysis.refresh_from_db()
self.assertFalse(analysis.is_reachable)

@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
@mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
def test_product_portfolio_import_packages_from_scio_importer_is_reachable_without_cdx(
self, mock_fetch_packages, mock_fetch_dependencies
):
# When cdx_vulnerability_data is absent, a minimal VulnerabilityAnalysis is still
# created to record the is_reachable value from the scan.
mock_fetch_packages.return_value = [
{
"purl": "pkg:maven/abc/abc@1.0",
"type": "maven",
"namespace": "abc",
"name": "abc",
"version": "1.0",
"affected_by_vulnerabilities": [
{
"advisory_uid": "github_osv/GHSA-no-cdx",
"summary": "A vulnerability",
"is_reachable": "yes",
}
],
}
]
mock_fetch_dependencies.return_value = []

importer = ImportPackageFromScanCodeIO(
user=self.super_user,
project_uuid=uuid.uuid4(),
product=self.product1,
)
importer.save()

analysis = VulnerabilityAnalysis.objects.get(
vulnerability__advisory_uid="github_osv/GHSA-no-cdx"
)
self.assertTrue(analysis.is_reachable)
self.assertFalse(analysis.state)
4 changes: 0 additions & 4 deletions product_portfolio/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -2989,10 +2989,6 @@ def apply_analysis_preset_view(request, productpackage_uuid, advisory_uid, prese
dataspace=dataspace,
)
preset.apply_to_analysis(analysis)

if not analysis.has_content_fields():
return JsonResponse({"error": "This preset has no content fields to apply."}, status=400)

analysis.applied_by_preset = preset
analysis.save()

Expand Down
28 changes: 9 additions & 19 deletions vulnerabilities/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -353,18 +353,6 @@ class Response(models.TextChoices):
),
)

def has_content_fields(self):
return any([self.state, self.justification, self.responses, self.detail])

def save(self, *args, **kwargs):
# At least one of those fields must be provided.
if not self.has_content_fields():
raise ValueError(
"At least one of state, justification, responses or detail must be provided."
)

super().save(*args, **kwargs)

class Meta:
abstract = True

Expand All @@ -385,13 +373,15 @@ class Meta:
abstract = True

def as_cyclonedx(self):
state = None
if self.state:
state = cdx_vulnerability.ImpactAnalysisState(self.state)

justification = None
if self.justification:
justification = cdx_vulnerability.ImpactAnalysisJustification(self.justification)
if not any([self.state, self.justification, self.responses, self.detail]):
return None

state = cdx_vulnerability.ImpactAnalysisState(self.state) if self.state else None
justification = (
cdx_vulnerability.ImpactAnalysisJustification(self.justification)
if self.justification
else None
)

return cdx_vulnerability.VulnerabilityAnalysis(
state=state,
Expand Down
7 changes: 1 addition & 6 deletions vulnerabilities/tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -327,13 +327,8 @@ def test_vulnerability_model_vulnerability_analysis_save(self):
product_package=product_package1,
vulnerability=vulnerability1,
dataspace=self.dataspace,
state=VulnerabilityAnalysis.State.RESOLVED,
)

msg = "At least one of state, justification, responses or detail must be provided."
with self.assertRaisesMessage(ValueError, msg):
analysis.save()

analysis.state = VulnerabilityAnalysis.State.RESOLVED
analysis.save()

# Refresh from db
Expand Down
2 changes: 0 additions & 2 deletions vulnerabilities/triage/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,8 +106,6 @@ def apply_preset_for_vulnerabilities(preset, product, vulnerability_ids):
dataspace_id=product.dataspace_id,
)
preset.apply_to_analysis(analysis)
if not analysis.has_content_fields():
continue # Preset has no content fields - cannot save a new analysis
else:
analysis = existing
preset.apply_to_analysis(analysis)
Expand Down
7 changes: 7 additions & 0 deletions vulnerabilities/triage/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,13 @@ class Meta:
def __str__(self):
return self.name

def save(self, *args, **kwargs):
if not any([self.state, self.justification, self.responses, self.detail]):
raise ValueError(
"At least one of state, justification, responses or detail must be provided."
)
super().save(*args, **kwargs)

def apply_to_analysis(self, analysis):
"""Copy non-blank preset fields onto the analysis instance (does not save)."""
for field_name in ("state", "justification", "responses", "detail"):
Expand Down
8 changes: 0 additions & 8 deletions vulnerabilities/triage/tests/test_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,14 +145,6 @@ def test_updates_an_existing_preset_owned_analysis(self):
self.assertEqual(second_preset, analysis.applied_by_preset)
self.assertEqual(1, VulnerabilityAnalysis.objects.count())

def test_skips_creation_when_the_preset_has_no_content_field_set(self):
# An AnalysisPreset always requires at least one content field to be saved (see
# VulnerabilityAnalysisContentMixin.save), so this can only happen with an in-memory
# preset. This exercises the defensive guard against saving a content-less analysis.
content_less_preset = AnalysisPreset(dataspace=self.dataspace, is_reachable=True)
apply_preset_for_vulnerabilities(content_less_preset, self.product, [self.vulnerability.pk])
self.assertFalse(VulnerabilityAnalysis.objects.exists())

def test_does_nothing_when_no_product_package_carries_the_vulnerability(self):
other_package = make_package(self.dataspace)
other_vulnerability = make_vulnerability(self.dataspace, affecting=other_package)
Expand Down
4 changes: 1 addition & 3 deletions vulnerabilities/triage/tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,7 @@ def setUp(self):
self.dataspace = Dataspace.objects.create(name="nexB")

def test_save_requires_at_least_one_content_field(self):
# AnalysisPreset shares its `save` validation with VulnerabilityAnalysis through
# VulnerabilityAnalysisContentMixin: a preset that only sets `is_reachable` has no
# content to apply and must be rejected the same way a bare analysis would be.
# A preset that carries no content fields is useless: it has nothing to apply.
preset = AnalysisPreset(dataspace=self.dataspace, name="No content", is_reachable=True)
with self.assertRaises(ValueError):
preset.save()
Expand Down
Loading