From c7ecab95b8396f46c90cfdaea10262d9e094e65f Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 14:09:45 +0400 Subject: [PATCH 01/19] add detected_date on AffectedByVulnerabilityRelationship Signed-off-by: tdruez --- ...ected_date_to_affected_by_vulnerability.py | 43 +++++++++++++++++++ vulnerabilities/models.py | 8 ++++ 2 files changed, 51 insertions(+) create mode 100644 component_catalog/migrations/0014_add_detected_date_to_affected_by_vulnerability.py diff --git a/component_catalog/migrations/0014_add_detected_date_to_affected_by_vulnerability.py b/component_catalog/migrations/0014_add_detected_date_to_affected_by_vulnerability.py new file mode 100644 index 00000000..a790c3f6 --- /dev/null +++ b/component_catalog/migrations/0014_add_detected_date_to_affected_by_vulnerability.py @@ -0,0 +1,43 @@ +# Generated by Django 6.0 on 2026-07-20 + +import django.utils.timezone +from django.db import migrations +from django.db import models + + +class Migration(migrations.Migration): + + dependencies = [ + ("component_catalog", "0013_package_package_content"), + ] + + operations = [ + migrations.AddField( + model_name="componentaffectedbyvulnerability", + name="detected_date", + field=models.DateTimeField( + auto_now_add=True, + default=django.utils.timezone.now, + help_text=( + "Date and time when this vulnerability was first detected on this object. " + "Used to measure how long a vulnerability has remained unaddressed. " + "Defaults to the time the record was created." + ), + ), + preserve_default=False, + ), + migrations.AddField( + model_name="packageaffectedbyvulnerability", + name="detected_date", + field=models.DateTimeField( + auto_now_add=True, + default=django.utils.timezone.now, + help_text=( + "Date and time when this vulnerability was first detected on this object. " + "Used to measure how long a vulnerability has remained unaddressed. " + "Defaults to the time the record was created." + ), + ), + preserve_default=False, + ), + ] diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py index 0c82ee21..b0d25a84 100644 --- a/vulnerabilities/models.py +++ b/vulnerabilities/models.py @@ -393,6 +393,14 @@ class AffectedByVulnerabilityRelationship(DataspacedModel): to="vulnerabilities.Vulnerability", on_delete=models.CASCADE, ) + detected_date = models.DateTimeField( + auto_now_add=True, + help_text=_( + "Date and time when this vulnerability was first detected on this object. " + "Used to measure how long a vulnerability has remained unaddressed. " + "Defaults to the time the record was created." + ), + ) class Meta: abstract = True From 5ee2ceac6002f5ea0c18dc90bc760af3f9d8d401 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 14:19:41 +0400 Subject: [PATCH 02/19] add new VulnerabilityDetectedRule Signed-off-by: tdruez --- policy/rules.py | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/policy/rules.py b/policy/rules.py index bc89931c..ca4051ce 100644 --- a/policy/rules.py +++ b/policy/rules.py @@ -6,8 +6,12 @@ # See https://aboutcode.org for more information about AboutCode FOSS projects. # + from django.apps import apps +# VulnerabilityAnalysis states that indicate a vulnerability has been triaged and addressed. +TERMINAL_VULNERABILITY_STATES = ["resolved", "resolved_with_pedigree", "not_affected"] + class BaseRule: """Base class for policy rule handlers.""" @@ -100,10 +104,39 @@ class LicenseCoverageGapRule(PackageBaseRule): package_filter = {"license_expression": ""} +class VulnerabilityDetectedRule(BaseRule): + rule_type = "vulnerability_detected" + label = "Vulnerability Detected" + severity = "error" + description = "Detects packages with at least one known vulnerability (non-null risk score)." + parameters_schema = { + "min_risk_score": "Minimum risk score (0.0-10.0). Default: any vulnerability.", + } + + def get_package_filter(self): + return {"package__risk_score__isnull": False} + + def count_violations(self, product, threshold, parameters): + Package = apps.get_model("component_catalog", "package") + + packages = Package.objects.filter( + productpackages__product=product, + risk_score__isnull=False, + ) + + min_risk_score = parameters.get("min_risk_score") + if min_risk_score is not None: + packages = packages.filter(risk_score__gte=min_risk_score) + + count = packages.count() + return count if count > threshold else 0 + + RULE_REGISTRY = { UsagePolicyErrorRule.rule_type: UsagePolicyErrorRule(), UsagePolicyWarningRule.rule_type: UsagePolicyWarningRule(), LicensePolicyErrorRule.rule_type: LicensePolicyErrorRule(), LicensePolicyWarningRule.rule_type: LicensePolicyWarningRule(), LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(), + VulnerabilityDetectedRule.rule_type: VulnerabilityDetectedRule(), } From e0018cc4adcc40498946115adcfe3141b6aba863 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 15:48:48 +0400 Subject: [PATCH 03/19] add UnresolvedVulnerabilityCountRule Signed-off-by: tdruez --- policy/rules.py | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/policy/rules.py b/policy/rules.py index ca4051ce..63594c0c 100644 --- a/policy/rules.py +++ b/policy/rules.py @@ -8,6 +8,8 @@ from django.apps import apps +from django.db.models import Exists +from django.db.models import OuterRef # VulnerabilityAnalysis states that indicate a vulnerability has been triaged and addressed. TERMINAL_VULNERABILITY_STATES = ["resolved", "resolved_with_pedigree", "not_affected"] @@ -132,6 +134,41 @@ def count_violations(self, product, threshold, parameters): return count if count > threshold else 0 +class UnresolvedVulnerabilityCountRule(BaseRule): + rule_type = "unresolved_vulnerability_count" + label = "Unresolved Vulnerability Count" + severity = "warning" + description = ( + "Counts individual package-vulnerability links in the product that have not been " + "addressed via a VulnerabilityAnalysis with a terminal state " + "(resolved, resolved_with_pedigree, or not_affected)." + ) + + def count_violations(self, product, threshold, parameters): + PackageAffectedByVulnerability = apps.get_model( + "component_catalog", "packageaffectedbyvulnerability" + ) + VulnerabilityAnalysis = apps.get_model("vulnerabilities", "vulnerabilityanalysis") + + terminal_analysis = VulnerabilityAnalysis.objects.filter( + product=product, + state__in=TERMINAL_VULNERABILITY_STATES, + package=OuterRef("package"), + vulnerability=OuterRef("vulnerability"), + ) + + count = ( + PackageAffectedByVulnerability.objects.filter( + package__productpackages__product=product, + ) + .annotate(has_terminal_analysis=Exists(terminal_analysis)) + .filter(has_terminal_analysis=False) + .distinct() + .count() + ) + return count if count > threshold else 0 + + RULE_REGISTRY = { UsagePolicyErrorRule.rule_type: UsagePolicyErrorRule(), UsagePolicyWarningRule.rule_type: UsagePolicyWarningRule(), @@ -139,4 +176,5 @@ def count_violations(self, product, threshold, parameters): LicensePolicyWarningRule.rule_type: LicensePolicyWarningRule(), LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(), VulnerabilityDetectedRule.rule_type: VulnerabilityDetectedRule(), + UnresolvedVulnerabilityCountRule.rule_type: UnresolvedVulnerabilityCountRule(), } From c7e97657e2f5b70a1b7a4e12e7a97ea9fefe962f Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 16:00:40 +0400 Subject: [PATCH 04/19] add Stale Vulnerability rule Signed-off-by: tdruez --- policy/rules.py | 55 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/policy/rules.py b/policy/rules.py index 63594c0c..7ccd3d7e 100644 --- a/policy/rules.py +++ b/policy/rules.py @@ -6,10 +6,12 @@ # See https://aboutcode.org for more information about AboutCode FOSS projects. # +from datetime import timedelta from django.apps import apps from django.db.models import Exists from django.db.models import OuterRef +from django.utils import timezone # VulnerabilityAnalysis states that indicate a vulnerability has been triaged and addressed. TERMINAL_VULNERABILITY_STATES = ["resolved", "resolved_with_pedigree", "not_affected"] @@ -169,6 +171,58 @@ def count_violations(self, product, threshold, parameters): return count if count > threshold else 0 +class StaleVulnerabilityRule(BaseRule): + rule_type = "stale_vulnerability" + label = "Stale Vulnerability" + severity = "error" + description = ( + "Detects packages with a high-risk vulnerability that has remained unaddressed " + "beyond a configured delay. The delay is measured from the detected_date on the " + "package-vulnerability link, i.e. when the vulnerability was first imported for " + "that specific package." + ) + parameters_schema = { + "max_days": ( + "Maximum number of days a vulnerability may remain unaddressed before flagging. " + "Default: 30." + ), + "min_risk_score": ( + "Only consider vulnerabilities with at least this risk score. Default: 8.0." + ), + } + + def count_violations(self, product, threshold, parameters): + PackageAffectedByVulnerability = apps.get_model( + "component_catalog", "packageaffectedbyvulnerability" + ) + VulnerabilityAnalysis = apps.get_model("vulnerabilities", "vulnerabilityanalysis") + + max_days = parameters.get("max_days", 30) + min_risk_score = parameters.get("min_risk_score", 8.0) + cutoff_date = timezone.now() - timedelta(days=max_days) + + terminal_analysis = VulnerabilityAnalysis.objects.filter( + product=product, + state__in=TERMINAL_VULNERABILITY_STATES, + package=OuterRef("package"), + vulnerability=OuterRef("vulnerability"), + ) + + count = ( + PackageAffectedByVulnerability.objects.filter( + package__productpackages__product=product, + vulnerability__risk_score__gte=min_risk_score, + detected_date__lte=cutoff_date, + ) + .annotate(has_terminal_analysis=Exists(terminal_analysis)) + .filter(has_terminal_analysis=False) + .values("package_id") + .distinct() + .count() + ) + return count if count > threshold else 0 + + RULE_REGISTRY = { UsagePolicyErrorRule.rule_type: UsagePolicyErrorRule(), UsagePolicyWarningRule.rule_type: UsagePolicyWarningRule(), @@ -177,4 +231,5 @@ def count_violations(self, product, threshold, parameters): LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(), VulnerabilityDetectedRule.rule_type: VulnerabilityDetectedRule(), UnresolvedVulnerabilityCountRule.rule_type: UnresolvedVulnerabilityCountRule(), + StaleVulnerabilityRule.rule_type: StaleVulnerabilityRule(), } From f55e9bd0d3d62c58b7ea8d548ed029f98f5bfbe5 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 17:33:59 +0400 Subject: [PATCH 05/19] add unit test for the new rules Signed-off-by: tdruez --- dje/models.py | 1 + policy/tests/test_rules.py | 250 +++++++++++++++++++++++++++++++++++++ 2 files changed, 251 insertions(+) create mode 100644 policy/tests/test_rules.py diff --git a/dje/models.py b/dje/models.py index 996195ce..259a980b 100644 --- a/dje/models.py +++ b/dje/models.py @@ -1143,6 +1143,7 @@ def get_exclude_candidates_fields(self): field.related_model is Dataspace, isinstance(field, models.AutoField), isinstance(field, models.UUIDField), + getattr(field, "auto_now_add", False) or getattr(field, "auto_now", False), not field.null and not field.blank and not field.has_default(), field.name in ALWAYS_EXCLUDE, ] diff --git a/policy/tests/test_rules.py b/policy/tests/test_rules.py new file mode 100644 index 00000000..596f70b4 --- /dev/null +++ b/policy/tests/test_rules.py @@ -0,0 +1,250 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + +from datetime import timedelta + +from django.contrib.contenttypes.models import ContentType +from django.test import TestCase +from django.utils import timezone + +from component_catalog.models import Package +from component_catalog.models import PackageAffectedByVulnerability +from component_catalog.tests import make_package +from dje.models import Dataspace +from license_library.models import License +from license_library.tests import make_license +from policy.models import UsagePolicy +from policy.rules import LicenseCoverageGapRule +from policy.rules import LicensePolicyErrorRule +from policy.rules import LicensePolicyWarningRule +from policy.rules import StaleVulnerabilityRule +from policy.rules import UnresolvedVulnerabilityCountRule +from policy.rules import UsagePolicyErrorRule +from policy.rules import UsagePolicyWarningRule +from policy.rules import VulnerabilityDetectedRule +from product_portfolio.tests import make_product +from product_portfolio.tests import make_product_package +from vulnerabilities.tests import make_vulnerability +from vulnerabilities.tests import make_vulnerability_analysis + + +class UsagePolicyErrorRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + self.policy = UsagePolicy.objects.create( + label="Prohibited", + icon="icon", + content_type=ContentType.objects.get_for_model(Package), + compliance_alert=UsagePolicy.Compliance.ERROR, + dataspace=self.dataspace, + ) + + def test_counts_packages_with_error_usage_policy(self): + package = make_package(self.dataspace, usage_policy=self.policy) + make_product_package(self.product, package=package) + make_product_package(self.product, package=make_package(self.dataspace)) + count = UsagePolicyErrorRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_returns_zero_when_no_matching_packages(self): + make_product_package(self.product, package=make_package(self.dataspace)) + count = UsagePolicyErrorRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + +class UsagePolicyWarningRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + self.policy = UsagePolicy.objects.create( + label="Restricted", + icon="icon", + content_type=ContentType.objects.get_for_model(Package), + compliance_alert=UsagePolicy.Compliance.WARNING, + dataspace=self.dataspace, + ) + + def test_counts_packages_with_warning_usage_policy(self): + package = make_package(self.dataspace, usage_policy=self.policy) + make_product_package(self.product, package=package) + count = UsagePolicyWarningRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + +class LicensePolicyErrorRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + self.policy = UsagePolicy.objects.create( + label="Prohibited", + icon="icon", + content_type=ContentType.objects.get_for_model(License), + compliance_alert=UsagePolicy.Compliance.ERROR, + dataspace=self.dataspace, + ) + + def test_counts_packages_with_license_error_policy(self): + license_ = make_license(self.dataspace, key="gpl-3.0", usage_policy=self.policy) + package = make_package(self.dataspace) + package.licenses.add(license_, through_defaults={"dataspace": self.dataspace}) + make_product_package(self.product, package=package) + count = LicensePolicyErrorRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_returns_zero_when_license_has_no_policy(self): + license_ = make_license(self.dataspace, key="mit") + package = make_package(self.dataspace) + package.licenses.add(license_, through_defaults={"dataspace": self.dataspace}) + make_product_package(self.product, package=package) + count = LicensePolicyErrorRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + +class LicensePolicyWarningRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + self.policy = UsagePolicy.objects.create( + label="Restricted", + icon="icon", + content_type=ContentType.objects.get_for_model(License), + compliance_alert=UsagePolicy.Compliance.WARNING, + dataspace=self.dataspace, + ) + + def test_counts_packages_with_license_warning_policy(self): + license_ = make_license(self.dataspace, key="lgpl-2.1", usage_policy=self.policy) + package = make_package(self.dataspace) + package.licenses.add(license_, through_defaults={"dataspace": self.dataspace}) + make_product_package(self.product, package=package) + count = LicensePolicyWarningRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + +class LicenseCoverageGapRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + + def test_counts_packages_with_no_license_expression(self): + make_product_package( + self.product, package=make_package(self.dataspace, license_expression="") + ) + make_product_package( + self.product, package=make_package(self.dataspace, license_expression="mit") + ) + count = LicenseCoverageGapRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_returns_zero_when_all_packages_have_license(self): + make_product_package( + self.product, package=make_package(self.dataspace, license_expression="mit") + ) + count = LicenseCoverageGapRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + +class VulnerabilityDetectedRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + + def test_counts_packages_with_any_vulnerability(self): + package = make_package(self.dataspace) + make_vulnerability(self.dataspace, affecting=package, risk_score=5.0) + make_product_package(self.product, package=package) + make_product_package(self.product, package=make_package(self.dataspace)) + count = VulnerabilityDetectedRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_min_risk_score_excludes_packages_below_threshold(self): + package = make_package(self.dataspace) + make_vulnerability(self.dataspace, affecting=package, risk_score=4.0) + make_product_package(self.product, package=package) + count = VulnerabilityDetectedRule().count_violations( + self.product, 0, {"min_risk_score": 7.0} + ) + self.assertEqual(0, count) + + def test_min_risk_score_includes_packages_at_or_above_threshold(self): + package = make_package(self.dataspace) + make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + make_product_package(self.product, package=package) + count = VulnerabilityDetectedRule().count_violations( + self.product, 0, {"min_risk_score": 7.0} + ) + self.assertEqual(1, count) + + +class UnresolvedVulnerabilityCountRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + + def test_counts_unanalyzed_package_vulnerability_links(self): + package = make_package(self.dataspace) + make_vulnerability(self.dataspace, affecting=package) + make_product_package(self.product, package=package) + count = UnresolvedVulnerabilityCountRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_does_not_count_links_with_terminal_analysis(self): + package = make_package(self.dataspace) + vulnerability = make_vulnerability(self.dataspace, affecting=package) + product_package = make_product_package(self.product, package=package) + make_vulnerability_analysis(product_package, vulnerability, state="resolved") + count = UnresolvedVulnerabilityCountRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + +class StaleVulnerabilityRuleTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.product = make_product(self.dataspace) + + def test_counts_old_unresolved_high_risk_links(self): + package = make_package(self.dataspace) + vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + make_product_package(self.product, package=package) + old_date = timezone.now() - timedelta(days=60) + PackageAffectedByVulnerability.objects.filter( + package=package, vulnerability=vulnerability + ).update(detected_date=old_date) + count = StaleVulnerabilityRule().count_violations(self.product, 0, {}) + self.assertEqual(1, count) + + def test_does_not_count_recent_links(self): + package = make_package(self.dataspace) + make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + make_product_package(self.product, package=package) + count = StaleVulnerabilityRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + def test_does_not_count_links_below_min_risk_score(self): + package = make_package(self.dataspace) + vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=3.0) + make_product_package(self.product, package=package) + old_date = timezone.now() - timedelta(days=60) + PackageAffectedByVulnerability.objects.filter( + package=package, vulnerability=vulnerability + ).update(detected_date=old_date) + count = StaleVulnerabilityRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) + + def test_does_not_count_stale_links_with_terminal_analysis(self): + package = make_package(self.dataspace) + vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + product_package = make_product_package(self.product, package=package) + old_date = timezone.now() - timedelta(days=60) + PackageAffectedByVulnerability.objects.filter( + package=package, vulnerability=vulnerability + ).update(detected_date=old_date) + make_vulnerability_analysis(product_package, vulnerability, state="resolved") + count = StaleVulnerabilityRule().count_violations(self.product, 0, {}) + self.assertEqual(0, count) From cb8f916e65c105f3d3eddefab7a74ab2862cec0b Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 17:39:46 +0400 Subject: [PATCH 06/19] add model migration Signed-off-by: tdruez --- ...ctaffectedbyvulnerability_detected_date.py | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 product_portfolio/migrations/0019_productaffectedbyvulnerability_detected_date.py diff --git a/product_portfolio/migrations/0019_productaffectedbyvulnerability_detected_date.py b/product_portfolio/migrations/0019_productaffectedbyvulnerability_detected_date.py new file mode 100644 index 00000000..74e8ee2e --- /dev/null +++ b/product_portfolio/migrations/0019_productaffectedbyvulnerability_detected_date.py @@ -0,0 +1,20 @@ +# Generated by Django 6.0.6 on 2026-07-22 13:35 + +import django.utils.timezone +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('product_portfolio', '0018_productpolicyviolation'), + ] + + operations = [ + migrations.AddField( + model_name='productaffectedbyvulnerability', + name='detected_date', + field=models.DateTimeField(auto_now_add=True, default=django.utils.timezone.now, help_text='Date and time when this vulnerability was first detected on this object. Used to measure how long a vulnerability has remained unaddressed. Defaults to the time the record was created.'), + preserve_default=False, + ), + ] From d46f510a114883fb34defc9771bdd58d4bd09bf1 Mon Sep 17 00:00:00 2001 From: tdruez Date: Wed, 22 Jul 2026 17:47:25 +0400 Subject: [PATCH 07/19] simplify descriptions Signed-off-by: tdruez --- policy/rules.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/policy/rules.py b/policy/rules.py index 7ccd3d7e..961b406c 100644 --- a/policy/rules.py +++ b/policy/rules.py @@ -141,9 +141,7 @@ class UnresolvedVulnerabilityCountRule(BaseRule): label = "Unresolved Vulnerability Count" severity = "warning" description = ( - "Counts individual package-vulnerability links in the product that have not been " - "addressed via a VulnerabilityAnalysis with a terminal state " - "(resolved, resolved_with_pedigree, or not_affected)." + "Detects packages with known vulnerabilities that have not been triaged or addressed." ) def count_violations(self, product, threshold, parameters): @@ -176,10 +174,8 @@ class StaleVulnerabilityRule(BaseRule): label = "Stale Vulnerability" severity = "error" description = ( - "Detects packages with a high-risk vulnerability that has remained unaddressed " - "beyond a configured delay. The delay is measured from the detected_date on the " - "package-vulnerability link, i.e. when the vulnerability was first imported for " - "that specific package." + "Detects packages with high-risk vulnerabilities that have remained unaddressed " + "beyond a configured number of days." ) parameters_schema = { "max_days": ( From 5b6c9b7fdc853976341181648bd165060faca389 Mon Sep 17 00:00:00 2001 From: tdruez Date: Thu, 23 Jul 2026 14:16:06 +0400 Subject: [PATCH 08/19] refine rules filtering system Signed-off-by: tdruez --- policy/rules.py | 80 +++++++++++++++++++----------------- policy/tests/test_rules.py | 2 +- product_portfolio/filters.py | 2 +- 3 files changed, 45 insertions(+), 39 deletions(-) diff --git a/policy/rules.py b/policy/rules.py index 961b406c..d223c244 100644 --- a/policy/rules.py +++ b/policy/rules.py @@ -35,6 +35,13 @@ def get_package_filter(self): """Return queryset filter kwargs for ProductPackage to identify violating packages.""" return {} + def filter_queryset(self, queryset, parameters=None): + """Filter a ProductPackage queryset to packages that violate this rule.""" + package_filter = self.get_package_filter() + if package_filter: + return queryset.filter(**package_filter) + return queryset + class PackageBaseRule(BaseRule): """Base for rules that count packages matching a fixed filter within a product.""" @@ -112,27 +119,24 @@ class VulnerabilityDetectedRule(BaseRule): rule_type = "vulnerability_detected" label = "Vulnerability Detected" severity = "error" - description = "Detects packages with at least one known vulnerability (non-null risk score)." + description = "Detects packages with at least one known vulnerability." parameters_schema = { "min_risk_score": "Minimum risk score (0.0-10.0). Default: any vulnerability.", } - def get_package_filter(self): - return {"package__risk_score__isnull": False} - - def count_violations(self, product, threshold, parameters): - Package = apps.get_model("component_catalog", "package") - - packages = Package.objects.filter( - productpackages__product=product, - risk_score__isnull=False, - ) - + def filter_queryset(self, queryset, parameters=None): + parameters = parameters or {} min_risk_score = parameters.get("min_risk_score") if min_risk_score is not None: - packages = packages.filter(risk_score__gte=min_risk_score) + return queryset.filter( + package__affected_by_vulnerabilities__risk_score__gte=min_risk_score + ) + return queryset.filter(package__affected_by_vulnerabilities__isnull=False) - count = packages.count() + def count_violations(self, product, threshold, parameters): + ProductPackage = apps.get_model("product_portfolio", "productpackage") + product_packages = ProductPackage.objects.filter(product=product) + count = self.filter_queryset(product_packages, parameters).distinct().count() return count if count > threshold else 0 @@ -144,28 +148,28 @@ class UnresolvedVulnerabilityCountRule(BaseRule): "Detects packages with known vulnerabilities that have not been triaged or addressed." ) - def count_violations(self, product, threshold, parameters): + def filter_queryset(self, queryset, parameters=None): PackageAffectedByVulnerability = apps.get_model( "component_catalog", "packageaffectedbyvulnerability" ) VulnerabilityAnalysis = apps.get_model("vulnerabilities", "vulnerabilityanalysis") terminal_analysis = VulnerabilityAnalysis.objects.filter( - product=product, + product_package=OuterRef(OuterRef("pk")), state__in=TERMINAL_VULNERABILITY_STATES, - package=OuterRef("package"), vulnerability=OuterRef("vulnerability"), ) - - count = ( - PackageAffectedByVulnerability.objects.filter( - package__productpackages__product=product, - ) - .annotate(has_terminal_analysis=Exists(terminal_analysis)) - .filter(has_terminal_analysis=False) - .distinct() - .count() + unresolved_link = ( + PackageAffectedByVulnerability.objects.filter(package=OuterRef("package")) + .annotate(has_terminal=Exists(terminal_analysis)) + .filter(has_terminal=False) ) + return queryset.filter(Exists(unresolved_link)) + + def count_violations(self, product, threshold, parameters): + ProductPackage = apps.get_model("product_portfolio", "productpackage") + product_packages = ProductPackage.objects.filter(product=product) + count = self.filter_queryset(product_packages).distinct().count() return count if count > threshold else 0 @@ -187,35 +191,37 @@ class StaleVulnerabilityRule(BaseRule): ), } - def count_violations(self, product, threshold, parameters): + def filter_queryset(self, queryset, parameters=None): PackageAffectedByVulnerability = apps.get_model( "component_catalog", "packageaffectedbyvulnerability" ) VulnerabilityAnalysis = apps.get_model("vulnerabilities", "vulnerabilityanalysis") + parameters = parameters or {} max_days = parameters.get("max_days", 30) min_risk_score = parameters.get("min_risk_score", 8.0) cutoff_date = timezone.now() - timedelta(days=max_days) terminal_analysis = VulnerabilityAnalysis.objects.filter( - product=product, + product_package=OuterRef(OuterRef("pk")), state__in=TERMINAL_VULNERABILITY_STATES, - package=OuterRef("package"), vulnerability=OuterRef("vulnerability"), ) - - count = ( + stale_link = ( PackageAffectedByVulnerability.objects.filter( - package__productpackages__product=product, + package=OuterRef("package"), vulnerability__risk_score__gte=min_risk_score, detected_date__lte=cutoff_date, ) - .annotate(has_terminal_analysis=Exists(terminal_analysis)) - .filter(has_terminal_analysis=False) - .values("package_id") - .distinct() - .count() + .annotate(has_terminal=Exists(terminal_analysis)) + .filter(has_terminal=False) ) + return queryset.filter(Exists(stale_link)) + + def count_violations(self, product, threshold, parameters): + ProductPackage = apps.get_model("product_portfolio", "productpackage") + product_packages = ProductPackage.objects.filter(product=product) + count = self.filter_queryset(product_packages, parameters).distinct().count() return count if count > threshold else 0 diff --git a/policy/tests/test_rules.py b/policy/tests/test_rules.py index 596f70b4..3843cdc0 100644 --- a/policy/tests/test_rules.py +++ b/policy/tests/test_rules.py @@ -157,7 +157,7 @@ def setUp(self): def test_counts_packages_with_any_vulnerability(self): package = make_package(self.dataspace) - make_vulnerability(self.dataspace, affecting=package, risk_score=5.0) + make_vulnerability(self.dataspace, affecting=package) make_product_package(self.product, package=package) make_product_package(self.product, package=make_package(self.dataspace)) count = VulnerabilityDetectedRule().count_violations(self.product, 0, {}) diff --git a/product_portfolio/filters.py b/product_portfolio/filters.py index f6521aae..858a5113 100644 --- a/product_portfolio/filters.py +++ b/product_portfolio/filters.py @@ -429,7 +429,7 @@ def filter_by_policy_rule(self, queryset, name, value): handler = RULE_REGISTRY.get(value) if not handler: return queryset - return queryset.filter(**handler.get_package_filter()) + return handler.filter_queryset(queryset).distinct() def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) From 95fb3dc183c46a4def4f822277f1057f33b28271 Mon Sep 17 00:00:00 2001 From: tdruez Date: Thu, 23 Jul 2026 14:19:44 +0400 Subject: [PATCH 09/19] refine wording Signed-off-by: tdruez --- .../compliance/compliance_panels.html | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html b/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html index 550bc451..e5444438 100644 --- a/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html +++ b/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html @@ -11,8 +11,8 @@

{% trans "Policy violations" %}

@@ -32,16 +32,16 @@

{% trans "Policy violations" %}

{% endif %} - {{ policy_violation_count }} {% trans "active" %} + {{ policy_violation_count }} {% trans "triggered" %} -