125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
"""Ed25519 signed configuration wrapper.
|
|
|
|
Unsigned files are rejected. After an admin console change, sign again and
|
|
append prev + new + diff to AdminHistory (Verae DataCube-shaped JSONL).
|
|
|
|
Config key ``crypto.mode`` selects npe | lab-xor | plain-lab (see System-Git-Sync).
|
|
"""
|
|
|
|
from __future__ import annotations # postpone evaluation of annotations
|
|
|
|
import json # canonical JSON for signatures
|
|
from dataclasses import dataclass # SignedConfig container
|
|
from pathlib import Path # filesystem paths
|
|
from typing import Any, Dict, Mapping, Optional # types for payload maps
|
|
|
|
from cryptography.exceptions import InvalidSignature # bad sig → reject
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import ( # Ed25519
|
|
Ed25519PrivateKey,
|
|
Ed25519PublicKey,
|
|
)
|
|
from cryptography.hazmat.primitives.serialization import ( # PEM/raw helpers
|
|
Encoding,
|
|
NoEncryption,
|
|
PrivateFormat,
|
|
PublicFormat,
|
|
)
|
|
|
|
|
|
class UnsignedConfig(ValueError):
|
|
"""Raised when a config file has no wrapper or a bad signature."""
|
|
|
|
|
|
def canonical(payload: Mapping[str, Any]) -> bytes:
|
|
"""Stable bytes for signing: sorted keys, no extra spaces."""
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") # one canonical form
|
|
|
|
|
|
@dataclass
|
|
class SignedConfig:
|
|
"""A payload plus Ed25519 signature metadata."""
|
|
|
|
payload: Dict[str, Any] # the configuration object
|
|
key_id: str # fingerprint of the public key
|
|
signature: bytes # 64-byte Ed25519 signature
|
|
|
|
def wrapper(self) -> Dict[str, Any]:
|
|
"""On-disk JSON object (hex signature)."""
|
|
return { # never put the private key here
|
|
"payload": json.loads(json.dumps(self.payload)),
|
|
"sig": {
|
|
"alg": "ed25519",
|
|
"key_id": self.key_id,
|
|
"signature": self.signature.hex(),
|
|
},
|
|
}
|
|
|
|
def dumps(self) -> str:
|
|
"""Pretty JSON for files (payload still signed as canonical)."""
|
|
return json.dumps(self.wrapper(), indent=2) + "\n" # human-readable file
|
|
|
|
|
|
def key_id_from_public(pub: Ed25519PublicKey) -> str:
|
|
"""First 16 hex chars of SHA-256(raw 32-byte public key)."""
|
|
import hashlib # local import keeps module top smaller
|
|
|
|
raw = pub.public_bytes(Encoding.Raw, PublicFormat.Raw) # 32 bytes
|
|
return hashlib.sha256(raw).hexdigest()[:16] # short id for configs
|
|
|
|
|
|
def generate_signing_key() -> Ed25519PrivateKey:
|
|
"""Create a new Ed25519 key (lab). Production: load from HSM."""
|
|
return Ed25519PrivateKey.generate() # OS CSPRNG
|
|
|
|
|
|
def sign(payload: Mapping[str, Any], priv: Ed25519PrivateKey) -> SignedConfig:
|
|
"""Sign canonical JSON of payload."""
|
|
pub = priv.public_key() # derive public
|
|
kid = key_id_from_public(pub) # id printed in wrapper
|
|
sig = priv.sign(canonical(payload)) # 64 bytes
|
|
return SignedConfig(payload=dict(payload), key_id=kid, signature=sig) # copy payload
|
|
|
|
|
|
def verify(wrapper: Mapping[str, Any], pub: Ed25519PublicKey) -> Dict[str, Any]:
|
|
"""Return payload if signature matches; else UnsignedConfig."""
|
|
sig = wrapper.get("sig") or {} # missing sig → fail
|
|
if sig.get("alg") != "ed25519": # only this alg this round
|
|
raise UnsignedConfig("unsupported sig.alg")
|
|
payload = wrapper.get("payload") # unsigned body
|
|
if not isinstance(payload, dict): # must be object
|
|
raise UnsignedConfig("payload must be an object")
|
|
try: # decode hex
|
|
signature = bytes.fromhex(str(sig.get("signature") or ""))
|
|
except ValueError as exc: # not hex
|
|
raise UnsignedConfig("signature not hex") from exc
|
|
if key_id_from_public(pub) != str(sig.get("key_id") or ""): # wrong key
|
|
raise UnsignedConfig("key_id mismatch")
|
|
try: # cryptography raises InvalidSignature
|
|
pub.verify(signature, canonical(payload))
|
|
except InvalidSignature as exc:
|
|
raise UnsignedConfig("bad signature") from exc
|
|
return dict(payload) # verified copy
|
|
|
|
|
|
def load_signed(path: Path, pub: Ed25519PublicKey) -> Dict[str, Any]:
|
|
"""Read a file and reject if unsigned or tampered."""
|
|
raw = Path(path).read_text() # entire file
|
|
wrapper = json.loads(raw) # must be JSON
|
|
if not isinstance(wrapper, dict) or "sig" not in wrapper: # no wrapper
|
|
raise UnsignedConfig("missing sig wrapper")
|
|
return verify(wrapper, pub) # verified payload
|
|
|
|
|
|
def save_signed(path: Path, signed: SignedConfig) -> None:
|
|
"""Write wrapper JSON (does not write the private key)."""
|
|
Path(path).write_text(signed.dumps()) # replace file atomically enough for lab
|
|
|
|
|
|
def pem_private(priv: Ed25519PrivateKey) -> bytes:
|
|
"""PKCS8 PEM for gitignored key files."""
|
|
return priv.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()) # lab only
|
|
|
|
|
|
def pem_public(pub: Ed25519PublicKey) -> bytes:
|
|
"""SubjectPublicKeyInfo PEM for distribution in signed payload."""
|
|
return pub.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo) # public
|