use crate::rules::RuleRegistry;
use crate::Finding;
use std::collections::HashMap;
pub mod deadlines {
pub const SOFTWARE_FIRMWARE_SIGNING: &str = "2030";
pub const NETWORKING_EQUIPMENT: &str = "2030";
pub const WEB_AND_CLOUD: &str = "2033";
pub const OPERATING_SYSTEMS: &str = "2033";
pub const NICHE_LEGACY: &str = "2033";
pub const NSS_WIDE_OUTER_LIMIT: &str = "2035";
}
pub fn deadline_for_rule_id<'a>(registry: &'a RuleRegistry, rule_id: &str) -> Option<&'a str> {
registry
.all_rules()
.iter()
.find(|r| r.id() == rule_id)
.and_then(|r| r.cnsa2_deadline())
}
pub fn annotate_cnsa2_deadlines(findings: &mut [Finding], registry: &RuleRegistry) {
let map: HashMap<&str, &'static str> = registry
.all_rules()
.iter()
.filter_map(|r| r.cnsa2_deadline().map(|d| (r.id(), d)))
.collect();
if map.is_empty() {
return;
}
for f in findings.iter_mut() {
if f.cnsa2_deadline.is_some() {
continue;
}
if let Some(deadline) = map.get(f.rule_id.as_str()) {
f.cnsa2_deadline = Some((*deadline).to_string());
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MigrationLevel {
Clean,
OnTrack,
AtRisk,
}
impl std::fmt::Display for MigrationLevel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
MigrationLevel::Clean => write!(f, "clean"),
MigrationLevel::OnTrack => write!(f, "on-track"),
MigrationLevel::AtRisk => write!(f, "at-risk"),
}
}
}
#[derive(Debug, Clone)]
pub struct MigrationReport {
pub annotated: usize,
pub pq_total: usize,
pub by_deadline: HashMap<String, usize>,
pub level: MigrationLevel,
pub pq_ready: usize,
pub pq_algorithms: Vec<String>,
}
impl MigrationReport {
pub const WARN_THRESHOLD: f32 = 0.5;
pub fn from_findings(findings: &[Finding]) -> Self {
let mut annotated = 0usize;
let mut by_deadline: HashMap<String, usize> = HashMap::new();
let mut pq_ready = 0usize;
let mut pq_algorithms: Vec<String> = Vec::new();
for f in findings {
if f.is_pq_ready() {
pq_ready += 1;
if let Some(algo) = &f.crypto_algorithm {
if !pq_algorithms.iter().any(|a| a == algo) {
pq_algorithms.push(algo.clone());
}
}
continue;
}
if let Some(d) = &f.cnsa2_deadline {
annotated += 1;
*by_deadline.entry(d.clone()).or_insert(0) += 1;
}
}
pq_algorithms.sort();
let pq_total = annotated;
let level = if pq_total == 0 {
MigrationLevel::Clean
} else {
let ratio = annotated as f32 / pq_total as f32;
if ratio >= Self::WARN_THRESHOLD {
MigrationLevel::AtRisk
} else {
MigrationLevel::OnTrack
}
};
Self {
annotated,
pq_total,
by_deadline,
level,
pq_ready,
pq_algorithms,
}
}
pub fn readiness_percent(&self) -> Option<u32> {
let total = self.pq_ready + self.annotated;
if total == 0 {
return None;
}
Some(((self.pq_ready as f64 / total as f64) * 100.0).round() as u32)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{default_confidence, Severity};
fn mk(rule_id: &str, deadline: Option<&str>) -> Finding {
Finding {
rule_id: rule_id.to_string(),
severity: Severity::High,
cwe: None,
description: String::new(),
file: String::new(),
line: 1,
column: 1,
end_line: 1,
end_column: 1,
snippet: String::new(),
source_line: None,
source_description: None,
sink_line: None,
sink_description: None,
fix_suggestion: None,
sink_start_byte: None,
sink_end_byte: None,
confidence: default_confidence(),
taint_hops: None,
tags: vec![],
crypto_algorithm: None,
cnsa2_deadline: deadline.map(String::from),
dep_name: None,
dep_version: None,
dep_ecosystem: None,
dep_purl: None,
dep_vulnerability_id: None,
dep_fixed_version: None,
dep_source: None,
dep_vulnerability_severity: None,
dep_path: vec![],
crypto_material: None,
}
}
#[test]
fn deadlines_are_sourced_strings() {
assert_eq!(deadlines::SOFTWARE_FIRMWARE_SIGNING, "2030");
assert_eq!(deadlines::NETWORKING_EQUIPMENT, "2030");
assert_eq!(deadlines::WEB_AND_CLOUD, "2033");
assert_eq!(deadlines::OPERATING_SYSTEMS, "2033");
assert_eq!(deadlines::NICHE_LEGACY, "2033");
assert_eq!(deadlines::NSS_WIDE_OUTER_LIMIT, "2035");
}
#[test]
fn annotate_uses_registry_not_substring_matching() {
let registry = RuleRegistry::new();
let mut findings = vec![
mk("py/pq-vulnerable-crypto", None),
mk("py/no-weak-crypto", None),
mk("unknown/fake-rule", None),
];
annotate_cnsa2_deadlines(&mut findings, ®istry);
assert_eq!(
findings[0].cnsa2_deadline.as_deref(),
Some(deadlines::WEB_AND_CLOUD)
);
assert!(
findings[1].cnsa2_deadline.is_none(),
"no-weak-crypto must not be annotated just because id contains 'crypto'"
);
assert!(
findings[2].cnsa2_deadline.is_none(),
"unknown rule must not be annotated"
);
}
#[test]
fn deadline_lookup_returns_none_for_unknown_rule() {
let registry = RuleRegistry::new();
assert!(deadline_for_rule_id(®istry, "definitely/not-a-rule").is_none());
}
#[test]
fn deadline_lookup_returns_expected_class_for_known_rule() {
let registry = RuleRegistry::new();
assert_eq!(
deadline_for_rule_id(®istry, "config/nginx-pq-vulnerable-tls"),
Some(deadlines::WEB_AND_CLOUD)
);
}
#[test]
fn annotate_preserves_existing_deadline_if_set() {
let registry = RuleRegistry::new();
let mut findings = vec![mk("py/pq-vulnerable-crypto", Some("2099"))];
annotate_cnsa2_deadlines(&mut findings, ®istry);
assert_eq!(findings[0].cnsa2_deadline.as_deref(), Some("2099"));
}
#[test]
fn migration_report_empty_findings_is_clean() {
let report = MigrationReport::from_findings(&[]);
assert_eq!(report.annotated, 0);
assert_eq!(report.pq_total, 0);
assert_eq!(report.level, MigrationLevel::Clean);
assert!(report.by_deadline.is_empty());
}
#[test]
fn migration_report_all_pq_safe_is_clean() {
let findings = vec![mk("py/no-eval", None), mk("js/no-sql-injection", None)];
let report = MigrationReport::from_findings(&findings);
assert_eq!(report.level, MigrationLevel::Clean);
assert_eq!(report.annotated, 0);
assert_eq!(report.pq_total, 0);
}
#[test]
fn migration_report_all_pq_vulnerable_is_at_risk() {
let findings = vec![
mk("py/pq-vulnerable-crypto", Some("2033")),
mk("js/pq-vulnerable-crypto", Some("2033")),
];
let report = MigrationReport::from_findings(&findings);
assert_eq!(report.level, MigrationLevel::AtRisk);
assert_eq!(report.annotated, 2);
assert_eq!(report.by_deadline.get("2033").copied(), Some(2));
}
#[test]
fn migration_report_tallies_per_deadline() {
let findings = vec![
mk("py/pq-vulnerable-crypto", Some("2033")),
mk("config/nginx-pq-vulnerable-tls", Some("2033")),
mk("other/signing", Some("2030")),
];
let report = MigrationReport::from_findings(&findings);
assert_eq!(report.annotated, 3);
assert_eq!(report.by_deadline.get("2033").copied(), Some(2));
assert_eq!(report.by_deadline.get("2030").copied(), Some(1));
}
fn mk_pq_ready(algo: &str) -> Finding {
let mut f = mk("py/pq-ready-crypto", None);
f.tags = vec![crate::PQ_READY_TAG.to_string()];
f.crypto_algorithm = Some(algo.to_string());
f
}
#[test]
fn pq_ready_findings_counted_separately_from_vulnerable() {
let findings = vec![
mk("py/pq-vulnerable-crypto", Some("2033")),
mk_pq_ready("ML-KEM"),
mk_pq_ready("ML-KEM"),
mk_pq_ready("X25519MLKEM768"),
];
let report = MigrationReport::from_findings(&findings);
assert_eq!(report.annotated, 1);
assert_eq!(report.pq_ready, 3);
assert_eq!(report.pq_algorithms, vec!["ML-KEM", "X25519MLKEM768"]);
assert_eq!(report.readiness_percent(), Some(75));
}
#[test]
fn readiness_is_none_without_any_crypto() {
let report = MigrationReport::from_findings(&[mk("py/no-eval", None)]);
assert_eq!(report.pq_ready, 0);
assert_eq!(report.readiness_percent(), None);
}
#[test]
fn readiness_is_full_when_only_post_quantum() {
let report = MigrationReport::from_findings(&[mk_pq_ready("ML-KEM")]);
assert_eq!(report.annotated, 0);
assert_eq!(report.readiness_percent(), Some(100));
assert_eq!(report.level, MigrationLevel::Clean);
}
#[test]
fn readiness_is_zero_when_migration_not_started() {
let report = MigrationReport::from_findings(&[mk("py/pq-vulnerable-crypto", Some("2033"))]);
assert_eq!(report.readiness_percent(), Some(0));
}
#[test]
fn migration_level_display_is_stable() {
assert_eq!(MigrationLevel::Clean.to_string(), "clean");
assert_eq!(MigrationLevel::OnTrack.to_string(), "on-track");
assert_eq!(MigrationLevel::AtRisk.to_string(), "at-risk");
}
#[test]
fn every_registered_rule_with_deadline_uses_a_documented_class() {
let registry = RuleRegistry::new();
let valid: std::collections::HashSet<&'static str> = [
deadlines::SOFTWARE_FIRMWARE_SIGNING,
deadlines::NETWORKING_EQUIPMENT,
deadlines::WEB_AND_CLOUD,
deadlines::OPERATING_SYSTEMS,
deadlines::NICHE_LEGACY,
deadlines::NSS_WIDE_OUTER_LIMIT,
]
.into_iter()
.collect();
for rule in registry.all_rules() {
if let Some(d) = rule.cnsa2_deadline() {
assert!(
valid.contains(d),
"rule {} has non-canonical deadline {:?}",
rule.id(),
d
);
}
}
}
}