feat(cra): coarse priority engine — P0 floor + customer weights + quick wins
Deterministic prioritisation on top of the mapper (cra_prioritizer.py): a non-negotiable P0 floor (safety-function compromise / actively exploited / CRITICAL — customer weights cannot demote) plus a discretionary tier ranked by severity x the customer's weight (high/medium/low) for the 5 business objectives (access/data/network_api/supply_updates/monitoring). Quick-win flag (high impact, low effort) for a second view; each finding carries a short priority reason. Endpoint accepts weights + per-finding safety_impact/exploited. Rough pre-sort only (devs re-sort in Jira). No DB. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -38,3 +38,26 @@ def test_assess_requires_finding_id():
|
||||
# id is required by the schema -> 422
|
||||
r = client.post("/api/v1/cra/assess", json={"findings": [{"title": "no id"}]})
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_assess_prioritizes_with_weights():
|
||||
r = client.post("/api/v1/cra/assess", json={
|
||||
"findings": [
|
||||
{"id": "mfa", "cwe": "CWE-306", "severity": "high"},
|
||||
{"id": "log", "cwe": "CWE-778", "severity": "high"},
|
||||
],
|
||||
"weights": {"access": "high", "monitoring": "low"},
|
||||
})
|
||||
assert r.status_code == 200
|
||||
d = r.json()
|
||||
order = [m["finding_id"] for m in d["mapped"]]
|
||||
assert order.index("mfa") < order.index("log")
|
||||
assert all("priority_tier" in m for m in d["mapped"])
|
||||
|
||||
|
||||
def test_assess_p0_floor_on_safety_impact():
|
||||
r = client.post("/api/v1/cra/assess", json={"findings": [
|
||||
{"id": "s", "cwe": "CWE-319", "severity": "low", "safety_impact": True},
|
||||
]})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["mapped"][0]["priority_tier"] == "P0"
|
||||
|
||||
@@ -77,7 +77,8 @@ def test_payload_entry_is_json_serializable_and_deterministic():
|
||||
assert r1 == r2 # deterministic
|
||||
assert r1["findings_total"] == 2
|
||||
assert isinstance(r1["mapped"], list) and isinstance(r1["mapped"][0], dict)
|
||||
assert r1["mapped"][0]["primary_requirement"] == "CRA-AI-9"
|
||||
by_id = {m["finding_id"]: m for m in r1["mapped"]} # order is now priority-sorted
|
||||
assert by_id["x"]["primary_requirement"] == "CRA-AI-9"
|
||||
|
||||
|
||||
def test_empty_payload_is_safe():
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Tests for the coarse CRA prioritisation (P0 floor + weighted tier + quick wins)."""
|
||||
from compliance.services.cra_finding_mapper import ScannerFinding, assess_findings
|
||||
|
||||
|
||||
def test_safety_impact_forces_p0():
|
||||
a = assess_findings([ScannerFinding(id="s", title="TLS 1.0", cwe="CWE-319", severity="medium", safety_impact=True)])
|
||||
m = a.mapped[0]
|
||||
assert m.priority_tier == "P0"
|
||||
assert "Personenschaden" in m.priority_reason
|
||||
|
||||
|
||||
def test_exploited_forces_p0():
|
||||
a = assess_findings([ScannerFinding(id="e", title="outdated dep", category="dependency", severity="medium", exploited=True)])
|
||||
assert a.mapped[0].priority_tier == "P0"
|
||||
|
||||
|
||||
def test_critical_is_p0():
|
||||
a = assess_findings([ScannerFinding(id="c", title="default password", cwe="CWE-259", severity="critical")])
|
||||
assert a.mapped[0].priority_tier == "P0"
|
||||
|
||||
|
||||
def test_weights_order_the_discretionary_tier():
|
||||
findings = [
|
||||
ScannerFinding(id="log", title="no security logging", cwe="CWE-778", severity="high"), # monitoring
|
||||
ScannerFinding(id="mfa", title="missing authentication", cwe="CWE-306", severity="high"), # access
|
||||
]
|
||||
a = assess_findings(findings, weights={"access": "high", "monitoring": "low"})
|
||||
order = [m.finding_id for m in a.mapped]
|
||||
assert order.index("mfa") < order.index("log")
|
||||
assert a.mapped[0].priority_tier != "P0" # neither is a floor finding
|
||||
|
||||
|
||||
def test_quick_win_flag_and_view():
|
||||
a = assess_findings([ScannerFinding(id="tls", title="TLS 1.0", cwe="CWE-319", severity="high")])
|
||||
m = a.mapped[0]
|
||||
assert m.primary_requirement == "CRA-AI-15" # effort 2 days
|
||||
assert m.quick_win is True
|
||||
assert "tls" in a.quick_wins
|
||||
|
||||
|
||||
def test_p0_sorts_above_discretionary():
|
||||
findings = [
|
||||
ScannerFinding(id="low", title="missing logging", cwe="CWE-778", severity="low"), # P3
|
||||
ScannerFinding(id="crit", title="default password", cwe="CWE-259", severity="critical"), # P0
|
||||
]
|
||||
a = assess_findings(findings)
|
||||
assert a.mapped[0].finding_id == "crit"
|
||||
|
||||
|
||||
def test_objectives_exposed():
|
||||
a = assess_findings([])
|
||||
assert a.objectives == ["access", "data", "network_api", "supply_updates", "monitoring"]
|
||||
Reference in New Issue
Block a user