Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions core/seal/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
"""core/seal/ — the CROSS-3 bounded-autonomy SEAL: a detached, offline-verifiable receipt
binding the model identity (#4 wedge) + the bounded authority (#5 ceiling) over a specific
evidence bundle, signed with a pinned key (a YubiKey PIV slot in production)."""
from __future__ import annotations

from .seal import (
SEAL_VERSION,
SealError,
SealVerdict,
build_claims,
canonical_claims_bytes,
seal_bundle,
verify_seal,
)
from .signing import Ed25519Signer, Ed25519Verifier, Signer, Verifier, key_id

__all__ = [
"SEAL_VERSION",
"SealError",
"SealVerdict",
"build_claims",
"canonical_claims_bytes",
"seal_bundle",
"verify_seal",
"Signer",
"Verifier",
"Ed25519Signer",
"Ed25519Verifier",
"key_id",
]
53 changes: 53 additions & 0 deletions core/seal/schema/seal.schema.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"$id": "https://aegis.local/schema/seal.schema.json",
"title": "AEGIS CROSS-3 Bounded-Autonomy Seal (detached receipt)",
"type": "object",
"additionalProperties": false,
"required": ["seal_version", "claims", "signature"],
"properties": {
"seal_version": { "const": "1.0" },
"claims": {
"type": "object",
"additionalProperties": false,
"required": ["seal_version", "bundle_sha256", "model", "authority", "sealed_at_utc"],
"properties": {
"seal_version": { "const": "1.0" },
"bundle_sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" },
"model": {
"type": "object",
"additionalProperties": false,
"required": ["provider", "model", "model_hash", "model_hash_kind"],
"properties": {
"provider": { "type": "string" },
"model": { "type": "string" },
"model_hash": { "type": ["string", "null"], "pattern": "^[a-f0-9]{64}$" },
"model_hash_kind": { "enum": ["weights-sha256", "identity-claim"] }
}
},
"authority": {
"type": "object",
"additionalProperties": false,
"required": ["required", "max_authorized", "allowed", "effective"],
"properties": {
"required": { "enum": ["auto", "hitl", "hotl", "block"] },
"max_authorized": { "enum": ["auto", "hitl", "hotl", "block"] },
"allowed": { "type": "boolean" },
"effective": { "enum": ["auto", "hitl", "hotl", "block"] }
}
},
"sealed_at_utc": { "type": "string" }
}
},
"signature": {
"type": "object",
"additionalProperties": false,
"required": ["alg", "key_id", "value"],
"properties": {
"alg": { "enum": ["ed25519", "piv-ecdsa-p256"] },
"key_id": { "type": "string" },
"value": { "type": "string" }
}
}
}
}
150 changes: 150 additions & 0 deletions core/seal/seal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
"""core/seal/seal.py — the CROSS-3 bounded-autonomy SEAL.

A DETACHED receipt that fuses, into one offline-verifiable object, three facts no
competitor binds together:
(#4 wedge) WHICH model produced the change — provider + model + model-hash
(#5 ceiling) that it was BOUNDED — required authority <= max_authorized
(integrity) over THIS exact evidence bundle — bundle_sha256
signed with a pinned key (a YubiKey PIV slot in production).

The receipt is DETACHED — it does not mutate the bundle, so the bundle's own integrity
hash stays stable — and binds to the bundle by its sha256. VERIFY runs OFFLINE against
only the public key (no secret), and a VALID verdict is a portable cryptographic proof
that a named+hashed model produced this exact change AND that it was bounded by the ceiling
— the thing competitors assert but cannot hand an auditor offline.
"""
from __future__ import annotations

import base64
import json
from dataclasses import dataclass

from ...evidence.bundler import verify as verify_bundle
from ..risk import Tier
from .signing import Signer, Verifier

SEAL_VERSION = "1.0"


class SealError(Exception):
"""Refused to produce a seal (the bundle is unbounded or fails integrity)."""


def _canonical(obj) -> bytes:
return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")


def canonical_claims_bytes(claims: dict) -> bytes:
return _canonical(claims)


def build_claims(bundle: dict, sealed_at_utc: str) -> dict:
"""The exact claim set the seal signs — binds model + authority + the bundle hash."""
mi = bundle["change"]["model_identity"]
au = bundle["change"]["authority"]
return {
"seal_version": SEAL_VERSION,
"bundle_sha256": bundle["integrity"]["sha256"],
"model": {
"provider": mi["provider"], "model": mi["model"],
"model_hash": mi["model_hash"], "model_hash_kind": mi["model_hash_kind"],
},
"authority": {
"required": au["required"], "max_authorized": au["max_authorized"],
"allowed": au["allowed"], "effective": au["effective"],
},
"sealed_at_utc": sealed_at_utc,
}


def _is_bounded(authority: dict) -> bool:
if not authority.get("allowed") or authority.get("effective") == "block":
return False
try:
return Tier[str(authority["required"]).upper()] <= Tier[str(authority["max_authorized"]).upper()]
except KeyError:
return False


def seal_bundle(bundle: dict, signer: Signer, *, sealed_at_utc: str) -> dict:
"""Produce a detached seal receipt for a BOUNDED, integrity-valid bundle. Refuses
(SealError) to seal a tampered bundle or one whose change exceeds the autonomy ceiling
— you never certify a self-escalation."""
if not verify_bundle(bundle):
raise SealError("refuse to seal: bundle fails its own integrity check")
if not _is_bounded(bundle["change"]["authority"]):
raise SealError("refuse to seal: change is not within the autonomy ceiling "
"(no-self-escalation) — nothing to certify")
claims = build_claims(bundle, sealed_at_utc)
sig = signer.sign(canonical_claims_bytes(claims))
return {
"seal_version": SEAL_VERSION,
"claims": claims,
"signature": {
"alg": signer.alg(),
"key_id": signer.key_id(),
"value": base64.b64encode(sig).decode("ascii"),
},
}


@dataclass(frozen=True)
class SealVerdict:
valid: bool
gate: str # the gate that decided (G0..G6) or "OK"
reason: str


def verify_seal(bundle: dict, seal: dict, verifier: Verifier) -> SealVerdict:
"""Offline verification — uses ONLY the pinned public key. Runs gates G0..G6; the first
failure decides."""
# G0 — structure
try:
claims = seal["claims"]
sig = seal["signature"]
alg, kid, value = sig["alg"], sig["key_id"], sig["value"]
b_sha, model, authority = claims["bundle_sha256"], claims["model"], claims["authority"]
except (KeyError, TypeError):
Comment on lines +98 to +107

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): Consider explicitly validating seal_version in the verifier to catch incompatible or unexpected seal formats.

Right now we never compare seal["seal_version"] / claims["seal_version"] to SEAL_VERSION, so a seal with an older or newer format could still verify as long as the shape mostly matches and the signature is valid. Please add a version check (e.g., in G0 or a dedicated gate) that rejects unknown major versions so format changes can’t be silently accepted without explicit migration handling.

Suggested implementation:

def verify_seal(bundle: dict, seal: dict, verifier: Verifier) -> SealVerdict:
    """Offline verification — uses ONLY the pinned public key. Runs gates G0..G6; the first
    failure decides."""
    # G0 — structure
    try:
        claims = seal["claims"]
        sig = seal["signature"]
        alg, kid, value = sig["alg"], sig["key_id"], sig["value"]
        b_sha, model, authority = claims["bundle_sha256"], claims["model"], claims["authority"]
        claims_seal_version = claims["seal_version"]
    except (KeyError, TypeError):
        return SealVerdict(False, "G0", "malformed seal")

    # G0 — seal version compatibility
    # We require a known major version so that format changes cannot be silently accepted.
    if not isinstance(claims_seal_version, str):
        return SealVerdict(False, "G0", "seal_version must be a string")

    seal_seal_version = seal.get("seal_version", claims_seal_version)
    if not isinstance(seal_seal_version, str):
        return SealVerdict(False, "G0", "seal.seal_version must be a string if present")

    if seal_seal_version != claims_seal_version:
        return SealVerdict(False, "G0", "seal_version mismatch between envelope and claims")

    # Compare major versions only (e.g., "1.x" compatible with "1.y"), reject unknown majors.
    expected_major = str(SEAL_VERSION).split(".", 1)[0]
    actual_major = claims_seal_version.split(".", 1)[0]
    if expected_major != actual_major:
        return SealVerdict(False, "G0", f"unsupported seal_version {claims_seal_version}")

    # G1 — key pinning: only the pinned key + alg are trusted
    if alg != verifier.alg() or kid != verifier.key_id():
        return SealVerdict(False, "G1", f"unpinned key/alg ({alg}/{kid})")
  1. Ensure that SEAL_VERSION is defined at module scope (e.g., SEAL_VERSION = "1.0" or similar). If it is not yet defined, add it near the top of core/seal/seal.py and update its value according to your current seal format versioning scheme.
  2. If your project already has a helper for parsing/handling version strings, you may want to replace the simple split(".", 1)[0] logic with that helper for consistency.

return SealVerdict(False, "G0", "malformed seal")

# G1 — key pinning: only the pinned key + alg are trusted
if alg != verifier.alg() or kid != verifier.key_id():
return SealVerdict(False, "G1", f"unpinned key/alg ({alg}/{kid})")

# G2 — signature over the canonical claims
try:
ok = verifier.verify(canonical_claims_bytes(claims), base64.b64decode(value))
except Exception: # noqa: BLE001 — malformed base64, etc.
return SealVerdict(False, "G2", "signature decode error")
if not ok:
return SealVerdict(False, "G2", "signature does not verify against the pinned key")

# G3 — the seal covers THIS bundle
if b_sha != bundle.get("integrity", {}).get("sha256"):
return SealVerdict(False, "G3", "seal bundle_sha256 does not match the bundle")

# G4 — model identity bound + honest (an identity-claim never carries a faked weight hash)
mi = bundle.get("change", {}).get("model_identity", {})
if (model.get("provider") != mi.get("provider") or model.get("model") != mi.get("model")
or model.get("model_hash") != mi.get("model_hash")
or model.get("model_hash_kind") != mi.get("model_hash_kind")):
return SealVerdict(False, "G4", "sealed model identity does not match the bundle")
if model.get("model_hash_kind") == "identity-claim" and model.get("model_hash") is not None:
return SealVerdict(False, "G4", "identity-claim must not carry a weight hash")

# G5 — bounded autonomy: matches the bundle AND is within the ceiling
au = bundle.get("change", {}).get("authority", {})
if (authority.get("required") != au.get("required")
or authority.get("max_authorized") != au.get("max_authorized")
or authority.get("allowed") != au.get("allowed")
or authority.get("effective") != au.get("effective")):
return SealVerdict(False, "G5", "sealed authority does not match the bundle")
if not _is_bounded(authority):
return SealVerdict(False, "G5", "sealed change exceeds the autonomy ceiling")

# G6 — the bundle itself integrity-verifies
if not verify_bundle(bundle):
return SealVerdict(False, "G6", "bundle fails its own integrity check")

return SealVerdict(True, "OK", "named+hashed model and bounded authority sealed over "
"this bundle; signature valid against the pinned key")
106 changes: 106 additions & 0 deletions core/seal/signing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
"""core/seal/signing.py — detached-signature primitives for the CROSS-3 seal.

A `Signer` produces a detached signature over canonical claim bytes; a `Verifier` checks
one against a PINNED public key. The seal is verifiable OFFLINE with no secret — the
verifier holds only the public key.

Ed25519Signer/Verifier are the software implementation (real asymmetric crypto via the
`cryptography` library). The HARDWARE path — a YubiKey PIV slot via PKCS#11 C_Sign, with a
slot-9a attestation chain pinned to the Yubico PIV root CA — implements the SAME `Signer`
protocol and is the documented swap-in (it needs the physical token, so it is not built
here; this module ships the seal mechanics + offline verifier the hardware signer plugs
into). No hardware signer is stubbed-as-working — that would be a false-green crypto claim.
"""
from __future__ import annotations

import hashlib
from typing import Protocol, runtime_checkable

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)

_ALG = "ed25519"


def key_id(public_raw: bytes) -> str:
"""Short, stable id of a public key — sha256(raw)[:16]. The verifier PINS this; a seal
signed by any other key is rejected at gate G1."""
return hashlib.sha256(public_raw).hexdigest()[:16]


@runtime_checkable
class Signer(Protocol):
def alg(self) -> str: ...
def key_id(self) -> str: ...
def public_key_bytes(self) -> bytes: ...
def sign(self, payload: bytes) -> bytes: ...


@runtime_checkable
class Verifier(Protocol):
def alg(self) -> str: ...
def key_id(self) -> str: ...
def verify(self, payload: bytes, signature: bytes) -> bool: ...


class Ed25519Signer:
"""Software Ed25519 signer (real asymmetric crypto). Production replaces this with a
PIV/PKCS#11 signer implementing the same protocol; the seal format is identical."""

def __init__(self, private_key: Ed25519PrivateKey) -> None:
self._sk = private_key
self._pub = private_key.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw)

@classmethod
def generate(cls) -> "Ed25519Signer":
return cls(Ed25519PrivateKey.generate())

@classmethod
def from_private_bytes(cls, raw: bytes) -> "Ed25519Signer":
return cls(Ed25519PrivateKey.from_private_bytes(raw))

def alg(self) -> str:
return _ALG

def key_id(self) -> str:
return key_id(self._pub)

def public_key_bytes(self) -> bytes:
return self._pub

def sign(self, payload: bytes) -> bytes:
return self._sk.sign(payload)

def verifier(self) -> "Ed25519Verifier":
"""The matching offline verifier (public key only)."""
return Ed25519Verifier.from_public_bytes(self._pub)


class Ed25519Verifier:
"""Offline verifier — holds only the PINNED public key (no secret material)."""

def __init__(self, public_key: Ed25519PublicKey, public_raw: bytes) -> None:
self._pk = public_key
self._pub = public_raw

@classmethod
def from_public_bytes(cls, raw: bytes) -> "Ed25519Verifier":
return cls(Ed25519PublicKey.from_public_bytes(raw), raw)

def alg(self) -> str:
return _ALG

def key_id(self) -> str:
return key_id(self._pub)

def verify(self, payload: bytes, signature: bytes) -> bool:
try:
self._pk.verify(signature, payload)
return True
except InvalidSignature:
return False
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
jsonschema==3.2.0
reportlab>=4.0
flask>=3.0
cryptography>=42
Loading
Loading