feat: [five-c] Create extended policy rules set #411

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,
+        ),
+    ]

component_catalog/tests/test_models.py

@@ -36,6 +36,7 @@
 from component_catalog.models import ComponentType
 from component_catalog.models import LicenseExpressionMixin
 from component_catalog.models import Package
+from component_catalog.models import PackageAffectedByVulnerability
 from component_catalog.models import PackageAlreadyExistsWarning
 from component_catalog.models import Subcomponent
 from component_catalog.tests import make_package
@@ -1372,9 +1373,15 @@ def test_component_catalog_models_get_exclude_candidates_fields(self):
         )
 
         for model_class, expected in input_data:
-            results = [f.name for f in model_class().get_exclude_candidates_fields()]
+            results = [field.name for field in model_class().get_exclude_candidates_fields()]
             self.assertEqual(sorted(expected), sorted(results))
 
+    def test_package_affected_by_vulnerability_excludes_auto_now_add_fields(self):
+        field_names = [
+            field.name for field in PackageAffectedByVulnerability().get_exclude_candidates_fields()
+        ]
+        self.assertNotIn("detected_date", field_names)
+
     def test_component_create_with_or_and_and_in_license_name_and_key(self):
         or_license = License.objects.create(
             key="orrible",

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,
             ]

policy/rules.py

@@ -6,7 +6,15 @@
 # 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"]
 
 
 class BaseRule:
@@ -27,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."""
@@ -55,17 +70,15 @@ class UsagePolicyErrorRule(PackageBaseRule):
     rule_type = "usage_policy_error"
     label = "Usage Policy Error"
     severity = "error"
-    description = (
-        "Detects packages assigned a usage policy with a compliance alert level of 'error'."
-    )
+    description = "Detects packages assigned a usage policy flagged with an error compliance alert."
     package_filter = {"usage_policy__compliance_alert": "error"}
 
 
 class UsagePolicyWarningRule(PackageBaseRule):
     rule_type = "usage_policy_warning"
     label = "Usage Policy Warning"
     description = (
-        "Detects packages assigned a usage policy with a compliance alert level of 'warning'."
+        "Detects packages assigned a usage policy flagged with a warning compliance alert."
     )
     package_filter = {"usage_policy__compliance_alert": "warning"}
 
@@ -76,7 +89,7 @@ class LicensePolicyErrorRule(PackageBaseRule):
     severity = "error"
     description = (
         "Detects packages whose licenses are assigned a usage policy"
-        " with a compliance alert level of 'error'."
+        " flagged with an error compliance alert."
     )
     package_filter = {"licenses__usage_policy__compliance_alert": "error"}
 
@@ -86,18 +99,126 @@ class LicensePolicyWarningRule(PackageBaseRule):
     label = "License Policy Warning"
     description = (
         "Detects packages whose licenses are assigned a usage policy"
-        " with a compliance alert level of 'warning'."
+        " flagged with a warning compliance alert."
     )
     package_filter = {"licenses__usage_policy__compliance_alert": "warning"}
 
 
 class LicenseCoverageGapRule(PackageBaseRule):
     rule_type = "license_coverage_gap"
     label = "License Coverage Gap"
+    description = "Detects packages with no license expression."
+    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."
+    parameters_schema = {
+        "min_risk_score": "Minimum risk score (0.0-10.0). Default: any vulnerability.",
+    }
+
+    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:
+            return queryset.filter(
+                package__affected_by_vulnerabilities__risk_score__gte=min_risk_score
+            )
+        return queryset.filter(package__affected_by_vulnerabilities__isnull=False)
+
+    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
+
+
+class UnresolvedVulnerabilityRule(BaseRule):
+    rule_type = "vulnerability_unresolved"
+    label = "Vulnerability Unresolved"
+    severity = "warning"
     description = (
-        "Detects packages with no license expression, indicating a gap in license coverage."
+        "Detects packages with known vulnerabilities and no completed vulnerability analysis."
     )
-    package_filter = {"license_expression": ""}
+
+    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_package=OuterRef(OuterRef("pk")),
+            state__in=TERMINAL_VULNERABILITY_STATES,
+            vulnerability=OuterRef("vulnerability"),
+        )
+        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
+
+
+class StaleVulnerabilityRule(BaseRule):
+    rule_type = "vulnerability_stale"
+    label = "Vulnerability Stale"
+    severity = "error"
+    description = (
+        "Detects packages with high-risk vulnerabilities unaddressed"
+        " for more than the configured number of days."
+    )
+    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 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_package=OuterRef(OuterRef("pk")),
+            state__in=TERMINAL_VULNERABILITY_STATES,
+            vulnerability=OuterRef("vulnerability"),
+        )
+        stale_link = (
+            PackageAffectedByVulnerability.objects.filter(
+                package=OuterRef("package"),
+                vulnerability__risk_score__gte=min_risk_score,
+                detected_date__lte=cutoff_date,
+            )
+            .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
 
 
 RULE_REGISTRY = {
@@ -106,4 +227,7 @@ class LicenseCoverageGapRule(PackageBaseRule):
     LicensePolicyErrorRule.rule_type: LicensePolicyErrorRule(),
     LicensePolicyWarningRule.rule_type: LicensePolicyWarningRule(),
     LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(),
+    VulnerabilityDetectedRule.rule_type: VulnerabilityDetectedRule(),
+    UnresolvedVulnerabilityRule.rule_type: UnresolvedVulnerabilityRule(),
+    StaleVulnerabilityRule.rule_type: StaleVulnerabilityRule(),
 }