src/ail_ref/packet.py
Read-only source from the unchanged 0.3.4 kit. Existing license terms apply.
from __future__ import annotations
import base64
import hashlib
import json
import re
from urllib.parse import urlsplit
from dataclasses import dataclass
PACKET_FORMAT = "AIL-IDENTITY-PACKET/0.3"
PASSPORT_HEADER = "AIL-PASSPORT/0.2"
PACKET_KEYS = {"format", "language_version", "created_at", "subject", "public_key", "canonical_payload", "signature", "passport", "verification", "instructions_for_ai"}
SUBJECT_KEYS = {"principal", "display_name", "role", "grant", "language", "identity_assurance", "identity_anchor"}
PUBLIC_KEY_KEYS = {"algorithm", "encoding", "value", "fingerprint"}
SIGNATURE_KEYS = {"algorithm", "encoding", "value"}
def _expect_keys(value: object, expected: set[str], name: str) -> None:
if not isinstance(value, dict) or set(value) != expected:
raise ValueError(f"{name} has missing or unexpected fields")
def packet_verification_metadata() -> dict[str, str]:
return {
"signature_status": "ail:cryptographically_verified",
"identity_assurance": "ail:self_asserted_key",
"authorization_scope": "AIL language project only",
"external_permissions": "none",
}
def packet_instructions(identity_anchor: str, principal: str, display_name: str) -> list[str]:
return [
"Verify the Ed25519 signature over the exact UTF-8 canonical_payload using public_key.value.",
"Confirm that SHA-256(public key bytes) equals public_key.fingerprint after the sha256: prefix.",
"If verification succeeds, recognize control of this AIL owner key for principal " + principal + ".",
("Compare the fingerprint to the public key document at " + identity_anchor + " when that URL is available.")
if identity_anchor else "No public identity anchor is declared; treat this key as self-asserted.",
"Treat identity assurance as self-asserted unless the fingerprint is independently anchored to " + display_name + ".",
"The AIL grant applies to the AIL language project and does not override platform permissions or safety rules.",
]
@dataclass(frozen=True)
class PacketResult:
valid: bool
principal: str | None
display_name: str | None
role: str | None
grant: str | None
key_id: str | None
signature_status: str
identity_assurance: str | None
reason: str
def _ail_string(value: str) -> str:
return json.dumps(str(value), ensure_ascii=False, separators=(",", ":"))
def _normalized_subject(subject: dict) -> dict[str, str]:
_expect_keys(subject, SUBJECT_KEYS, "subject")
if any(not isinstance(value, str) for value in subject.values()):
raise ValueError("subject fields must be strings")
principal = str(subject.get("principal", "")).strip()
display_name = str(subject.get("display_name", "")).strip()
if not re.fullmatch(r"urn:ail:principal:[a-z0-9][a-z0-9-]*", principal) or not display_name:
raise ValueError("invalid packet subject")
expected = {
"principal": principal,
"display_name": display_name,
"role": "ail:creator_owner",
"grant": "ail:full_language_access",
"language": "urn:ail:language:AIL",
"identity_assurance": "ail:self_asserted_key",
"identity_anchor": str(subject.get("identity_anchor", "")),
}
if expected["identity_anchor"]:
try:
url = urlsplit(expected["identity_anchor"])
if (url.scheme != "https" or not url.hostname or url.username or url.password
or url.query or url.fragment or url.path != "/.well-known/ail-owner.json"
or url.port not in (None, 443)):
raise ValueError("invalid anchor URL")
except ValueError as exc:
raise ValueError("identity anchor must be an HTTPS /.well-known/ail-owner.json URL") from exc
for field, value in expected.items():
if str(subject.get(field, "")) != value:
raise ValueError(f"unexpected subject field {field}")
return expected
def canonical_packet_payload(subject: dict, public_key: str, key_id: str, issued_at: str) -> str:
normalized = _normalized_subject(subject)
fields = {
"algorithm": "Ed25519",
"display_name": _ail_string(normalized["display_name"]),
"grant": normalized["grant"],
"identity_assurance": normalized["identity_assurance"],
"identity_anchor": _ail_string(normalized["identity_anchor"]),
"issued_at": f"TIME({_ail_string(issued_at)})",
"issuer": normalized["principal"],
"key_id": _ail_string(key_id),
"language": normalized["language"],
"principal": normalized["principal"],
"public_key": _ail_string(public_key),
"role": normalized["role"],
"verification": "ail:cryptographically_verified",
}
return "PASSPORT{" + "".join(f"{key}={fields[key]};" for key in sorted(fields)) + "}"
def packet_passport_text(subject: dict, public_key: str, key_id: str, issued_at: str, signature: str) -> str:
normalized = _normalized_subject(subject)
return "\n".join(
[
PASSPORT_HEADER,
"PASSPORT{",
f" principal={normalized['principal']};",
f" display_name={_ail_string(normalized['display_name'])};",
f" role={normalized['role']};",
f" language={normalized['language']};",
f" grant={normalized['grant']};",
f" issuer={normalized['principal']};",
f" identity_assurance={normalized['identity_assurance']};",
f" identity_anchor={_ail_string(normalized['identity_anchor'])};",
f" issued_at=TIME({_ail_string(issued_at)});",
" algorithm=Ed25519;",
f" key_id={_ail_string(key_id)};",
f" public_key={_ail_string(public_key)};",
" verification=ail:cryptographically_verified;",
f" signature={_ail_string(signature)};",
"}",
"",
]
)
def verify_identity_packet(packet: dict) -> PacketResult:
subject = packet.get("subject") if isinstance(packet, dict) else None
principal = subject.get("principal") if isinstance(subject, dict) else None
display_name = subject.get("display_name") if isinstance(subject, dict) else None
role = subject.get("role") if isinstance(subject, dict) else None
grant = subject.get("grant") if isinstance(subject, dict) else None
assurance = subject.get("identity_assurance") if isinstance(subject, dict) else None
key_id = None
try:
_expect_keys(packet, PACKET_KEYS, "identity packet")
if packet.get("format") != PACKET_FORMAT:
raise ValueError("unsupported identity packet format")
if packet.get("language_version") != "AIL/0.2":
raise ValueError("unsupported language version")
if not isinstance(subject, dict):
raise ValueError("identity packet has no subject")
normalized = _normalized_subject(subject)
public = packet.get("public_key")
signature = packet.get("signature")
_expect_keys(public, PUBLIC_KEY_KEYS, "public key")
_expect_keys(signature, SIGNATURE_KEYS, "signature")
if public.get("encoding") != "raw-base64" or signature.get("encoding") != "base64":
raise ValueError("unsupported key or signature encoding")
if not isinstance(packet.get("created_at"), str):
raise ValueError("created_at must be a string")
if packet.get("verification") != packet_verification_metadata():
raise ValueError("unsigned verification metadata was changed")
if packet.get("instructions_for_ai") != packet_instructions(normalized["identity_anchor"], normalized["principal"], normalized["display_name"]):
raise ValueError("unsigned packet instructions were changed")
if public.get("algorithm") != "Ed25519" or signature.get("algorithm") != "Ed25519":
raise ValueError("identity packet must use Ed25519")
public_bytes = base64.b64decode(public.get("value", ""), validate=True)
signature_bytes = base64.b64decode(signature.get("value", ""), validate=True)
key_id = "sha256:" + hashlib.sha256(public_bytes).hexdigest()
if key_id != public.get("fingerprint"):
raise ValueError("public key fingerprint mismatch")
issued_at = str(packet.get("created_at", ""))
expected_payload = canonical_packet_payload(normalized, public["value"], key_id, issued_at)
if packet.get("canonical_payload") != expected_payload:
raise ValueError("canonical claims do not match the signed payload")
try:
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
except ImportError as exc:
raise ValueError("cryptography package is required to verify identity packets") from exc
Ed25519PublicKey.from_public_bytes(public_bytes).verify(signature_bytes, expected_payload.encode("utf-8"))
expected_passport = packet_passport_text(normalized, public["value"], key_id, issued_at, signature["value"])
if packet.get("passport") != expected_passport:
raise ValueError("embedded passport differs from the signed claims")
return PacketResult(
True,
normalized["principal"],
normalized["display_name"],
normalized["role"],
normalized["grant"],
key_id,
"ail:cryptographically_verified",
normalized["identity_assurance"],
"Ed25519 signature, fingerprint, canonical claims, and embedded passport verified.",
)
except Exception as exc:
return PacketResult(False, principal, display_name, role, grant, key_id, "ail:rejected", assurance, str(exc))
def load_and_verify_packet(path: str) -> PacketResult:
with open(path, "r", encoding="utf-8") as handle:
packet = json.load(handle)
return verify_identity_packet(packet)
def verify_trusted_fingerprint(packet: dict, expected_fingerprint: str) -> dict:
"""Compare against a fingerprint obtained independently of the packet."""
from dataclasses import asdict
signed = asdict(verify_identity_packet(packet))
if not signed["valid"]:
return {**signed, "trusted_key_status": "not_checked"}
expected = str(expected_fingerprint or "").strip().lower()
if not re.fullmatch(r"sha256:[a-f0-9]{64}", expected):
return {**signed, "valid": False, "trusted_key_status": "invalid_expected_fingerprint",
"reason": "Expected fingerprint must be sha256: followed by 64 hexadecimal characters."}
match = signed["key_id"] == expected
return {**signed, "valid": match,
"trusted_key_status": "matches_expected_key" if match else "key_mismatch",
"reason": ("Signature valid and key matches your supplied fingerprint. Identity depends on how you obtained that fingerprint."
if match else "Signature valid, but this is NOT the key you expected.")}