diff --git a/dejacode/settings.py b/dejacode/settings.py
index 8adfdf39..7c9af8c3 100644
--- a/dejacode/settings.py
+++ b/dejacode/settings.py
@@ -475,7 +475,9 @@ def gettext_noop(s):
# Cron jobs (scheduler)
daily_at_3am = "0 3 * * *"
+hourly = "0 * * * *"
DEJACODE_VULNERABILITIES_CRON = env.str("DEJACODE_VULNERABILITIES_CRON", default=daily_at_3am)
+DEJACODE_POLICY_RULES_CRON = env.str("DEJACODE_POLICY_RULES_CRON", default=hourly)
def enable_rq_eager_mode():
diff --git a/dje/admin.py b/dje/admin.py
index 6aa131b5..5a4762b8 100644
--- a/dje/admin.py
+++ b/dje/admin.py
@@ -96,6 +96,7 @@
from dje.views import manage_tab_permissions_view
from dje.views import object_compare_view
from dje.views import object_copy_view
+from policy.rules import RULE_REGISTRY
EXTERNAL_SOURCE_LOOKUP = "external_references__external_source_id"
@@ -1046,12 +1047,26 @@ def render(self, name, value, attrs=None, renderer=None):
class DataspaceConfigurationForm(forms.ModelForm):
- """
- Configure Dataspace settings.
+ """Configure Dataspace integration settings, with sensitive values hidden in the UI."""
- This form includes fields for various API keys, with sensitive values
- hidden in the UI using the HiddenValueWidget.
- """
+ class Meta:
+ model = DataspaceConfiguration
+ fields = [
+ "homepage_layout",
+ "scancodeio_url",
+ "scancodeio_api_key",
+ "vulnerablecode_url",
+ "vulnerablecode_api_key",
+ "vulnerabilities_risk_threshold",
+ "purldb_url",
+ "purldb_api_key",
+ "forgejo_token",
+ "github_token",
+ "gitlab_token",
+ "jira_user",
+ "jira_token",
+ "sourcehut_token",
+ ]
hidden_value_fields = [
"scancodeio_api_key",
@@ -1078,6 +1093,73 @@ def clean(self):
del self.cleaned_data[field_name]
+class PolicyRulesConfigurationForm(forms.ModelForm):
+ """Form for configuring policy rule overrides stored in policy_rules_config."""
+
+ class Meta:
+ model = DataspaceConfiguration
+ fields = []
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.add_policy_rule_config_fields()
+
+ def add_policy_rule_config_fields(self):
+ """Inject per-rule form fields with initial values from the saved policy_rules_config."""
+ config = getattr(self.instance, "policy_rules_config", {}) or {}
+ for rule_type, handler in RULE_REGISTRY.items():
+ rule_config = config.get(rule_type, {})
+ self.fields[f"rule_{rule_type}_enabled"] = forms.BooleanField(
+ label="Enable this rule",
+ required=False,
+ initial=rule_config.get("is_active", False),
+ )
+ self.fields[f"rule_{rule_type}_threshold"] = forms.IntegerField(
+ label="Threshold",
+ required=False,
+ min_value=0,
+ initial=rule_config.get("threshold"),
+ widget=forms.NumberInput(
+ attrs={"placeholder": f"Default: {handler.default_threshold}"}
+ ),
+ help_text="Minimum violations to trigger the rule. Leave blank to use the default.",
+ )
+ for param_name, param_desc in handler.parameters_schema.items():
+ self.fields[f"rule_{rule_type}_param_{param_name}"] = forms.FloatField(
+ label=param_name.replace("_", " ").title(),
+ required=False,
+ initial=(rule_config.get("parameters") or {}).get(param_name),
+ help_text=param_desc,
+ )
+
+ def build_policy_rules_config(self):
+ """Serialize the per-rule form fields back into the policy_rules_config dict."""
+ policy_rules_config = {}
+ for rule_type, handler in RULE_REGISTRY.items():
+ rule_config = {}
+ if self.cleaned_data.get(f"rule_{rule_type}_enabled"):
+ rule_config["is_active"] = True
+ threshold = self.cleaned_data.get(f"rule_{rule_type}_threshold")
+ if threshold is not None:
+ rule_config["threshold"] = threshold
+ parameters = {}
+ for param_name in handler.parameters_schema:
+ param_value = self.cleaned_data.get(f"rule_{rule_type}_param_{param_name}")
+ if param_value is not None:
+ parameters[param_name] = param_value
+ if parameters:
+ rule_config["parameters"] = parameters
+ if rule_config:
+ policy_rules_config[rule_type] = rule_config
+ return policy_rules_config
+
+ def save(self, commit=True):
+ self.instance.policy_rules_config = self.build_policy_rules_config()
+ if commit:
+ self.instance.save(update_fields=["policy_rules_config"])
+ return self.instance
+
+
class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
model = DataspaceConfiguration
form = DataspaceConfigurationForm
@@ -1142,12 +1224,13 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
]
# Do not include the Dataspace related FKs on addition as the Dataspace does not exist yet
fieldsets = [("", {"fields": ("homepage_layout",)})] + add_fieldsets
+ inline_classes = ("grp-collapse grp-open",)
can_delete = False
def get_fieldsets(self, request, obj=None):
if not obj:
return self.add_fieldsets
- return super().get_fieldsets(request, obj)
+ return [("", {"fields": ("homepage_layout",)})] + self.add_fieldsets
def get_readonly_fields(self, request, obj=None):
"""Only a user from the current Dataspace can edit Dataspace related FKs."""
@@ -1160,6 +1243,40 @@ def get_readonly_fields(self, request, obj=None):
return readonly_fields
+class PolicyRulesConfigurationInline(DataspacedFKMixin, admin.StackedInline):
+ model = DataspaceConfiguration
+ form = PolicyRulesConfigurationForm
+ verbose_name_plural = _("Policy Rules Configuration")
+ verbose_name = _("Policy Rules Configuration")
+ classes = ("grp-collapse grp-open",)
+ inline_classes = ("grp-collapse grp-open",)
+ can_delete = False
+
+ def get_fieldsets(self, request, obj=None):
+ if not obj:
+ return []
+ rule_fieldsets = []
+ for rule_type, handler in RULE_REGISTRY.items():
+ fields = [f"rule_{rule_type}_enabled", f"rule_{rule_type}_threshold"]
+ for param_name in handler.parameters_schema:
+ fields.append(f"rule_{rule_type}_param_{param_name}")
+ rule_fieldsets.append(
+ (
+ handler.label,
+ {
+ "fields": fields,
+ "description": handler.description,
+ "classes": ("grp-collapse grp-open",),
+ },
+ )
+ )
+ return rule_fieldsets
+
+ def get_formset(self, request, obj=None, **kwargs):
+ kwargs["fields"] = []
+ return super().get_formset(request, obj, **kwargs)
+
+
@admin.register(Dataspace, site=dejacode_site)
class DataspaceAdmin(
ReferenceOnlyPermissions,
@@ -1239,7 +1356,7 @@ class DataspaceAdmin(
),
)
search_fields = ("name",)
- inlines = [DataspaceConfigurationInline]
+ inlines = [DataspaceConfigurationInline, PolicyRulesConfigurationInline]
form = DataspaceAdminForm
change_form_template = "admin/dje/dataspace/change_form.html"
change_list_template = "admin/change_list_extended.html"
diff --git a/dje/cron_jobs.py b/dje/cron_jobs.py
index 38cd35fc..1631a15c 100644
--- a/dje/cron_jobs.py
+++ b/dje/cron_jobs.py
@@ -11,6 +11,7 @@
from rq import cron
from dje.tasks import update_vulnerabilities
+from policy.tasks import evaluate_all_products_rules_task
two_hour = 7200
@@ -20,3 +21,10 @@
cron=settings.DEJACODE_VULNERABILITIES_CRON, # Daily at 3am by default
job_timeout=two_hour,
)
+
+cron.register(
+ func=evaluate_all_products_rules_task,
+ queue_name="default",
+ cron=settings.DEJACODE_POLICY_RULES_CRON, # Hourly by default
+ job_timeout=two_hour,
+)
diff --git a/dje/migrations/0016_dataspaceconfiguration_policy_rules_config.py b/dje/migrations/0016_dataspaceconfiguration_policy_rules_config.py
new file mode 100644
index 00000000..340b8226
--- /dev/null
+++ b/dje/migrations/0016_dataspaceconfiguration_policy_rules_config.py
@@ -0,0 +1,18 @@
+# Generated by Django 6.0.6 on 2026-07-15 08:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('dje', '0015_alter_dataspaceconfiguration_purldb_api_key_and_more'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='dataspaceconfiguration',
+ name='policy_rules_config',
+ field=models.JSONField(blank=True, default=dict, help_text='Override default policy rule settings for this dataspace.'),
+ ),
+ ]
diff --git a/dje/models.py b/dje/models.py
index 2d87316d..996195ce 100644
--- a/dje/models.py
+++ b/dje/models.py
@@ -614,6 +614,12 @@ class DataspaceConfiguration(DataspaceForeignKeyValidationMixin, models.Model):
),
)
+ policy_rules_config = models.JSONField(
+ blank=True,
+ default=dict,
+ help_text=_("Override default policy rule settings for this dataspace."),
+ )
+
def __str__(self):
return f"{self.dataspace}"
diff --git a/dje/tests/test_admin.py b/dje/tests/test_admin.py
index 83e00311..ceb3aa1e 100644
--- a/dje/tests/test_admin.py
+++ b/dje/tests/test_admin.py
@@ -13,10 +13,12 @@
from django.test import override_settings
from django.urls import reverse
+from dje.admin import PolicyRulesConfigurationForm
from dje.copier import copy_object
from dje.filters import DataspaceFilter
from dje.filters import MissingInFilter
from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
from dje.models import History
from dje.search import advanced_search
from dje.tests import add_perm
@@ -25,6 +27,7 @@
from dje.tests import create_superuser
from dje.tests import create_user
from organization.models import Owner
+from policy.rules import RULE_REGISTRY
class DataspacedModelAdminTestCase(TestCase):
@@ -178,6 +181,8 @@ def test_dataspace_admin_changeform_update_packages_from_scan_field_validation(s
"update_packages_from_scan": True,
"configuration-TOTAL_FORMS": 0,
"configuration-INITIAL_FORMS": 0,
+ "configuration-2-TOTAL_FORMS": 0,
+ "configuration-2-INITIAL_FORMS": 0,
}
response = self.client.post(url, data)
@@ -611,3 +616,78 @@ def test_admin_group_permission_export_csv(self):
'attachment; filename="dejacode_group_permission.csv"', response["Content-Disposition"]
)
self.assertEqual(b",change_license\r\nchange license,X\r\n", response.content)
+
+
+class PolicyRulesConfigurationFormTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.config = DataspaceConfiguration.objects.create(dataspace=self.dataspace)
+ self.Form = PolicyRulesConfigurationForm
+
+ def _bound_form(self, extra_data=None):
+ data = {}
+ for rule_type in RULE_REGISTRY:
+ data[f"rule_{rule_type}_enabled"] = False
+ data[f"rule_{rule_type}_threshold"] = ""
+ if extra_data:
+ data.update(extra_data)
+ return self.Form(data=data, instance=self.config)
+
+ def test_enabled_rule_included_in_config(self):
+ form = self._bound_form({"rule_usage_policy_error_enabled": True})
+ self.assertTrue(form.is_valid(), form.errors)
+ result = form.build_policy_rules_config()
+ self.assertIn("usage_policy_error", result)
+ self.assertTrue(result["usage_policy_error"]["is_active"])
+
+ def test_disabled_rule_not_in_config(self):
+ form = self._bound_form()
+ self.assertTrue(form.is_valid(), form.errors)
+ result = form.build_policy_rules_config()
+ self.assertEqual({}, result)
+
+ def test_threshold_saved_when_set(self):
+ form = self._bound_form(
+ {
+ "rule_usage_policy_error_enabled": True,
+ "rule_usage_policy_error_threshold": "3",
+ }
+ )
+ self.assertTrue(form.is_valid(), form.errors)
+ result = form.build_policy_rules_config()
+ self.assertEqual(3, result["usage_policy_error"]["threshold"])
+
+ def test_initial_values_loaded_from_existing_config(self):
+ self.config.policy_rules_config = {
+ "usage_policy_error": {"is_active": True, "threshold": 7}
+ }
+ self.config.save()
+ form = self.Form(instance=self.config)
+ self.assertTrue(form.fields["rule_usage_policy_error_enabled"].initial)
+ self.assertEqual(7, form.fields["rule_usage_policy_error_threshold"].initial)
+ self.assertFalse(form.fields["rule_license_coverage_gap_enabled"].initial)
+
+ def test_save_persists_policy_rules_config_to_db(self):
+ form = self._bound_form({"rule_usage_policy_error_enabled": True})
+ self.assertTrue(form.is_valid(), form.errors)
+ form.save()
+ self.config.refresh_from_db()
+ self.assertTrue(self.config.policy_rules_config["usage_policy_error"]["is_active"])
+
+ def test_inline_shows_rule_fieldsets_on_existing_dataspace(self):
+ self.super_user = create_superuser("super_user", self.dataspace)
+ self.client.login(username="super_user", password="secret")
+ url = reverse("admin:dje_dataspace_change", args=[self.dataspace.pk])
+ response = self.client.get(url)
+ self.assertEqual(200, response.status_code)
+ self.assertContains(response, "Policy Rules Configuration")
+ for rule_type in RULE_REGISTRY:
+ self.assertContains(response, f"rule_{rule_type}_enabled")
+
+ def test_inline_not_shown_on_dataspace_add(self):
+ self.super_user = create_superuser("super_user", self.dataspace)
+ self.client.login(username="super_user", password="secret")
+ url = reverse("admin:dje_dataspace_add")
+ response = self.client.get(url)
+ self.assertEqual(200, response.status_code)
+ self.assertNotContains(response, "rule_usage_policy_error_enabled")
diff --git a/dje/tests/test_history.py b/dje/tests/test_history.py
index b9182bd2..cbba8700 100644
--- a/dje/tests/test_history.py
+++ b/dje/tests/test_history.py
@@ -58,6 +58,8 @@ def test_history_on_admin_dataspace_add(self):
"name": "new_dataspace",
"configuration-TOTAL_FORMS": 0,
"configuration-INITIAL_FORMS": 0,
+ "configuration-2-TOTAL_FORMS": 0,
+ "configuration-2-INITIAL_FORMS": 0,
}
self.client.post(url, params)
diff --git a/notification/admin.py b/notification/admin.py
index d683bec2..a9d74656 100644
--- a/notification/admin.py
+++ b/notification/admin.py
@@ -64,3 +64,9 @@ class WebhookSubscriptionAdmin(ProhibitDataspaceLookupMixin, DataspacedAdmin):
actions_to_remove = ["copy_to", "compare_with"]
email_notification_on = ()
inlines = [WebhookDeliveryInline]
+
+ def get_inlines(self, request, obj=None):
+ """Exclude delivery history on the add form as no deliveries exist yet."""
+ if obj is None:
+ return []
+ return super().get_inlines(request, obj)
diff --git a/notification/models.py b/notification/models.py
index 661ef5a4..b8676a70 100644
--- a/notification/models.py
+++ b/notification/models.py
@@ -28,6 +28,8 @@
"user.added_or_updated",
"user.locked_out",
"vulnerability.data_update",
+ "policy.violation_detected",
+ "policy.violation_resolved",
]
diff --git a/notification/tasks.py b/notification/tasks.py
index 7c42c97e..e8222e33 100644
--- a/notification/tasks.py
+++ b/notification/tasks.py
@@ -12,6 +12,7 @@
from django_rq import job
+from dje.models import get_unsecured_manager
from notification.models import WebhookSubscription
logger = logging.getLogger("dje")
@@ -36,7 +37,7 @@ def deliver_webhook_task(
if instance_app_label and instance_model_name and instance_pk:
try:
model_class = apps.get_model(instance_app_label, instance_model_name)
- instance = model_class.objects.get(pk=instance_pk)
+ instance = get_unsecured_manager(model_class).get(pk=instance_pk)
except Exception:
logger.error(
f"Instance {instance_app_label}.{instance_model_name} pk={instance_pk} not found."
diff --git a/notification/tests/test_tasks.py b/notification/tests/test_tasks.py
index 6ee03ad3..ce0227ec 100644
--- a/notification/tests/test_tasks.py
+++ b/notification/tests/test_tasks.py
@@ -7,6 +7,7 @@
#
import json
+import uuid
from unittest.mock import patch
from django.conf import settings
@@ -16,6 +17,8 @@
from dje.models import Dataspace
from dje.tests import create_superuser
from notification.models import WebhookSubscription
+from notification.tasks import deliver_webhook_task
+from product_portfolio.tests import make_product
from workflow.models import Priority
from workflow.models import Question
from workflow.models import Request
@@ -23,6 +26,45 @@
from workflow.models import RequestTemplate
+class DeliverWebhookTaskTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.webhook = WebhookSubscription.objects.create(
+ dataspace=self.dataspace,
+ target_url="http://127.0.0.1:8000/",
+ event="policy.violation_detected",
+ )
+
+ @patch("notification.models.WebhookSubscription.deliver")
+ def test_deliver_webhook_task_resolves_secured_model_instance(self, mock_deliver):
+ product = make_product(self.dataspace)
+ deliver_webhook_task(
+ webhook_subscription_uuid=self.webhook.uuid,
+ payload_override={"text": "test"},
+ instance_app_label="product_portfolio",
+ instance_model_name="product",
+ instance_pk=product.pk,
+ )
+ mock_deliver.assert_called_once()
+ delivered_instance = mock_deliver.call_args[0][0]
+ self.assertEqual(product, delivered_instance)
+
+ def test_deliver_webhook_task_missing_subscription_logs_error(self):
+ with self.assertLogs("dje", level="ERROR") as captured:
+ deliver_webhook_task(webhook_subscription_uuid=uuid.uuid4())
+ self.assertTrue(any("not found" in line for line in captured.output))
+
+ def test_deliver_webhook_task_missing_instance_logs_error(self):
+ with self.assertLogs("dje", level="ERROR") as captured:
+ deliver_webhook_task(
+ webhook_subscription_uuid=self.webhook.uuid,
+ instance_app_label="product_portfolio",
+ instance_model_name="product",
+ instance_pk=99999999,
+ )
+ self.assertTrue(any("not found" in line for line in captured.output))
+
+
class NotificationTasksTestCase(TestCase):
def setUp(self):
self.nexb_dataspace = Dataspace.objects.create(name="nexB")
diff --git a/policy/apps.py b/policy/apps.py
index e67945dc..392564e7 100755
--- a/policy/apps.py
+++ b/policy/apps.py
@@ -13,3 +13,6 @@
class PolicyConfig(AppConfig):
name = "policy"
verbose_name = _("Policy")
+
+ def ready(self):
+ import policy.signals # noqa: F401
diff --git a/policy/engine.py b/policy/engine.py
new file mode 100644
index 00000000..e1b9c450
--- /dev/null
+++ b/policy/engine.py
@@ -0,0 +1,129 @@
+#
+# 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 django.utils import timezone
+
+from notification.models import fire_webhooks
+from policy.rules import RULE_REGISTRY
+from product_portfolio.models import ProductPolicyViolation
+
+
+def fire_policy_webhooks(product, new_violations, resolved_count):
+ """Fire policy webhooks for newly detected or resolved violations."""
+ if new_violations:
+ lines = [
+ f"- {violation.rule_label}: {violation.violation_count} violation(s)"
+ for violation in new_violations
+ ]
+ payload = {
+ "text": (f"[DejaCode] Policy violations detected for {product}\n" + "\n".join(lines))
+ }
+ fire_webhooks("policy.violation_detected", instance=product, payload_override=payload)
+
+ if resolved_count:
+ payload = {
+ "text": (f"[DejaCode] {resolved_count} policy violation(s) resolved for {product}")
+ }
+ fire_webhooks("policy.violation_resolved", instance=product, payload_override=payload)
+
+
+def get_effective_config(rule_type, dataspace):
+ """
+ Resolve threshold, parameters, and is_active for a rule type in a given dataspace.
+
+ Reads the dataspace-level override from DataspaceConfiguration.policy_rules_config,
+ falling back to the code defaults defined on the rule handler.
+ """
+ handler = RULE_REGISTRY[rule_type]
+ try:
+ rule_config = dataspace.configuration.policy_rules_config.get(rule_type, {})
+ except AttributeError:
+ rule_config = {}
+
+ return {
+ "is_active": rule_config.get("is_active", False),
+ "threshold": rule_config.get("threshold", handler.default_threshold),
+ "parameters": rule_config.get("parameters", {}),
+ }
+
+
+def evaluate_rule(rule_type, product, threshold, parameters):
+ """
+ Evaluate a single rule against a product and record the violation if triggered.
+
+ Returns a 3-tuple: (violation_or_none, created, resolved_count).
+ """
+ handler = RULE_REGISTRY[rule_type]
+ violation_count = handler.count_violations(product, threshold, parameters)
+
+ lookup = {"rule_type": rule_type, "product": product}
+
+ if violation_count > 0:
+ # update_or_create on the unique (rule_type, product) pair so that a previously
+ # resolved violation can be re-activated without hitting the unique constraint.
+ violation, created = ProductPolicyViolation.objects.update_or_create(
+ **lookup,
+ defaults={
+ "dataspace": product.dataspace,
+ "violation_count": violation_count,
+ "resolved": False,
+ "resolved_date": None,
+ },
+ )
+ return violation, created, 0
+
+ now = timezone.now()
+ resolved_count = (
+ ProductPolicyViolation.objects.filter(**lookup)
+ .unresolved()
+ .update(
+ resolved=True,
+ resolved_date=now,
+ last_checked=now,
+ )
+ )
+ return None, False, resolved_count
+
+
+def evaluate_rules(product):
+ """
+ Evaluate all rules in RULE_REGISTRY for the given product.
+
+ Returns a 2-tuple: (new_violations, resolved_count).
+ new_violations is a list of newly created ProductPolicyViolation instances.
+ resolved_count is the total number of violations resolved during this run.
+ """
+ new_violations = []
+ resolved_count = 0
+ now = timezone.now()
+
+ for rule_type in RULE_REGISTRY:
+ config = get_effective_config(rule_type, product.dataspace)
+ if not config["is_active"]:
+ # Explicitly resolve open violations so disabling a rule clears its history
+ # rather than leaving stale unresolved records.
+ rows = (
+ ProductPolicyViolation.objects.filter(
+ rule_type=rule_type,
+ product=product,
+ )
+ .unresolved()
+ .update(resolved=True, resolved_date=now, last_checked=now)
+ )
+ resolved_count += rows
+ continue
+
+ violation, created, resolved = evaluate_rule(
+ rule_type, product, config["threshold"], config["parameters"]
+ )
+ if created:
+ new_violations.append(violation)
+ resolved_count += resolved
+
+ fire_policy_webhooks(product, new_violations, resolved_count)
+ return new_violations, resolved_count
diff --git a/policy/models.py b/policy/models.py
index cc731855..ed649d0d 100755
--- a/policy/models.py
+++ b/policy/models.py
@@ -244,3 +244,32 @@ def save(self, *args, **kwargs):
if self.from_policy.content_type == self.to_policy.content_type:
raise AssertionError
super().save(*args, **kwargs)
+
+
+class AbstractPolicyViolation(models.Model):
+ """Shared fields for all concrete policy violation models. No DB table."""
+
+ violation_count = models.PositiveIntegerField(
+ default=0,
+ help_text=_("Number of objects currently violating the rule."),
+ )
+ detected_date = models.DateTimeField(
+ auto_now_add=True,
+ help_text=_("The date and time when this violation was first detected."),
+ )
+ last_checked = models.DateTimeField(
+ auto_now=True,
+ help_text=_("The date and time of the last evaluation."),
+ )
+ resolved = models.BooleanField(
+ default=False,
+ help_text=_("Indicates whether this violation has been resolved."),
+ )
+ resolved_date = models.DateTimeField(
+ null=True,
+ blank=True,
+ help_text=_("The date and time when this violation was resolved."),
+ )
+
+ class Meta:
+ abstract = True
diff --git a/policy/rules.py b/policy/rules.py
new file mode 100644
index 00000000..bc89931c
--- /dev/null
+++ b/policy/rules.py
@@ -0,0 +1,109 @@
+#
+# 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 django.apps import apps
+
+
+class BaseRule:
+ """Base class for policy rule handlers."""
+
+ rule_type = None
+ label = None
+ description = None
+ severity = "warning"
+ default_threshold = 0
+ parameters_schema = {}
+
+ def count_violations(self, product, threshold, parameters):
+ """Count objects violating the rule for the given product."""
+ raise NotImplementedError
+
+ def get_package_filter(self):
+ """Return queryset filter kwargs for ProductPackage to identify violating packages."""
+ return {}
+
+
+class PackageBaseRule(BaseRule):
+ """Base for rules that count packages matching a fixed filter within a product."""
+
+ package_filter = {}
+
+ def count_violations(self, product, threshold, parameters):
+ Package = apps.get_model("component_catalog", "package")
+
+ count = (
+ Package.objects.filter(
+ productpackages__product=product,
+ **self.package_filter,
+ )
+ .distinct()
+ .count()
+ )
+
+ return count if count > threshold else 0
+
+ def get_package_filter(self):
+ return {f"package__{key}": value for key, value in self.package_filter.items()}
+
+
+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'."
+ )
+ 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'."
+ )
+ package_filter = {"usage_policy__compliance_alert": "warning"}
+
+
+class LicensePolicyErrorRule(PackageBaseRule):
+ rule_type = "license_policy_error"
+ label = "License Policy Error"
+ severity = "error"
+ description = (
+ "Detects packages whose licenses are assigned a usage policy"
+ " with a compliance alert level of 'error'."
+ )
+ package_filter = {"licenses__usage_policy__compliance_alert": "error"}
+
+
+class LicensePolicyWarningRule(PackageBaseRule):
+ rule_type = "license_policy_warning"
+ label = "License Policy Warning"
+ description = (
+ "Detects packages whose licenses are assigned a usage policy"
+ " with a compliance alert level of 'warning'."
+ )
+ 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, indicating a gap in license coverage."
+ )
+ package_filter = {"license_expression": ""}
+
+
+RULE_REGISTRY = {
+ UsagePolicyErrorRule.rule_type: UsagePolicyErrorRule(),
+ UsagePolicyWarningRule.rule_type: UsagePolicyWarningRule(),
+ LicensePolicyErrorRule.rule_type: LicensePolicyErrorRule(),
+ LicensePolicyWarningRule.rule_type: LicensePolicyWarningRule(),
+ LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(),
+}
diff --git a/policy/signals.py b/policy/signals.py
new file mode 100644
index 00000000..4734945b
--- /dev/null
+++ b/policy/signals.py
@@ -0,0 +1,65 @@
+#
+# 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.
+#
+
+import logging
+
+from django.apps import apps
+from django.db.models.signals import post_delete
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from dje.models import get_unsecured_manager
+from policy.tasks import evaluate_all_products_rules_task
+from policy.tasks import evaluate_product_rules_task
+
+logger = logging.getLogger(__name__)
+
+
+@receiver(post_save, sender="product_portfolio.Product")
+def evaluate_product_rules_on_product_save(sender, instance, **kwargs):
+ """Queue a policy rule evaluation whenever a product is saved."""
+ evaluate_product_rules_task.delay(product_uuid=instance.uuid)
+
+
+@receiver(post_save, sender="product_portfolio.ProductPackage")
+def evaluate_product_rules_on_productpackage_save(sender, instance, **kwargs):
+ """Queue a policy rule evaluation whenever a package is added or updated in a product."""
+ evaluate_product_rules_task.delay(product_uuid=instance.product.uuid)
+
+
+@receiver(post_delete, sender="product_portfolio.ProductPackage")
+def evaluate_product_rules_on_productpackage_delete(sender, instance, **kwargs):
+ """Queue a policy rule evaluation whenever a package is removed from a product."""
+ evaluate_product_rules_task.delay(product_uuid=instance.product.uuid)
+
+
+@receiver(post_save, sender="component_catalog.Package")
+def evaluate_product_rules_on_package_save(sender, instance, created, **kwargs):
+ """Queue a policy rule evaluation for all products containing this package."""
+ if created:
+ return # A newly created package has no ProductPackage relations yet.
+
+ product_uuids = list(instance.productpackages.values_list("product__uuid", flat=True))
+ if product_uuids:
+ evaluate_all_products_rules_task.delay(product_uuids=product_uuids)
+
+
+@receiver(post_save, sender="dje.DataspaceConfiguration")
+def evaluate_products_on_policy_rules_config_save(sender, instance, update_fields, **kwargs):
+ """Queue re-evaluation for all products in the dataspace when policy rules config changes."""
+ if update_fields is not None and "policy_rules_config" not in update_fields:
+ return
+
+ Product = apps.get_model("product_portfolio", "product")
+ product_uuids = list(
+ get_unsecured_manager(Product)
+ .filter(dataspace=instance.dataspace)
+ .values_list("uuid", flat=True)
+ )
+ if product_uuids:
+ evaluate_all_products_rules_task.delay(product_uuids=product_uuids)
diff --git a/policy/tasks.py b/policy/tasks.py
new file mode 100644
index 00000000..77f8114d
--- /dev/null
+++ b/policy/tasks.py
@@ -0,0 +1,73 @@
+#
+# 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.
+#
+
+import logging
+
+from django.apps import apps
+
+from django_rq import job
+
+from dje.models import get_unsecured_manager
+from policy.engine import evaluate_rules
+
+logger = logging.getLogger(__name__)
+
+
+@job
+def evaluate_product_rules_task(product_uuid):
+ """Evaluate all active policy rules for the given product and fire webhooks on changes."""
+ Product = apps.get_model("product_portfolio", "product")
+
+ try:
+ product = (
+ get_unsecured_manager(Product)
+ .select_related("dataspace__configuration")
+ .get(uuid=product_uuid)
+ )
+ except Product.DoesNotExist:
+ logger.error(f"evaluate_product_rules_task: product {product_uuid} not found, skipping.")
+ return
+
+ logger.info(f"Evaluating policy rules for product {product}")
+ new_violations, resolved_count = evaluate_rules(product)
+ logger.info(
+ f"Policy rules evaluated for {product}: "
+ f"{len(new_violations)} new violation(s), {resolved_count} resolved."
+ )
+
+
+@job
+def evaluate_all_products_rules_task(include_locked=False, product_uuids=None):
+ """
+ Evaluate policy rules for products, skipping locked ones by default.
+ When product_uuids is provided, only those products are evaluated.
+ """
+ Product = apps.get_model("product_portfolio", "product")
+
+ products = get_unsecured_manager(Product).select_related("dataspace__configuration")
+ if product_uuids is not None:
+ products = products.filter(uuid__in=product_uuids)
+ if not include_locked:
+ products = products.exclude_locked()
+
+ count = products.count()
+ logger.info(f"Starting policy rule evaluation for {count} product(s).")
+
+ for product in products:
+ logger.info(f"Evaluating policy rules for product {product}")
+ try:
+ new_violations, resolved_count = evaluate_rules(product)
+ except Exception:
+ logger.exception(f"Policy rule evaluation failed for product {product}, skipping.")
+ continue
+ logger.info(
+ f"Policy rules evaluated for {product}: "
+ f"{len(new_violations)} new violation(s), {resolved_count} resolved."
+ )
+
+ logger.info(f"Policy rule evaluation complete for {count} product(s).")
diff --git a/policy/tests/test_engine.py b/policy/tests/test_engine.py
new file mode 100644
index 00000000..6fe11b35
--- /dev/null
+++ b/policy/tests/test_engine.py
@@ -0,0 +1,142 @@
+#
+# 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 unittest.mock import patch
+
+from django.test import TestCase
+
+from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
+from policy.engine import evaluate_rule
+from policy.engine import evaluate_rules
+from policy.engine import get_effective_config
+from product_portfolio.models import ProductPolicyViolation
+from product_portfolio.tests import make_product
+
+RULE_TYPE = "usage_policy_error"
+
+
+class EvaluateRuleTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ def _count_violations(self, count):
+ """Patch count_violations for RULE_TYPE to return a fixed count."""
+ return patch(
+ "policy.rules.UsagePolicyErrorRule.count_violations",
+ return_value=count,
+ )
+
+ def test_evaluate_rule_creates_violation_on_first_trigger(self):
+ with self._count_violations(3):
+ violation, created, resolved_count = evaluate_rule(RULE_TYPE, self.product, 0, {})
+ self.assertTrue(created)
+ self.assertEqual(0, resolved_count)
+ self.assertEqual(3, violation.violation_count)
+ self.assertFalse(violation.resolved)
+ self.assertEqual(1, ProductPolicyViolation.objects.count())
+
+ def test_evaluate_rule_resolves_violation_when_count_drops_to_zero(self):
+ with self._count_violations(3):
+ evaluate_rule(RULE_TYPE, self.product, 0, {})
+ with self._count_violations(0):
+ violation, created, resolved_count = evaluate_rule(RULE_TYPE, self.product, 0, {})
+ self.assertIsNone(violation)
+ self.assertFalse(created)
+ self.assertEqual(1, resolved_count)
+ db_violation = ProductPolicyViolation.objects.get()
+ self.assertTrue(db_violation.resolved)
+ self.assertIsNotNone(db_violation.resolved_date)
+
+ def test_evaluate_rule_retrigger_after_resolution_does_not_raise(self):
+ with self._count_violations(3):
+ evaluate_rule(RULE_TYPE, self.product, 0, {})
+ with self._count_violations(0):
+ evaluate_rule(RULE_TYPE, self.product, 0, {})
+ with self._count_violations(5):
+ violation, created, resolved_count = evaluate_rule(RULE_TYPE, self.product, 0, {})
+ self.assertFalse(violation.resolved)
+ self.assertIsNone(violation.resolved_date)
+ self.assertEqual(5, violation.violation_count)
+ self.assertEqual(1, ProductPolicyViolation.objects.count())
+
+ def test_evaluate_rule_updates_count_on_existing_active_violation(self):
+ with self._count_violations(2):
+ evaluate_rule(RULE_TYPE, self.product, 0, {})
+ with self._count_violations(7):
+ violation, created, resolved_count = evaluate_rule(RULE_TYPE, self.product, 0, {})
+ self.assertFalse(created)
+ self.assertEqual(7, violation.violation_count)
+ self.assertEqual(1, ProductPolicyViolation.objects.count())
+
+
+class GetEffectiveConfigTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ def test_returns_defaults_when_no_dataspace_configuration(self):
+ config = get_effective_config("usage_policy_error", self.dataspace)
+ self.assertFalse(config["is_active"])
+ self.assertEqual(0, config["threshold"])
+ self.assertEqual({}, config["parameters"])
+
+ def test_reads_dataspace_override(self):
+ DataspaceConfiguration.objects.create(
+ dataspace=self.dataspace,
+ policy_rules_config={"usage_policy_error": {"is_active": True, "threshold": 5}},
+ )
+ self.dataspace.refresh_from_db()
+ config = get_effective_config("usage_policy_error", self.dataspace)
+ self.assertTrue(config["is_active"])
+ self.assertEqual(5, config["threshold"])
+
+ def test_falls_back_to_code_defaults_for_partial_config(self):
+ DataspaceConfiguration.objects.create(
+ dataspace=self.dataspace,
+ policy_rules_config={"usage_policy_error": {"is_active": True}},
+ )
+ self.dataspace.refresh_from_db()
+ config = get_effective_config("usage_policy_error", self.dataspace)
+ self.assertTrue(config["is_active"])
+ self.assertEqual(0, config["threshold"])
+
+
+class EvaluateRulesTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ DataspaceConfiguration.objects.create(
+ dataspace=self.dataspace,
+ policy_rules_config={"usage_policy_error": {"is_active": True}},
+ )
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.engine.fire_policy_webhooks")
+ @patch("policy.rules.UsagePolicyErrorRule.count_violations", return_value=3)
+ def test_evaluate_rules_creates_violation_for_active_rule(self, _mock_count, mock_fire):
+ new_violations, resolved_count = evaluate_rules(self.product)
+ self.assertEqual(1, len(new_violations))
+ self.assertEqual(0, resolved_count)
+ self.assertEqual(1, ProductPolicyViolation.objects.filter(resolved=False).count())
+ mock_fire.assert_called_once()
+
+ @patch("policy.engine.fire_policy_webhooks")
+ def test_evaluate_rules_resolves_violation_when_rule_inactive(self, _mock_fire):
+ ProductPolicyViolation.objects.create(
+ product=self.product,
+ dataspace=self.dataspace,
+ rule_type="license_coverage_gap",
+ violation_count=2,
+ )
+ new_violations, resolved_count = evaluate_rules(self.product)
+ self.assertEqual(0, len(new_violations))
+ self.assertGreater(resolved_count, 0)
+ self.assertTrue(
+ ProductPolicyViolation.objects.get(rule_type="license_coverage_gap").resolved
+ )
diff --git a/policy/tests/test_signals.py b/policy/tests/test_signals.py
new file mode 100644
index 00000000..a3d81620
--- /dev/null
+++ b/policy/tests/test_signals.py
@@ -0,0 +1,113 @@
+#
+# 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 unittest.mock import patch
+
+from django.test import TestCase
+
+from component_catalog.tests import make_package
+from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
+from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_package
+
+
+class ProductSaveSignalTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+
+ @patch("policy.signals.evaluate_product_rules_task.delay")
+ def test_product_save_queues_task(self, mock_delay):
+ product = make_product(self.dataspace)
+ mock_delay.assert_called_with(product_uuid=product.uuid)
+
+ @patch("policy.signals.evaluate_product_rules_task.delay")
+ def test_product_update_queues_task(self, mock_delay):
+ product = make_product(self.dataspace)
+ mock_delay.reset_mock()
+ product.save()
+ mock_delay.assert_called_once_with(product_uuid=product.uuid)
+
+
+class ProductPackageSignalTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.signals.evaluate_product_rules_task.delay")
+ def test_productpackage_save_queues_task(self, mock_delay):
+ make_product_package(self.product)
+ mock_delay.assert_called_with(product_uuid=self.product.uuid)
+
+ @patch("policy.signals.evaluate_product_rules_task.delay")
+ def test_productpackage_delete_queues_task(self, mock_delay):
+ pp = make_product_package(self.product)
+ mock_delay.reset_mock()
+ pp.delete()
+ mock_delay.assert_called_once_with(product_uuid=self.product.uuid)
+
+
+class PackageSaveSignalTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_package_create_does_not_queue_task(self, mock_delay):
+ make_package(self.dataspace)
+ mock_delay.assert_not_called()
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_package_update_with_products_queues_task(self, mock_delay):
+ package = make_package(self.dataspace)
+ make_product_package(self.product, package=package)
+ mock_delay.reset_mock()
+ package.save()
+ mock_delay.assert_called_once()
+ called_uuids = mock_delay.call_args[1]["product_uuids"]
+ self.assertIn(self.product.uuid, called_uuids)
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_package_update_without_products_does_not_queue_task(self, mock_delay):
+ package = make_package(self.dataspace)
+ mock_delay.reset_mock()
+ package.save()
+ mock_delay.assert_not_called()
+
+
+class DataspaceConfigurationSignalTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.config = DataspaceConfiguration.objects.create(dataspace=self.dataspace)
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_full_save_queues_task(self, mock_delay):
+ self.config.save()
+ mock_delay.assert_called_once()
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_policy_rules_config_update_fields_queues_task(self, mock_delay):
+ self.config.policy_rules_config = {"usage_policy_error": {"is_active": True}}
+ self.config.save(update_fields=["policy_rules_config"])
+ mock_delay.assert_called_once()
+ called_uuids = mock_delay.call_args[1]["product_uuids"]
+ self.assertIn(self.product.uuid, called_uuids)
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_unrelated_update_fields_does_not_queue_task(self, mock_delay):
+ self.config.save(update_fields=["scancodeio_url"])
+ mock_delay.assert_not_called()
+
+ @patch("policy.signals.evaluate_all_products_rules_task.delay")
+ def test_no_products_in_dataspace_does_not_queue_task(self, mock_delay):
+ empty_dataspace = Dataspace.objects.create(name="Empty")
+ empty_config = DataspaceConfiguration.objects.create(dataspace=empty_dataspace)
+ mock_delay.reset_mock()
+ empty_config.save(update_fields=["policy_rules_config"])
+ mock_delay.assert_not_called()
diff --git a/policy/tests/test_tasks.py b/policy/tests/test_tasks.py
new file mode 100644
index 00000000..dcc470e7
--- /dev/null
+++ b/policy/tests/test_tasks.py
@@ -0,0 +1,139 @@
+#
+# 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.
+#
+
+import uuid
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from dje.models import Dataspace
+from policy.engine import fire_policy_webhooks
+from policy.tasks import evaluate_all_products_rules_task
+from policy.tasks import evaluate_product_rules_task
+from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_status
+
+
+class FirePolicyWebhooksTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.engine.fire_webhooks")
+ def test_fire_policy_webhooks_dispatches_violation_detected(self, mock_fire):
+ violation = MagicMock()
+ violation.rule_label = "Usage Policy Error"
+ violation.violation_count = 3
+ fire_policy_webhooks(self.product, new_violations=[violation], resolved_count=0)
+ mock_fire.assert_called_once()
+ event_name, kwargs = mock_fire.call_args[0][0], mock_fire.call_args[1]
+ self.assertEqual("policy.violation_detected", event_name)
+ self.assertIn("Policy violations detected", kwargs["payload_override"]["text"])
+ self.assertIn("Usage Policy Error", kwargs["payload_override"]["text"])
+
+ @patch("policy.engine.fire_webhooks")
+ def test_fire_policy_webhooks_dispatches_violation_resolved(self, mock_fire):
+ fire_policy_webhooks(self.product, new_violations=[], resolved_count=2)
+ mock_fire.assert_called_once()
+ event_name, kwargs = mock_fire.call_args[0][0], mock_fire.call_args[1]
+ self.assertEqual("policy.violation_resolved", event_name)
+ self.assertIn("2 policy violation(s) resolved", kwargs["payload_override"]["text"])
+
+ @patch("policy.engine.fire_webhooks")
+ def test_fire_policy_webhooks_dispatches_both_events(self, mock_fire):
+ violation = MagicMock()
+ violation.rule_label = "License Coverage Gap"
+ violation.violation_count = 1
+ fire_policy_webhooks(self.product, new_violations=[violation], resolved_count=1)
+ self.assertEqual(2, mock_fire.call_count)
+ events_fired = [c[0][0] for c in mock_fire.call_args_list]
+ self.assertIn("policy.violation_detected", events_fired)
+ self.assertIn("policy.violation_resolved", events_fired)
+
+ @patch("policy.engine.fire_webhooks")
+ def test_fire_policy_webhooks_silent_when_no_changes(self, mock_fire):
+ fire_policy_webhooks(self.product, new_violations=[], resolved_count=0)
+ mock_fire.assert_not_called()
+
+
+class EvaluateProductRulesTaskTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_product_rules_task_runs_evaluation(self, mock_evaluate):
+ mock_evaluate.return_value = ([], 0)
+ evaluate_product_rules_task(product_uuid=self.product.uuid)
+ mock_evaluate.assert_called_once_with(self.product)
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_product_rules_task_unknown_uuid_logs_error(self, mock_evaluate):
+ with self.assertLogs("policy.tasks", level="ERROR") as captured:
+ evaluate_product_rules_task(product_uuid=uuid.uuid4())
+ mock_evaluate.assert_not_called()
+ self.assertTrue(any("not found" in line for line in captured.output))
+
+
+class EvaluateAllProductsRulesTaskTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_all_products_excludes_locked_by_default(self, mock_evaluate):
+ # Regression: previously called .exclude_locked() on DataspacedQuerySet which lacks that
+ # method. Now uses .exclude(configuration_status__is_locked=True) inline.
+ mock_evaluate.return_value = ([], 0)
+ active_product = make_product(self.dataspace)
+ locked_status = make_product_status(self.dataspace, is_locked=True)
+ locked_product = make_product(self.dataspace, configuration_status=locked_status)
+ mock_evaluate.reset_mock()
+
+ evaluate_all_products_rules_task()
+
+ evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+ self.assertIn(active_product, evaluated_products)
+ self.assertNotIn(locked_product, evaluated_products)
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_all_products_includes_locked_when_requested(self, mock_evaluate):
+ mock_evaluate.return_value = ([], 0)
+ locked_status = make_product_status(self.dataspace, is_locked=True)
+ locked_product = make_product(self.dataspace, configuration_status=locked_status)
+ mock_evaluate.reset_mock()
+
+ evaluate_all_products_rules_task(include_locked=True)
+
+ evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+ self.assertIn(locked_product, evaluated_products)
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_all_products_filters_by_uuids(self, mock_evaluate):
+ mock_evaluate.return_value = ([], 0)
+ product_a = make_product(self.dataspace)
+ product_b = make_product(self.dataspace)
+ mock_evaluate.reset_mock()
+
+ evaluate_all_products_rules_task(product_uuids=[product_a.uuid])
+
+ evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+ self.assertIn(product_a, evaluated_products)
+ self.assertNotIn(product_b, evaluated_products)
+
+ @patch("policy.tasks.evaluate_rules")
+ def test_evaluate_all_products_uuid_filter_still_excludes_locked(self, mock_evaluate):
+ mock_evaluate.return_value = ([], 0)
+ locked_status = make_product_status(self.dataspace, is_locked=True)
+ locked_product = make_product(self.dataspace, configuration_status=locked_status)
+ mock_evaluate.reset_mock()
+
+ evaluate_all_products_rules_task(product_uuids=[locked_product.uuid])
+
+ evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+ self.assertNotIn(locked_product, evaluated_products)
diff --git a/product_portfolio/admin.py b/product_portfolio/admin.py
index c9624c07..36ece772 100644
--- a/product_portfolio/admin.py
+++ b/product_portfolio/admin.py
@@ -46,6 +46,7 @@
from dje.permissions import assign_all_object_permissions
from dje.permissions import get_limited_perms_for_model
from dje.utils import is_purl_fragment
+from policy.tasks import evaluate_all_products_rules_task
from product_portfolio.filters import ComponentCompletenessListFilter
from product_portfolio.forms import ProductAdminForm
from product_portfolio.forms import ProductComponentAdminForm
@@ -384,7 +385,7 @@ class ProductAdmin(
ProductPackageInline,
]
form = ProductAdminForm
- actions = []
+ actions = ["evaluate_policy_rules"]
actions_to_remove = ["copy_to", "compare_with", "delete_selected"]
navigation_buttons = True
activity_log = False
@@ -395,6 +396,15 @@ class ProductAdmin(
awesomplete_data = {"primary_language": PROGRAMMING_LANGUAGES}
readonly_fields = DataspacedAdmin.readonly_fields + ("get_feature_datalist",)
+ def evaluate_policy_rules(self, request, queryset):
+ product_uuids = list(queryset.values_list("uuid", flat=True))
+ evaluate_all_products_rules_task.delay(product_uuids=product_uuids)
+ self.message_user(
+ request, f"Policy rules evaluation enqueued for {len(product_uuids)} product(s)."
+ )
+
+ evaluate_policy_rules.short_description = _("Evaluate policy rules")
+
def get_feature_datalist(self, obj):
if obj.pk:
return obj.get_feature_datalist()
diff --git a/product_portfolio/api.py b/product_portfolio/api.py
index c8d9998f..5c74d9c0 100644
--- a/product_portfolio/api.py
+++ b/product_portfolio/api.py
@@ -48,6 +48,7 @@
from product_portfolio.models import ProductComponent
from product_portfolio.models import ProductDependency
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.models import ScanCodeProject
from vulnerabilities.api import VulnerabilityAnalysisSerializer
@@ -343,6 +344,25 @@ class Meta:
)
+class ProductPolicyViolationSerializer(serializers.ModelSerializer):
+ rule_label = serializers.ReadOnlyField()
+ rule_description = serializers.ReadOnlyField()
+ rule_severity = serializers.ReadOnlyField()
+
+ class Meta:
+ model = ProductPolicyViolation
+ fields = (
+ "rule_type",
+ "rule_label",
+ "rule_description",
+ "rule_severity",
+ "violation_count",
+ "detected_date",
+ "resolved",
+ "resolved_date",
+ )
+
+
class ProductViewSet(
ObjectPermissionsMixin,
SendAboutFilesMixin,
@@ -413,6 +433,14 @@ def imports(self, request, uuid):
projects_data = ScanCodeProjectSerializer(scancode_projects, many=True).data
return Response(projects_data)
+ @action(detail=True, url_path="policy_violations")
+ def policy_violations(self, request, uuid):
+ """List active policy violations for this product, with rule details and counts."""
+ product = self.get_object()
+ violations = product.policy_violations.unresolved()
+ serializer = ProductPolicyViolationSerializer(violations, many=True)
+ return Response(serializer.data)
+
@action(detail=True, methods=["post"], serializer_class=LoadSBOMsFormSerializer)
def load_sboms(self, request, *args, **kwargs):
"""
diff --git a/product_portfolio/filters.py b/product_portfolio/filters.py
index 98f6f369..f6521aae 100644
--- a/product_portfolio/filters.py
+++ b/product_portfolio/filters.py
@@ -31,11 +31,13 @@
from dje.widgets import DropDownRightWidget
from dje.widgets import DropDownWidget
from license_library.models import License
+from policy.rules import RULE_REGISTRY
from product_portfolio.models import CodebaseResource
from product_portfolio.models import Product
from product_portfolio.models import ProductComponent
from product_portfolio.models import ProductDependency
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.models import ProductStatus
from vulnerabilities.filters import ScoreRangeFilter
from vulnerabilities.models import RISK_SCORE_RANGES
@@ -149,6 +151,10 @@ class ProductFilterSet(DataspacedFilterSet):
label=_("License issues"),
method="filter_license_compliance_issues",
)
+ policy_violations = django_filters.BooleanFilter(
+ label=_("Policy violations"),
+ method="filter_policy_violations",
+ )
class Meta:
model = Product
@@ -189,6 +195,16 @@ def filter_license_compliance_issues(self, queryset, name, value):
condition = Exists(has_alert)
return queryset.filter(condition if value else ~condition)
+ def filter_policy_violations(self, queryset, name, value):
+ if value is None:
+ return queryset
+ has_violation = ProductPolicyViolation.objects.filter(
+ product_id=OuterRef("pk"),
+ rule_type__in=RULE_REGISTRY.keys(),
+ ).unresolved()
+ condition = Exists(has_violation)
+ return queryset.filter(condition if value else ~condition)
+
class BaseProductRelationFilterSet(DataspacedFilterSet):
field_name_prefix = None
@@ -392,6 +408,7 @@ class ProductPackageFilterSet(BaseProductRelationFilterSet):
field_name="package__usage_policy__compliance_alert",
distinct=True,
)
+ policy_rule = django_filters.CharFilter(method="filter_by_policy_rule")
class Meta:
model = ProductPackage
@@ -407,6 +424,13 @@ class Meta:
"exploitability",
]
+ def filter_by_policy_rule(self, queryset, name, value):
+ """Filter packages that triggered the given policy rule type."""
+ handler = RULE_REGISTRY.get(value)
+ if not handler:
+ return queryset
+ return queryset.filter(**handler.get_package_filter())
+
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.filters["vulnerability_analyses__state"].extra["null_label"] = "(No values)"
diff --git a/product_portfolio/migrations/0018_productpolicyviolation.py b/product_portfolio/migrations/0018_productpolicyviolation.py
new file mode 100644
index 00000000..9b38d420
--- /dev/null
+++ b/product_portfolio/migrations/0018_productpolicyviolation.py
@@ -0,0 +1,37 @@
+# Generated by Django 6.0.6 on 2026-07-15 08:25
+
+import django.db.models.deletion
+import dje.models
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('dje', '0016_dataspaceconfiguration_policy_rules_config'),
+ ('product_portfolio', '0017_scancodeproject_import_options'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='ProductPolicyViolation',
+ fields=[
+ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+ ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, verbose_name='UUID')),
+ ('violation_count', models.PositiveIntegerField(default=0, help_text='Number of objects currently violating the rule.')),
+ ('detected_date', models.DateTimeField(auto_now_add=True, help_text='The date and time when this violation was first detected.')),
+ ('last_checked', models.DateTimeField(auto_now=True, help_text='The date and time of the last evaluation.')),
+ ('resolved', models.BooleanField(default=False, help_text='Indicates whether this violation has been resolved.')),
+ ('resolved_date', models.DateTimeField(blank=True, help_text='The date and time when this violation was resolved.', null=True)),
+ ('rule_type', models.CharField(help_text='The rule type from the rule registry that triggered this violation.', max_length=50)),
+ ('dataspace', models.ForeignKey(editable=False, help_text='A Dataspace is an independent, exclusive set of DejaCode data, which can be either nexB master reference data or installation-specific data.', on_delete=django.db.models.deletion.PROTECT, to='dje.dataspace')),
+ ('product', models.ForeignKey(help_text='The product in the context of which this violation was detected.', on_delete=django.db.models.deletion.CASCADE, related_name='policy_violations', to='product_portfolio.product')),
+ ],
+ options={
+ 'ordering': ['-detected_date'],
+ 'unique_together': {('dataspace', 'uuid'), ('rule_type', 'product')},
+ },
+ bases=(dje.models.DataspaceForeignKeyValidationMixin, models.Model),
+ ),
+ ]
diff --git a/product_portfolio/models.py b/product_portfolio/models.py
index 9fb8e6c9..70ddb03d 100644
--- a/product_portfolio/models.py
+++ b/product_portfolio/models.py
@@ -24,6 +24,7 @@
from django.db.models import Max
from django.db.models import OuterRef
from django.db.models import Q
+from django.db.models import Subquery
from django.db.models import Value
from django.db.models import When
from django.db.models.functions import Coalesce
@@ -56,6 +57,8 @@
from dje.validators import generic_uri_validator
from dje.validators import validate_url_segment
from dje.validators import validate_version
+from policy.models import AbstractPolicyViolation
+from policy.rules import RULE_REGISTRY
from vulnerabilities.fetch import fetch_for_packages
from vulnerabilities.models import AffectedByVulnerabilityMixin
from vulnerabilities.models import AffectedByVulnerabilityRelationship
@@ -139,6 +142,9 @@ class Meta(BaseStatusMixin.Meta):
class ProductQuerySet(DataspacedQuerySet):
+ def exclude_locked(self):
+ return self.exclude(configuration_status__is_locked=True)
+
def with_risk_threshold(self):
return self.annotate(
risk_threshold=Coalesce(
@@ -213,6 +219,7 @@ def with_compliance_data(self):
self.with_risk_threshold()
.with_vulnerability_counts()
.with_license_compliance_counts()
+ .with_policy_violation_count()
.annotate(package_count=Count("productpackages", distinct=True))
.order_by(
F("max_risk_score").desc(nulls_last=True),
@@ -224,12 +231,13 @@ def with_compliance_data(self):
)
def with_compliance_issues(self):
- """Filter to products that have license or critical/high vulnerability issues."""
+ """Filter to products that have license, vulnerability, or policy violation issues."""
return self.filter(
Q(license_error_count__gt=0)
| Q(license_warning_count__gt=0)
| Q(critical_count__gt=0)
| Q(high_count__gt=0)
+ | Q(policy_violation_count__gt=0)
)
def with_has_vulnerable_packages(self):
@@ -240,6 +248,22 @@ def with_has_vulnerable_packages(self):
has_vulnerable_packages=Exists(vulnerable_productpackage_qs),
)
+ def with_policy_violation_count(self):
+ subquery = (
+ ProductPolicyViolation.objects.filter(
+ product=OuterRef("pk"),
+ )
+ .unresolved()
+ .values("product")
+ .annotate(violation_count=models.Count("id"))
+ .values("violation_count")
+ )
+ return self.annotate(
+ policy_violation_count=Coalesce(
+ Subquery(subquery, output_field=models.IntegerField()), Value(0)
+ ),
+ )
+
class ProductSecuredManager(DataspacedManager):
"""
@@ -286,7 +310,7 @@ def get_queryset(
).scope(user.dataspace)
if exclude_locked:
- queryset = queryset.exclude(configuration_status__is_locked=True)
+ queryset = queryset.exclude_locked()
if include_inactive:
return queryset
@@ -408,7 +432,7 @@ class Product(
# WARNING: Bypass the security system implemented in ProductSecuredManager.
# This is to be used only in a few cases where the User scoping is not appropriated.
# For example: `self.dataspace.product_set(manager='unsecured_objects').count()`
- unsecured_objects = DataspacedManager()
+ unsecured_objects = DataspacedManager.from_queryset(ProductQuerySet)()
class Meta(BaseProductMixin.Meta):
permissions = (("view_product", "Can view product"),)
@@ -467,6 +491,9 @@ def get_export_license_compliance_url(self):
def get_export_security_compliance_url(self):
return self.get_url("export_security_compliance")
+ def get_evaluate_policy_rules_url(self):
+ return self.get_url("evaluate_policy_rules")
+
@property
def cyclonedx_bom_ref(self):
return str(self.uuid)
@@ -1901,3 +1928,48 @@ def save(self, *args, **kwargs):
"The 'for_package' cannot be the same as 'resolved_to_package'."
)
super().save(*args, **kwargs)
+
+
+class ProductPolicyViolationQuerySet(ProductSecuredQuerySet):
+ def unresolved(self):
+ return self.filter(resolved=False)
+
+
+class ProductPolicyViolation(DataspacedModel, AbstractPolicyViolation):
+ """Concrete policy violation scoped to a product."""
+
+ product = models.ForeignKey(
+ to="product_portfolio.Product",
+ on_delete=models.CASCADE,
+ related_name="policy_violations",
+ help_text=_("The product in the context of which this violation was detected."),
+ )
+ rule_type = models.CharField(
+ max_length=50,
+ help_text=_("The rule type from the rule registry that triggered this violation."),
+ )
+
+ objects = DataspacedManager.from_queryset(ProductPolicyViolationQuerySet)()
+
+ class Meta:
+ unique_together = (("dataspace", "uuid"), ("rule_type", "product"))
+ ordering = ["-detected_date"]
+
+ def __str__(self):
+ return f"{self.rule_type} / {self.product}: {self.violation_count} violation(s)"
+
+ @cached_property
+ def rule_handler(self):
+ return RULE_REGISTRY.get(self.rule_type)
+
+ @property
+ def rule_label(self):
+ return self.rule_handler.label if self.rule_handler else self.rule_type
+
+ @property
+ def rule_description(self):
+ return self.rule_handler.description if self.rule_handler else ""
+
+ @property
+ def rule_severity(self):
+ return self.rule_handler.severity if self.rule_handler else "warning"
diff --git a/product_portfolio/templates/product_portfolio/compliance/compliance_dashboard.html b/product_portfolio/templates/product_portfolio/compliance/compliance_dashboard.html
index bd258e85..cda97072 100644
--- a/product_portfolio/templates/product_portfolio/compliance/compliance_dashboard.html
+++ b/product_portfolio/templates/product_portfolio/compliance/compliance_dashboard.html
@@ -68,36 +68,42 @@
-
{% trans "Security issues" %}
-
- {{ products_with_critical_or_high }}
-
+
{% trans "License compliance" %}
+ {% if products_with_license_issues %}
+
+ {{ products_with_license_issues }}
+
+ {% else %}
+
+ {{ products_with_license_issues }}
+
+ {% endif %}
- {% if products_with_critical_or_high %}
- {% trans "products with critical/high vulnerabilities" %}
+ {% if products_with_license_issues %}
+ {% trans "products with license compliance alerts" %}
{% else %}
- {% trans "No critical or high vulnerabilities" %}
+ {% trans "All products within policy" %}
{% endif %}
@@ -127,8 +133,8 @@
{% trans "Product" %}
{% trans "Packages" %}
+ {% trans "Policy violations" %}
{% trans "License compliance" %}
- {% trans "Security compliance" %}
{% trans "Vulnerabilities" %}
@@ -146,6 +152,15 @@
{{ product.package_count|intcomma }}
+
+ {% if product.policy_violation_count %}
+
+ {{ product.policy_violation_count }} {% trans "rule" %}{{ product.policy_violation_count|pluralize }} {% trans "triggered" %}
+
+ {% else %}
+ {% trans "None" %}
+ {% endif %}
+
{% if product.license_error_count %}
@@ -161,19 +176,6 @@
-
- {% if product.max_risk_level == "critical" %}
- {% trans "Critical" %}
- {% elif product.max_risk_level == "high" %}
- {% trans "High" %}
- {% elif product.max_risk_level == "medium" %}
- {% trans "Medium" %}
- {% elif product.max_risk_level == "low" %}
- {% trans "Low" %}
- {% else %}
- {% trans "OK" %}
- {% endif %}
-
{% if product.risk_threshold and product.vulnerability_count %}
diff --git a/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html b/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html
index 92c0fe29..550bc451 100644
--- a/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html
+++ b/product_portfolio/templates/product_portfolio/compliance/compliance_panels.html
@@ -1,5 +1,116 @@
{% load i18n %}
{% url product.get_absolute_url as product_url %}
+
+{% if policy_violations %}
+
+
+
+
+
+
{% trans "Policy violations" %}
+
+
+
+
+
+ {% if has_change_permission %}
+
+ {% endif %}
+
+ {{ policy_violation_count }} {% trans "active" %}
+
+
+
+
+
+
+
+
+
+ {% for rule in all_rules %}
+
+
+ {% if rule.severity == "error" %}
+
{{ rule.label }}
+ {% else %}
+
{{ rule.label }}
+ {% endif %}
+
{{ rule.description }}
+
+
+ {% if not rule.is_active %}
+ {% trans "Disabled" %}
+ {% elif rule.is_violated %}
+ {% if rule.severity == "error" %}
+ {% trans "Triggered" %}
+ {% else %}
+ {% trans "Triggered" %}
+ {% endif %}
+ {% else %}
+ {% trans "OK" %}
+ {% endif %}
+
+
+ {% endfor %}
+
+
+
+
+
+
+
+ {% trans "Rule" %}
+ {% trans "Description" %}
+ {% trans "In violation" %}
+ {% trans "Detected" %}
+
+
+
+ {% for violation in policy_violations %}
+
+
+ {% if violation.rule_severity == "error" %}
+ {{ violation.rule_label }}
+ {% else %}
+ {{ violation.rule_label }}
+ {% endif %}
+
+ {{ violation.rule_description }}
+
+
+ {{ violation.violation_count }}
+
+
+ {{ violation.detected_date|date:"N j, Y" }}
+
+ {% endfor %}
+
+
+
+
+
+
+{% endif %}
+
{# License panel #}
@@ -170,38 +281,48 @@
{% trans "Security compliance" %}
{% else %}
{{ vulnerability_count }}
{% trans "vulnerabilit" %}{{ vulnerability_count|pluralize:"y,ies" }}
- — {% trans "no risk threshold set" %}
+ - {% trans "no risk threshold set" %}
{% endif %}
{% endif %}
{# Vulnerability list #}
- {% for vulnerability in vulnerabilities %}
-
-
- {% if vulnerability.risk_level == "critical" %}
- {% trans "Critical" %}
- {% elif vulnerability.risk_level == "high" %}
- {% trans "High" %}
- {% elif vulnerability.risk_level == "medium" %}
- {% trans "Medium" %}
- {% elif vulnerability.risk_level == "low" %}
- {% trans "Low" %}
- {% else %}
- {% trans "Unknown" %}
- {% endif %}
-
-
- {{ vulnerability.advisory_id }}
-
-
{{ vulnerability.summary|truncatechars:70 }}
-
- {% empty %}
-
-
- {% trans "No known vulnerabilities" %}
-
- {% endfor %}
+
+
+ {% for vulnerability in vulnerabilities %}
+
+
+
+ {% if vulnerability.risk_level == "critical" %}
+ {% trans "Critical" %}
+ {% elif vulnerability.risk_level == "high" %}
+ {% trans "High" %}
+ {% elif vulnerability.risk_level == "medium" %}
+ {% trans "Medium" %}
+ {% elif vulnerability.risk_level == "low" %}
+ {% trans "Low" %}
+ {% else %}
+ {% trans "Unknown" %}
+ {% endif %}
+
+
+
+
+ {{ vulnerability.advisory_id }}
+
+
+ {{ vulnerability.summary|truncatechars:70 }}
+
+ {% empty %}
+
+
+
+ {% trans "No known vulnerabilities" %}
+
+
+ {% endfor %}
+
+
{# View all link #}
{% if vulnerabilities|length < vulnerability_count %}
diff --git a/product_portfolio/templates/product_portfolio/compliance/metric_cards.html b/product_portfolio/templates/product_portfolio/compliance/metric_cards.html
index 05a4ecba..14bd04e8 100644
--- a/product_portfolio/templates/product_portfolio/compliance/metric_cards.html
+++ b/product_portfolio/templates/product_portfolio/compliance/metric_cards.html
@@ -1,21 +1,22 @@
{% load i18n humanize %}
-
+
-
{% trans "Total packages" %}
-
{{ total_packages|intcomma }}
-
+ {% endif %}
-
+
{% trans "License compliance" %}
@@ -32,7 +33,7 @@
-
+
{% trans "License coverage" %}
@@ -49,7 +50,7 @@
-
+
{% trans "Vulnerabilities" %}
{% if vulnerability_count == 0 %}
@@ -63,14 +64,18 @@
{% elif risk_threshold_number %}
{{ above_threshold_count }}
- {% trans "of" %} {{ vulnerability_count }} {% trans "above threshold of" %} {{ risk_threshold_number }}
+
+ {% trans "of" %} {{ vulnerability_count }} {% trans "above threshold of" %} {{ risk_threshold_number }}
+
{% else %}
{{ vulnerability_count }}
- {% if critical_count %}{{ critical_count }} {% trans "critical" %}{% endif %}{% if critical_count and high_count %}, {% endif %}{% if high_count %}{{ high_count }} {% trans "high" %}{% endif %}{% if medium_count and critical_count or medium_count and high_count %}, {% endif %}{% if medium_count %}{{ medium_count }} {% trans "medium" %}{% endif %}{% if low_count and vulnerability_count == low_count %}{{ low_count }} {% trans "low" %}{% endif %}
+
+ {% if critical_count %}{{ critical_count }} {% trans "critical" %}{% endif %}{% if critical_count and high_count %}, {% endif %}{% if high_count %}{{ high_count }} {% trans "high" %}{% endif %}{% if medium_count and critical_count or medium_count and high_count %}, {% endif %}{% if medium_count %}{{ medium_count }} {% trans "medium" %}{% endif %}{% if low_count and vulnerability_count == low_count %}{{ low_count }} {% trans "low" %}{% endif %}
+
{% endif %}
diff --git a/product_portfolio/tests/test_admin.py b/product_portfolio/tests/test_admin.py
index 8caeaf65..32836d5c 100644
--- a/product_portfolio/tests/test_admin.py
+++ b/product_portfolio/tests/test_admin.py
@@ -6,6 +6,8 @@
# See https://aboutcode.org for more information about AboutCode FOSS projects.
#
+from unittest.mock import patch
+
from django.core.exceptions import NON_FIELD_ERRORS
from django.test import TestCase
from django.urls import NoReverseMatch
@@ -101,7 +103,11 @@ def test_product_admin_form_clean_license_expression_in_alternate_dataspace(self
def test_product_security_admin_changelist_available_actions(self):
self.client.login(username=self.user.username, password="secret")
response = self.client.get(self.product_changelist_url)
- expected = [("", "---------"), ("mass_update", "Mass update")]
+ expected = [
+ ("", "---------"),
+ ("evaluate_policy_rules", "Evaluate policy rules"),
+ ("mass_update", "Mass update"),
+ ]
self.assertEqual(expected, response.context_data["action_form"].fields["action"].choices)
with self.assertRaises(NoReverseMatch):
@@ -539,3 +545,26 @@ def test_product_dependencies_add_view(self):
dependency = self.product1.dependencies.get()
self.assertEqual(self.package1, dependency.for_package)
self.assertEqual(package2, dependency.resolved_to_package)
+
+
+class EvaluatePolicyRulesActionTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.super_user = create_superuser("nexb_user", self.dataspace)
+ self.product1 = Product.objects.create(name="Product1", dataspace=self.dataspace)
+ self.product2 = Product.objects.create(name="Product2", dataspace=self.dataspace)
+
+ @patch("product_portfolio.admin.evaluate_all_products_rules_task.delay")
+ def test_evaluate_policy_rules_action_queues_task_for_selected_products(self, mock_delay):
+ self.client.login(username="nexb_user", password="secret")
+ url = reverse("admin:product_portfolio_product_changelist")
+ data = {
+ "action": "evaluate_policy_rules",
+ "_selected_action": [self.product1.pk, self.product2.pk],
+ }
+ response = self.client.post(url, data, follow=True)
+ self.assertEqual(200, response.status_code)
+ mock_delay.assert_called_once()
+ called_uuids = set(mock_delay.call_args[1]["product_uuids"])
+ self.assertIn(self.product1.uuid, called_uuids)
+ self.assertIn(self.product2.uuid, called_uuids)
diff --git a/product_portfolio/tests/test_admin_guardian.py b/product_portfolio/tests/test_admin_guardian.py
index e23a6997..12de6067 100644
--- a/product_portfolio/tests/test_admin_guardian.py
+++ b/product_portfolio/tests/test_admin_guardian.py
@@ -58,7 +58,11 @@ def test_product_guardian_admin_security_attributes(self):
# actions = []
# actions_to_remove = ['copy_to', 'compare_with', 'delete_selected']
- expected = [("", "---------"), ("mass_update", "Mass update")]
+ expected = [
+ ("", "---------"),
+ ("evaluate_policy_rules", "Evaluate policy rules"),
+ ("mass_update", "Mass update"),
+ ]
self.assertEqual(expected, response.context_data["action_form"].fields["action"].choices)
# activity_log = False
diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py
index ce43a34e..23803db8 100644
--- a/product_portfolio/tests/test_api.py
+++ b/product_portfolio/tests/test_api.py
@@ -43,6 +43,7 @@
from product_portfolio.models import ProductComponent
from product_portfolio.models import ProductItemPurpose
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.models import ProductRelationStatus
from product_portfolio.models import ProductStatus
from product_portfolio.models import ScanCodeProject
@@ -716,6 +717,47 @@ def test_api_product_endpoint_manage_permissions_action(self):
self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
self.assertIn("errors", response.data)
+ def test_api_product_endpoint_policy_violations_action(self):
+ url = reverse("api_v2:product-policy-violations", args=[self.product1.uuid])
+
+ self.client.login(username=self.base_user.username, password="secret")
+ response = self.client.get(url)
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
+ add_perm(self.base_user, "add_product")
+ assign_perm("view_product", self.base_user, self.product1)
+
+ response = self.client.get(url)
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual([], response.data)
+
+ violation = ProductPolicyViolation.objects.create(
+ product=self.product1,
+ dataspace=self.dataspace,
+ rule_type="usage_policy_error",
+ violation_count=5,
+ )
+
+ response = self.client.get(url)
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual(1, len(response.data))
+ entry = response.data[0]
+ self.assertEqual("usage_policy_error", entry["rule_type"])
+ self.assertEqual(5, entry["violation_count"])
+ self.assertFalse(entry["resolved"])
+ self.assertIsNone(entry["resolved_date"])
+ self.assertIn("rule_label", entry)
+ self.assertIn("rule_description", entry)
+ self.assertIn("rule_severity", entry)
+ self.assertIn("detected_date", entry)
+
+ violation.resolved = True
+ violation.save()
+
+ response = self.client.get(url)
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual([], response.data)
+
class ProductRelatedAPITestCase(TestCase):
def setUp(self):
diff --git a/product_portfolio/tests/test_filters.py b/product_portfolio/tests/test_filters.py
index ce167547..024a68e2 100644
--- a/product_portfolio/tests/test_filters.py
+++ b/product_portfolio/tests/test_filters.py
@@ -6,15 +6,22 @@
# See https://aboutcode.org for more information about AboutCode FOSS projects.
#
+from django.contrib.contenttypes.models import ContentType
from django.test import TestCase
+from component_catalog.models import Package
from component_catalog.tests import make_package
from dje.models import Dataspace
from license_library.models import License
from organization.models import Owner
+from policy.models import UsagePolicy
+from product_portfolio.filters import ProductFilterSet
from product_portfolio.filters import ProductPackageFilterSet
+from product_portfolio.models import Product
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_package
class ProductPackageFilterSetTestCase(TestCase):
@@ -81,3 +88,79 @@ def test_product_package_filterset_has_licenses(self):
data = {"has_licenses": "no"}
filterset = ProductPackageFilterSet(dataspace=self.dataspace, data=data)
self.assertQuerySetEqual(filterset.qs, [self.pp3])
+
+
+class ProductFilterSetPolicyTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product_with = make_product(self.dataspace)
+ self.product_without = make_product(self.dataspace)
+ ProductPolicyViolation.objects.create(
+ product=self.product_with,
+ dataspace=self.dataspace,
+ rule_type="usage_policy_error",
+ violation_count=1,
+ )
+
+ def _make_filterset(self, data):
+ return ProductFilterSet(
+ dataspace=self.dataspace,
+ data=data,
+ queryset=Product.unsecured_objects.filter(dataspace=self.dataspace),
+ )
+
+ def test_filter_policy_violations_true_returns_products_with_violations(self):
+ filterset = self._make_filterset({"policy_violations": "true"})
+ self.assertIn(self.product_with, filterset.qs)
+ self.assertNotIn(self.product_without, filterset.qs)
+
+ def test_filter_policy_violations_false_returns_products_without_violations(self):
+ filterset = self._make_filterset({"policy_violations": "false"})
+ self.assertIn(self.product_without, filterset.qs)
+ self.assertNotIn(self.product_with, filterset.qs)
+
+ def test_filter_policy_violations_excludes_stale_rule_types(self):
+ product_stale = make_product(self.dataspace)
+ ProductPolicyViolation.objects.create(
+ product=product_stale,
+ dataspace=self.dataspace,
+ rule_type="removed_rule",
+ violation_count=1,
+ )
+ filterset = self._make_filterset({"policy_violations": "true"})
+ self.assertNotIn(product_stale, filterset.qs)
+
+
+class ProductPackageFilterByRuleTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.owner = Owner.objects.create(name="Owner", dataspace=self.dataspace)
+ self.usage_policy = UsagePolicy.objects.create(
+ label="Error Policy",
+ icon="icon",
+ content_type=ContentType.objects.get_for_model(Package),
+ compliance_alert="error",
+ dataspace=self.dataspace,
+ )
+ self.product = make_product(self.dataspace)
+ self.pp_with_policy = make_product_package(
+ self.product,
+ make_package(self.dataspace, usage_policy=self.usage_policy),
+ )
+ self.pp_without_policy = make_product_package(self.product, make_package(self.dataspace))
+
+ def test_filter_by_policy_rule_filters_matching_packages(self):
+ filterset = ProductPackageFilterSet(
+ dataspace=self.dataspace,
+ data={"policy_rule": "usage_policy_error"},
+ )
+ self.assertIn(self.pp_with_policy, filterset.qs)
+ self.assertNotIn(self.pp_without_policy, filterset.qs)
+
+ def test_filter_by_unknown_rule_returns_all(self):
+ filterset = ProductPackageFilterSet(
+ dataspace=self.dataspace,
+ data={"policy_rule": "nonexistent_rule"},
+ )
+ self.assertIn(self.pp_with_policy, filterset.qs)
+ self.assertIn(self.pp_without_policy, filterset.qs)
diff --git a/product_portfolio/tests/test_models.py b/product_portfolio/tests/test_models.py
index f5abe9d2..9158613f 100644
--- a/product_portfolio/tests/test_models.py
+++ b/product_portfolio/tests/test_models.py
@@ -35,6 +35,7 @@
from product_portfolio.models import ProductComponentAssignedLicense
from product_portfolio.models import ProductInventoryItem
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.models import ProductRelationStatus
from product_portfolio.models import ProductSecuredManager
from product_portfolio.models import ScanCodeProject
@@ -1327,3 +1328,75 @@ def test_product_dependency_prackage_queryset_declared_dependencies_count(self):
qs = Package.objects.declared_dependencies_count(product=self.product1)
annotated_package1 = qs.filter(pk=package1.pk)[0]
self.assertEqual(1, annotated_package1.declared_dependencies_count)
+
+
+class ProductPolicyViolationTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.product = make_product(self.dataspace)
+
+ def _make_violation(self, rule_type="usage_policy_error", **kwargs):
+ return ProductPolicyViolation.objects.create(
+ product=self.product,
+ dataspace=self.dataspace,
+ rule_type=rule_type,
+ violation_count=kwargs.pop("violation_count", 1),
+ **kwargs,
+ )
+
+ def test_rule_label_returns_handler_label_for_known_rule(self):
+ violation = self._make_violation("usage_policy_error")
+ self.assertEqual("Usage Policy Error", violation.rule_label)
+
+ def test_rule_severity_returns_error_for_error_rule(self):
+ violation = self._make_violation("usage_policy_error")
+ self.assertEqual("error", violation.rule_severity)
+
+ def test_rule_severity_returns_warning_for_unknown_rule(self):
+ violation = self._make_violation("nonexistent_rule")
+ self.assertEqual("warning", violation.rule_severity)
+
+ def test_rule_label_falls_back_to_rule_type_for_unknown_rule(self):
+ violation = self._make_violation("nonexistent_rule")
+ self.assertEqual("nonexistent_rule", violation.rule_label)
+
+ def test_rule_description_empty_for_unknown_rule(self):
+ violation = self._make_violation("nonexistent_rule")
+ self.assertEqual("", violation.rule_description)
+
+ def test_with_policy_violation_count_returns_zero_when_no_violations(self):
+ qs = Product.unsecured_objects.filter(pk=self.product.pk).with_policy_violation_count()
+ self.assertEqual(0, qs.get().policy_violation_count)
+
+ def test_with_policy_violation_count_counts_unresolved_violations(self):
+ self._make_violation("usage_policy_error")
+ self._make_violation("license_coverage_gap")
+ self._make_violation("usage_policy_warning", resolved=True)
+ qs = Product.unsecured_objects.filter(pk=self.product.pk).with_policy_violation_count()
+ self.assertEqual(2, qs.get().policy_violation_count)
+
+ def test_with_compliance_issues_includes_product_with_policy_violation(self):
+ self._make_violation("usage_policy_error")
+ qs = (
+ Product.unsecured_objects.filter(pk=self.product.pk)
+ .with_compliance_data()
+ .with_compliance_issues()
+ )
+ self.assertIn(self.product, qs)
+
+ def test_with_compliance_issues_excludes_product_with_no_issues(self):
+ qs = (
+ Product.unsecured_objects.filter(pk=self.product.pk)
+ .with_compliance_data()
+ .with_compliance_issues()
+ )
+ self.assertNotIn(self.product, qs)
+
+ def test_exclude_locked_removes_locked_products(self):
+ locked_status = make_product_status(self.dataspace, is_locked=True)
+ locked_product = make_product(self.dataspace, configuration_status=locked_status)
+ qs = Product.unsecured_objects.filter(
+ pk__in=[self.product.pk, locked_product.pk]
+ ).exclude_locked()
+ self.assertIn(self.product, qs)
+ self.assertNotIn(locked_product, qs)
diff --git a/product_portfolio/tests/test_views.py b/product_portfolio/tests/test_views.py
index fa4b20ad..8effb74a 100644
--- a/product_portfolio/tests/test_views.py
+++ b/product_portfolio/tests/test_views.py
@@ -9,6 +9,7 @@
import io
import json
from unittest import mock
+from unittest.mock import patch
from urllib.parse import quote
from django.contrib.contenttypes.models import ContentType
@@ -31,6 +32,7 @@
from component_catalog.tests import make_package
from dejacode_toolkit import scancodeio
from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
from dje.models import History
from dje.outputs import get_spdx_extracted_licenses
from dje.tests import MaxQueryMixin
@@ -50,6 +52,7 @@
from product_portfolio.models import ProductComponent
from product_portfolio.models import ProductItemPurpose
from product_portfolio.models import ProductPackage
+from product_portfolio.models import ProductPolicyViolation
from product_portfolio.models import ProductRelationStatus
from product_portfolio.models import ProductStatus
from product_portfolio.models import ScanCodeProject
@@ -4241,3 +4244,116 @@ def test_product_portfolio_product_security_compliance_export_view_respects_perm
data = json.loads(response.content)
advisory_ids = [entry["advisory_id"] for entry in data]
self.assertIn(vulnerability.advisory_id, advisory_ids)
+
+
+class EvaluatePolicyRulesViewTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.super_user = create_superuser("nexb_user", self.dataspace)
+ self.basic_user = create_user("basic_user", self.dataspace)
+ self.product1 = Product.objects.create(
+ name="Product1", version="1.0", dataspace=self.dataspace
+ )
+
+ def test_get_returns_405(self):
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_evaluate_policy_rules_url()
+ response = self.client.get(url)
+ self.assertEqual(405, response.status_code)
+
+ def test_post_without_login_redirects(self):
+ url = self.product1.get_evaluate_policy_rules_url()
+ response = self.client.post(url)
+ self.assertEqual(302, response.status_code)
+
+ def test_post_without_change_perm_returns_404(self):
+ self.client.login(username="basic_user", password="secret")
+ url = self.product1.get_evaluate_policy_rules_url()
+ response = self.client.post(url)
+ self.assertEqual(404, response.status_code)
+
+ @patch("product_portfolio.views.evaluate_rules")
+ def test_post_with_change_perm_evaluates_and_returns_hx_refresh(self, mock_evaluate):
+ mock_evaluate.return_value = ([], 0)
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_evaluate_policy_rules_url()
+ response = self.client.post(url)
+ self.assertEqual(200, response.status_code)
+ self.assertEqual("true", response.headers["HX-Refresh"])
+ mock_evaluate.assert_called_once_with(self.product1)
+
+
+class TabCompliancePolicyContextTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.super_user = create_superuser("nexb_user", self.dataspace)
+ self.product1 = Product.objects.create(
+ name="Product1", version="1.0", dataspace=self.dataspace
+ )
+
+ def test_no_violations_context_is_empty(self):
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_url("tab_compliance")
+ response = self.client.get(url)
+ self.assertEqual(200, response.status_code)
+ self.assertEqual([], response.context["policy_violations"])
+ self.assertEqual(0, response.context["policy_violation_count"])
+
+ def test_active_violation_appears_in_context(self):
+ ProductPolicyViolation.objects.create(
+ product=self.product1,
+ dataspace=self.dataspace,
+ rule_type="usage_policy_error",
+ violation_count=3,
+ )
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_url("tab_compliance")
+ response = self.client.get(url)
+ self.assertEqual(1, response.context["policy_violation_count"])
+ self.assertEqual(1, len(response.context["policy_violations"]))
+
+ def test_stale_rule_type_filtered_from_context(self):
+ ProductPolicyViolation.objects.create(
+ product=self.product1,
+ dataspace=self.dataspace,
+ rule_type="removed_rule",
+ violation_count=1,
+ )
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_url("tab_compliance")
+ response = self.client.get(url)
+ self.assertEqual(0, response.context["policy_violation_count"])
+
+ def test_all_rules_context_reflects_active_status(self):
+ DataspaceConfiguration.objects.create(
+ dataspace=self.dataspace,
+ policy_rules_config={"usage_policy_error": {"is_active": True}},
+ )
+ self.client.login(username="nexb_user", password="secret")
+ url = self.product1.get_url("tab_compliance")
+ response = self.client.get(url)
+ all_rules = {rule["rule_type"]: rule for rule in response.context["all_rules"]}
+ self.assertTrue(all_rules["usage_policy_error"]["is_active"])
+ self.assertFalse(all_rules["license_coverage_gap"]["is_active"])
+
+
+class ComplianceDashboardPolicyViolationsTestCase(TestCase):
+ def setUp(self):
+ self.dataspace = Dataspace.objects.create(name="nexB")
+ self.super_user = create_superuser("nexb_user", self.dataspace)
+ self.product1 = Product.objects.create(
+ name="Product1", version="1.0", dataspace=self.dataspace
+ )
+
+ def test_products_with_policy_violations_count(self):
+ ProductPolicyViolation.objects.create(
+ product=self.product1,
+ dataspace=self.dataspace,
+ rule_type="usage_policy_error",
+ violation_count=2,
+ )
+ self.client.login(username="nexb_user", password="secret")
+ url = reverse("product_portfolio:compliance_dashboard")
+ response = self.client.get(url)
+ self.assertEqual(1, response.context["products_with_policy_violations"])
+ self.assertEqual(1, response.context["products_with_issues"])
diff --git a/product_portfolio/urls.py b/product_portfolio/urls.py
index 1c84b51f..8231594a 100644
--- a/product_portfolio/urls.py
+++ b/product_portfolio/urls.py
@@ -42,6 +42,7 @@
from product_portfolio.views import check_package_version_ajax_view
from product_portfolio.views import delete_scan_htmx_view
from product_portfolio.views import edit_productrelation_ajax_view
+from product_portfolio.views import evaluate_policy_rules_view
from product_portfolio.views import import_from_scan_view
from product_portfolio.views import import_packages_from_scancodeio_view
from product_portfolio.views import improve_packages_from_purldb_view
@@ -129,6 +130,7 @@ def product_path(path_segment, view):
*product_path("add_customcomponent_ajax", add_customcomponent_ajax_view),
*product_path("vulnerability_analysis_form", vulnerability_analysis_form_view),
*product_path("scan_all_packages", scan_all_packages_view),
+ *product_path("evaluate_policy_rules", evaluate_policy_rules_view),
*product_path("improve_packages_from_purldb", improve_packages_from_purldb_view),
*product_path("about_files", ProductSendAboutFilesView.as_view()),
*product_path("export_spdx", ProductExportSPDXDocumentView.as_view()),
diff --git a/product_portfolio/views.py b/product_portfolio/views.py
index f8e50d5f..a06cc416 100644
--- a/product_portfolio/views.py
+++ b/product_portfolio/views.py
@@ -112,6 +112,8 @@
from license_library.filters import LicenseFilterSet
from license_library.models import License
from license_library.models import LicenseAssignedTag
+from policy.engine import evaluate_rules
+from policy.rules import RULE_REGISTRY
from product_portfolio.filters import CodebaseResourceFilterSet
from product_portfolio.filters import DependencyFilterSet
from product_portfolio.filters import ProductComponentFilterSet
@@ -422,7 +424,10 @@ def tab_terms(self):
if self.object.notice_text:
notice_field = self.get_tab_fields([TabField("notice_text")])[0]
- tab_data["fields"].append(notice_field)
+ if tab_data is None:
+ tab_data = {"fields": [notice_field]}
+ else:
+ tab_data["fields"].append(notice_field)
return tab_data
@@ -2063,6 +2068,22 @@ def scan_all_packages_view(request, dataspace, name, version=""):
return redirect(product)
+@require_POST
+@login_required
+def evaluate_policy_rules_view(request, dataspace, name, version=""):
+ guarded_qs = Product.objects.get_queryset(request.user, perms="change_product")
+ product = get_object_or_404(
+ guarded_qs,
+ name=unquote_plus(name),
+ version=unquote_plus(version),
+ dataspace__name=dataspace,
+ )
+
+ evaluate_rules(product)
+
+ return HttpResponse(headers={"HX-Refresh": "true"})
+
+
@login_required
def import_from_scan_view(request, dataspace, name, version=""):
"""
@@ -2766,12 +2787,15 @@ def get_context_data(self, **kwargs):
product = self.object
productpackages = product.productpackages.all()
licenses = License.objects.filter(productpackage__in=productpackages)
+ user_perms = guardian_get_perms(self.request.user, product)
context.update(
{
**self.get_package_compliance_context(productpackages),
**self.get_license_compliance_context(licenses),
**self.get_security_compliance_context(product),
+ **self.get_policy_compliance_context(product),
+ "has_change_permission": "change_product" in user_perms,
}
)
@@ -2842,6 +2866,37 @@ def get_license_compliance_context(licenses, distribution_limit=10):
"remaining_license_count": max(0, len(license_distribution) - distribution_limit),
}
+ @staticmethod
+ def get_policy_compliance_context(product):
+ policy_violations = list(
+ product.policy_violations.filter(rule_type__in=RULE_REGISTRY.keys())
+ .unresolved()
+ .order_by("rule_type")
+ )
+ violated_rule_types = {violation.rule_type for violation in policy_violations}
+
+ try:
+ rules_config = product.dataspace.configuration.policy_rules_config or {}
+ except AttributeError:
+ rules_config = {}
+
+ all_rules = [
+ {
+ "label": handler.label,
+ "description": handler.description,
+ "rule_type": rule_type,
+ "severity": handler.severity,
+ "is_active": rules_config.get(rule_type, {}).get("is_active", False),
+ "is_violated": rule_type in violated_rule_types,
+ }
+ for rule_type, handler in RULE_REGISTRY.items()
+ ]
+ return {
+ "policy_violations": policy_violations,
+ "policy_violation_count": len(policy_violations),
+ "all_rules": all_rules,
+ }
+
@staticmethod
def get_security_compliance_context(product, display_limit=10):
risk_threshold = product.get_vulnerabilities_risk_threshold()
@@ -3049,6 +3104,7 @@ class ComplianceDashboardView(LoginRequiredMixin, ExportComplianceMixin, Dataspa
"medium_count": "Medium",
"low_count": "Low",
"vulnerability_count": "Total vulnerabilities",
+ "policy_violation_count": "Policy violations",
}
def get_queryset(self):
@@ -3069,6 +3125,8 @@ def get_context_data(self, **kwargs):
Q(license_error_count__gt=0) | Q(license_warning_count__gt=0)
).count()
+ products_with_policy_violations = products.filter(policy_violation_count__gt=0).count()
+
products_with_critical_or_high = products.filter(
Q(critical_count__gt=0) | Q(high_count__gt=0)
).count()
@@ -3086,6 +3144,7 @@ def get_context_data(self, **kwargs):
"total_products": context["paginator"].count,
"products_with_issues": products_with_issues,
"products_with_license_issues": products_with_license_issues,
+ "products_with_policy_violations": products_with_policy_violations,
"products_with_critical_or_high": products_with_critical_or_high,
"total_vulnerabilities": totals["total_vulnerabilities"] or 0,
"total_critical": totals["total_critical"] or 0,