feat: [five-c] Create rules-based engine #409

policy/admin.py

@@ -23,9 +23,12 @@
 from dje.admin import dejacode_site
 from dje.list_display import AsColored
 from policy.forms import AssociatedPolicyForm
+from policy.forms import PolicyRuleForm
 from policy.forms import UsagePolicyForm
 from policy.models import AssociatedPolicy
+from policy.models import PolicyRule
 from policy.models import UsagePolicy
+from policy.rules import RULE_REGISTRY
 
 License = apps.get_model("license_library", "license")
 
@@ -159,3 +162,37 @@ def download_license_dump_view(self, request):
         response["Content-Disposition"] = 'attachment; filename="license_policies.yml"'
 
         return response
+
+
+@admin.register(PolicyRule, site=dejacode_site)
+class PolicyRuleAdmin(DataspacedAdmin):
+    form = PolicyRuleForm
+    list_display = ("name", "rule_type", "threshold", "is_active", "get_dataspace")
+    list_filter = DataspacedAdmin.list_filter + ("rule_type", "is_active")
+    readonly_fields = DataspacedAdmin.readonly_fields + ("parameters_schema_hint",)
+    activity_log = False
+    actions = []
+    actions_to_remove = ["copy_to", "compare_with"]
+    email_notification_on = ()
+
+    short_description = (
+        "You can define Policy Rules that automatically detect compliance violations "
+        "across your products and trigger notifications."
+    )
+
+    long_description = linebreaksbr(
+        "A Policy Rule defines a type of automated check to run against your products. "
+        "When the number of detected issues exceeds the configured threshold, a "
+        "ProductPolicyViolation is recorded.\n"
+        "Set the rule type to match a registered evaluation handler and configure the "
+        "threshold (0 means any violation triggers the rule). "
+    )
+
+    def parameters_schema_hint(self, obj):
+        handler = RULE_REGISTRY.get(obj.rule_type)
+        if not handler or not handler.parameters_schema:
+            return "No parameters supported for this rule type."
+        lines = [f"{key}: {desc}" for key, desc in handler.parameters_schema.items()]
+        return mark_safe("<br>".join(lines))
+
+    parameters_schema_hint.short_description = "Supported parameters"

policy/engine.py

@@ -0,0 +1,58 @@
+#
+# 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 policy.models import PolicyRule
+from policy.rules import RULE_REGISTRY
+from product_portfolio.models import ProductPolicyViolation
+
+
+def evaluate_rule(policy_rule, product):
+    """
+    Evaluate a single PolicyRule against a product, create or update the
+    ProductPolicyViolation record.
+    Returns the ProductPolicyViolation instance, or None if no violation exists.
+    """
+    rule_handler = RULE_REGISTRY.get(policy_rule.rule_type)
+    if not rule_handler:
+        return
+
+    violation_count = rule_handler.count_violations(policy_rule, product)
+
+    lookup = {"policy_rule": policy_rule, "product": product, "resolved": False}
+
+    if violation_count > 0:
+        violation, created = ProductPolicyViolation.objects.get_or_create(
+            **lookup,
+            defaults={"dataspace": policy_rule.dataspace, "violation_count": violation_count},
+        )
+        if not created:
+            violation.violation_count = violation_count
+            violation.save()
+        return violation
+    else:
+        ProductPolicyViolation.objects.filter(**lookup).update(
+            resolved=True,
+            resolved_date=timezone.now(),
+        )
+        return
+
+
+def evaluate_rules(product):
+    """
+    Evaluate all active PolicyRules for the given product.
+    Returns the list of active ProductPolicyViolation instances.
+    """
+    violations = []
+    for policy_rule in PolicyRule.objects.scope(product.dataspace).active():
+        violation = evaluate_rule(policy_rule, product)
+        if violation:
+            violations.append(violation)
+
+    return violations

policy/forms.py

@@ -11,6 +11,8 @@
 
 from dje.forms import ColorCodeFormMixin
 from dje.forms import DataspacedAdminForm
+from policy.models import PolicyRule
+from policy.rules import RULE_REGISTRY
 
 
 class UsagePolicyForm(ColorCodeFormMixin, DataspacedAdminForm):
@@ -92,3 +94,21 @@ def get_ct(app_label, model):
             self.add_error("to_policy", msg)
 
         return cleaned_data
+
+
+class PolicyRuleForm(DataspacedAdminForm):
+    class Meta:
+        model = PolicyRule
+        fields = "__all__"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields["rule_type"].widget = forms.Select(
+            choices=[(key, handler.label) for key, handler in RULE_REGISTRY.items()]
+        )
+
+    def clean_rule_type(self):
+        value = self.cleaned_data["rule_type"]
+        if value not in RULE_REGISTRY:
+            raise forms.ValidationError(f"Unknown rule type: {value}")
+        return value

policy/migrations/0003_policyrule.py

@@ -0,0 +1,35 @@
+# Generated by Django 6.0.6 on 2026-07-08 08:39
+
+import django.db.models.deletion
+import dje.models
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dje', '0015_alter_dataspaceconfiguration_purldb_api_key_and_more'),
+        ('policy', '0002_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='PolicyRule',
+            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')),
+                ('name', models.CharField(help_text='Descriptive name for this policy rule.', max_length=100)),
+                ('rule_type', models.CharField(help_text='The type of evaluation performed by this rule.', max_length=50)),
+                ('threshold', models.PositiveIntegerField(default=0, help_text='Minimum number of violations required to trigger this rule (0 means any).')),
+                ('is_active', models.BooleanField(default=True, help_text='Only active rules are evaluated.')),
+                ('parameters', models.JSONField(blank=True, default=dict, help_text='Optional rule-specific parameters as a JSON object. Supported keys depend on the chosen rule type.')),
+                ('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')),
+            ],
+            options={
+                'ordering': ['name'],
+                'unique_together': {('dataspace', 'name'), ('dataspace', 'uuid')},
+            },
+            bases=(dje.models.DataspaceForeignKeyValidationMixin, models.Model),
+        ),
+    ]

policy/models.py

@@ -20,6 +20,7 @@
 from dje.models import DataspacedModel
 from dje.models import DataspacedQuerySet
 from dje.models import colored_icon_mixin_factory
+from policy.rules import RULE_REGISTRY
 
 ColoredIconMixin = colored_icon_mixin_factory(
     verbose_name="usage policy",
@@ -244,3 +245,83 @@ def save(self, *args, **kwargs):
         if self.from_policy.content_type == self.to_policy.content_type:
             raise AssertionError
         super().save(*args, **kwargs)
+
+
+class PolicyRuleQuerySet(DataspacedQuerySet):
+    def active(self):
+        return self.filter(is_active=True)
+
+
+class PolicyRule(DataspacedModel):
+    name = models.CharField(
+        max_length=100,
+        help_text=_("Descriptive name for this policy rule."),
+    )
+    rule_type = models.CharField(
+        max_length=50,
+        help_text=_("The type of evaluation performed by this rule."),
+    )
+    threshold = models.PositiveIntegerField(
+        default=0,
+        help_text=_("Minimum number of violations required to trigger this rule (0 means any)."),
+    )
+    is_active = models.BooleanField(
+        default=True,
+        help_text=_("Only active rules are evaluated."),
+    )
+    parameters = models.JSONField(
+        blank=True,
+        default=dict,
+        help_text=_(
+            "Optional rule-specific parameters as a JSON object. "
+            "Supported keys depend on the chosen rule type."
+        ),
+    )
+
+    objects = PolicyRuleQuerySet.as_manager()
+
+    class Meta:
+        unique_together = (("dataspace", "uuid"), ("dataspace", "name"))
+        ordering = ["name"]
+
+    def __str__(self):
+        return self.name
+
+    @property
+    def rule_label(self):
+        handler = RULE_REGISTRY.get(self.rule_type)
+        return handler.label if handler else self.rule_type
+
+    @property
+    def rule_description(self):
+        handler = RULE_REGISTRY.get(self.rule_type)
+        return handler.description if handler else ""
+
+
+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

policy/rules.py

@@ -0,0 +1,97 @@
+#
+# 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
+    parameters_schema = {}
+
+    def count_violations(self, policy_rule, product):
+        """Count objects violating the rule for the given product."""
+        raise NotImplementedError
+
+
+class PackageBaseRule(BaseRule):
+    """Base for rules that count packages matching a fixed filter within a product."""
+
+    package_filter = {}
+
+    def count_violations(self, policy_rule, product):
+        Package = apps.get_model("component_catalog", "package")
+
+        count = Package.objects.filter(
+            productpackages__product=product,
+            **self.package_filter,
+        ).count()
+
+        return count if count > policy_rule.threshold else 0
+
+
+class LicensePolicyErrorRule(PackageBaseRule):
+    rule_type = "license_policy_error"
+    label = "License Policy Error"
+    description = (
+        "Detects packages assigned a usage policy with a compliance alert level of 'error'."
+    )
+    package_filter = {"usage_policy__compliance_alert": "error"}
+
+
+class LicensePolicyWarningRule(PackageBaseRule):
+    rule_type = "license_policy_warning"
+    label = "License Policy Warning"
+    description = (
+        "Detects packages assigned a usage policy with a compliance alert level of 'warning'."
+    )
+    package_filter = {"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": ""}
+
+
+class VulnerabilityDetectedRule(BaseRule):
+    rule_type = "vulnerability_detected"
+    label = "Vulnerability Detected"
+    description = "Detects packages with at least one known vulnerability (non-null risk score)."
+    parameters_schema = {
+        "min_risk_score": "Minimum risk score (0.0-10.0). Default: any vulnerability.",
+    }
+
+    def count_violations(self, policy_rule, product):
+        Package = apps.get_model("component_catalog", "package")
+
+        packages = Package.objects.filter(
+            productpackages__product=product,
+            risk_score__isnull=False,
+        )
+
+        min_risk_score = policy_rule.parameters.get("min_risk_score")
+        if min_risk_score is not None:
+            packages = packages.filter(risk_score__gte=min_risk_score)
+
+        count = packages.count()
+        return count if count > policy_rule.threshold else 0
+
+
+RULE_REGISTRY = {
+    LicensePolicyErrorRule.rule_type: LicensePolicyErrorRule(),
+    LicensePolicyWarningRule.rule_type: LicensePolicyWarningRule(),
+    LicenseCoverageGapRule.rule_type: LicenseCoverageGapRule(),
+    VulnerabilityDetectedRule.rule_type: VulnerabilityDetectedRule(),
+}