diff --git a/.github/workflows/pqc-compliance-gate.yml b/.github/workflows/pqc-compliance-gate.yml index b6bb248..2814e4c 100644 --- a/.github/workflows/pqc-compliance-gate.yml +++ b/.github/workflows/pqc-compliance-gate.yml @@ -141,6 +141,7 @@ jobs: --schemafile schemas/crypto-inventory.schema.json \ examples/inventory/mixed-platform-inventory.json \ examples/inventory/scoring-ready-inventory.json \ + examples/inventory/calendar-anchored-inventory.json \ build/evidence/iac-inventory.json - name: Exercise schema-compatible risk enrichment and scoring diff --git a/policies/scoring/risk_score.rego b/policies/scoring/risk_score.rego index 3de8860..592232f 100644 --- a/policies/scoring/risk_score.rego +++ b/policies/scoring/risk_score.rego @@ -150,12 +150,43 @@ has_deadline_input(asset) if { object.get(asset, "regulatory_category", null) in regulatory_category_values } else := false +# Optional enrichment must be validated whenever present. Otherwise an +# unsupported category can hide behind an explicit numeric deadline, and a +# fractional fan-in count can be reported as though it satisfied the schema. +valid_optional_regulatory_category(asset) if { + object.get(asset, "regulatory_category", null) == null +} else if { + object.get(asset, "regulatory_category", null) in regulatory_category_values +} else := false + +valid_optional_dependent_asset_count(asset) if { + object.get(asset, "dependent_asset_count", null) == null +} else if { + count := object.get(asset, "dependent_asset_count", null) + is_number(count) + count >= 0 + count == floor(count) +} else := false + +# An asset can be scored by itself without an inventory envelope. During an +# inventory assessment, however, every asset_id must be unique before queue +# construction: migration_work_queue later keys its ordering by asset_id. +duplicate_asset_id(asset) if { + asset_id := object.get(asset, "asset_id", null) + count([candidate | + some candidate in input.assets + object.get(candidate, "asset_id", null) == asset_id + ]) > 1 +} + metadata_checks(asset) := { "asset_id must be a non-empty string": valid_nonempty_string(object.get(asset, "asset_id", null)), "secrecy_lifetime_years must be a non-negative number": valid_nonnegative_number(object.get(asset, "secrecy_lifetime_years", null)), "data_classification is missing or unsupported": object.get(asset, "data_classification", null) in classification_values, "impact is missing or unsupported": object.get(asset, "impact", null) in impact_values, "migration_deadline_months must be a number, or regulatory_category must be set": has_deadline_input(asset), + "regulatory_category is set but unsupported": valid_optional_regulatory_category(asset), + "dependent_asset_count must be a non-negative integer when set": valid_optional_dependent_asset_count(asset), "remediation_effort is missing or unsupported": object.get(asset, "remediation_effort", null) in effort_values, "evidence_confidence is missing or unsupported": object.get(asset, "evidence_confidence", null) in confidence_values, } @@ -167,6 +198,14 @@ metadata_errors(asset) := {message | is_valid(asset) if count(metadata_errors(asset)) == 0 +# This is intentionally separate from is_valid: standalone scoring has no +# inventory to compare against, while collection assessment must reject +# ambiguous identities before it builds queue entries keyed by asset_id. +is_valid_inventory_asset(asset) if { + is_valid(asset) + not duplicate_asset_id(asset) +} + inherent_risk_score(asset) := total if { is_valid(asset) total := (hndl_weight(asset.secrecy_lifetime_years) + classification_weight(asset.data_classification)) + impact_weight(asset.impact) @@ -278,7 +317,7 @@ display_deadline_months(asset) := round(effective_deadline_months(asset) * 10) / scored_inventory contains entry if { some asset in input.assets - is_valid(asset) + is_valid_inventory_asset(asset) entry := { "asset_id": asset.asset_id, "inherent_risk_score": inherent_risk_score(asset), @@ -302,6 +341,15 @@ invalid_inventory contains entry if { } } +invalid_inventory contains entry if { + some asset in input.assets + duplicate_asset_id(asset) + entry := { + "asset_id": object.get(asset, "asset_id", "unknown"), + "errors": ["asset_id must be unique within the inventory"], + } +} + priority_matrix := { "critical": sort([e.asset_id | some e in scored_inventory; e.tier == "critical"]), "high": sort([e.asset_id | some e in scored_inventory; e.tier == "high"]), @@ -338,13 +386,13 @@ work_queue_sort_key(asset) := [ work_queue_keys := [work_queue_sort_key(asset) | some asset in input.assets - is_valid(asset) + is_valid_inventory_asset(asset) ] work_queue_entry(asset_id) := entry if { some asset in input.assets asset.asset_id == asset_id - is_valid(asset) + is_valid_inventory_asset(asset) entry := { "asset_id": asset.asset_id, "inherent_risk_score": inherent_risk_score(asset), diff --git a/policies/scoring/risk_score_test.rego b/policies/scoring/risk_score_test.rego index e3c1c5b..b3bb9b3 100644 --- a/policies/scoring/risk_score_test.rego +++ b/policies/scoring/risk_score_test.rego @@ -170,6 +170,19 @@ test_missing_both_deadline_inputs_is_invalid if { not is_valid(asset) } +test_malformed_optional_enrichment_is_invalid if { + asset := object.union(base_asset, { + # The numeric deadline makes the category unnecessary for scoring, but + # an explicitly supplied optional field must still satisfy the schema. + "regulatory_category": "not_a_cnsa_category", + "dependent_asset_count": 1.5, + }) + errors := metadata_errors(asset) + "regulatory_category is set but unsupported" in errors + "dependent_asset_count must be a non-negative integer when set" in errors + not is_valid(asset) +} + # --- Gap #4: classification/impact divergence (reported only) -------------- test_aligned_classification_and_impact_is_not_flagged if { @@ -221,6 +234,22 @@ test_work_queue_uses_effort_only_as_tiebreaker if { queue[3].inherent_risk_score < queue[2].inherent_risk_score } +test_duplicate_asset_ids_are_invalid_and_do_not_break_the_queue if { + duplicate := object.union(base_asset, { + "secrecy_lifetime_years": 2, + "data_classification": "public", + "impact": "informational", + "migration_deadline_months": 60, + "remediation_effort": "low", + }) + result := assessment with input as {"assets": [base_asset, duplicate]} + result.valid_asset_count == 0 + count(result.invalid_inventory) == 1 + count(result.migration_work_queue) == 0 + some invalid in result.invalid_inventory + "asset_id must be unique within the inventory" in invalid.errors +} + # --- Gap #7: dependency-informed impact hint (advisory only) --------------- test_high_fan_in_flags_impact_as_possibly_underrated if {