From 0903578e99837bb4be6d539fac5f1ffa332e90b3 Mon Sep 17 00:00:00 2001 From: Georgi Gaydarov Date: Mon, 1 Jun 2026 10:36:29 -0400 Subject: [PATCH] feat(risk): deterministic authority gate + no-self-escalation [#5-core] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-risk-tier authority model — two orthogonal axes (severity vs authority), hard-force overrides, and the no-self-escalation ceiling. Additive new package (core/risk/); no pipeline or schema change. - Severity (NONE..CRITICAL) and Tier (AUTO BLOCK, spine/underlay -> >= HOTL — most-restrictive wins. A twin-clean LOW-severity AS edit still BLOCKs: severity != authority is the real product change. - authorize(): the no-self-escalation ceiling (required <= max_authorized, else BLOCK). max_authorized is a plain input here; binding it to a PIV-signed, compiled-in value is the #5-ceiling / CROSS-3 follow-up — no crypto is claimed. tests/authority_test.py: 14 tests incl. the differential vs both legacy scorers, two hard-force PROPERTY tests (fabric-identity always BLOCK; spine/underlay never AUTO across all severities), and the no-self-escalation 0-violations property. 14/14 green; stress/ wedge(10)/promote/contract pass; PR-1 egress 29/29. --- core/risk/__init__.py | 25 ++++++ core/risk/authority.py | 171 ++++++++++++++++++++++++++++++++++++ tests/authority_test.py | 190 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 386 insertions(+) create mode 100644 core/risk/__init__.py create mode 100644 core/risk/authority.py create mode 100644 tests/authority_test.py diff --git a/core/risk/__init__.py b/core/risk/__init__.py new file mode 100644 index 0000000..a2405ee --- /dev/null +++ b/core/risk/__init__.py @@ -0,0 +1,25 @@ +"""core/risk/ — deterministic authority model (#5): severity vs authority, hard-force +overrides for fabric-identity / spine / underlay, and the no-self-escalation ceiling.""" +from __future__ import annotations + +from .authority import ( + AuthorityDecision, + ChangeClass, + Severity, + Tier, + authorize, + classify_change, + required_authority, + unify_severity, +) + +__all__ = [ + "Severity", + "Tier", + "ChangeClass", + "AuthorityDecision", + "classify_change", + "unify_severity", + "required_authority", + "authorize", +] diff --git a/core/risk/authority.py b/core/risk/authority.py new file mode 100644 index 0000000..5ccd0e8 --- /dev/null +++ b/core/risk/authority.py @@ -0,0 +1,171 @@ +"""core/risk/authority.py — the deterministic per-risk-tier authority model (#5). + +Two ORTHOGONAL axes, both pure and auditable: + + Severity (how bad if it goes wrong) : NONE < LOW < MEDIUM < HIGH < CRITICAL + Authority (how much autonomy is allowed): AUTO < HITL < HOTL < BLOCK + +The product insight is that these are NOT the same axis. A LOW-severity change to a +fabric-identity construct (AS number, route-distinguisher / route-target) is still BLOCK, +and a spine/underlay touch is never fully automated no matter how clean the twin looked. +`required_authority` derives a base tier from severity, then applies hard-force overrides +from the change class, and returns the MOST RESTRICTIVE of the two. + +`authorize` is the no-self-escalation ceiling: the computed `required` authority must be +<= a `max_authorized` ceiling, else the change is BLOCKED — an agent can never grant +itself more autonomy than its ceiling allows. (Binding `max_authorized` to a hardware-PIV +-signed, compiled-in value is the #5-ceiling / CROSS-3 follow-up; this module ships the +deterministic invariant that signature will protect — no crypto is claimed here.) + +`unify_severity` reconciles AEGIS's two existing severity scorers — guards.risk_tier's +3-bucket scale and bundler._severity's 5-bucket scale — into one canonical 5-bucket scale. +The richer engine wins; tests/authority_test.py carries the differential vs both. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import IntEnum + +from ..backends.base import GeneratedConfig + + +class Severity(IntEnum): + NONE = 0 + LOW = 1 + MEDIUM = 2 + HIGH = 3 + CRITICAL = 4 + + +class Tier(IntEnum): + """Authority tier — how much autonomy a verified change is allowed. Ordered so a + HIGHER value is MORE restrictive (less autonomy).""" + + AUTO = 0 # fully automated, no human + HITL = 1 # human-in-the-loop: explicit approval before apply + HOTL = 2 # human-on-the-loop: applied under monitoring, human can intervene/abort + BLOCK = 3 # never auto-promotable + + +@dataclass(frozen=True) +class ChangeClass: + """What fabric construct the change touches — drives the hard-force overrides. + + AS/RD/RT detection is regex over the candidate config and is reliable across vendors. + The spine/underlay flags are a documented HEURISTIC (device-name + underlay-protocol + patterns); a production deployment should back them with an authoritative topology + role source (containerlab labels / NetBox device role). Until then they bias toward + FAIL CLOSED — an IGP/underlay-looking change is treated as touching the underlay.""" + + touches_asn: bool = False + touches_rd_rt: bool = False + touches_spine: bool = False + touches_underlay: bool = False + + @property + def is_fabric_identity(self) -> bool: + """AS number / RD / RT — the identity of the fabric. Never auto-changed.""" + return self.touches_asn or self.touches_rd_rt + + +# ── change classification (regex, fail-closed) ────────────────────────────────── + +_ASN_RE = re.compile( + r"(?:autonomous-system|remote-as|peer-as|local-as)\s+\d+|router\s+bgp\s+\d+", + re.IGNORECASE, +) +_RD_RT_RE = re.compile( + r"route-(?:distinguisher|target)|\brd\s+\d+:\d+|\brt\s+\d+:\d+|target:\d+:\d+", + re.IGNORECASE, +) +_UNDERLAY_CONFIG_RE = re.compile(r"\b(?:underlay|ospf|isis|is-is)\b", re.IGNORECASE) +_SPINE_NAME_RE = re.compile(r"\bspine\b", re.IGNORECASE) +_UNDERLAY_NAME_RE = re.compile(r"\b(?:underlay|super-?spine)\b", re.IGNORECASE) + + +def classify_change(configs: list[GeneratedConfig]) -> ChangeClass: + """Classify candidate configs into fabric-construct flags (the hard-force triggers).""" + text = "\n".join(c.get("config", "") for c in configs) + names = " ".join(c.get("device", "") for c in configs) + return ChangeClass( + touches_asn=bool(_ASN_RE.search(text)), + touches_rd_rt=bool(_RD_RT_RE.search(text)), + touches_spine=bool(_SPINE_NAME_RE.search(names)), + touches_underlay=bool(_UNDERLAY_NAME_RE.search(names) or _UNDERLAY_CONFIG_RE.search(text)), + ) + + +# ── canonical severity (reconciles guards.risk_tier + bundler._severity) ───────── + +def unify_severity(*, batfish_errors: int, devices_affected: int, + sessions_dropped: int, converged: bool) -> Severity: + """One canonical 5-bucket severity. Mirrors bundler._severity (the richer engine); + the differential test shows where guards.risk_tier's 3-bucket scale collapses.""" + if not converged or batfish_errors > 0: + return Severity.CRITICAL + if sessions_dropped > 0: + return Severity.HIGH + if devices_affected >= 4: + return Severity.HIGH + if devices_affected >= 2: + return Severity.MEDIUM + if devices_affected >= 1: + return Severity.LOW + return Severity.NONE + + +# ── severity x change-class -> required authority ─────────────────────────────── + +_SEVERITY_BASE: dict[Severity, Tier] = { + Severity.NONE: Tier.AUTO, + Severity.LOW: Tier.AUTO, + Severity.MEDIUM: Tier.HITL, + Severity.HIGH: Tier.HOTL, + Severity.CRITICAL: Tier.BLOCK, +} + + +def required_authority(severity: Severity, change_class: ChangeClass) -> Tier: + """The minimum authority tier a change requires. Base tier from severity, then + hard-force overrides from the change class; the result is the MOST RESTRICTIVE. + + Hard-force (regardless of how clean the twin looked): + * AS number / RD / RT touch -> BLOCK (fabric identity is never auto-changed) + * spine or underlay touch -> >= HOTL (never fully automated) + """ + tier = _SEVERITY_BASE[severity] + if change_class.is_fabric_identity: + tier = max(tier, Tier.BLOCK) + if change_class.touches_spine or change_class.touches_underlay: + tier = max(tier, Tier.HOTL) + return tier + + +@dataclass(frozen=True) +class AuthorityDecision: + required: Tier + max_authorized: Tier + allowed: bool + effective: Tier + reason: str + + +def authorize(required: Tier, max_authorized: Tier) -> AuthorityDecision: + """No-self-escalation ceiling: a change may proceed at `required` ONLY if + required <= max_authorized; otherwise it is BLOCKED. An agent can never grant itself + more autonomy than its ceiling allows. + + `max_authorized` is a plain input here. The #5-ceiling / CROSS-3 follow-up binds it to + a hardware-PIV-signed, compiled-in value so the ceiling itself is tamper-proof; this + function is the deterministic invariant that signature protects.""" + allowed = required <= max_authorized + effective = required if allowed else Tier.BLOCK + if allowed: + reason = (f"required {required.name} <= ceiling {max_authorized.name}: " + f"allowed at {required.name}") + else: + reason = (f"NO-SELF-ESCALATION: required {required.name} exceeds ceiling " + f"{max_authorized.name} -> BLOCK") + return AuthorityDecision(required=required, max_authorized=max_authorized, + allowed=allowed, effective=effective, reason=reason) diff --git a/tests/authority_test.py b/tests/authority_test.py new file mode 100644 index 0000000..29790c2 --- /dev/null +++ b/tests/authority_test.py @@ -0,0 +1,190 @@ +"""aegis/tests/authority_test.py — #5 deterministic authority gate + no-self-escalation. + +House style: pytest-discoverable test_* + a dep-light __main__ runner +(CI: python3 -m aegis.tests.authority_test). Pure logic, no network. Includes the CROSS-2 +differential of the canonical severity engine vs the two legacy AEGIS scorers +(guards.risk_tier 3-bucket, bundler._severity 5-bucket). +""" +from __future__ import annotations + +import itertools +import sys + +from aegis.core.backends.base import GeneratedConfig +from aegis.core.orchestrator.guards import risk_tier +from aegis.core.risk import ( + ChangeClass, + Severity, + Tier, + authorize, + classify_change, + required_authority, + unify_severity, +) +from aegis.evidence.bundler import _severity # legacy 5-bucket scorer (differential ref) + + +def _cfg(config: str, device: str = "leaf-1") -> GeneratedConfig: + return GeneratedConfig(device=device, vendor="frr", config=config, grounded_commands=[]) + + +# ── unify_severity reconciles the two legacy scorers ──────────────────────────── + +def test_unify_severity_table() -> None: + cases = [ + # (errors, devices, dropped, converged) -> canonical severity + (0, 0, 0, True, Severity.NONE), + (0, 1, 0, True, Severity.LOW), + (0, 3, 0, True, Severity.MEDIUM), + (0, 4, 0, True, Severity.HIGH), + (0, 1, 1, True, Severity.HIGH), # a dropped session is HIGH + (1, 1, 0, True, Severity.CRITICAL), # batfish error + (0, 1, 0, False, Severity.CRITICAL), # non-convergence + ] + for e, d, dr, c, exp in cases: + got = unify_severity(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c) + assert got == exp, f"{(e, d, dr, c)} -> {got.name}, want {exp.name}" + + +def test_differential_vs_legacy_scorers() -> None: + """CROSS-2 differential: canonical vs guards.risk_tier (3-bucket) and bundler._severity + (5-bucket). The canonical equals the richer (bundler) engine by construction; guards + disagrees in documented ways (it has no NONE/CRITICAL and over-tiers a clean 1-device).""" + map3 = {"low": Severity.LOW, "medium": Severity.MEDIUM, "high": Severity.HIGH} + map5 = {"none": Severity.NONE, "low": Severity.LOW, "medium": Severity.MEDIUM, + "high": Severity.HIGH, "critical": Severity.CRITICAL} + disagreements = [] + for e, d, dr, c in itertools.product((0, 1), (0, 1, 2, 4), (0, 1), (True, False)): + canon = unify_severity(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c) + bundler = map5[_severity(d, dr, e, c)] + guards = map3[risk_tier(batfish_errors=e, devices_affected=d, sessions_dropped=dr, converged=c)] + assert canon == bundler, f"canon {canon.name} != bundler {bundler.name} for {(e, d, dr, c)}" + if canon != guards: + disagreements.append((e, d, dr, c, canon.name, guards.name)) + assert disagreements, "expected guards-vs-canonical disagreements (the reason to unify)" + # the headline disagreement: one clean device -> canonical LOW, guards MEDIUM + assert (0, 1, 0, True, "LOW", "MEDIUM") in disagreements + # guards collapses non-convergence (CRITICAL) into HIGH + assert (0, 1, 0, False, "CRITICAL", "HIGH") in disagreements + + +# ── classify_change: AS/RD/RT (reliable) + spine/underlay (heuristic) ──────────── + +def test_classify_detects_asn() -> None: + assert classify_change([_cfg("router bgp 65001")]).touches_asn + assert classify_change([_cfg("set protocols bgp group X peer-as 65002")]).touches_asn + assert classify_change([_cfg("neighbor 10.0.0.1 remote-as 65003")]).touches_asn + assert not classify_change([_cfg("interface eth0\n ip address 10.0.0.1/31")]).touches_asn + + +def test_classify_detects_rd_rt() -> None: + assert classify_change([_cfg("set routing-instances V route-distinguisher 65000:1")]).touches_rd_rt + assert classify_change([_cfg("route-target target:65000:100")]).touches_rd_rt + assert not classify_change([_cfg("interface eth0")]).touches_rd_rt + + +def test_classify_spine_underlay_heuristic() -> None: + assert classify_change([_cfg("x", device="dc1-spine-01")]).touches_spine + assert classify_change([_cfg("router ospf 1\n network 10.0.0.0/24 area 0")]).touches_underlay + leaf = classify_change([_cfg("# add vlan 10", device="leaf-1")]) + assert not leaf.touches_spine and not leaf.touches_underlay + + +# ── required_authority: severity base + hard-force overrides ───────────────────── + +def test_severity_base_mapping() -> None: + none = ChangeClass() + assert required_authority(Severity.NONE, none) == Tier.AUTO + assert required_authority(Severity.LOW, none) == Tier.AUTO + assert required_authority(Severity.MEDIUM, none) == Tier.HITL + assert required_authority(Severity.HIGH, none) == Tier.HOTL + assert required_authority(Severity.CRITICAL, none) == Tier.BLOCK + + +def test_hardforce_fabric_identity_is_always_block() -> None: + assert required_authority(Severity.LOW, ChangeClass(touches_asn=True)) == Tier.BLOCK + assert required_authority(Severity.NONE, ChangeClass(touches_rd_rt=True)) == Tier.BLOCK + + +def test_hardforce_spine_underlay_never_auto() -> None: + assert required_authority(Severity.LOW, ChangeClass(touches_spine=True)) == Tier.HOTL + assert required_authority(Severity.NONE, ChangeClass(touches_underlay=True)) == Tier.HOTL + # severity can still escalate above the spine/underlay floor + assert required_authority(Severity.CRITICAL, ChangeClass(touches_spine=True)) == Tier.BLOCK + + +def test_hardforce_property_fabric_identity_never_below_block() -> None: + for sev in Severity: + for asn, rdrt in itertools.product((True, False), repeat=2): + t = required_authority(sev, ChangeClass(touches_asn=asn, touches_rd_rt=rdrt)) + if asn or rdrt: + assert t == Tier.BLOCK, f"fabric-identity must BLOCK (sev={sev.name})" + + +def test_spine_underlay_never_auto_property() -> None: + for sev in Severity: + for sp, ul in itertools.product((True, False), repeat=2): + t = required_authority(sev, ChangeClass(touches_spine=sp, touches_underlay=ul)) + if sp or ul: + assert t >= Tier.HOTL, f"spine/underlay must be >= HOTL (sev={sev.name})" + + +# ── no-self-escalation ceiling ────────────────────────────────────────────────── + +def test_authorize_allows_within_ceiling() -> None: + d = authorize(Tier.HITL, Tier.HOTL) + assert d.allowed and d.effective == Tier.HITL + + +def test_authorize_blocks_self_escalation() -> None: + d = authorize(Tier.HOTL, Tier.HITL) + assert not d.allowed and d.effective == Tier.BLOCK and "NO-SELF-ESCALATION" in d.reason + + +def test_no_self_escalation_zero_violations_property() -> None: + """The invariant: effective authority is NEVER above the ceiling. 0 violations over + every (required, ceiling) pair.""" + violations = 0 + for required in Tier: + for ceiling in Tier: + d = authorize(required, ceiling) + if d.effective != Tier.BLOCK and d.effective > ceiling: + violations += 1 + assert violations == 0, f"{violations} self-escalation violations" + + +def test_end_to_end_low_severity_asn_change_is_blocked() -> None: + """A twin-clean, single-device change that edits an AS number must still BLOCK: + severity alone (LOW) would say AUTO; the change class forces BLOCK, and an AUTO-ceiling + agent cannot push it.""" + configs = [_cfg("router bgp 65010\n neighbor 10.0.0.1 remote-as 65011", device="leaf-1")] + sev = unify_severity(batfish_errors=0, devices_affected=1, sessions_dropped=0, converged=True) + assert sev == Severity.LOW + cc = classify_change(configs) + assert cc.touches_asn + req = required_authority(sev, cc) + assert req == Tier.BLOCK + assert not authorize(req, Tier.AUTO).allowed + + +# ── runner ─────────────────────────────────────────────────────────────────────── + +def _run_all() -> int: + funcs = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] + failures = 0 + for fn in funcs: + try: + fn() + print(f"PASS {fn.__name__}") + except AssertionError as e: + failures += 1 + print(f"FAIL {fn.__name__}: {e}") + except Exception as e: # noqa: BLE001 + failures += 1 + print(f"ERROR {fn.__name__}: {type(e).__name__}: {e}") + print(f"\n{len(funcs) - failures}/{len(funcs)} passed") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(_run_all())