"""Node identities: Ed25519 keypairs used to sign operations. A node's identity lives in ``identity.json`` inside the node directory, storing the 32-byte signing seed as hex plus a human label. The file is written with mode 0600. Author identifiers on the wire are ``ed25519:<64 hex chars of the public key>``. """ from __future__ import annotations import base64 import json import os from pathlib import Path from typing import Union from nacl.exceptions import BadSignatureError from nacl.signing import SigningKey, VerifyKey from .errors import SignatureError from .wire import AUTHOR_RE class Identity: """An Ed25519 signing identity bound to a human-readable label.""" def __init__(self, signing_key: SigningKey, label: str): self._sk = signing_key self.label = label # -- construction ------------------------------------------------------- @classmethod def create(cls, label: str) -> "Identity": return cls(SigningKey.generate(), label) @classmethod def from_seed_hex(cls, seed_hex: str, label: str) -> "Identity": try: seed = bytes.fromhex(seed_hex) except ValueError as exc: raise SignatureError(f"invalid identity seed: {exc}") from exc if len(seed) != 32: raise SignatureError("identity seed must be 32 bytes") return cls(SigningKey(seed), label) @classmethod def load(cls, path: Union[str, Path]) -> "Identity": path = Path(path) try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise SignatureError(f"cannot read identity file {path}: {exc}") from exc if data.get("scheme") != "ed25519": raise SignatureError( f"identity file {path} uses unsupported scheme " f"{data.get('scheme')!r}" ) ident = cls.from_seed_hex(data["seed_hex"], data.get("label", "node")) recorded = data.get("author") if recorded and recorded != ident.author: raise SignatureError( f"identity file {path} is inconsistent: recorded author " f"{recorded} does not match the stored seed" ) return ident def save(self, path: Union[str, Path]) -> None: path = Path(path) payload = { "scheme": "ed25519", "label": self.label, "seed_hex": self._sk.encode().hex(), "author": self.author, } path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") try: os.chmod(path, 0o600) except OSError: pass # best effort on platforms without POSIX permissions # -- identity & signing -------------------------------------------------- @property def author(self) -> str: return "ed25519:" + bytes(self._sk.verify_key).hex() def sign(self, message: bytes) -> str: """Sign *message*; return the detached signature as base64.""" sig = self._sk.sign(message).signature return base64.b64encode(sig).decode("ascii") def verify_signature(author: str, message: bytes, sig_b64: str) -> None: """Verify a detached signature. Raises SignatureError on any failure.""" if not AUTHOR_RE.match(author): raise SignatureError(f"malformed author identifier {author!r}") pub_hex = author.split(":", 1)[1] try: sig = base64.b64decode(sig_b64, validate=True) except Exception as exc: raise SignatureError(f"signature is not valid base64: {exc}") from exc try: VerifyKey(bytes.fromhex(pub_hex)).verify(message, sig) except BadSignatureError as exc: raise SignatureError(f"signature verification failed for {author}") from exc except Exception as exc: # malformed key material raise SignatureError(f"cannot verify signature: {exc}") from exc