AEGIS bounded-autonomy upgrade — egress → CROSS-3 seal (squashed release)#9
Merged
Merged
Conversation
…authority gate, ceiling, CROSS-3 seal Squashed release of the 8-PR stack (#2..#8): the offline, hardware-rootable bounded-autonomy receipt and everything under it. AUTO disabled by default; the only un-built piece is the hardware-PIV signer (docs/PIV_HARDWARE_SIGNER_PLAN.md) — software Ed25519 runs the whole demo. - core/llm: single shared LLM egress + adapter (local-first fallback chain, retry/cooldown, 4-event hook bus; raw httpx, no anthropic SDK; air-gap fail-closed). [#2] - evidence: model-identity wedge — change.model_identity sealed into the bundle hash; honest UNKNOWN / unattested states, never a faked attribution. [#3] - core/risk: deterministic AUTO/HITL/HOTL/BLOCK authority model (severity != authority; AS/RD/RT -> BLOCK; spine/underlay -> >= HOTL) + the no-self-escalation ceiling. [#4] - core/promote: gate rule G5 enforces the ceiling from the sealed required tier; change.authority sealed (bundle_version 1.1->1.2). [#5] - core/seal: CROSS-3 detached, offline-verifiable bounded-autonomy receipt (Ed25519; gates G0..G6; refuses to seal an unbounded change). [#6] - serve: /api/preflight/run emits the seal; GET /api/seal/pubkey + POST /api/seal/verify; examiner PDF shows the seal. [#7] - docs: PIV hardware-signer plan — the final, hardware-gated CROSS-3 pass. [#8] Property-tested safety invariants: no-self-escalation 0-violations, every-seal-is-bounded, tamper detection, offline-keyless verify, air-gap fail-closed.
Reviewer's GuideIntroduces a shared, policy-aware LLM egress layer, deterministic authority model, and a detached Ed25519-based CROSS-3 seal that produces offline-verifiable bounded-autonomy receipts, then wires them through the preflight pipeline, promotion gate, API, simulator backend, PDF evidence, and tests, updating the evidence bundle schema/version accordingly. Sequence diagram for preflight run with bounded-autonomy seal emissionsequenceDiagram
actor Operator
participant FlaskApp as serve.preflight_run
participant Pipeline as run_preflight
participant Backend as SimulatorBackend
participant Authority as authority_record
participant Bundler as build_bundle
participant Signer as Ed25519Signer
Operator->>FlaskApp: POST /api/preflight/run
FlaskApp->>Pipeline: run_preflight(intent, backend, lab, operator,...)
Pipeline->>Backend: generate_config
Backend-->>Pipeline: configs
Pipeline->>Backend: model_identity
Backend-->>Pipeline: model_identity dict
Pipeline->>Authority: authority_record(unify_severity(...), classify_change(configs), load_max_authorized())
Authority-->>Pipeline: authority dict
Pipeline->>Bundler: build_bundle(..., model_identity, authority)
Bundler-->>Pipeline: bundle
Pipeline-->>FlaskApp: bundle
FlaskApp->>Signer: seal_response(bundle, _SIGNER, now)
Signer-->>FlaskApp: seal or error_reason
FlaskApp->>FlaskApp: bundle["seal"] = seal
FlaskApp-->>Operator: 200 {bundle with detached seal}
Operator->>FlaskApp: POST /api/seal/verify {bundle, seal}
FlaskApp->>FlaskApp: verify_seal(bundle, seal, _VERIFIER)
FlaskApp-->>Operator: {valid, gate, reason}
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- In
core/llm/airgap.is_loopbackyou currently treat any*.localhostname as in-perimeter; that can still resolve to a different machine on the LAN and undermines the air-gap guarantee, so consider restricting to literal loopback IPs andlocalhostonly (no.localsuffix wildcard). - There are now two slightly different "unknown identity" helpers (
evidence.bundler._unknown_identityandcore.llm.identity.unknown_identityplusunattested_identity), which makes the semantics easy to drift; consider consolidating them into a single well-documented helper to keep the identity states consistent across the pipeline.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `core/llm/airgap.is_loopback` you currently treat any `*.local` hostname as in-perimeter; that can still resolve to a different machine on the LAN and undermines the air-gap guarantee, so consider restricting to literal loopback IPs and `localhost` only (no `.local` suffix wildcard).
- There are now two slightly different "unknown identity" helpers (`evidence.bundler._unknown_identity` and `core.llm.identity.unknown_identity` plus `unattested_identity`), which makes the semantics easy to drift; consider consolidating them into a single well-documented helper to keep the identity states consistent across the pipeline.
## Individual Comments
### Comment 1
<location path="core/llm/airgap.py" line_range="48-53" />
<code_context>
+ return False # reject empty / relative — never auto-allow
+ if host in _LOOPBACK_NAMES:
+ return True
+ try:
+ return ip_address(host).is_loopback # 127.0.0.0/8 and ::1, strictly
+ except ValueError:
+ # Not a literal IP. Only the reserved, non-internet-routable mDNS suffix is
+ # treated as in-perimeter (RFC 6762); everything else is non-loopback.
+ return host.endswith(".local")
+
+
</code_context>
<issue_to_address>
**🚨 question (security):** Allowing arbitrary `*.local` hosts in air-gap mode may be looser than intended for a strict loopback-only policy.
The `is_loopback` docstring promises “genuine loopback” or the literal `localhost` / reserved `.local` mDNS host, but this implementation lets any `.local` name through. That allows `AEGIS_LLM_URL` to point at a non-loopback LAN host and still satisfy the air-gap check. To keep air-gap mode strictly loopback-only, consider allowing only literal loopback IPs plus explicit aliases like `localhost` / `ip6-localhost`, and removing the generic `.local` suffix handling so it matches the `assert_airgap_ok` "non-loopback egress forbidden" invariant.
</issue_to_address>
### Comment 2
<location path="core/llm/egress.py" line_range="48" />
<code_context>
+ hints: Mapping[str, object] = field(default_factory=dict)
+
+
+class HookBus:
+ """Deterministic, in-process, synchronous hook dispatch — the ENTIRE extension surface.
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider tightening the HookBus API around explicit transform/observe events and simplifying the sync bridge implementation to reduce cognitive and threading complexity without changing behavior.
You can keep all current features but trim some complexity in two focused spots: the hook bus API and the sync bridge.
---
### 1. Make `HookBus` event types explicit
Right now `HookBus` treats all events uniformly as `HookEvent`, even though only some are transforming and some are observing. Splitting these at the type/API level keeps the extension surface but makes control flow easier to follow and reduces misuse surface.
```python
# narrow the event types
TransformEvent = Literal["before_request", "build_payload"]
ObserveEvent = Literal["after_response", "on_result"]
class HookBus:
def __init__(self) -> None:
self._transform_hooks: dict[TransformEvent, list[Callable]] = defaultdict(list)
self._observe_hooks: dict[ObserveEvent, list[Callable]] = defaultdict(list)
def register_transform(self, event: TransformEvent, fn: Callable) -> None:
self._transform_hooks[event].append(fn)
def register_observe(self, event: ObserveEvent, fn: Callable) -> None:
self._observe_hooks[event].append(fn)
def transform(self, event: TransformEvent, value, *args):
for fn in self._transform_hooks[event]:
try:
out = fn(value, *args)
if out is not None:
value = out
except Exception as exc: # noqa: BLE001
_log_hook_error(event, fn, exc)
return value
def observe(self, event: ObserveEvent, *args) -> None:
for fn in self._observe_hooks[event]:
try:
fn(*args)
except Exception as exc: # noqa: BLE001
_log_hook_error(event, fn, exc)
```
Then `LLMEgress.on` can route based on the event instead of going through a fully generic bus:
```python
def on(self, event: HookEvent, fn: Callable) -> "LLMEgress":
if event in ("before_request", "build_payload"):
self._hooks.register_transform(event, fn)
else:
self._hooks.register_observe(event, fn)
return self
```
This keeps the four-event surface you want but removes the “arbitrary event bus” feel, aligning better with the current usage.
---
### 2. Simplify `complete_sync` threading logic
`complete_sync` currently creates a new `ThreadPoolExecutor` per call and nests `asyncio.run` inside that thread. You can keep the loop-safe behavior but use the existing loop’s `run_in_executor`, avoiding a per-call executor and making the threading + loop interaction easier to reason about:
```python
def complete_sync(self, req: LLMRequest) -> LLMResult:
"""Sync bridge for legacy callers. Loop-safe."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
# no event loop: safe to run directly
return asyncio.run(self.complete(req))
# already in an event loop: offload to default executor
return loop.run_until_complete(
loop.run_in_executor(None, lambda: asyncio.run(self.complete(req)))
)
```
If `run_until_complete` isn’t available in your embedding context, another option is to move this compatibility wrapper closer to legacy call sites (e.g., a small `complete_sync` helper in the adapter layer) and keep `LLMEgress` purely async. That keeps the core egress simpler while preserving the sync API.
</issue_to_address>
### Comment 3
<location path="core/risk/authority.py" line_range="42" />
<code_context>
+ 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)."""
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating Tier comparison semantics and adding a single high-level orchestration function so callers don’t need to understand enum ordering or manually compose multiple helpers.
You can reduce the cognitive load without changing behavior by:
1. **Encapsulating Tier comparisons** so callers never rely on the inverted IntEnum ordering.
2. **Providing a single “entry point”** that goes from raw inputs to the final authority record, so readers don’t need to chase multiple helpers.
### 1. Encapsulate Tier comparisons
Right now, callers must know that larger `Tier` values mean *more restrictive* and use `max()` / `<=` accordingly. You can centralize that:
```python
def is_more_restrictive(a: Tier, b: Tier) -> bool:
"""True if `a` is strictly more restrictive than `b`."""
return a > b
def most_restrictive(*tiers: Tier) -> Tier:
"""Return the most restrictive tier from the given values."""
return max(tiers)
def can_satisfy(required: Tier, ceiling: Tier) -> bool:
"""True if the ceiling is sufficient for the required tier."""
return required <= ceiling
```
Then use them in the existing functions instead of raw comparisons:
```python
def required_authority(severity: Severity, change_class: ChangeClass) -> Tier:
tier = _SEVERITY_BASE[severity]
if change_class.is_fabric_identity:
tier = most_restrictive(tier, Tier.BLOCK)
if change_class.touches_spine or change_class.touches_underlay:
tier = most_restrictive(tier, Tier.HOTL)
return tier
```
```python
def authorize(required: Tier, max_authorized: Tier) -> AuthorityDecision:
allowed = can_satisfy(required, max_authorized)
effective = required if allowed else Tier.BLOCK
# ... keep the rest as-is ...
```
This keeps behavior identical but makes the unusual ordering explicit and harder to misuse in future changes.
### 2. Provide a single “front door” to the authority record
Right now, readers must mentally chain:
`unify_severity -> required_authority -> authorize -> authority_record`
You can keep these helpers intact but add one orchestrating function a typical caller uses, with the helpers clearly “inside” it:
```python
@dataclass(frozen=True)
class AuthorityInputs:
batfish_errors: int
devices_affected: int
sessions_dropped: int
converged: bool
configs: list[GeneratedConfig]
max_authorized: Tier | None = None # None means "use deployment default"
```
```python
def classify_and_authorize(inputs: AuthorityInputs) -> dict:
"""Single entry point: raw risk + configs -> sealed authority record."""
severity = unify_severity(
batfish_errors=inputs.batfish_errors,
devices_affected=inputs.devices_affected,
sessions_dropped=inputs.sessions_dropped,
converged=inputs.converged,
)
change_class = classify_change(inputs.configs)
max_authorized = (
inputs.max_authorized if inputs.max_authorized is not None else load_max_authorized()
)
return authority_record(severity, change_class, max_authorized)
```
Call sites that “just want the final dictionary” can now use only `classify_and_authorize`, while `authority_record`, `authorize`, `required_authority`, and `unify_severity` remain as composable internals for tests and advanced use.
This keeps all behavior and structure, but significantly reduces the number of concepts a reader has to juggle at each call site.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+48
to
+53
| try: | ||
| return ip_address(host).is_loopback # 127.0.0.0/8 and ::1, strictly | ||
| except ValueError: | ||
| # Not a literal IP. Only the reserved, non-internet-routable mDNS suffix is | ||
| # treated as in-perimeter (RFC 6762); everything else is non-loopback. | ||
| return host.endswith(".local") |
There was a problem hiding this comment.
🚨 question (security): Allowing arbitrary *.local hosts in air-gap mode may be looser than intended for a strict loopback-only policy.
The is_loopback docstring promises “genuine loopback” or the literal localhost / reserved .local mDNS host, but this implementation lets any .local name through. That allows AEGIS_LLM_URL to point at a non-loopback LAN host and still satisfy the air-gap check. To keep air-gap mode strictly loopback-only, consider allowing only literal loopback IPs plus explicit aliases like localhost / ip6-localhost, and removing the generic .local suffix handling so it matches the assert_airgap_ok "non-loopback egress forbidden" invariant.
| hints: Mapping[str, object] = field(default_factory=dict) | ||
|
|
||
|
|
||
| class HookBus: |
There was a problem hiding this comment.
issue (complexity): Consider tightening the HookBus API around explicit transform/observe events and simplifying the sync bridge implementation to reduce cognitive and threading complexity without changing behavior.
You can keep all current features but trim some complexity in two focused spots: the hook bus API and the sync bridge.
1. Make HookBus event types explicit
Right now HookBus treats all events uniformly as HookEvent, even though only some are transforming and some are observing. Splitting these at the type/API level keeps the extension surface but makes control flow easier to follow and reduces misuse surface.
# narrow the event types
TransformEvent = Literal["before_request", "build_payload"]
ObserveEvent = Literal["after_response", "on_result"]
class HookBus:
def __init__(self) -> None:
self._transform_hooks: dict[TransformEvent, list[Callable]] = defaultdict(list)
self._observe_hooks: dict[ObserveEvent, list[Callable]] = defaultdict(list)
def register_transform(self, event: TransformEvent, fn: Callable) -> None:
self._transform_hooks[event].append(fn)
def register_observe(self, event: ObserveEvent, fn: Callable) -> None:
self._observe_hooks[event].append(fn)
def transform(self, event: TransformEvent, value, *args):
for fn in self._transform_hooks[event]:
try:
out = fn(value, *args)
if out is not None:
value = out
except Exception as exc: # noqa: BLE001
_log_hook_error(event, fn, exc)
return value
def observe(self, event: ObserveEvent, *args) -> None:
for fn in self._observe_hooks[event]:
try:
fn(*args)
except Exception as exc: # noqa: BLE001
_log_hook_error(event, fn, exc)Then LLMEgress.on can route based on the event instead of going through a fully generic bus:
def on(self, event: HookEvent, fn: Callable) -> "LLMEgress":
if event in ("before_request", "build_payload"):
self._hooks.register_transform(event, fn)
else:
self._hooks.register_observe(event, fn)
return selfThis keeps the four-event surface you want but removes the “arbitrary event bus” feel, aligning better with the current usage.
2. Simplify complete_sync threading logic
complete_sync currently creates a new ThreadPoolExecutor per call and nests asyncio.run inside that thread. You can keep the loop-safe behavior but use the existing loop’s run_in_executor, avoiding a per-call executor and making the threading + loop interaction easier to reason about:
def complete_sync(self, req: LLMRequest) -> LLMResult:
"""Sync bridge for legacy callers. Loop-safe."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
# no event loop: safe to run directly
return asyncio.run(self.complete(req))
# already in an event loop: offload to default executor
return loop.run_until_complete(
loop.run_in_executor(None, lambda: asyncio.run(self.complete(req)))
)If run_until_complete isn’t available in your embedding context, another option is to move this compatibility wrapper closer to legacy call sites (e.g., a small complete_sync helper in the adapter layer) and keep LLMEgress purely async. That keeps the core egress simpler while preserving the sync API.
| CRITICAL = 4 | ||
|
|
||
|
|
||
| class Tier(IntEnum): |
There was a problem hiding this comment.
issue (complexity): Consider encapsulating Tier comparison semantics and adding a single high-level orchestration function so callers don’t need to understand enum ordering or manually compose multiple helpers.
You can reduce the cognitive load without changing behavior by:
- Encapsulating Tier comparisons so callers never rely on the inverted IntEnum ordering.
- Providing a single “entry point” that goes from raw inputs to the final authority record, so readers don’t need to chase multiple helpers.
1. Encapsulate Tier comparisons
Right now, callers must know that larger Tier values mean more restrictive and use max() / <= accordingly. You can centralize that:
def is_more_restrictive(a: Tier, b: Tier) -> bool:
"""True if `a` is strictly more restrictive than `b`."""
return a > b
def most_restrictive(*tiers: Tier) -> Tier:
"""Return the most restrictive tier from the given values."""
return max(tiers)
def can_satisfy(required: Tier, ceiling: Tier) -> bool:
"""True if the ceiling is sufficient for the required tier."""
return required <= ceilingThen use them in the existing functions instead of raw comparisons:
def required_authority(severity: Severity, change_class: ChangeClass) -> Tier:
tier = _SEVERITY_BASE[severity]
if change_class.is_fabric_identity:
tier = most_restrictive(tier, Tier.BLOCK)
if change_class.touches_spine or change_class.touches_underlay:
tier = most_restrictive(tier, Tier.HOTL)
return tierdef authorize(required: Tier, max_authorized: Tier) -> AuthorityDecision:
allowed = can_satisfy(required, max_authorized)
effective = required if allowed else Tier.BLOCK
# ... keep the rest as-is ...This keeps behavior identical but makes the unusual ordering explicit and harder to misuse in future changes.
2. Provide a single “front door” to the authority record
Right now, readers must mentally chain:
unify_severity -> required_authority -> authorize -> authority_record
You can keep these helpers intact but add one orchestrating function a typical caller uses, with the helpers clearly “inside” it:
@dataclass(frozen=True)
class AuthorityInputs:
batfish_errors: int
devices_affected: int
sessions_dropped: int
converged: bool
configs: list[GeneratedConfig]
max_authorized: Tier | None = None # None means "use deployment default"def classify_and_authorize(inputs: AuthorityInputs) -> dict:
"""Single entry point: raw risk + configs -> sealed authority record."""
severity = unify_severity(
batfish_errors=inputs.batfish_errors,
devices_affected=inputs.devices_affected,
sessions_dropped=inputs.sessions_dropped,
converged=inputs.converged,
)
change_class = classify_change(inputs.configs)
max_authorized = (
inputs.max_authorized if inputs.max_authorized is not None else load_max_authorized()
)
return authority_record(severity, change_class, max_authorized)Call sites that “just want the final dictionary” can now use only classify_and_authorize, while authority_record, authorize, required_authority, and unify_severity remain as composable internals for tests and advanced use.
This keeps all behavior and structure, but significantly reduces the number of concepts a reader has to juggle at each call site.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
AEGIS bounded-autonomy upgrade — the whole arc, squashed for review
One commit that collapses the 8-PR stack (#2–#8) onto current
main, so the entire feature reviews as a single diff (30 files, +3.4k). The offline, hardware-rootable bounded-autonomy receipt and everything under it. AUTO ships disabled by default.What it delivers
core/llm/anthropicSDK); air-gap fail-closedevidence/change.model_identitysealed into the bundle hash; honest UNKNOWN / unattested states, never a faked attributioncore/risk/core/promote/requiredtier;change.authoritysealed (bundle 1.1→1.2)core/seal/serve.py,evidence/pdf.py/api/preflight/runemits the seal ·/api/seal/{pubkey,verify}· examiner PDF shows itdocs/PIV_HARDWARE_SIGNER_PLAN.mdTests — full matrix green (verified on this branch, base + flask/reportlab venv)
api23 checks / 7 routes ·pdf·seal15 ·stress2000 ·promote·ceiling8 ·authority14 ·twin·contract14 ·wedge10 ·egress29 — all PASS. Property-tested invariants: no-self-escalation 0-violations, every-seal-is-bounded, tamper detection, offline-keyless verify, air-gap fail-closed.Notes
main(which carries the launch/marketing docs), so it merges clean and up to date; this PR's diff is exactly the stack's code/tests/docs.docs/PIV_HARDWARE_SIGNER_PLAN.md) — sameSignerprotocol, needs a physical YubiKey; software Ed25519 runs the whole demo today.Try it:
python -m aegis.serve→POST /api/preflight/run {"intent":"add vlan 10 to leaf-1"}returns a bundle with aseal;POST /api/seal/verifyconfirms it offline.Summary by Sourcery
Introduce a shared, air-gapped LLM egress layer, deterministic authority/ceiling model, and a detached, offline-verifiable seal that binds model identity and bounded autonomy into evidence bundles, exposing seal APIs and wiring them through the preflight pipeline and PDF evidence output.
New Features:
Enhancements:
Build:
Documentation:
Tests: