cayce-ops/cayceops/signing.py

"""Operator approval signing — mandatory, content- and deployment-bound.

Every human approval (executor approve / maild approve-draft / publisher
approve) must carry a BIP-340 signature made with the operator key:

    message = sha256("cayce-ops:<type>:<deployment_id>:<id>:<content_hash>")

where content_hash binds the approval to the exact approved content:

    proposal: canonical json of {kind, title, url, detail}
    mail:     canonical json of {to_addr, subject, body}
    content:  canonical json of {channel, caption, link}

There is NO degraded mode: without CAYCE_OPS_OPERATOR_PUBKEY and
CAYCE_OPS_DEPLOYMENT_ID the daemons refuse to execute/send/publish
(see cayceops.config.Config.signing_credentials).
"""

from __future__ import annotations

import hashlib
import json


class SignatureError(RuntimeError):
    """Raised when an approval signature is missing, invalid, or foreign."""


def _canonical(obj: dict) -> str:
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def _sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def proposal_content_hash(kind: str, title: str, url: str | None, detail: dict | None) -> str:
    return _sha256_hex(
        _canonical(
            {
                "kind": kind,
                "title": title,
                "url": url,
                "detail": detail or {},
            }
        )
    )


def mail_content_hash(to_addr: str | None, subject: str | None, body: str | None) -> str:
    return _sha256_hex(
        _canonical(
            {
                "to_addr": to_addr,
                "subject": subject,
                "body": body,
            }
        )
    )


def content_content_hash(channel: str, caption: str, link: str | None) -> str:
    return _sha256_hex(
        _canonical(
            {
                "channel": channel,
                "caption": caption,
                "link": link,
            }
        )
    )


def approval_message(
    item_type: str, deployment_id: str, item_id: int, content_hash: str
) -> bytes:
    """The exact 32-byte message the operator signs."""
    raw = f"cayce-ops:{item_type}:{deployment_id}:{item_id}:{content_hash}"
    return hashlib.sha256(raw.encode("utf-8")).digest()


def verify_signature(
    item_type: str,
    deployment_id: str,
    item_id: int,
    content_hash: str,
    signature_hex: str | None,
    operator_pubkey_hex: str,
) -> bool:
    """Verify an operator approval signature. Foreign deployments or edited
    content verify False because both are inside the signed message."""
    if not signature_hex or not operator_pubkey_hex or not deployment_id:
        return False
    from cayceops.components.publisher.nostr import _schnorr_verify

    try:
        return _schnorr_verify(
            bytes.fromhex(signature_hex),
            approval_message(item_type, deployment_id, item_id, content_hash),
            bytes.fromhex(operator_pubkey_hex),
        )
    except (ValueError, TypeError):
        return False


def pubkey_from_private_hex(private_key_hex: str) -> str:
    """Derive the 64-char x-only public key from a 32-byte hex private key.

    Raises ValueError on malformed input.
    """
    if not private_key_hex:
        raise ValueError("empty private key")
    try:
        from coincurve import PublicKey

        return PublicKey.from_secret(bytes.fromhex(private_key_hex)).format().hex()[2:]
    except (ImportError, ValueError) as exc:
        raise ValueError(f"cannot derive pubkey: {exc}") from exc


class SelfApprovalError(RuntimeError):
    """Raised when the configured operator pubkey is the identity of any
    key held by the system itself (e.g. Cayce nostr identity). The design
    requires a *separate human* operator key — never an agent key."""


FORBIDDEN_IDENTITIES_ENV = "CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS"


def forbidden_identities(cfg) -> list[str]:
    """Pubkeys that may never act as the operator key.

    Includes:
      1. Any private key loaded from the config (nostr_private_key, or any
         future agent identity).
      2. An explicit deny-list in CAYCE_OPS_FORBIDDEN_OPERATOR_PUBKEYS
         (comma-separated hex pubkeys) for keys held outside this process.
    """
    listed: list[str] = []
    if cfg.nostr_private_key:
        try:
            listed.append(pubkey_from_private_hex(cfg.nostr_private_key))
        except ValueError:
            pass  # malformed nostr key: _signing() already catches it elsewhere
    import os

    raw = os.environ.get(FORBIDDEN_IDENTITIES_ENV, "")
    for tok in raw.split(","):
        tok = tok.strip().lower()
        if len(tok) == 64:
            listed.append(tok)
    return listed


def validate_no_self_approval(cfg) -> None:
    """Refuse to operate if the operator pubkey equals any identity held by
    the system. Raises SelfApprovalError.

    Empty operator_pubkey is fine here (that's the 'not configured' state
    which _signing() rejects elsewhere); this check is about *wrong* keys.
    """
    if not cfg.operator_pubkey:
        return
    me = cfg.operator_pubkey.strip().lower()
    for forbidden in forbidden_identities(cfg):
        if me == forbidden:
            raise SelfApprovalError(
                "CAYCE_OPS_OPERATOR_PUBKEY is an agent identity (Cayce or "
                "another in-process key). The operator key must belong to a "
                "human and must never reside on this server. Refusing to "
                "operate. See docs/DECISIONS.md D16."
            )