Skip to content

fix(scoring): validate optional metadata and duplicate queue IDs #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ethanolivertroy wants to merge 1 commit into
base: risk-scoring-gaps-1-7
Choose a base branch
from
Draft

fix(scoring): validate optional metadata and duplicate queue IDs #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/pqc-compliance-gate.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
54 changes: 51 additions & 3 deletions policies/scoring/risk_score.rego
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}
Expand All @@ -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)
Expand Down Expand Up @@ -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),
Expand All @@ -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"]),
Expand Down Expand Up @@ -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),
Expand Down
29 changes: 29 additions & 0 deletions policies/scoring/risk_score_test.rego
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down