emergent-security/src/emergent_security/scales.py

81 lines
3.2 KiB
Python

"""Ordinal risk scales (canon B1 risk dimensions, A3 step 8).
Every comparison fails safe: a value that cannot be ranked is treated as exceeding.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping
import yaml
from emergent_security.model import same_value
@dataclass(frozen=True)
class RiskScales:
scales: Mapping[str, tuple[Any, ...]]
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> RiskScales:
return cls(scales={dim: tuple(levels) for dim, levels in data["scales"].items()})
@classmethod
def load(cls, path: Path) -> RiskScales:
with Path(path).open(encoding="utf-8") as handle:
return cls.from_dict(yaml.safe_load(handle))
@property
def dimensions(self) -> tuple[str, ...]:
return tuple(self.scales)
def rank(self, dimension: str, value: Any) -> int | None:
for index, level in enumerate(self.scales.get(dimension, ())):
if same_value(level, value):
return index
return None
def within(self, dimension: str, value: Any, ceiling: Any) -> bool:
value_rank, ceiling_rank = self.rank(dimension, value), self.rank(dimension, ceiling)
return value_rank is not None and ceiling_rank is not None and value_rank <= ceiling_rank
def narrower(self, dimension: str, left: Any, right: Any) -> Any:
"""The more restrictive of two ceilings; an unrankable value wins so that checks fail safe."""
left_rank, right_rank = self.rank(dimension, left), self.rank(dimension, right)
if left_rank is None:
return left
if right_rank is None:
return right
return left if left_rank <= right_rank else right
def intersect(self, *ceilings: Mapping[str, Any]) -> dict[str, Any]:
result: dict[str, Any] = {}
for mapping in ceilings:
for dimension, value in mapping.items():
result[dimension] = self.narrower(dimension, result[dimension], value) if dimension in result else value
return result
def exceeded(self, risk: Mapping[str, Any], ceilings: Mapping[str, Any], *, declared_only: bool = False) -> tuple[str, ...]:
"""Dimensions where `risk` is above `ceilings`, in scale order.
By default a dimension the ceilings declare but the request does not classify
counts as exceeded (A3 step 8). With declared_only, only dimensions the request
states are checked.
"""
ordered = [d for d in self.dimensions if d in ceilings] + sorted(d for d in ceilings if d not in self.scales)
return tuple(
dimension
for dimension in ordered
if not (declared_only and risk.get(dimension) is None)
and not self.within(dimension, risk.get(dimension), ceilings[dimension])
)
def check(self, mapping: Mapping[str, Any], where: str) -> list[str]:
"""Validation errors for a ceilings mapping."""
return [
f"{where}: unknown value for {dimension}: {value}"
for dimension, value in mapping.items()
if self.rank(dimension, value) is None
]