feat(promote): enforce the no-self-escalation ceiling at the gate — #5-ceiling

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,14 +104,25 @@ 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,
             compliance=compliance, tier=tier, needs_approval=needs_approval,
             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:

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")

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",
 ]

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,
+        },
+    }

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},
+    }

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<hitl<hotl<block). A fabric-identity (AS/RD/RT) or critical change is forced to block regardless of severity. 'allowed' is the no-self-escalation decision (required <= max_authorized). Sealed into the bundle hash; re-derived + enforced at gate rule G5.",
+          "required": ["severity", "required", "max_authorized", "allowed", "effective", "change_class"],
+          "properties": {
+            "severity":       { "enum": ["none", "low", "medium", "high", "critical"] },
+            "required":       { "enum": ["auto", "hitl", "hotl", "block"] },
+            "max_authorized": { "enum": ["auto", "hitl", "hotl", "block"] },
+            "allowed":        { "type": "boolean" },
+            "effective":      { "enum": ["auto", "hitl", "hotl", "block"] },
+            "change_class": {
+              "type": "object",
+              "additionalProperties": false,
+              "required": ["touches_asn", "touches_rd_rt", "touches_spine", "touches_underlay"],
+              "properties": {
+                "touches_asn":      { "type": "boolean" },
+                "touches_rd_rt":    { "type": "boolean" },
+                "touches_spine":    { "type": "boolean" },
+                "touches_underlay": { "type": "boolean" }
+              }
+            }
+          }
         }
       }
     },