From 75207c993abf00e0a995fcf994753bb2c43c636a Mon Sep 17 00:00:00 2001 From: Tushar Goel Date: Tue, 28 Jul 2026 18:14:04 +0530 Subject: [PATCH 1/2] POC for vulnerability triaging using rules Signed-off-by: Tushar Goel --- dejacode/settings.py | 1 + triage_rules/README.md | 460 +++++++++ triage_rules/__init__.py | 0 triage_rules/admin.py | 3 + triage_rules/apps.py | 5 + triage_rules/engine.py | 253 +++++ triage_rules/loader.py | 332 +++++++ triage_rules/migrations/0001_initial.py | 82 ++ triage_rules/migrations/__init__.py | 0 triage_rules/models.py | 214 +++++ triage_rules/tests.py | 882 ++++++++++++++++++ .../tests/data/comprehensive_policy.yaml | 158 ++++ triage_rules/tests/data/sample_policy.yaml | 63 ++ triage_rules/views.py | 3 + 14 files changed, 2456 insertions(+) create mode 100644 triage_rules/README.md create mode 100644 triage_rules/__init__.py create mode 100644 triage_rules/admin.py create mode 100644 triage_rules/apps.py create mode 100644 triage_rules/engine.py create mode 100644 triage_rules/loader.py create mode 100644 triage_rules/migrations/0001_initial.py create mode 100644 triage_rules/migrations/__init__.py create mode 100644 triage_rules/models.py create mode 100644 triage_rules/tests.py create mode 100644 triage_rules/tests/data/comprehensive_policy.yaml create mode 100644 triage_rules/tests/data/sample_policy.yaml create mode 100644 triage_rules/views.py diff --git a/dejacode/settings.py b/dejacode/settings.py index 7c9af8c3..c6f76932 100644 --- a/dejacode/settings.py +++ b/dejacode/settings.py @@ -345,6 +345,7 @@ def gettext_noop(s): "policy", "notification", "vulnerabilities", + "triage_rules", ] EXTRA_APPS = env.list("EXTRA_APPS", default=[]) diff --git a/triage_rules/README.md b/triage_rules/README.md new file mode 100644 index 00000000..1de8ab48 --- /dev/null +++ b/triage_rules/README.md @@ -0,0 +1,460 @@ +# Triage Rules + +Product-centric policy evaluation for DejaCode. A **Product** is evaluated against +configurable rules; the engine returns a single **triage decision** (upgrade, +downgrade, reachability analysis, or forensics) plus a boolean **decision vector** +showing which predicates matched. + +Policies can be defined in YAML and loaded into the database. Decision-point +predicates reuse the existing `reporting.Query` and `reporting.Filter` models. + +## Contents + +- [Architecture](#architecture) +- [Data model](#data-model) +- [Evaluation engine](#evaluation-engine) +- [Policy loader](#policy-loader) +- [YAML policy format](#yaml-policy-format) +- [Usage](#usage) +- [Tests](#tests) +- [Current limitations](#current-limitations) +- [Open design questions](#open-design-questions) + +## Architecture + +``` +Policy YAML + │ + ▼ +loader.py ──► TriageDecision, DecisionPoint, Ruleset, Rule, RuleCondition + └── reporting.Query + reporting.Filter (per decision point) + +Product + assigned Rulesets + │ + ▼ +engine.py ──► DecisionPointEvaluator (build vector) + └── RulesetEvaluator + RuleMatcher (pick decision) + │ + ▼ +EvaluationResult(decision, vector) +``` + +Evaluation is **product-centric**: one product in, one product-level decision out. +Package and vulnerability decision points use **existential** semantics today +(`True` if **any** matching package or vulnerability exists in the product). + +## Data model + +### DecisionPoint + +Reusable boolean predicate evaluated against a product. + +| Field | Description | +|-------|-------------| +| `name` | Unique key; appears in the evaluation vector | +| `target` | `package`, `vulnerability`, or `product` | +| `query` | FK to `reporting.Query` (filter logic via `Filter` rows) | +| `enabled` | Disabled points are skipped | + +### TriageDecision + +Outcome label when a rule matches. + +| Field | Description | +|-------|-------------| +| `name` | Unique label (e.g. `upgrade-immediately`) | +| `action` | `upgrade`, `downgrade`, `reachability`, `forensics` | +| `timeline_days` | Recommended response window | +| `description` | Optional text | + +### Ruleset + +Collection of rules. Multiple rulesets can be evaluated for one product; the +ruleset with the **highest `precedence`** wins. + +| Field | Description | +|-------|-------------| +| `precedence` | Higher value wins across rulesets | +| `default_decision` | Used when no rule in the ruleset matches | +| `enabled` | Disabled rulesets are ignored | + +### Rule + +One row in a decision table inside a ruleset. + +| Field | Description | +|-------|-------------| +| `priority` | Lower values are evaluated first | +| `decision` | `TriageDecision` returned when this rule matches | + +### RuleCondition + +One column in a decision table row. + +| Field | Description | +|-------|-------------| +| `decision_point` | Which predicate to check | +| `expected` | `TRUE` (1) or `FALSE` (0) | + +There is no wildcard / `ANY` value; every condition is strictly true or false. + +### Relationship to reporting + +Each `DecisionPoint` owns a `reporting.Query` with one or more `reporting.Filter` +rows: + +```text +Query + ├── content_type → Package, Vulnerability, or Product + ├── operator → and | or + └── filters[] + ├── field_name + ├── lookup → exact, gte, icontains, … + ├── value + └── negate → optional +``` + +This reuses DejaCode reporting queries so the same filters work in triage and in +reports. + +## Evaluation engine + +Module: `triage_rules/engine.py` + +### Entry point + +```python +from triage_rules.engine import EvaluationEngine + +engine = EvaluationEngine() +result = engine.evaluate(product, rulesets, user=None) + +result.decision # TriageDecision or None +result.vector # {"decision_point_name": True/False, ...} +``` + +### Evaluation flow + +1. **Collect decision points** — gather every `DecisionPoint` referenced by rule + conditions in the enabled rulesets (skip disabled decision points). + +2. **Build decision vector** — for each decision point, evaluate `True`/`False`: + + | Target | Current semantics | + |--------|-------------------| + | `package` | `True` if **any** package in `product.all_packages` matches the query | + | `vulnerability` | `True` if **any** vulnerability from package-linked vulns **or** `product.affected_by_vulnerabilities` matches | + | `product` | `True` if the product itself matches the query filters | + +3. **Evaluate each ruleset** — walk rules in ascending `priority`; first matching + rule wins, otherwise use `default_decision`. + +4. **Pick winning ruleset** — among enabled rulesets, the highest `precedence` + supplies the final `decision`. + +### Rule matching + +For each `RuleCondition` on a rule: + +- Skip if the decision point is disabled. +- If `expected == TRUE` and vector value is `False` → rule does not match. +- If `expected == FALSE` and vector value is `True` → rule does not match. + +### User context + +`user` is passed to `Query.get_qs(user=...)` for package and vulnerability +queries so secured reporting behavior applies when a user is provided. + +Product-targeted decision points evaluate the current product via +`Product.unsecured_objects` because the default Product manager returns an empty +queryset without a user. + +## Policy loader + +Module: `triage_rules/loader.py` + +### Entry point + +```python +from triage_rules.loader import load_policy_yaml + +loaded = load_policy_yaml("path/to/policy.yaml", dataspace=dataspace) + +loaded["decisions"] # dict name → TriageDecision +loaded["decision_points"] # dict name → DecisionPoint +loaded["rulesets"] # dict name → Ruleset +``` + +`source` may be a file path, `Path`, or an already-parsed dict. + +Loads run inside `transaction.atomic()`. Reloading the same YAML updates objects +in place (`update_or_create` for decisions, decision points, and rulesets; rules +are replaced on ruleset reload). + +### Load order + +1. `TriageDecision` +2. `DecisionPoint` (+ `Query` + `Filter`) +3. `Ruleset` → `Rule` → `RuleCondition` + +## YAML policy format + +### Minimal example + +See `tests/data/sample_policy.yaml`. + +```yaml +decisions: + upgrade-now: + action: upgrade + timeline_days: 7 + description: Upgrade affected packages immediately. + + forensics: + action: forensics + timeline_days: 30 + +decision_points: + matching_package: + target: package + query: + operator: and + filters: + - field: filename + lookup: exact + value: package-match.tar.gz + + critical_vulnerability: + target: vulnerability + query: + filters: + - field: risk_score + lookup: gte + value: "8.0" + + matching_product: + target: product + query: + filters: + - field: name + lookup: exact + value: Product Match + +rulesets: + security-policy: + precedence: 100 + default_decision: forensics + enabled: true + rules: + - name: Upgrade immediately + priority: 10 + decision: upgrade-now + conditions: + matching_package: true + critical_vulnerability: true + matching_product: true +``` + +### Comprehensive example + +See `tests/data/comprehensive_policy.yaml` (9 decision points, 2 rulesets, 4 +rules). + +### Field reference + +**`decisions`** + +| Key | Required | Description | +|-----|----------|-------------| +| `action` | yes | `upgrade`, `downgrade`, `reachability`, `forensics` | +| `timeline_days` | yes | Integer | +| `description` | no | Text | + +**`decision_points`** + +| Key | Required | Description | +|-----|----------|-------------| +| `target` | yes | `package`, `vulnerability`, `product` | +| `query` | yes | Query definition (see below) | +| `description` | no | Text | +| `enabled` | no | Default `true` | + +**`query`** + +| Key | Required | Description | +|-----|----------|-------------| +| `filters` | yes | List of filter dicts | +| `operator` | no | `and` (default) or `or` | +| `name` | no | Query name; default `triage:{decision_point_name}` | +| `description` | no | Text | + +**Filter dict** + +| Key | Required | Description | +|-----|----------|-------------| +| `field` or `field_name` | yes | Model field path | +| `lookup` | no | Default `exact` | +| `value` | no | Filter value (string) | +| `negate` | no | Default `false` | + +**`rulesets`** + +| Key | Required | Description | +|-----|----------|-------------| +| `precedence` | no | Default `100` | +| `default_decision` | no | Decision name | +| `enabled` | no | Default `true` | +| `rules` | no | List of rules | + +**Rule** + +| Key | Required | Description | +|-----|----------|-------------| +| `name` | yes | Rule label | +| `decision` | yes | Decision name | +| `priority` | no | Default `100` (lower = first) | +| `conditions` | no | Map or list (see below) | + +**Conditions** — dict form: + +```yaml +conditions: + high_risk: true + reachable: false +``` + +List form: + +```yaml +conditions: + - decision_point: high_risk + expected: true +``` + +## Usage + +### Load policy and evaluate + +```python +from triage_rules.loader import load_policy_yaml +from triage_rules.engine import EvaluationEngine + +loaded = load_policy_yaml("/path/to/policy.yaml", dataspace=product.dataspace) + +rulesets = list(loaded["rulesets"].values()) +result = EvaluationEngine().evaluate(product, rulesets, user=request.user) + +if result.decision: + print(result.decision.action, result.decision.timeline_days) + print(result.vector) +``` + +### Run tests + +```bash +.venv/bin/python manage.py test triage_rules --verbosity=2 --noinput +``` + +Ensure `triage_rules` is in `INSTALLED_APPS` (`dejacode/settings.py` → +`PROJECT_APPS`). + +## Tests + +| Area | Location | +|------|----------| +| Engine unit / integration | `tests.py` | +| Policy loader | `tests.py` → `PolicyLoaderTest` | +| Sample YAML | `tests/data/sample_policy.yaml` | +| Comprehensive E2E YAML | `tests/data/comprehensive_policy.yaml` | + +The comprehensive E2E test loads a full policy, verifies DB objects (`Query`, +`Filter`, triage models), and runs multiple evaluation scenarios (upgrade, +reachability default, forensics, downgrade). + +## Current limitations + +- **No Product ↔ Ruleset assignment** — rulesets must be passed into + `evaluate()` manually. +- **No persistence** — `EvaluationResult` is not stored on the product. +- **No admin UI** — models are not registered in Django admin yet. +- **No workflow hooks** — decisions do not create tickets or notifications. +- **Package/vulnerability aggregation is fixed to `any`** — see open questions. +- **Decisions are product-level only** — no per-package action lists yet. + +## Open design questions + +These need product decisions before the next implementation phase. + +### Q1 — Package and vulnerability aggregation: `any` vs `all`? + +**Current behavior:** package and vulnerability decision points use **any** +(existential) semantics. + +| Mode | Meaning | Example | +|------|---------|---------| +| **any** (current) | At least one item matches | “Any package has critical risk” | +| **all** | Every item must match | “All packages use approved licenses” | + +**Options:** + +- **A.** Keep `any` only. +- **B.** Add `aggregation: any | all` on `DecisionPoint` for package/vulnerability + targets (recommended if both are needed). +- **C.** Encode only via query design (limited for `all` cases). + +**Recommendation:** add explicit `aggregation` on `DecisionPoint`, default `any`. + +--- + +### Q2 — Decision scope: product label vs actionable targets? + +**Current behavior:** one `TriageDecision` for the whole product. The engine does +not record which packages or vulnerabilities caused the match. + +**Proposed split:** + +| Action | Scope | Rationale | +|--------|-------|-----------| +| `upgrade` | Affected **packages** (optionally target version) | Inventory-specific remediation | +| `downgrade` | Affected **packages** | Same | +| `reachability` | **Product** | Analysis is product-wide | +| `forensics` | **Product** | Investigation is product-wide | + +**Follow-up questions:** + +1. Should `TriageDecision` include a **target version** for upgrade/downgrade? +2. One product label plus a list of affected packages, or one decision per package? +3. If multiple packages breach different predicates, same action for all or per-package actions? + +**Recommendation:** keep one product-level decision as the primary label; extend +`EvaluationResult` with `affected_packages` / `affected_vulnerabilities` for +`upgrade`/`downgrade`; use product scope for `reachability` and `forensics`. + +--- + +### Q3 — Should rulesets be assigned to products in the database? + +**Current:** rulesets are passed at evaluation time. + +**Question:** should `Product` have a M2M to `Ruleset` so evaluation is +`engine.evaluate(product)` without external lookup? + +--- + +### Q4 — Should evaluation results be persisted? + +**Current:** results are volatile. + +**Question:** store last decision, vector, timestamp, and user on the product for +audit and UI? + +--- + +### Q5 — Should evaluation always require a user? + +**Current:** `user` is optional; product queries use `unsecured_objects`. + +**Question:** in production, should evaluation always pass `user` for permission-aware +reporting queries? + +--- diff --git a/triage_rules/__init__.py b/triage_rules/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/triage_rules/admin.py b/triage_rules/admin.py new file mode 100644 index 00000000..8c38f3f3 --- /dev/null +++ b/triage_rules/admin.py @@ -0,0 +1,3 @@ +from django.contrib import admin + +# Register your models here. diff --git a/triage_rules/apps.py b/triage_rules/apps.py new file mode 100644 index 00000000..1630cfb8 --- /dev/null +++ b/triage_rules/apps.py @@ -0,0 +1,5 @@ +from django.apps import AppConfig + + +class TriageRulesConfig(AppConfig): + name = 'triage_rules' diff --git a/triage_rules/engine.py b/triage_rules/engine.py new file mode 100644 index 00000000..07ce2e9f --- /dev/null +++ b/triage_rules/engine.py @@ -0,0 +1,253 @@ +import operator +from functools import reduce + +from django.db.models import Q + +from product_portfolio.models import Product +from triage_rules.models import DecisionPoint +from triage_rules.models import ExpectedValue + + +class DecisionPointEvaluator: + + def __init__(self, product, user=None): + + self.product = product + self.user = user + + def evaluate(self, decision_point): + + target = decision_point.target + + if target == DecisionPoint.TARGET_PACKAGE: + queryset = decision_point.query.get_qs(user=self.user) + return queryset.filter( + pk__in=self.product.all_packages.values("pk") + ).exists() + + if target == DecisionPoint.TARGET_VULNERABILITY: + queryset = decision_point.query.get_qs(user=self.user) + return queryset.filter( + Q( + pk__in=self.product.get_vulnerability_qs().values("pk") + ) + | Q( + pk__in=self.product.affected_by_vulnerabilities.values("pk") + ) + ).exists() + + if target == DecisionPoint.TARGET_PRODUCT: + return self._product_matches_query(decision_point.query) + + raise ValueError( + f"Unknown target {target}" + ) + + def _product_matches_query(self, query): + q_objects = [] + for filter_ in query.filters.all(): + q_object = filter_.get_q(None, self.user) + if q_object: + q_objects.append(q_object) + + if not q_objects: + return False + + queryset = ( + Product.unsecured_objects + .scope(self.product.dataspace) + .filter(pk=self.product.pk) + ) + operator_type = operator.or_ if query.operator == "or" else operator.and_ + return queryset.filter(reduce(operator_type, q_objects)).exists() + + +class RuleMatcher: + + @staticmethod + def matches(rule, vector): + + """ + vector + + { + dp_high_risk: True, + dp_known_exploit: False + } + """ + + for condition in rule.conditions.all(): + if not condition.decision_point.enabled: + continue + + expected = condition.expected + + actual = vector[ + condition.decision_point.name + ] + + if ( + expected == ExpectedValue.TRUE + and not actual + ): + return False + + if ( + expected == ExpectedValue.FALSE + and actual + ): + return False + + return True + + +class RulesetEvaluator: + + def __init__(self, ruleset): + + self.ruleset = ruleset + + def evaluate(self, vector): + + rules = ( + self.ruleset.rules + .prefetch_related( + "conditions", + "decision", + ) + .order_by("priority") + ) + + for rule in rules: + + if RuleMatcher.matches( + rule, + vector, + ): + return rule.decision + + return self.ruleset.default_decision + + +class EvaluationEngine: + + """ + Public entry point. + + engine = EvaluationEngine() + + decision = engine.evaluate( + product, + rulesets, + ) + """ + + def evaluate( + self, + product, + rulesets, + user=None, + ): + + decision_point_evaluator = ( + DecisionPointEvaluator( + product, + user=user, + ) + ) + + vector = {} + + decision_points = ( + self.get_all_decision_points( + rulesets + ) + ) + + for dp in decision_points: + + vector[dp.name] = ( + decision_point_evaluator.evaluate(dp) + ) + + winning = None + + winning_precedence = -1 + + for ruleset in rulesets: + if not ruleset.enabled: + continue + + decision = ( + RulesetEvaluator( + ruleset + ).evaluate(vector) + ) + + if ( + winning is None + or ruleset.precedence + > winning_precedence + ): + winning = decision + winning_precedence = ( + ruleset.precedence + ) + + return EvaluationResult( + decision=winning, + decision_vector=vector, + ) + + @staticmethod + def get_all_decision_points( + rulesets, + ): + + seen = {} + + for ruleset in rulesets: + if not ruleset.enabled: + continue + + for rule in ( + ruleset.rules + .prefetch_related( + "conditions__decision_point" + ) + ): + + for condition in ( + rule.conditions.all() + ): + dp = condition.decision_point + if not dp.enabled: + continue + + seen[dp.id] = dp + + return list(seen.values()) + + +class EvaluationResult: + + def __init__( + self, + decision, + decision_vector, + ): + + self.decision = decision + + self.vector = decision_vector + + def __repr__(self): + + return ( + f"" + ) + + +# Have policies that are more related to real life examples \ No newline at end of file diff --git a/triage_rules/loader.py b/triage_rules/loader.py new file mode 100644 index 00000000..fdb81ccc --- /dev/null +++ b/triage_rules/loader.py @@ -0,0 +1,332 @@ +from pathlib import Path + +import yaml +from django.apps import apps +from django.contrib.contenttypes.models import ContentType +from django.db import transaction + +from reporting.models import Filter +from reporting.models import Query +from triage_rules.models import DecisionPoint +from triage_rules.models import ExpectedValue +from triage_rules.models import Rule +from triage_rules.models import RuleCondition +from triage_rules.models import Ruleset +from triage_rules.models import TriageDecision + +TARGET_MODELS = { + DecisionPoint.TARGET_PACKAGE: ("component_catalog", "package"), + DecisionPoint.TARGET_VULNERABILITY: ("vulnerabilities", "vulnerability"), + DecisionPoint.TARGET_PRODUCT: ("product_portfolio", "product"), +} + +EXPECTED_VALUES = { + "true": ExpectedValue.TRUE, + "false": ExpectedValue.FALSE, + "1": ExpectedValue.TRUE, + "0": ExpectedValue.FALSE, + 1: ExpectedValue.TRUE, + 0: ExpectedValue.FALSE, + True: ExpectedValue.TRUE, + False: ExpectedValue.FALSE, +} + + +class PolicyLoadError(Exception): + pass + + +def load_policy_yaml(source, dataspace): + """ + Load a triage policy YAML file into concrete model objects. + + Expected top-level keys: + - decisions + - decision_points + - rulesets + + ``source`` may be a file path, Path, or already-parsed dict. + """ + if isinstance(source, dict): + data = source + else: + path = Path(source) + data = yaml.safe_load(path.read_text()) + + if not isinstance(data, dict): + raise PolicyLoadError("Policy YAML must define a mapping at the top level.") + + with transaction.atomic(): + decisions = _load_decisions(data.get("decisions") or {}, dataspace) + decision_points = _load_decision_points( + data.get("decision_points") or {}, + dataspace, + ) + rulesets = _load_rulesets( + data.get("rulesets") or {}, + dataspace, + decisions=decisions, + decision_points=decision_points, + ) + + return { + "decisions": decisions, + "decision_points": decision_points, + "rulesets": rulesets, + } + + +def _load_decisions(decisions_data, dataspace): + decisions = {} + + for name, config in decisions_data.items(): + if not isinstance(config, dict): + raise PolicyLoadError(f"Decision '{name}' must be a mapping.") + + action = config.get("action") + if action not in dict(TriageDecision.ACTION_CHOICES): + raise PolicyLoadError( + f"Decision '{name}' has invalid action '{action}'." + ) + + timeline_days = config.get("timeline_days") + if timeline_days is None: + raise PolicyLoadError( + f"Decision '{name}' requires 'timeline_days'." + ) + + decision, _ = TriageDecision.objects.update_or_create( + name=name, + defaults={ + "action": action, + "timeline_days": timeline_days, + "description": config.get("description", ""), + }, + ) + decisions[name] = decision + + return decisions + + +def _load_decision_points(decision_points_data, dataspace): + decision_points = {} + + for name, config in decision_points_data.items(): + if not isinstance(config, dict): + raise PolicyLoadError(f"Decision point '{name}' must be a mapping.") + + target = config.get("target") + if target not in TARGET_MODELS: + raise PolicyLoadError( + f"Decision point '{name}' has invalid target '{target}'." + ) + + query_config = config.get("query") + if not isinstance(query_config, dict): + raise PolicyLoadError( + f"Decision point '{name}' requires a 'query' mapping." + ) + + query = _load_query( + query_config, + dataspace=dataspace, + target=target, + default_name=f"triage:{name}", + ) + + decision_point, _ = DecisionPoint.objects.update_or_create( + name=name, + defaults={ + "target": target, + "query": query, + "description": config.get("description", ""), + "enabled": config.get("enabled", True), + }, + ) + decision_points[name] = decision_point + + return decision_points + + +def _load_query(query_config, dataspace, target, default_name): + app_label, model_name = TARGET_MODELS[target] + model = apps.get_model(app_label, model_name) + content_type = ContentType.objects.get_for_model(model) + + query_name = query_config.get("name", default_name) + operator = query_config.get("operator", "and") + if operator not in dict(Query.OPERATOR_CHOICES): + raise PolicyLoadError( + f"Query '{query_name}' has invalid operator '{operator}'." + ) + + query, _ = Query.objects.update_or_create( + dataspace=dataspace, + name=query_name, + defaults={ + "description": query_config.get("description", ""), + "content_type": content_type, + "operator": operator, + }, + ) + + filters_data = query_config.get("filters") or [] + if not filters_data: + raise PolicyLoadError(f"Query '{query_name}' requires at least one filter.") + + query.filters.all().delete() + for filter_config in filters_data: + _load_filter(query, filter_config, dataspace) + + return query + + +def _load_filter(query, filter_config, dataspace): + if not isinstance(filter_config, dict): + raise PolicyLoadError("Each query filter must be a mapping.") + + field_name = filter_config.get("field") or filter_config.get("field_name") + if not field_name: + raise PolicyLoadError("Each query filter requires 'field'.") + + lookup = filter_config.get("lookup", "exact") + if lookup not in dict(Filter.LOOKUP_CHOICES): + raise PolicyLoadError( + f"Filter on '{field_name}' has invalid lookup '{lookup}'." + ) + + Filter.objects.create( + dataspace=dataspace, + query=query, + field_name=field_name, + lookup=lookup, + value=str(filter_config.get("value", "")), + negate=filter_config.get("negate", False), + runtime_parameter=filter_config.get("runtime_parameter", False), + ) + + +def _load_rulesets(rulesets_data, dataspace, decisions, decision_points): + rulesets = {} + + for name, config in rulesets_data.items(): + if not isinstance(config, dict): + raise PolicyLoadError(f"Ruleset '{name}' must be a mapping.") + + default_decision_name = config.get("default_decision") + default_decision = None + if default_decision_name: + default_decision = decisions.get(default_decision_name) + if default_decision is None: + raise PolicyLoadError( + f"Ruleset '{name}' references unknown decision " + f"'{default_decision_name}'." + ) + + ruleset, _ = Ruleset.objects.update_or_create( + name=name, + defaults={ + "description": config.get("description", ""), + "precedence": config.get("precedence", 100), + "default_decision": default_decision, + "enabled": config.get("enabled", True), + }, + ) + + ruleset.rules.all().delete() + for rule_config in config.get("rules") or []: + _load_rule( + ruleset=ruleset, + rule_config=rule_config, + decisions=decisions, + decision_points=decision_points, + ) + + rulesets[name] = ruleset + + return rulesets + + +def _load_rule(ruleset, rule_config, decisions, decision_points): + if not isinstance(rule_config, dict): + raise PolicyLoadError(f"Rule in ruleset '{ruleset.name}' must be a mapping.") + + rule_name = rule_config.get("name") + if not rule_name: + raise PolicyLoadError( + f"Each rule in ruleset '{ruleset.name}' requires 'name'." + ) + + decision_name = rule_config.get("decision") + decision = decisions.get(decision_name) + if decision is None: + raise PolicyLoadError( + f"Rule '{rule_name}' references unknown decision '{decision_name}'." + ) + + rule = Rule.objects.create( + ruleset=ruleset, + name=rule_name, + priority=rule_config.get("priority", 100), + decision=decision, + ) + + conditions_data = rule_config.get("conditions") or {} + for decision_point_name, expected_value in _iter_conditions(conditions_data): + decision_point = decision_points.get(decision_point_name) + if decision_point is None: + raise PolicyLoadError( + f"Rule '{rule_name}' references unknown decision point " + f"'{decision_point_name}'." + ) + + expected = _parse_expected(expected_value, decision_point_name) + RuleCondition.objects.create( + rule=rule, + decision_point=decision_point, + expected=expected, + ) + + return rule + + +def _iter_conditions(conditions_data): + if isinstance(conditions_data, dict): + return conditions_data.items() + + if isinstance(conditions_data, list): + items = [] + for condition in conditions_data: + if not isinstance(condition, dict): + raise PolicyLoadError("Each rule condition must be a mapping.") + decision_point_name = condition.get("decision_point") + if not decision_point_name: + raise PolicyLoadError( + "List-style rule conditions require 'decision_point'." + ) + items.append( + (decision_point_name, condition.get("expected")) + ) + return items + + raise PolicyLoadError("Rule conditions must be a mapping or list.") + + +def _parse_expected(value, decision_point_name): + if value is None: + raise PolicyLoadError( + f"Decision point '{decision_point_name}' is missing an expected value." + ) + + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in EXPECTED_VALUES: + return EXPECTED_VALUES[normalized] + + if value in EXPECTED_VALUES: + return EXPECTED_VALUES[value] + + raise PolicyLoadError( + f"Decision point '{decision_point_name}' has invalid expected value " + f"'{value}'. Use true or false." + ) diff --git a/triage_rules/migrations/0001_initial.py b/triage_rules/migrations/0001_initial.py new file mode 100644 index 00000000..bb4727fa --- /dev/null +++ b/triage_rules/migrations/0001_initial.py @@ -0,0 +1,82 @@ +# Generated by Django 6.0.4 on 2026-07-28 09:49 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ('reporting', '0002_alter_columntemplate_content_type_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='TriageDecision', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=100, unique=True)), + ('action', models.CharField(choices=[('upgrade', 'Upgrade'), ('downgrade', 'Downgrade'), ('reachability', 'Reachability Analysis'), ('forensics', 'Forensic Analysis')], max_length=30)), + ('timeline_days', models.PositiveIntegerField()), + ('description', models.TextField(blank=True)), + ], + options={ + 'ordering': ('name',), + }, + ), + migrations.CreateModel( + name='DecisionPoint', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=100, unique=True)), + ('description', models.TextField(blank=True)), + ('target', models.CharField(choices=[('package', 'Package'), ('vulnerability', 'Vulnerability'), ('product', 'Product')], max_length=20)), + ('enabled', models.BooleanField(default=True)), + ('query', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='decision_points', to='reporting.query')), + ], + options={ + 'ordering': ('name',), + }, + ), + migrations.CreateModel( + name='Ruleset', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=100, unique=True)), + ('description', models.TextField(blank=True)), + ('precedence', models.PositiveIntegerField(default=100)), + ('enabled', models.BooleanField(default=True)), + ('default_decision', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='triage_rules.triagedecision')), + ], + options={ + 'ordering': ('-precedence', 'name'), + }, + ), + migrations.CreateModel( + name='Rule', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=100)), + ('priority', models.PositiveIntegerField(default=100, help_text='Lower values are evaluated first.')), + ('ruleset', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='rules', to='triage_rules.ruleset')), + ('decision', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='rules', to='triage_rules.triagedecision')), + ], + options={ + 'ordering': ('priority', 'id'), + }, + ), + migrations.CreateModel( + name='RuleCondition', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('expected', models.PositiveSmallIntegerField(choices=[(0, 'False'), (1, 'True')])), + ('decision_point', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='triage_rules.decisionpoint')), + ('rule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='conditions', to='triage_rules.rule')), + ], + options={ + 'unique_together': {('rule', 'decision_point')}, + }, + ), + ] diff --git a/triage_rules/migrations/__init__.py b/triage_rules/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/triage_rules/models.py b/triage_rules/models.py new file mode 100644 index 00000000..7e5611de --- /dev/null +++ b/triage_rules/models.py @@ -0,0 +1,214 @@ +from django.db import models + +from reporting.models import Query + +class DecisionPoint(models.Model): + """ + A reusable boolean predicate. + + Example: + Name: High Risk Vulnerability + + Query: + risk_score >= 8 + exploitability >= 2 + + During evaluation this resolves to True or False. + """ + + TARGET_PACKAGE = "package" + TARGET_VULNERABILITY = "vulnerability" + TARGET_PRODUCT = "product" + + TARGET_CHOICES = ( + (TARGET_PACKAGE, "Package"), + (TARGET_VULNERABILITY, "Vulnerability"), + (TARGET_PRODUCT, "Product"), + ) + + name = models.CharField( + max_length=100, + unique=True, + ) + + description = models.TextField(blank=True) + + target = models.CharField( + max_length=20, + choices=TARGET_CHOICES, + ) + + query = models.ForeignKey( + Query, + on_delete=models.CASCADE, + related_name="decision_points", + ) + + enabled = models.BooleanField(default=True) + + class Meta: + ordering = ("name",) + + def __str__(self): + return self.name + + +class TriageDecision(models.Model): + """ + The outcome produced by a matching rule. + """ + + ACTION_UPGRADE = "upgrade" + ACTION_DOWNGRADE = "downgrade" + ACTION_REACHABILITY = "reachability" + ACTION_FORENSICS = "forensics" + + ACTION_CHOICES = ( + (ACTION_UPGRADE, "Upgrade"), + (ACTION_DOWNGRADE, "Downgrade"), + (ACTION_REACHABILITY, "Reachability Analysis"), + (ACTION_FORENSICS, "Forensic Analysis"), + ) + + name = models.CharField( + max_length=100, + unique=True, + ) + + action = models.CharField( + max_length=30, + choices=ACTION_CHOICES, + ) + + timeline_days = models.PositiveIntegerField() + + description = models.TextField(blank=True) + + class Meta: + ordering = ("name",) + + def __str__(self): + return self.name + + +class Ruleset(models.Model): + """ + Collection of Rules. + + Multiple Rulesets can be assigned to a Product. + """ + + name = models.CharField( + max_length=100, + unique=True, + ) + + description = models.TextField(blank=True) + + precedence = models.PositiveIntegerField( + default=100, + ) + + default_decision = models.ForeignKey( + TriageDecision, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="+", + ) + + enabled = models.BooleanField(default=True) + + class Meta: + ordering = ( + "-precedence", + "name", + ) + + def __str__(self): + return self.name + + +class Rule(models.Model): + """ + Maps a set of DecisionPoint values to a Decision. + """ + + ruleset = models.ForeignKey( + Ruleset, + on_delete=models.CASCADE, + related_name="rules", + ) + + name = models.CharField( + max_length=100, + ) + + priority = models.PositiveIntegerField( + default=100, + help_text="Lower values are evaluated first.", + ) + + decision = models.ForeignKey( + TriageDecision, + on_delete=models.PROTECT, + related_name="rules", + ) + + class Meta: + ordering = ( + "priority", + "id", + ) + + def __str__(self): + return f"{self.ruleset} :: {self.name}" + + +class ExpectedValue(models.IntegerChoices): + FALSE = 0, "False" + TRUE = 1, "True" + + +class RuleCondition(models.Model): + """ + Represents one column inside a decision table. + + Example + + Rule: + Upgrade Immediately + + High Risk = TRUE + Known Exploit = TRUE + Reachable = FALSE + """ + + FALSE = 0, "False" + TRUE = 1, "True" + + rule = models.ForeignKey( + Rule, + on_delete=models.CASCADE, + related_name="conditions", + ) + + decision_point = models.ForeignKey( + DecisionPoint, + on_delete=models.CASCADE, + ) + + expected = models.PositiveSmallIntegerField( + choices=ExpectedValue.choices, + ) + + class Meta: + unique_together = ( + "rule", + "decision_point", + ) + + def __str__(self): + return ( + f"{self.decision_point.name} = {self.expected}" + ) diff --git a/triage_rules/tests.py b/triage_rules/tests.py new file mode 100644 index 00000000..176dfe64 --- /dev/null +++ b/triage_rules/tests.py @@ -0,0 +1,882 @@ +from pathlib import Path +from types import SimpleNamespace + +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 product_portfolio.models import Product +from product_portfolio.tests import make_product +from triage_rules.engine import DecisionPointEvaluator +from triage_rules.engine import EvaluationEngine +from triage_rules.engine import RuleMatcher +from triage_rules.models import DecisionPoint +from triage_rules.models import ExpectedValue +from triage_rules.models import Rule +from triage_rules.models import RuleCondition +from triage_rules.models import Ruleset +from triage_rules.models import TriageDecision +from vulnerabilities.models import Vulnerability +from vulnerabilities.tests import make_vulnerability +from reporting.models import Filter +from reporting.models import Query + + +class BaseTriageRulesTestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.package = make_package( + self.dataspace, + filename="package-match.tar.gz", + ) + self.other_package = make_package( + self.dataspace, + filename="package-other.tar.gz", + ) + self.vulnerability = make_vulnerability( + self.dataspace, + vulnerability_id="CVE-2026-0001", + risk_score=9.2, + affecting=self.package, + ) + self.other_vulnerability = make_vulnerability( + self.dataspace, + vulnerability_id="CVE-2026-0002", + risk_score=2.5, + affecting=self.other_package, + ) + self.product = make_product( + self.dataspace, + name="Product Match", + inventory=[self.package], + ) + self.other_product = make_product( + self.dataspace, + name="Product Other", + inventory=[self.other_package], + ) + + def make_query(self, name, model, field_name, lookup, value): + query = Query.objects.create( + dataspace=self.dataspace, + name=name, + content_type=ContentType.objects.get_for_model(model), + operator="and", + ) + Filter.objects.create( + dataspace=self.dataspace, + query=query, + field_name=field_name, + lookup=lookup, + value=value, + ) + return query + + def make_decision_point(self, name, target, query): + return DecisionPoint.objects.create( + name=name, + target=target, + query=query, + ) + + def make_decision(self, name): + return TriageDecision.objects.create( + name=name, + action=TriageDecision.ACTION_UPGRADE, + timeline_days=7, + ) + + +class RuleMatcherTest(BaseTriageRulesTestCase): + def test_matches_respects_expected_values(self): + package_query = self.make_query( + name="Rule matcher package query", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + vulnerability_query = self.make_query( + name="Rule matcher vulnerability query", + model=Vulnerability, + field_name="risk_score", + lookup="gte", + value="8.0", + ) + package_dp = self.make_decision_point( + name="high_risk", + target=DecisionPoint.TARGET_PACKAGE, + query=package_query, + ) + vulnerability_dp = self.make_decision_point( + name="reachable", + target=DecisionPoint.TARGET_VULNERABILITY, + query=vulnerability_query, + ) + rule = Rule.objects.create( + ruleset=Ruleset.objects.create(name="Rule matcher ruleset"), + name="Rule matcher rule", + priority=1, + decision=self.make_decision("rule-matcher-decision"), + ) + RuleCondition.objects.create( + rule=rule, + decision_point=package_dp, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=vulnerability_dp, + expected=ExpectedValue.FALSE, + ) + + self.assertTrue( + RuleMatcher.matches( + rule, + { + "high_risk": True, + "reachable": False, + }, + ) + ) + self.assertFalse( + RuleMatcher.matches( + rule, + { + "high_risk": False, + "reachable": False, + }, + ) + ) + + +class DecisionPointEvaluatorTest(BaseTriageRulesTestCase): + def test_evaluate_matches_any_package_in_product(self): + query = self.make_query( + name="Package evaluator query", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + decision_point = self.make_decision_point( + name="package_match", + target=DecisionPoint.TARGET_PACKAGE, + query=query, + ) + evaluator = DecisionPointEvaluator( + product=self.product, + ) + + self.assertTrue(evaluator.evaluate(decision_point)) + + other_evaluator = DecisionPointEvaluator( + product=self.other_product, + ) + + self.assertFalse(other_evaluator.evaluate(decision_point)) + + def test_evaluate_matches_any_vulnerability_in_product(self): + query = self.make_query( + name="Vulnerability evaluator query", + model=Vulnerability, + field_name="risk_score", + lookup="gte", + value="8.0", + ) + decision_point = self.make_decision_point( + name="vulnerability_match", + target=DecisionPoint.TARGET_VULNERABILITY, + query=query, + ) + evaluator = DecisionPointEvaluator( + product=self.product, + ) + + self.assertTrue(evaluator.evaluate(decision_point)) + + other_evaluator = DecisionPointEvaluator( + product=self.other_product, + ) + + self.assertFalse(other_evaluator.evaluate(decision_point)) + + def test_evaluate_matches_direct_product_vulnerability(self): + direct_vulnerability = make_vulnerability( + self.dataspace, + vulnerability_id="CVE-2026-0003", + risk_score=9.8, + affecting=self.product, + ) + query = self.make_query( + name="Direct product vulnerability evaluator query", + model=Vulnerability, + field_name="vulnerability_id", + lookup="exact", + value=direct_vulnerability.vulnerability_id, + ) + decision_point = self.make_decision_point( + name="direct_product_vulnerability_match", + target=DecisionPoint.TARGET_VULNERABILITY, + query=query, + ) + + self.assertTrue( + DecisionPointEvaluator(product=self.product).evaluate(decision_point) + ) + self.assertFalse( + DecisionPointEvaluator(product=self.other_product).evaluate(decision_point) + ) + + def test_evaluate_matches_product_target_once(self): + query = self.make_query( + name="Product evaluator query", + model=Product, + field_name="name", + lookup="exact", + value=self.product.name, + ) + decision_point = self.make_decision_point( + name="product_match", + target=DecisionPoint.TARGET_PRODUCT, + query=query, + ) + evaluator = DecisionPointEvaluator( + product=self.product, + ) + + self.assertTrue(evaluator.evaluate(decision_point)) + + other_evaluator = DecisionPointEvaluator( + product=self.other_product, + ) + + self.assertFalse(other_evaluator.evaluate(decision_point)) + + def test_evaluate_rejects_unknown_target(self): + evaluator = DecisionPointEvaluator( + product=self.product, + ) + + with self.assertRaisesMessage( + ValueError, + "Unknown target unknown", + ): + evaluator.evaluate( + SimpleNamespace( + target="unknown", + query=None, + ) + ) + + +class EvaluationEngineIntegrationTest(BaseTriageRulesTestCase): + + def test_evaluate_builds_vector_for_product_package_and_vulnerability_targets(self): + package_query = self.make_query( + name="Package filename match", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + vulnerability_query = self.make_query( + name="Critical vulnerability", + model=Vulnerability, + field_name="risk_score", + lookup="gte", + value="8.0", + ) + product_query = self.make_query( + name="Product name match", + model=Product, + field_name="name", + lookup="exact", + value=self.product.name, + ) + package_dp = self.make_decision_point( + name="matching_package", + target=DecisionPoint.TARGET_PACKAGE, + query=package_query, + ) + vulnerability_dp = self.make_decision_point( + name="critical_vulnerability", + target=DecisionPoint.TARGET_VULNERABILITY, + query=vulnerability_query, + ) + product_dp = self.make_decision_point( + name="matching_product", + target=DecisionPoint.TARGET_PRODUCT, + query=product_query, + ) + decision = self.make_decision("upgrade-now") + ruleset = Ruleset.objects.create( + name="Security rules", + precedence=100, + default_decision=None, + ) + rule = Rule.objects.create( + ruleset=ruleset, + name="Upgrade immediately", + priority=10, + decision=decision, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=package_dp, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=vulnerability_dp, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=product_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[ruleset], + ) + + self.assertEqual(decision, result.decision) + self.assertEqual( + { + "matching_package": True, + "critical_vulnerability": True, + "matching_product": True, + }, + result.vector, + ) + + other_result = EvaluationEngine().evaluate( + product=self.other_product, + rulesets=[ruleset], + ) + + self.assertIsNone(other_result.decision) + self.assertEqual( + { + "matching_package": False, + "critical_vulnerability": False, + "matching_product": False, + }, + other_result.vector, + ) + + def test_evaluate_uses_higher_precedence_ruleset_winner_for_product(self): + package_query = self.make_query( + name="Package filename match for precedence", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + vulnerability_query = self.make_query( + name="High risk vulnerability for precedence", + model=Vulnerability, + field_name="risk_score", + lookup="gte", + value="8.0", + ) + package_dp = self.make_decision_point( + name="precedence_package_match", + target=DecisionPoint.TARGET_PACKAGE, + query=package_query, + ) + vulnerability_dp = self.make_decision_point( + name="precedence_vulnerability_match", + target=DecisionPoint.TARGET_VULNERABILITY, + query=vulnerability_query, + ) + low_precedence_decision = self.make_decision("forensics") + high_precedence_decision = TriageDecision.objects.create( + name="upgrade", + action=TriageDecision.ACTION_REACHABILITY, + timeline_days=2, + ) + low_precedence_ruleset = Ruleset.objects.create( + name="Fallback rules", + precedence=10, + default_decision=low_precedence_decision, + ) + high_precedence_ruleset = Ruleset.objects.create( + name="Priority rules", + precedence=100, + default_decision=None, + ) + rule = Rule.objects.create( + ruleset=high_precedence_ruleset, + name="Priority upgrade", + priority=1, + decision=high_precedence_decision, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=package_dp, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=vulnerability_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[ + low_precedence_ruleset, + high_precedence_ruleset, + ], + ) + + self.assertEqual(high_precedence_decision, result.decision) + + def test_evaluate_returns_default_decision_when_no_rule_matches(self): + package_query = self.make_query( + name="No package match", + model=Package, + field_name="filename", + lookup="exact", + value="missing-package.tar.gz", + ) + package_dp = self.make_decision_point( + name="no_package_match", + target=DecisionPoint.TARGET_PACKAGE, + query=package_query, + ) + default_decision = self.make_decision("default-decision") + ruleset = Ruleset.objects.create( + name="Defaulted rules", + precedence=100, + default_decision=default_decision, + ) + rule = Rule.objects.create( + ruleset=ruleset, + name="Never matches", + priority=1, + decision=self.make_decision("non-default-decision"), + ) + RuleCondition.objects.create( + rule=rule, + decision_point=package_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.other_product, + rulesets=[ruleset], + ) + + self.assertEqual(default_decision, result.decision) + self.assertEqual( + { + "no_package_match": False, + }, + result.vector, + ) + + def test_evaluate_uses_lowest_priority_matching_rule_within_ruleset(self): + product_query = self.make_query( + name="Priority product match", + model=Product, + field_name="name", + lookup="exact", + value=self.product.name, + ) + product_dp = self.make_decision_point( + name="priority_product_match", + target=DecisionPoint.TARGET_PRODUCT, + query=product_query, + ) + ruleset = Ruleset.objects.create( + name="Priority ruleset", + precedence=100, + ) + first_decision = self.make_decision("first-priority-decision") + second_decision = self.make_decision("second-priority-decision") + first_rule = Rule.objects.create( + ruleset=ruleset, + name="First match wins", + priority=1, + decision=first_decision, + ) + second_rule = Rule.objects.create( + ruleset=ruleset, + name="Second match loses", + priority=10, + decision=second_decision, + ) + RuleCondition.objects.create( + rule=first_rule, + decision_point=product_dp, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=second_rule, + decision_point=product_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[ruleset], + ) + + self.assertEqual(first_decision, result.decision) + + def test_evaluate_ignores_disabled_decision_points(self): + package_query = self.make_query( + name="Disabled package query", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + product_query = self.make_query( + name="Enabled product query", + model=Product, + field_name="name", + lookup="exact", + value=self.product.name, + ) + disabled_package_dp = self.make_decision_point( + name="disabled_package_match", + target=DecisionPoint.TARGET_PACKAGE, + query=package_query, + ) + disabled_package_dp.enabled = False + disabled_package_dp.save() + enabled_product_dp = self.make_decision_point( + name="enabled_product_match", + target=DecisionPoint.TARGET_PRODUCT, + query=product_query, + ) + ruleset = Ruleset.objects.create( + name="Disabled decision point ruleset", + precedence=100, + ) + decision = self.make_decision("enabled-product-decision") + rule = Rule.objects.create( + ruleset=ruleset, + name="Ignore disabled point", + priority=1, + decision=decision, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=disabled_package_dp, + expected=ExpectedValue.FALSE, + ) + RuleCondition.objects.create( + rule=rule, + decision_point=enabled_product_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[ruleset], + ) + + self.assertEqual(decision, result.decision) + self.assertEqual( + { + "enabled_product_match": True, + }, + result.vector, + ) + + def test_evaluate_ignores_disabled_rulesets(self): + product_query = self.make_query( + name="Disabled ruleset product query", + model=Product, + field_name="name", + lookup="exact", + value=self.product.name, + ) + product_dp = self.make_decision_point( + name="disabled_ruleset_product_match", + target=DecisionPoint.TARGET_PRODUCT, + query=product_query, + ) + enabled_decision = self.make_decision("enabled-ruleset-decision") + disabled_decision = self.make_decision("disabled-ruleset-decision") + enabled_ruleset = Ruleset.objects.create( + name="Enabled ruleset", + precedence=10, + default_decision=enabled_decision, + ) + disabled_ruleset = Ruleset.objects.create( + name="Disabled ruleset", + precedence=100, + default_decision=disabled_decision, + enabled=False, + ) + disabled_rule = Rule.objects.create( + ruleset=disabled_ruleset, + name="Disabled ruleset match", + priority=1, + decision=disabled_decision, + ) + RuleCondition.objects.create( + rule=disabled_rule, + decision_point=product_dp, + expected=ExpectedValue.TRUE, + ) + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[enabled_ruleset, disabled_ruleset], + ) + + self.assertEqual(enabled_decision, result.decision) + + def test_get_all_decision_points_deduplicates_shared_real_decision_points(self): + shared_query = self.make_query( + name="Shared package query", + model=Package, + field_name="filename", + lookup="exact", + value=self.package.filename, + ) + shared_decision_point = self.make_decision_point( + name="shared_point", + target=DecisionPoint.TARGET_PACKAGE, + query=shared_query, + ) + ruleset_one = Ruleset.objects.create( + name="Ruleset one", + precedence=10, + ) + ruleset_two = Ruleset.objects.create( + name="Ruleset two", + precedence=20, + ) + rule_one = Rule.objects.create( + ruleset=ruleset_one, + name="Rule one", + priority=1, + decision=self.make_decision("decision-one"), + ) + rule_two = Rule.objects.create( + ruleset=ruleset_two, + name="Rule two", + priority=1, + decision=self.make_decision("decision-two"), + ) + RuleCondition.objects.create( + rule=rule_one, + decision_point=shared_decision_point, + expected=ExpectedValue.TRUE, + ) + RuleCondition.objects.create( + rule=rule_two, + decision_point=shared_decision_point, + expected=ExpectedValue.FALSE, + ) + + decision_points = EvaluationEngine.get_all_decision_points( + [ruleset_one, ruleset_two] + ) + + self.assertEqual([shared_decision_point], decision_points) + + +class PolicyLoaderTest(BaseTriageRulesTestCase): + data = Path(__file__).parent / "tests" / "data" + + def test_load_policy_yaml_creates_reporting_queries_and_triage_objects(self): + from triage_rules.loader import load_policy_yaml + + loaded = load_policy_yaml( + self.data / "sample_policy.yaml", + dataspace=self.dataspace, + ) + + self.assertIn("upgrade-now", loaded["decisions"]) + self.assertIn("matching_package", loaded["decision_points"]) + self.assertIn("security-policy", loaded["rulesets"]) + + package_dp = loaded["decision_points"]["matching_package"] + self.assertEqual(DecisionPoint.TARGET_PACKAGE, package_dp.target) + self.assertEqual(1, package_dp.query.filters.count()) + self.assertEqual( + "filename", + package_dp.query.filters.get().field_name, + ) + + ruleset = loaded["rulesets"]["security-policy"] + rule = ruleset.rules.get() + self.assertEqual("Upgrade immediately", rule.name) + self.assertEqual(3, rule.conditions.count()) + + def test_load_policy_yaml_end_to_end_with_engine(self): + from triage_rules.loader import load_policy_yaml + + loaded = load_policy_yaml( + self.data / "sample_policy.yaml", + dataspace=self.dataspace, + ) + ruleset = loaded["rulesets"]["security-policy"] + decision = loaded["decisions"]["upgrade-now"] + + result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[ruleset], + ) + + self.assertEqual(decision, result.decision) + self.assertEqual( + { + "matching_package": True, + "critical_vulnerability": True, + "matching_product": True, + }, + result.vector, + ) + + def test_load_policy_yaml_is_idempotent(self): + from triage_rules.loader import load_policy_yaml + + path = self.data / "sample_policy.yaml" + first = load_policy_yaml(path, dataspace=self.dataspace) + second = load_policy_yaml(path, dataspace=self.dataspace) + + self.assertEqual( + first["rulesets"]["security-policy"].pk, + second["rulesets"]["security-policy"].pk, + ) + self.assertEqual(1, Ruleset.objects.filter(name="security-policy").count()) + self.assertEqual(1, first["rulesets"]["security-policy"].rules.count()) + + def test_comprehensive_policy_load_store_and_evaluate_end_to_end(self): + from product_portfolio.tests import make_product_package + from triage_rules.loader import load_policy_yaml + + loaded = load_policy_yaml( + self.data / "comprehensive_policy.yaml", + dataspace=self.dataspace, + ) + + self.assertEqual(4, TriageDecision.objects.count()) + self.assertEqual(9, DecisionPoint.objects.count()) + self.assertEqual(9, Query.objects.filter(dataspace=self.dataspace).count()) + self.assertEqual(9, Filter.objects.filter(dataspace=self.dataspace).count()) + self.assertEqual(2, Ruleset.objects.filter( + name__in=["critical-response", "baseline-response"] + ).count()) + self.assertEqual(4, Rule.objects.count()) + self.assertEqual(14, RuleCondition.objects.count()) + + critical_ruleset = loaded["rulesets"]["critical-response"] + baseline_ruleset = loaded["rulesets"]["baseline-response"] + upgrade_decision = loaded["decisions"]["upgrade-immediately"] + reachability_decision = loaded["decisions"]["reachability-analysis"] + downgrade_decision = loaded["decisions"]["downgrade-risk"] + forensics_decision = loaded["decisions"]["forensics-review"] + + legacy_package = make_package( + self.dataspace, + filename="legacy-lib.tar.gz", + download_url="https://example.com/legacy-lib.tar.gz", + ) + self.package.download_url = "http://example.com/package-match.tar.gz" + self.package.save() + make_product_package(self.product, package=legacy_package) + + make_vulnerability( + self.dataspace, + vulnerability_id="CVE-2026-0003", + risk_score=6.5, + exploitability=2.0, + affecting=self.package, + ) + make_vulnerability( + self.dataspace, + vulnerability_id="CVE-2026-0004", + risk_score=5.0, + affecting=self.product, + ) + + production_result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[critical_ruleset, baseline_ruleset], + ) + + self.assertEqual(upgrade_decision, production_result.decision) + self.assertEqual( + { + "watched_package": True, + "legacy_package_present": True, + "package_with_download_url": True, + "critical_vulnerability": True, + "known_exploit_vulnerability": True, + "low_risk_vulnerability": False, + "direct_product_vulnerability": True, + "production_product": True, + "non_production_product": False, + }, + production_result.vector, + ) + + other_product_result = EvaluationEngine().evaluate( + product=self.other_product, + rulesets=[critical_ruleset, baseline_ruleset], + ) + + self.assertEqual(reachability_decision, other_product_result.decision) + self.assertEqual( + { + "watched_package": False, + "legacy_package_present": False, + "package_with_download_url": False, + "critical_vulnerability": False, + "known_exploit_vulnerability": False, + "low_risk_vulnerability": True, + "direct_product_vulnerability": False, + "production_product": False, + "non_production_product": True, + }, + other_product_result.vector, + ) + + self.vulnerability.delete() + exploit_only_result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[critical_ruleset], + ) + + self.assertEqual(reachability_decision, exploit_only_result.decision) + self.assertTrue(exploit_only_result.vector["known_exploit_vulnerability"]) + self.assertFalse(exploit_only_result.vector["critical_vulnerability"]) + self.assertTrue(exploit_only_result.vector["watched_package"]) + + self.product.productpackages.filter( + package=self.package, + ).delete() + + legacy_result = EvaluationEngine().evaluate( + product=self.product, + rulesets=[baseline_ruleset], + ) + + self.assertEqual(forensics_decision, legacy_result.decision) + self.assertTrue(legacy_result.vector["legacy_package_present"]) + self.assertTrue(legacy_result.vector["production_product"]) + self.assertTrue(legacy_result.vector["direct_product_vulnerability"]) + self.assertNotIn("watched_package", legacy_result.vector) + + low_risk_only_result = EvaluationEngine().evaluate( + product=self.other_product, + rulesets=[baseline_ruleset], + ) + + self.assertEqual(downgrade_decision, low_risk_only_result.decision) + self.assertTrue(low_risk_only_result.vector["low_risk_vulnerability"]) + self.assertFalse(low_risk_only_result.vector["critical_vulnerability"]) diff --git a/triage_rules/tests/data/comprehensive_policy.yaml b/triage_rules/tests/data/comprehensive_policy.yaml new file mode 100644 index 00000000..139e3ca8 --- /dev/null +++ b/triage_rules/tests/data/comprehensive_policy.yaml @@ -0,0 +1,158 @@ +decisions: + upgrade-immediately: + action: upgrade + timeline_days: 1 + description: Upgrade affected packages without delay. + + reachability-analysis: + action: reachability + timeline_days: 14 + description: Analyze vulnerability reachability before remediation. + + downgrade-risk: + action: downgrade + timeline_days: 30 + description: Accept lower-priority risk for now. + + forensics-review: + action: forensics + timeline_days: 60 + description: Perform deeper forensic review. + +decision_points: + watched_package: + target: package + description: Product inventory includes the watched package filename. + query: + operator: and + filters: + - field: filename + lookup: exact + value: package-match.tar.gz + + legacy_package_present: + target: package + description: Product inventory includes a legacy package artifact. + query: + operator: and + filters: + - field: filename + lookup: icontains + value: legacy + + package_with_download_url: + target: package + description: At least one package has a download URL configured. + query: + operator: and + filters: + - field: download_url + lookup: icontains + value: example.com + + critical_vulnerability: + target: vulnerability + description: Any vulnerability with critical risk score. + query: + operator: and + filters: + - field: risk_score + lookup: gte + value: "8.0" + + known_exploit_vulnerability: + target: vulnerability + description: Any vulnerability with known exploits. + query: + operator: and + filters: + - field: exploitability + lookup: gte + value: "2.0" + + low_risk_vulnerability: + target: vulnerability + description: Any low-risk vulnerability in the product. + query: + operator: and + filters: + - field: risk_score + lookup: lte + value: "3.0" + + direct_product_vulnerability: + target: vulnerability + description: Vulnerability assigned directly to the product. + query: + operator: and + filters: + - field: vulnerability_id + lookup: icontains + value: DIRECT + + production_product: + target: product + description: Product is the production application under triage. + query: + operator: and + filters: + - field: name + lookup: exact + value: Product Match + + non_production_product: + target: product + description: Product is a non-production application. + query: + operator: and + filters: + - field: name + lookup: exact + value: Product Other + +rulesets: + critical-response: + description: High-precedence rules for critical production exposure. + precedence: 200 + enabled: true + default_decision: reachability-analysis + rules: + - name: Upgrade production critical exposure + priority: 1 + decision: upgrade-immediately + conditions: + watched_package: true + package_with_download_url: true + critical_vulnerability: true + production_product: true + + - name: Reachability for known exploit without critical score + priority: 10 + decision: reachability-analysis + conditions: + watched_package: true + known_exploit_vulnerability: true + critical_vulnerability: false + production_product: true + + baseline-response: + description: Fallback rules for lower-risk or non-production products. + precedence: 100 + enabled: true + default_decision: forensics-review + rules: + - name: Downgrade low-risk non-production exposure + priority: 1 + decision: downgrade-risk + conditions: + low_risk_vulnerability: true + critical_vulnerability: false + non_production_product: true + + - name: Forensics when legacy package is present + priority: 20 + decision: forensics-review + conditions: + legacy_package_present: true + production_product: true + direct_product_vulnerability: true diff --git a/triage_rules/tests/data/sample_policy.yaml b/triage_rules/tests/data/sample_policy.yaml new file mode 100644 index 00000000..bd6a0a73 --- /dev/null +++ b/triage_rules/tests/data/sample_policy.yaml @@ -0,0 +1,63 @@ +decisions: + upgrade-now: + action: upgrade + timeline_days: 7 + description: Upgrade affected packages immediately. + + forensics: + action: forensics + timeline_days: 30 + description: Perform forensic analysis. + +decision_points: + matching_package: + target: package + description: Product includes the watched package filename. + query: + operator: and + filters: + - field: filename + lookup: exact + value: package-match.tar.gz + + critical_vulnerability: + target: vulnerability + description: Any critical vulnerability in the product. + query: + operator: and + filters: + - field: risk_score + lookup: gte + value: "8.0" + + matching_product: + target: product + description: Product name matches the policy target. + query: + operator: and + filters: + - field: name + lookup: exact + value: Product Match + +rulesets: + security-policy: + description: Default security triage policy. + precedence: 100 + enabled: true + default_decision: forensics + rules: + - name: Upgrade immediately + priority: 10 + decision: upgrade-now + conditions: + matching_package: true + critical_vulnerability: true + matching_product: true + - name: Forensics for matching product + priority: 1 + decision: forensics + conditions: + matching_package: true + critical_vulnerability: true + matching_product: false diff --git a/triage_rules/views.py b/triage_rules/views.py new file mode 100644 index 00000000..91ea44a2 --- /dev/null +++ b/triage_rules/views.py @@ -0,0 +1,3 @@ +from django.shortcuts import render + +# Create your views here. From 00612554087870e691e9a3f110c43bde97af228c Mon Sep 17 00:00:00 2001 From: Tushar Goel Date: Wed, 29 Jul 2026 16:38:03 +0530 Subject: [PATCH 2/2] Run formatting Signed-off-by: Tushar Goel --- triage_rules/apps.py | 2 +- triage_rules/engine.py | 120 +++---------- triage_rules/loader.py | 16 +- triage_rules/migrations/0001_initial.py | 188 +++++++++++++++++---- triage_rules/models.py | 5 +- triage_rules/tests.py | 9 +- triage_rules/tests/data/sample_policy.yaml | 6 +- 7 files changed, 194 insertions(+), 152 deletions(-) diff --git a/triage_rules/apps.py b/triage_rules/apps.py index 1630cfb8..8f3df84f 100644 --- a/triage_rules/apps.py +++ b/triage_rules/apps.py @@ -2,4 +2,4 @@ class TriageRulesConfig(AppConfig): - name = 'triage_rules' + name = "triage_rules" diff --git a/triage_rules/engine.py b/triage_rules/engine.py index 07ce2e9f..fa1f3b87 100644 --- a/triage_rules/engine.py +++ b/triage_rules/engine.py @@ -28,20 +28,14 @@ def evaluate(self, decision_point): if target == DecisionPoint.TARGET_VULNERABILITY: queryset = decision_point.query.get_qs(user=self.user) return queryset.filter( - Q( - pk__in=self.product.get_vulnerability_qs().values("pk") - ) - | Q( - pk__in=self.product.affected_by_vulnerabilities.values("pk") - ) + Q(pk__in=self.product.get_vulnerability_qs().values("pk")) + | Q(pk__in=self.product.affected_by_vulnerabilities.values("pk")) ).exists() if target == DecisionPoint.TARGET_PRODUCT: return self._product_matches_query(decision_point.query) - raise ValueError( - f"Unknown target {target}" - ) + raise ValueError(f"Unknown target {target}") def _product_matches_query(self, query): q_objects = [] @@ -53,10 +47,8 @@ def _product_matches_query(self, query): if not q_objects: return False - queryset = ( - Product.unsecured_objects - .scope(self.product.dataspace) - .filter(pk=self.product.pk) + queryset = Product.unsecured_objects.scope(self.product.dataspace).filter( + pk=self.product.pk ) operator_type = operator.or_ if query.operator == "or" else operator.and_ return queryset.filter(reduce(operator_type, q_objects)).exists() @@ -66,7 +58,6 @@ class RuleMatcher: @staticmethod def matches(rule, vector): - """ vector @@ -82,20 +73,12 @@ def matches(rule, vector): expected = condition.expected - actual = vector[ - condition.decision_point.name - ] + actual = vector[condition.decision_point.name] - if ( - expected == ExpectedValue.TRUE - and not actual - ): + if expected == ExpectedValue.TRUE and not actual: return False - if ( - expected == ExpectedValue.FALSE - and actual - ): + if expected == ExpectedValue.FALSE and actual: return False return True @@ -109,14 +92,10 @@ def __init__(self, ruleset): def evaluate(self, vector): - rules = ( - self.ruleset.rules - .prefetch_related( - "conditions", - "decision", - ) - .order_by("priority") - ) + rules = self.ruleset.rules.prefetch_related( + "conditions", + "decision", + ).order_by("priority") for rule in rules: @@ -130,7 +109,6 @@ def evaluate(self, vector): class EvaluationEngine: - """ Public entry point. @@ -142,33 +120,20 @@ class EvaluationEngine: ) """ - def evaluate( - self, - product, - rulesets, - user=None, - ): - - decision_point_evaluator = ( - DecisionPointEvaluator( - product, - user=user, - ) + def evaluate(self, product, rulesets, user=None): + + decision_point_evaluator = DecisionPointEvaluator( + product, + user=user, ) vector = {} - decision_points = ( - self.get_all_decision_points( - rulesets - ) - ) + decision_points = self.get_all_decision_points(rulesets) for dp in decision_points: - vector[dp.name] = ( - decision_point_evaluator.evaluate(dp) - ) + vector[dp.name] = decision_point_evaluator.evaluate(dp) winning = None @@ -178,21 +143,11 @@ def evaluate( if not ruleset.enabled: continue - decision = ( - RulesetEvaluator( - ruleset - ).evaluate(vector) - ) + decision = RulesetEvaluator(ruleset).evaluate(vector) - if ( - winning is None - or ruleset.precedence - > winning_precedence - ): + if winning is None or ruleset.precedence > winning_precedence: winning = decision - winning_precedence = ( - ruleset.precedence - ) + winning_precedence = ruleset.precedence return EvaluationResult( decision=winning, @@ -200,9 +155,7 @@ def evaluate( ) @staticmethod - def get_all_decision_points( - rulesets, - ): + def get_all_decision_points(rulesets): seen = {} @@ -210,16 +163,9 @@ def get_all_decision_points( if not ruleset.enabled: continue - for rule in ( - ruleset.rules - .prefetch_related( - "conditions__decision_point" - ) - ): + for rule in ruleset.rules.prefetch_related("conditions__decision_point"): - for condition in ( - rule.conditions.all() - ): + for condition in rule.conditions.all(): dp = condition.decision_point if not dp.enabled: continue @@ -231,23 +177,11 @@ def get_all_decision_points( class EvaluationResult: - def __init__( - self, - decision, - decision_vector, - ): - + def __init__(self, decision, decision_vector): self.decision = decision - self.vector = decision_vector def __repr__(self): - return ( - f"" + f"" ) - - -# Have policies that are more related to real life examples \ No newline at end of file diff --git a/triage_rules/loader.py b/triage_rules/loader.py index fdb81ccc..ea49459f 100644 --- a/triage_rules/loader.py +++ b/triage_rules/loader.py @@ -85,15 +85,11 @@ def _load_decisions(decisions_data, dataspace): action = config.get("action") if action not in dict(TriageDecision.ACTION_CHOICES): - raise PolicyLoadError( - f"Decision '{name}' has invalid action '{action}'." - ) + raise PolicyLoadError(f"Decision '{name}' has invalid action '{action}'.") timeline_days = config.get("timeline_days") if timeline_days is None: - raise PolicyLoadError( - f"Decision '{name}' requires 'timeline_days'." - ) + raise PolicyLoadError(f"Decision '{name}' requires 'timeline_days'.") decision, _ = TriageDecision.objects.update_or_create( name=name, @@ -253,9 +249,7 @@ def _load_rule(ruleset, rule_config, decisions, decision_points): rule_name = rule_config.get("name") if not rule_name: - raise PolicyLoadError( - f"Each rule in ruleset '{ruleset.name}' requires 'name'." - ) + raise PolicyLoadError(f"Each rule in ruleset '{ruleset.name}' requires 'name'.") decision_name = rule_config.get("decision") decision = decisions.get(decision_name) @@ -304,9 +298,7 @@ def _iter_conditions(conditions_data): raise PolicyLoadError( "List-style rule conditions require 'decision_point'." ) - items.append( - (decision_point_name, condition.get("expected")) - ) + items.append((decision_point_name, condition.get("expected"))) return items raise PolicyLoadError("Rule conditions must be a mapping or list.") diff --git a/triage_rules/migrations/0001_initial.py b/triage_rules/migrations/0001_initial.py index bb4727fa..dd8029b2 100644 --- a/triage_rules/migrations/0001_initial.py +++ b/triage_rules/migrations/0001_initial.py @@ -9,74 +9,188 @@ class Migration(migrations.Migration): initial = True dependencies = [ - ('reporting', '0002_alter_columntemplate_content_type_and_more'), + ("reporting", "0002_alter_columntemplate_content_type_and_more"), ] operations = [ migrations.CreateModel( - name='TriageDecision', + name="TriageDecision", fields=[ - ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('name', models.CharField(max_length=100, unique=True)), - ('action', models.CharField(choices=[('upgrade', 'Upgrade'), ('downgrade', 'Downgrade'), ('reachability', 'Reachability Analysis'), ('forensics', 'Forensic Analysis')], max_length=30)), - ('timeline_days', models.PositiveIntegerField()), - ('description', models.TextField(blank=True)), + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("name", models.CharField(max_length=100, unique=True)), + ( + "action", + models.CharField( + choices=[ + ("upgrade", "Upgrade"), + ("downgrade", "Downgrade"), + ("reachability", "Reachability Analysis"), + ("forensics", "Forensic Analysis"), + ], + max_length=30, + ), + ), + ("timeline_days", models.PositiveIntegerField()), + ("description", models.TextField(blank=True)), ], options={ - 'ordering': ('name',), + "ordering": ("name",), }, ), migrations.CreateModel( - name='DecisionPoint', + name="DecisionPoint", fields=[ - ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('name', models.CharField(max_length=100, unique=True)), - ('description', models.TextField(blank=True)), - ('target', models.CharField(choices=[('package', 'Package'), ('vulnerability', 'Vulnerability'), ('product', 'Product')], max_length=20)), - ('enabled', models.BooleanField(default=True)), - ('query', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='decision_points', to='reporting.query')), + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("name", models.CharField(max_length=100, unique=True)), + ("description", models.TextField(blank=True)), + ( + "target", + models.CharField( + choices=[ + ("package", "Package"), + ("vulnerability", "Vulnerability"), + ("product", "Product"), + ], + max_length=20, + ), + ), + ("enabled", models.BooleanField(default=True)), + ( + "query", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="decision_points", + to="reporting.query", + ), + ), ], options={ - 'ordering': ('name',), + "ordering": ("name",), }, ), migrations.CreateModel( - name='Ruleset', + name="Ruleset", fields=[ - ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('name', models.CharField(max_length=100, unique=True)), - ('description', models.TextField(blank=True)), - ('precedence', models.PositiveIntegerField(default=100)), - ('enabled', models.BooleanField(default=True)), - ('default_decision', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='triage_rules.triagedecision')), + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("name", models.CharField(max_length=100, unique=True)), + ("description", models.TextField(blank=True)), + ("precedence", models.PositiveIntegerField(default=100)), + ("enabled", models.BooleanField(default=True)), + ( + "default_decision", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to="triage_rules.triagedecision", + ), + ), ], options={ - 'ordering': ('-precedence', 'name'), + "ordering": ("-precedence", "name"), }, ), migrations.CreateModel( - name='Rule', + name="Rule", fields=[ - ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('name', models.CharField(max_length=100)), - ('priority', models.PositiveIntegerField(default=100, help_text='Lower values are evaluated first.')), - ('ruleset', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='rules', to='triage_rules.ruleset')), - ('decision', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='rules', to='triage_rules.triagedecision')), + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("name", models.CharField(max_length=100)), + ( + "priority", + models.PositiveIntegerField( + default=100, help_text="Lower values are evaluated first." + ), + ), + ( + "ruleset", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="rules", + to="triage_rules.ruleset", + ), + ), + ( + "decision", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="rules", + to="triage_rules.triagedecision", + ), + ), ], options={ - 'ordering': ('priority', 'id'), + "ordering": ("priority", "id"), }, ), migrations.CreateModel( - name='RuleCondition', + name="RuleCondition", fields=[ - ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('expected', models.PositiveSmallIntegerField(choices=[(0, 'False'), (1, 'True')])), - ('decision_point', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='triage_rules.decisionpoint')), - ('rule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='conditions', to='triage_rules.rule')), + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "expected", + models.PositiveSmallIntegerField( + choices=[(0, "False"), (1, "True")] + ), + ), + ( + "decision_point", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="triage_rules.decisionpoint", + ), + ), + ( + "rule", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="conditions", + to="triage_rules.rule", + ), + ), ], options={ - 'unique_together': {('rule', 'decision_point')}, + "unique_together": {("rule", "decision_point")}, }, ), ] diff --git a/triage_rules/models.py b/triage_rules/models.py index 7e5611de..424252b1 100644 --- a/triage_rules/models.py +++ b/triage_rules/models.py @@ -2,6 +2,7 @@ from reporting.models import Query + class DecisionPoint(models.Model): """ A reusable boolean predicate. @@ -209,6 +210,4 @@ class Meta: ) def __str__(self): - return ( - f"{self.decision_point.name} = {self.expected}" - ) + return f"{self.decision_point.name} = {self.expected}" diff --git a/triage_rules/tests.py b/triage_rules/tests.py index 176dfe64..83fc82b6 100644 --- a/triage_rules/tests.py +++ b/triage_rules/tests.py @@ -768,9 +768,12 @@ def test_comprehensive_policy_load_store_and_evaluate_end_to_end(self): self.assertEqual(9, DecisionPoint.objects.count()) self.assertEqual(9, Query.objects.filter(dataspace=self.dataspace).count()) self.assertEqual(9, Filter.objects.filter(dataspace=self.dataspace).count()) - self.assertEqual(2, Ruleset.objects.filter( - name__in=["critical-response", "baseline-response"] - ).count()) + self.assertEqual( + 2, + Ruleset.objects.filter( + name__in=["critical-response", "baseline-response"] + ).count(), + ) self.assertEqual(4, Rule.objects.count()) self.assertEqual(14, RuleCondition.objects.count()) diff --git a/triage_rules/tests/data/sample_policy.yaml b/triage_rules/tests/data/sample_policy.yaml index bd6a0a73..6239d950 100644 --- a/triage_rules/tests/data/sample_policy.yaml +++ b/triage_rules/tests/data/sample_policy.yaml @@ -16,9 +16,9 @@ decision_points: query: operator: and filters: - - field: filename - lookup: exact - value: package-match.tar.gz + - field: risk + lookup: gte + value: "8.0" critical_vulnerability: target: vulnerability