diff --git a/core/orchestrator/pipeline.py b/core/orchestrator/pipeline.py index 9e5e9f2..8c7c68a 100644 --- a/core/orchestrator/pipeline.py +++ b/core/orchestrator/pipeline.py @@ -14,6 +14,7 @@ from ..backends.base import Backend, GeneratedConfig from . import guards, rollback from ...evidence.bundler import build_bundle, unattested_identity +from ..risk import authority_record, classify_change, load_max_authorized, unify_severity from ...evidence.compliance import map_controls @@ -103,6 +104,17 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn", attested = backend.model_identity() if hasattr(backend, "model_identity") else None mi = attested if attested is not None else unattested_identity() + # authority tier (#5): orthogonal to risk_tier -- a twin-clean LOW-severity change + # to a fabric-identity construct (AS/RD/RT) is still BLOCK. Sealed + enforced at G5. + authority = authority_record( + unify_severity(batfish_errors=bf["errors"], + devices_affected=diff["devices_affected"], + sessions_dropped=len(diff["sessions_dropped"]), + converged=twin["converged"]), + classify_change(configs), + load_max_authorized(), + ) + return build_bundle( run_id=run_id, created=created, operator=operator, intent=intent, source=source, configs=configs, batfish=bf, twin=twin, diff=diff, @@ -110,7 +122,7 @@ def run_preflight(intent: str, *, backend: Backend, lab: str = "clos-evpn", approver=approver if approval_granted else None, approval_utc=created if approval_granted else None, rollback_plan=plan, decision=decision, reason=reason, - model_identity=mi, + model_identity=mi, authority=authority, ) finally: if twin_id is not None: diff --git a/core/promote/gate.py b/core/promote/gate.py index 56cb77f..1ceacb0 100644 --- a/core/promote/gate.py +++ b/core/promote/gate.py @@ -12,6 +12,7 @@ from dataclasses import dataclass from ...evidence.bundler import verify +from ..risk import Tier, authorize, load_max_authorized @dataclass(frozen=True) @@ -43,4 +44,22 @@ def evaluate(bundle: dict, *, approver: str | None, approval_token: str | None, return GateDecision(False, "live connector blocked — set AEGIS_PROMOTE_ALLOW_LIVE=1 " "and wire an audited connector") + # G5 authority ceiling — NO SELF-ESCALATION. The change's REQUIRED authority is sealed + # in the bundle (covered by the G1 integrity check); re-derive the ceiling decision + # here so the gate enforces its OWN ceiling at promote time, not a recorded boolean. + # A change needing more autonomy than the ceiling allows -- or any hard-BLOCK change + # (AS/RD/RT / critical) -- is never promotable. + authority = bundle.get("change", {}).get("authority") + if not authority: + return GateDecision(False, "no authority record — cannot confirm the autonomy bound " + "(fail-closed)") + try: + required = Tier[str(authority.get("required", "")).upper()] + except KeyError: + return GateDecision(False, f"unrecognized required authority {authority.get('required')!r}") + ceiling = load_max_authorized() + if not authorize(required, ceiling).allowed: # G5 + return GateDecision(False, f"no-self-escalation: change requires {required.name} " + f"authority, above the {ceiling.name} ceiling — blocked") + return GateDecision(True, f"{tier or 'unknown'}-risk {decision} approved for promotion") diff --git a/core/risk/__init__.py b/core/risk/__init__.py index a2405ee..709a0e4 100644 --- a/core/risk/__init__.py +++ b/core/risk/__init__.py @@ -7,8 +7,10 @@ ChangeClass, Severity, Tier, + authority_record, authorize, classify_change, + load_max_authorized, required_authority, unify_severity, ) @@ -22,4 +24,6 @@ "unify_severity", "required_authority", "authorize", + "load_max_authorized", + "authority_record", ] diff --git a/core/risk/authority.py b/core/risk/authority.py index 5ccd0e8..95f01de 100644 --- a/core/risk/authority.py +++ b/core/risk/authority.py @@ -23,6 +23,7 @@ """ from __future__ import annotations +import os import re from dataclasses import dataclass from enum import IntEnum @@ -169,3 +170,51 @@ def authorize(required: Tier, max_authorized: Tier) -> AuthorityDecision: f"{max_authorized.name} -> BLOCK") return AuthorityDecision(required=required, max_authorized=max_authorized, allowed=allowed, effective=effective, reason=reason) + + +_DEFAULT_MAX_AUTHORIZED = Tier.HOTL + + +def load_max_authorized() -> Tier: + """The deployment's autonomy ceiling. Env AEGIS_MAX_AUTHORIZED_TIER (AUTO|HITL|HOTL), + default HOTL. BLOCK is rejected as a ceiling (it would authorize fabric-identity / + critical changes). An invalid value RAISES (fail-closed, loud) rather than silently + defaulting to something permissive. + + NOTE: env-configured today. The CROSS-3 follow-up binds it to a hardware-PIV signature + + compiled-in public key so it cannot be raised by editing the environment — the + tamper-proofing the no-self-escalation invariant ultimately needs. This ships the + deterministic enforcement that signature will protect.""" + raw = os.environ.get("AEGIS_MAX_AUTHORIZED_TIER") + if not raw: + return _DEFAULT_MAX_AUTHORIZED + try: + tier = Tier[raw.strip().upper()] + except KeyError: + raise ValueError(f"AEGIS_MAX_AUTHORIZED_TIER must be one of AUTO|HITL|HOTL (got {raw!r})") + if tier == Tier.BLOCK: + raise ValueError("AEGIS_MAX_AUTHORIZED_TIER=BLOCK is invalid — a ceiling cannot " + "authorize BLOCK-tier (fabric-identity / critical) changes") + return tier + + +def authority_record(severity: Severity, change_class: ChangeClass, + max_authorized: Tier) -> dict: + """The sealed change.authority object: severity (how bad if it fails) + the required + authority (how much autonomy), the ceiling it was checked against, and the + no-self-escalation decision. Lowercase strings to match the bundle vocabulary.""" + required = required_authority(severity, change_class) + decision = authorize(required, max_authorized) + return { + "severity": severity.name.lower(), + "required": required.name.lower(), + "max_authorized": max_authorized.name.lower(), + "allowed": decision.allowed, + "effective": decision.effective.name.lower(), + "change_class": { + "touches_asn": change_class.touches_asn, + "touches_rd_rt": change_class.touches_rd_rt, + "touches_spine": change_class.touches_spine, + "touches_underlay": change_class.touches_underlay, + }, + } diff --git a/evidence/bundler.py b/evidence/bundler.py index 963e58a..bed2681 100644 --- a/evidence/bundler.py +++ b/evidence/bundler.py @@ -28,12 +28,13 @@ def build_bundle(*, run_id, created, operator, intent, source, configs, batfish, twin, diff, compliance, tier, needs_approval, approver, approval_utc, rollback_plan, decision, reason, model_identity: dict | None = None, + authority: dict | None = None, signed: bool = False) -> dict: severity = _severity(diff["devices_affected"], len(diff["sessions_dropped"]), batfish["errors"], twin["converged"]) bundle = { - "bundle_version": "1.1", + "bundle_version": "1.2", "run_id": run_id, "created_utc": created, "operator": operator, @@ -53,6 +54,8 @@ def build_bundle(*, run_id, created, operator, intent, source, configs, batfish, }, "model_identity": (model_identity if model_identity is not None else _unknown_identity()), + "authority": (authority if authority is not None + else _failclosed_authority(severity)), }, "twin": { "engine": "containerlab", @@ -136,3 +139,18 @@ def unattested_identity() -> dict: "capabilities": [], "resolved_at_utc": _SYNTHETIC_RESOLVED_AT, } + + +def _failclosed_authority(severity: str) -> dict: + """Fallback when no authority was computed (a direct build_bundle call). Fail CLOSED: + an unknown authority is BLOCK, so it can never be promoted. run_preflight always + supplies a real record (core.risk.authority_record).""" + return { + "severity": severity, + "required": "block", + "max_authorized": "auto", + "allowed": False, + "effective": "block", + "change_class": {"touches_asn": False, "touches_rd_rt": False, + "touches_spine": False, "touches_underlay": False}, + } diff --git a/evidence/schema/evidence_bundle.schema.json b/evidence/schema/evidence_bundle.schema.json index 98a44e6..a2dca22 100644 --- a/evidence/schema/evidence_bundle.schema.json +++ b/evidence/schema/evidence_bundle.schema.json @@ -7,7 +7,7 @@ "required": ["bundle_version", "run_id", "created_utc", "change", "twin", "validation", "verdict", "integrity"], "properties": { - "bundle_version": { "const": "1.1" }, + "bundle_version": { "const": "1.2" }, "run_id": { "type": "string", "minLength": 8 }, "created_utc": { "type": "string", "format": "date-time" }, "operator": { "type": "string" }, @@ -16,7 +16,7 @@ "type": "object", "additionalProperties": false, "required": ["intent", "source", "generated_configs", "risk_tier", "approval", - "model_identity"], + "model_identity", "authority"], "properties": { "intent": { "type": "string" }, "source": { "enum": ["nl_intent", "config_import"] }, @@ -49,7 +49,7 @@ "model_identity": { "type": "object", "additionalProperties": false, - "description": "The proposing model, sealed into the bundle SHA-256 (the #4 wedge). LOCAL backends carry a real weights-sha256; CLOUD backends carry an honest identity-claim with model_hash=null; config_import/no-model runs carry _UNKNOWN_IDENTITY (provider 'none'); the simulator attests provider 'simulator'.", + "description": "The proposing model, sealed into the bundle SHA-256 (the #4 wedge). LOCAL backends carry a real weights-sha256; CLOUD backends an honest identity-claim (model_hash=null); a missing attestation seals provider 'unknown'/'unattested'; config_import seals 'none'/'operator-supplied'.", "required": ["provider", "model", "model_hash", "model_hash_kind", "api_version", "capabilities", "resolved_at_utc"], "properties": { @@ -61,6 +61,31 @@ "capabilities": { "type": "array", "items": { "enum": ["chat", "json", "tools"] } }, "resolved_at_utc": { "type": "string" } } + }, + + "authority": { + "type": "object", + "additionalProperties": false, + "description": "The orthogonal authority axis (#5): severity = how bad if it fails; required/effective = how much autonomy is allowed (auto dict: + return json.loads((_root() / "evidence" / "schema" / "evidence_bundle.schema.json").read_text("utf-8")) + + +def _validate(b: dict) -> None: + if _HAVE: + jsonschema.validate(b, _schema()) + + +def _b(intent: str, **kw) -> dict: + return run_preflight(intent, backend=SimulatorBackend(), lab="clos-evpn", **kw) + + +def _reseal(b: dict) -> dict: + b["integrity"]["sha256"] = "" + b["integrity"]["sha256"] = compute_sha256(b) + return b + + +# ── the bundle carries + seals the authority record (v1.2) ────────────────────── + +def test_bundle_carries_sealed_authority_v12() -> None: + b = _b("add vlan 10 to leaf-1") + assert b["bundle_version"] == "1.2" + a = b["change"]["authority"] + assert set(a) == {"severity", "required", "max_authorized", "allowed", "effective", "change_class"} + assert a["required"] in ("auto", "hitl", "hotl", "block") + assert set(a["change_class"]) == {"touches_asn", "touches_rd_rt", "touches_spine", "touches_underlay"} + assert verify(b) + _validate(b) + + +def test_authority_is_under_the_seal() -> None: + b = _b("add vlan 10 to leaf-1") + assert verify(b) + b["change"]["authority"]["required"] = "auto" # try to weaken the required tier + assert not verify(b), "tampering with the sealed authority must break the integrity hash" + + +def test_asn_change_forces_block_authority() -> None: + b = _b("peer bgp with neighbor 10.0.0.1 remote-as 65010") + a = b["change"]["authority"] + assert a["change_class"]["touches_asn"] + assert a["required"] == "block" and a["effective"] == "block" and a["allowed"] is False + _validate(b) + + +# ── gate rule G5: no self-escalation ──────────────────────────────────────────── + +def test_gate_g5_blocks_block_authority_even_when_otherwise_promotable() -> None: + """A twin-clean, low-risk AS change is promotable by G1-G4 yet G5 blocks it: the + change requires BLOCK authority, above any valid ceiling.""" + b = _b("peer bgp with neighbor 10.0.0.1 remote-as 65010") + assert b["change"]["authority"]["required"] == "block" + # make it otherwise fully promotable, keep the (block) authority, re-seal: + b["verdict"]["decision"] = "ship_ready" + b["change"]["risk_tier"] = "low" + _reseal(b) + d = gate.evaluate(b, approver=None, approval_token=None, connector_is_live=False, allow_live=False) + assert not d.allowed and "self-escalation" in d.reason.lower() + + +def test_gate_g5_allows_within_ceiling() -> None: + b = _b("shut interface ethernet-1/1") + if b["change"]["authority"]["required"] == "block": + return # extremely unlikely for this intent; skip rather than assert a false case + b["verdict"]["decision"] = "ship_ready" + b["change"]["risk_tier"] = "low" + _reseal(b) + d = gate.evaluate(b, approver="noc-lead", approval_token="t", connector_is_live=False, allow_live=False) + assert d.allowed, d.reason + + +def test_gate_g5_fail_closed_on_missing_authority() -> None: + b = _b("add vlan 10 to leaf-1") + b["verdict"]["decision"] = "ship_ready" + b["change"]["risk_tier"] = "low" + del b["change"]["authority"] + _reseal(b) + d = gate.evaluate(b, approver="noc-lead", approval_token="t", connector_is_live=False, allow_live=False) + assert not d.allowed and "authority" in d.reason.lower() + + +# ── 0-violation property: nothing promotes above the ceiling ──────────────────── + +def test_no_promoted_bundle_exceeds_ceiling_property() -> None: + ceiling = load_max_authorized() + intents = ["add vlan {n} to leaf-{n}", "peer bgp remote-as 6500{n}", + "enable ospf area 0 on edge-{n}", "shut interface ethernet-1/{n}"] + rng = random.Random(7) + promoted = denied_block = 0 + for _ in range(300): + b = _b(intents[rng.randrange(len(intents))].format(n=rng.randint(1, 9))) + try: + promote(b, connector=DryRunConnector(), approver="noc-lead", + approval_token="t", allow_live=False) + promoted += 1 + req = Tier[b["change"]["authority"]["required"].upper()] + assert authorize(req, ceiling).allowed, f"promoted above ceiling: {req.name}" + except PromoteDenied: + if b["change"]["authority"]["required"] == "block": + denied_block += 1 + assert promoted > 0, "some benign changes must still promote" + assert denied_block > 0, "some BLOCK-tier (AS) changes must be denied by the ceiling" + + +# ── ceiling loader: env override + fail-closed ────────────────────────────────── + +def test_load_max_authorized_env_and_failclosed() -> None: + old = os.environ.get("AEGIS_MAX_AUTHORIZED_TIER") + try: + assert load_max_authorized() == Tier.HOTL or old is not None # default HOTL + os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = "hitl" + assert load_max_authorized() == Tier.HITL + for bad in ("block", "BLOCK", "bogus", "5"): + os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = bad + try: + load_max_authorized() + raise AssertionError(f"ceiling {bad!r} must raise (fail-closed)") + except ValueError: + pass + finally: + if old is None: + os.environ.pop("AEGIS_MAX_AUTHORIZED_TIER", None) + else: + os.environ["AEGIS_MAX_AUTHORIZED_TIER"] = old + + +# ── runner ─────────────────────────────────────────────────────────────────────── + +def _root() -> Path: + here = Path(__file__).resolve() + for parent in here.parents: + if (parent / "evidence" / "schema").exists(): + return parent + return here.parent.parent + + +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()) diff --git a/tests/wedge_test.py b/tests/wedge_test.py index e7c23fc..128228d 100644 --- a/tests/wedge_test.py +++ b/tests/wedge_test.py @@ -63,7 +63,7 @@ def _local_identity_dict() -> dict: def test_nl_intent_seals_simulator_identity_v11() -> None: b = _bundle() - assert b["bundle_version"] == "1.1" + assert b["bundle_version"] == "1.2" mi = b["change"]["model_identity"] assert mi["provider"] == "simulator" and mi["model"] == "deterministic-sim-v1" assert verify(b)