Spaces:
Running on Zero
Running on Zero
File size: 2,542 Bytes
40b1357 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | """Cross-artifact validation for SignalBundle and CallDecision."""
from .decisions import signal_bundle_sha256
from .schemas import CallDecision, SignalBundle
def validate_decision_references(
decision: CallDecision,
bundle: SignalBundle,
) -> CallDecision:
"""Ensure every decision citation resolves inside its signal bundle."""
if decision.call_id != bundle.call_id:
raise ValueError("decision and signal bundle call_id values do not match")
if decision.signal_bundle_schema_version != bundle.schema_version:
raise ValueError("decision references an incompatible signal bundle schema")
if (
decision.signal_bundle_sha256
and decision.signal_bundle_sha256
!= signal_bundle_sha256(bundle)
):
raise ValueError("decision signal bundle hash does not match")
segment_ids = {segment.segment_id for segment in bundle.segments}
signal_ids = {signal.signal_id for signal in bundle.signals}
episode_ids = {episode.episode_id for episode in bundle.episodes}
findings = decision.triggered_findings + decision.positive_findings
for finding in findings:
unknown_rule_signals = (
set(finding.detection_rule.signal_ids) - signal_ids
)
if unknown_rule_signals:
raise ValueError(
f"finding {finding.finding_id!r} rule references unknown signals: "
f"{sorted(unknown_rule_signals)}"
)
for evidence in finding.evidence + finding.counter_evidence:
unknown_segments = set(evidence.segment_ids) - segment_ids
unknown_signals = set(evidence.signal_ids) - signal_ids
unknown_episodes = set(evidence.episode_ids) - episode_ids
if unknown_segments or unknown_signals or unknown_episodes:
raise ValueError(
f"evidence {evidence.evidence_id!r} has unresolved references"
)
for uncertainty in decision.uncertainties:
unknown_signals = set(uncertainty.affected_signal_ids) - signal_ids
if unknown_signals:
raise ValueError(
f"uncertainty {uncertainty.code!r} references unknown signals"
)
unknown_findings = (
set(uncertainty.affected_finding_ids)
- {finding.finding_id for finding in findings}
)
if unknown_findings:
raise ValueError(
f"uncertainty {uncertainty.code!r} references unknown findings"
)
return decision
|