diff --git a/alembic/versions/0003_operator_signatures.py b/alembic/versions/0003_operator_signatures.py new file mode 100644 index 0000000..06056b4 --- /dev/null +++ b/alembic/versions/0003_operator_signatures.py @@ -0,0 +1,42 @@ +"""approval hardening: mandatory operator signatures bound to content + +Revision ID: 0003 +Revises: 0002 +Create Date: 2026-10-20 + +- proposals.operator_signature +- mail_messages.operator_signature +- content_items.operator_signature + +Full signatures (128 hex chars) — never truncated. Existing NULL rows are +refused at execution/send/publish time by the signing gate (no degraded mode). +""" + +from __future__ import annotations + +import sqlalchemy as sa + +from alembic import op + +revision = "0003" +down_revision = "0002" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("proposals") as batch: + batch.add_column(sa.Column("operator_signature", sa.Text(), nullable=True)) + with op.batch_alter_table("mail_messages") as batch: + batch.add_column(sa.Column("operator_signature", sa.Text(), nullable=True)) + with op.batch_alter_table("content_items") as batch: + batch.add_column(sa.Column("operator_signature", sa.Text(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("content_items") as batch: + batch.drop_column("operator_signature") + with op.batch_alter_table("mail_messages") as batch: + batch.drop_column("operator_signature") + with op.batch_alter_table("proposals") as batch: + batch.drop_column("operator_signature") diff --git a/cayceops/cli.py b/cayceops/cli.py index eed4961..df1123f 100644 --- a/cayceops/cli.py +++ b/cayceops/cli.py @@ -74,10 +74,10 @@ def executor_approve( proposal_id: int, actor: str = typer.Option(..., help="Human approver id"), signature: str = typer.Option( - None, help="BIP-340 sig (required when operator pubkey is set)" + ..., help="BIP-340 sig over the approval message (mandatory)" ), ) -> None: - """Approve a proposal (human action, audited).""" + """Approve a proposal (human action, audited; signature mandatory).""" raise typer.Exit( _load_component("executor").approve(proposal_id, actor, signature=signature) ) @@ -101,9 +101,17 @@ def maild_poll() -> None: @maild_app.command("approve-draft") -def maild_approve(msg_id: int, actor: str = typer.Option(...)) -> None: - """Approve a drafted reply so SMTP may send it.""" - raise typer.Exit(_load_component("maild").approve_draft(msg_id, actor)) +def maild_approve( + msg_id: int, + actor: str = typer.Option(...), + signature: str = typer.Option( + ..., help="BIP-340 sig over the approval message (mandatory)" + ), +) -> None: + """Approve a drafted reply so SMTP may send it (signature mandatory).""" + raise typer.Exit( + _load_component("maild").approve_draft(msg_id, actor, signature) + ) @maild_app.command("send-approved") @@ -153,9 +161,17 @@ def publisher_image(item_id: int) -> None: @publisher_app.command("approve") -def publisher_approve(item_id: int, actor: str = typer.Option(...)) -> None: - """Approve a draft content item.""" - raise typer.Exit(_load_component("publisher").approve(item_id, actor)) +def publisher_approve( + item_id: int, + actor: str = typer.Option(...), + signature: str = typer.Option( + ..., help="BIP-340 sig over the approval message (mandatory)" + ), +) -> None: + """Approve a draft content item (signature mandatory).""" + raise typer.Exit( + _load_component("publisher").approve(item_id, actor, signature) + ) @publisher_app.command("publish-due") @@ -187,6 +203,12 @@ def monitor_run() -> None: raise typer.Exit(_load_component("monitor").build()) +@monitor_app.command("anchor") +def monitor_anchor() -> None: + """Publish today's audit-chain anchor as a signed Nostr note (1/day).""" + raise typer.Exit(_load_component("monitor").anchor()) + + def main() -> None: # pragma: no cover app() diff --git a/cayceops/components/executor/executor.py b/cayceops/components/executor/executor.py index 894dbb6..b1d8edf 100644 --- a/cayceops/components/executor/executor.py +++ b/cayceops/components/executor/executor.py @@ -37,11 +37,19 @@ def _load_cfg(): # --- human transitions ------------------------------------------------------- -# message signed by the operator key when CAYCE_OPS_OPERATOR_PUBKEY is set -def approval_message(proposal_id: int) -> bytes: - import hashlib +GATE_ERROR = 2 # missing/invalid configuration or signature - return hashlib.sha256(f"cayce-ops:approve:{proposal_id}".encode()).digest() + +def _signing(cfg: Config) -> tuple[str, str]: + """(deployment_id, operator_pubkey) or ConfigError. No degraded mode.""" + from cayceops.config import ConfigError + + if not cfg.operator_pubkey or not cfg.deployment_id: + raise ConfigError( + "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and " + "CAYCE_OPS_DEPLOYMENT_ID are both required" + ) + return cfg.deployment_id, cfg.operator_pubkey def approve( @@ -51,36 +59,54 @@ def approve( sess=None, signature: str | None = None, ) -> int: - """Human approves a proposal. Audited with the operator's identity. - - When CAYCE_OPS_OPERATOR_PUBKEY is configured, `signature` (hex, BIP-340 - over approval_message(id)) is REQUIRED and verified against that key — - 'actor' then is a label, the key is the proof (audit finding 1). - """ - from cayceops.components.publisher.nostr import verify_schnorr + """Human approves a proposal. Signature is MANDATORY and bound to the + proposal's current content and this deployment (audit finding 1).""" + from cayceops.config import ConfigError + from cayceops.signing import ( + SignatureError, + proposal_content_hash, + verify_signature, + ) cfg = cfg or _load_cfg() - if cfg.operator_pubkey: - ok = bool(signature) and verify_schnorr( - signature, approval_message(proposal_id), cfg.operator_pubkey - ) - if not ok and sess is not None: + try: + deployment_id, pubkey = _signing(cfg) + except ConfigError as exc: + if sess is not None: audit( sess, actor=actor, action="approve", proposal_id=proposal_id, result="error", - detail={ - "reason": ( - "operator pubkey configured: signature required" - if not signature - else "invalid approval signature" - ) - }, + detail={"reason": str(exc)}, ) - if not ok: - return 2 + return GATE_ERROR + + proposal = sess.get(Proposal, proposal_id) if sess is not None else None + if proposal is None: + return 1 + chash = proposal_content_hash( + proposal.kind, proposal.title, proposal.url, proposal.detail + ) + try: + if not signature: + raise SignatureError("signature required") + if not verify_signature( + "proposal", deployment_id, proposal_id, chash, signature, pubkey + ): + raise SignatureError("signature invalid for this content/deployment") + except SignatureError as exc: + audit( + sess, + actor=actor, + action="approve", + proposal_id=proposal_id, + result="error", + detail={"reason": str(exc), "content_hash": chash}, + ) + return GATE_ERROR + return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess, signature) @@ -122,8 +148,8 @@ def _decide( if decision == ProposalStatus.APPROVED: proposal.approved_by = actor proposal.approved_at = datetime.now(UTC) - if signature: - detail["signature"] = signature[:32] + "..." + proposal.operator_signature = signature # FULL signature, row + audit + detail["signature"] = signature or "" verb = "approve" if decision == ProposalStatus.APPROVED else "reject" audit(sess, actor=actor, action=verb, proposal_id=proposal_id, detail=detail) return 0 @@ -136,9 +162,20 @@ def _decide( def _auto_approve(cfg: Config, sess) -> int: - """Auto-approve proposed items per policy. Returns count approved.""" - if not cfg.autonomous_mode or not cfg.autonomous_kinds: - return 0 + """DISABLED (audit round 2, point 1a): approvals require operator + signatures bound to content — there is no non-human approval path. + AUTONOMOUS_MODE no longer grants approval powers (DECISIONS D13).""" + if cfg.autonomous_mode and cfg.autonomous_kinds: + audit( + sess, + actor="auto:policy", + action="autonomy_disabled", + result="error", + detail={ + "reason": "mandatory operator signatures: no non-human approvals", + }, + ) + return 0 midnight = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) used = ( sess.query(func.count(AuditEntry.id)) @@ -188,8 +225,67 @@ def _auto_approve(cfg: Config, sess) -> int: # --- execution ---------------------------------------------------------------- +def _verify_approval_signature(proposal: Proposal, cfg: Config, sess) -> bool: + """Run-time gate (audit round 2, 1d): a row in `approved` MUST carry a + valid operator signature for its CURRENT content and THIS deployment. + Rows approved by raw SQL (no signature) or edited after approval are + rejected, audited as error, and revert to `proposed`.""" + from cayceops.signing import ( + SignatureError, + proposal_content_hash, + verify_signature, + ) + + try: + deployment_id, pubkey = _signing(cfg) + except Exception as exc: # noqa: BLE001 — ConfigError at daemon boundary + audit( + sess, + actor="daemon:executor", + action="execute_refused", + proposal_id=proposal.id, + result="error", + detail={"reason": str(exc)}, + ) + return False + chash = proposal_content_hash( + proposal.kind, proposal.title, proposal.url, proposal.detail + ) + try: + if not verify_signature( + "proposal", + deployment_id, + proposal.id, + chash, + proposal.operator_signature, + pubkey, + ): + raise SignatureError( + "no valid operator signature for current content/deployment" + ) + except SignatureError as exc: + proposal.status = ProposalStatus.PROPOSED # back to pre-approval + proposal.operator_signature = None + audit( + sess, + actor="daemon:executor", + action="execute_refused", + proposal_id=proposal.id, + result="error", + detail={ + "reason": str(exc), + "content_hash": chash, + "had_signature": bool(proposal.operator_signature), + }, + ) + return False + return True + + def execute_one(proposal: Proposal, cfg: Config, sess) -> bool: """Run the plugin for one approved proposal. Returns True on success.""" + if not _verify_approval_signature(proposal, cfg, sess): + return False proposal.status = ProposalStatus.RUNNING proposal.attempts += 1 audit( @@ -256,6 +352,13 @@ def run(cfg: Config | None = None, sess=None) -> int: run_row = DaemonRun(daemon="executor") sess.add(run_row) sess.flush() + try: + _signing(cfg) # refuse to run without signing config (no degraded mode) + except Exception as exc: # noqa: BLE001 — ConfigError at boundary + run_row.status = "error" + run_row.detail = {"refused": str(exc)[:200]} + run_row.finished_at = datetime.now(UTC) + return 2 try: auto = _auto_approve(cfg, sess) pending = ( diff --git a/cayceops/components/maild/maild.py b/cayceops/components/maild/maild.py index 799bba4..0dd2da7 100644 --- a/cayceops/components/maild/maild.py +++ b/cayceops/components/maild/maild.py @@ -39,6 +39,21 @@ from cayceops.models import ( audit, ) +GATE_ERROR = 2 # missing/invalid configuration or signature + + +def _signing(cfg: Config) -> tuple[str, str]: + """(deployment_id, operator_pubkey) or ConfigError. No degraded mode.""" + from cayceops.config import ConfigError + + if not cfg.operator_pubkey or not cfg.deployment_id: + raise ConfigError( + "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and " + "CAYCE_OPS_DEPLOYMENT_ID are both required" + ) + return cfg.deployment_id, cfg.operator_pubkey + + # --- IMAP --------------------------------------------------------------------- @@ -252,13 +267,32 @@ def _poll(cfg: Config, sess, fetcher=None, marker=None) -> tuple[dict, list[byte # --- human gate: approve draft ------------------------------------------------- -def _approve_draft(mail_id: int, actor: str, sess) -> int: +def _mail_to_addr(record) -> str | None: + """The address a send would use for this draft.""" + if record.direction == MailDirection.INBOUND: + return record.from_addr + return record.to_addr + + +def _mail_content_hash(record) -> str: + from cayceops.signing import mail_content_hash + + return mail_content_hash( + _mail_to_addr(record), record.subject, record.draft_reply + ) + + +def _approve_draft( + mail_id: int, actor: str, sess, signature: str | None = None +) -> int: """Human approves a drafted reply (inbound-threaded or first-contact). - The approval is BOUND to the draft text via sha256 — a later edit breaks - the binding and the send refuses (audit finding: approved text was not - tied to its content). + Signature is MANDATORY and bound to (to_addr, subject, body) + deployment + (audit round 2, 1a/1b). The full signature is stored on the row. """ + from cayceops.config import ConfigError + from cayceops.signing import SignatureError, verify_signature + record = sess.get(MailMessage, mail_id) if record is None: return 1 @@ -272,15 +306,52 @@ def _approve_draft(mail_id: int, actor: str, sess) -> int: detail={"reason": f"invalid draft state: {record.draft_status.value}"}, ) return 1 + + from cayceops.config import load + + cfg = load() + try: + deployment_id, pubkey = _signing(cfg) + except ConfigError as exc: + audit( + sess, + actor=actor, + action="approve_draft", + proposal_id=record.proposal_id, + result="error", + detail={"reason": str(exc)}, + ) + return GATE_ERROR + + chash = _mail_content_hash(record) + try: + if not signature: + raise SignatureError("signature required") + if not verify_signature( + "mail", deployment_id, mail_id, chash, signature, pubkey + ): + raise SignatureError("signature invalid for this content/deployment") + except SignatureError as exc: + audit( + sess, + actor=actor, + action="approve_draft", + proposal_id=record.proposal_id, + result="error", + detail={"reason": str(exc), "content_hash": chash}, + ) + return GATE_ERROR + record.draft_status = DraftStatus.APPROVED record.approved_hash = _content_hash(record.draft_reply) record.approved_by = actor[:128] + record.operator_signature = signature # FULL signature audit( sess, actor=actor, action="approve_draft", proposal_id=record.proposal_id, - detail={"mail_id": record.id, "hash": record.approved_hash}, + detail={"mail_id": record.id, "hash": record.approved_hash, "signature": signature}, ) return 0 @@ -328,7 +399,9 @@ def _send_one(cfg: Config, sess, mail_id: int, sender=None) -> str: The caller owns the transaction: one message per commit, so an SMTP failure on message N never un-sends messages 1..N-1 (audit finding: duplicates). """ - from cayceops.models import _content_hash + + from cayceops.config import ConfigError + from cayceops.signing import verify_signature record = sess.get(MailMessage, mail_id) if record is None: @@ -336,6 +409,41 @@ def _send_one(cfg: Config, sess, mail_id: int, sender=None) -> str: if record.draft_status != DraftStatus.APPROVED: return "blocked" + # run-time signature gate (audit round 2, 1d): the row must carry a valid + # operator signature for its CURRENT to_addr/subject/body + deployment. + # Unsigned rows (raw SQL approval) or edited ones revert to DRAFTED. + try: + deployment_id, pubkey = _signing(cfg) + except ConfigError as exc: + audit( + sess, + actor="daemon:maild", + action="send_refused", + proposal_id=record.proposal_id, + result="error", + detail={"mail_id": record.id, "reason": str(exc)}, + ) + return "unsigned" + current_hash = _mail_content_hash(record) + if not verify_signature( + "mail", deployment_id, record.id, current_hash, record.operator_signature, pubkey + ): + record.draft_status = DraftStatus.DRAFTED # back to pre-approval + audit( + sess, + actor="daemon:maild", + action="send_refused", + proposal_id=record.proposal_id, + result="error", + detail={ + "mail_id": record.id, + "reason": "no valid operator signature for current content/deployment", + "content_hash": current_hash, + "had_signature": bool(record.operator_signature), + }, + ) + return "hash_mismatch" + if record.direction == MailDirection.INBOUND: # reply: goes back to the inbound sender, threads to the inbound mail, # and gets its own OUTBOUND row @@ -357,19 +465,6 @@ def _send_one(cfg: Config, sess, mail_id: int, sender=None) -> str: record.draft_status = DraftStatus.SKIPPED return "blocked" - # binding: the text must be exactly what the human approved - if record.approved_hash and _content_hash(record.draft_reply or "") != record.approved_hash: - record.draft_status = DraftStatus.DRAFTED # needs fresh approval - audit( - sess, - actor="daemon:maild", - action="send_refused_hash_mismatch", - proposal_id=record.proposal_id, - result="error", - detail={"mail_id": record.id}, - ) - return "hash_mismatch" - (sender or _smtp_send)( cfg, to_addr, @@ -436,10 +531,22 @@ def _send_approved(cfg: Config, sess, sender=None, factory=None) -> dict: outcome = _send_one(cfg, sess, mail_id, sender) except Exception as exc: # noqa: BLE001 — isolate the failed message outcome = "error" - if sess is not None: - detail = {"mail_id": mail_id, "error": str(exc)[:200]} - audit(sess, actor="daemon:maild", action="send_failed", - result="error", detail=detail) + # persist the failure in its OWN transaction (the failed message's + # transaction is dead) + try: + error_sess = factory() if factory is not None else None + if error_sess is not None: + audit( + error_sess, + actor="daemon:maild", + action="send_failed", + result="error", + detail={"mail_id": mail_id, "error": str(exc)[:200]}, + ) + error_sess.commit() + error_sess.close() + except Exception: # noqa: S110 — best-effort failure audit + pass stats[outcome] = stats.get(outcome, 0) + 1 return stats @@ -517,40 +624,48 @@ def poll(cfg: Config | None = None) -> int: engine().dispose() -def approve_draft(mail_id: int, actor: str) -> int: +def approve_draft(mail_id: int, actor: str, signature: str | None = None) -> int: from cayceops.db import engine try: with session_scope() as sess: - return _approve_draft(mail_id, actor, sess) + return _approve_draft(mail_id, actor, sess, signature) finally: engine().dispose() def send_approved(cfg: Config | None = None) -> int: + """CLI: same code path as the tests (_send_approved with factory). + + Failures are audited (send_failed), recorded in daemon_runs, and any + error yields a non-zero exit code (audit round 2, point 5). + """ from cayceops.config import load from cayceops.db import engine, session_factory cfg = cfg or load() try: - # ids first, then one transaction per message: a failure on message N - # cannot roll back already-sent messages (audit finding: duplicates) - with session_scope() as sess: - ids = [ - row.id - for row in sess.query(MailMessage.id) - .filter(MailMessage.draft_status == DraftStatus.APPROVED) - .all() - ] factory = session_factory() - for mail_id in ids: + with session_scope() as sess: + run_row = DaemonRun(daemon="maild-send") + sess.add(run_row) + sess.flush() try: - with factory() as one_sess: - _send_one(cfg, one_sess, mail_id) - one_sess.commit() - except Exception as exc: # noqa: BLE001 — isolate each send - print(f"send {mail_id} failed: {str(exc)[:160]}") - return 0 + _signing(cfg) # no degraded mode + except Exception as exc: # noqa: BLE001 — ConfigError + run_row.status = "error" + run_row.detail = {"refused": str(exc)[:200]} + run_row.finished_at = datetime.now(UTC) + print(f"send refused: {exc}") + return 2 + stats = _send_approved(cfg, sess, factory=factory) + run_row.status = "error" if stats.get("error") else "ok" + run_row.detail = stats + run_row.finished_at = datetime.now(UTC) + print(stats) + # any failure class is a non-zero exit (audit round 2, point 5) + failures = stats.get("error") or stats.get("hash_mismatch") or stats.get("unsigned") + return 1 if failures else 0 finally: engine().dispose() diff --git a/cayceops/components/publisher/publisher.py b/cayceops/components/publisher/publisher.py index 872fb00..6bde7d1 100644 --- a/cayceops/components/publisher/publisher.py +++ b/cayceops/components/publisher/publisher.py @@ -114,7 +114,33 @@ def _plan(cfg: Config, sess, days: int = 4) -> dict: # --- human gate ------------------------------------------------------------------- -def _approve(item_id: int, actor: str, sess) -> int: +GATE_ERROR = 2 # missing/invalid configuration or signature + + +def _signing(cfg: Config) -> tuple[str, str]: + from cayceops.config import ConfigError + + if not cfg.operator_pubkey or not cfg.deployment_id: + raise ConfigError( + "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and " + "CAYCE_OPS_DEPLOYMENT_ID are both required" + ) + return cfg.deployment_id, cfg.operator_pubkey + + +def _content_binding(item: ContentItem) -> str: + from cayceops.signing import content_content_hash + + return content_content_hash(item.channel, item.caption, item.link) + + +def _approve( + item_id: int, actor: str, sess, signature: str | None = None +) -> int: + """Signature MANDATORY, bound to (channel, caption, link) + deployment.""" + from cayceops.config import ConfigError + from cayceops.signing import SignatureError, verify_signature + item = sess.get(ContentItem, item_id) if item is None or item.status != ContentStatus.DRAFT: audit( @@ -125,17 +151,58 @@ def _approve(item_id: int, actor: str, sess) -> int: detail={"reason": f"item {item_id} not in draft state"}, ) return 1 + try: + deployment_id, pubkey = _signing(cfg_from(sess)) + except ConfigError as exc: + audit( + sess, + actor=actor, + action="approve_content", + result="error", + detail={"reason": str(exc)}, + ) + return GATE_ERROR + chash = _content_binding(item) + try: + if not signature: + raise SignatureError("signature required") + if not verify_signature( + "content", deployment_id, item_id, chash, signature, pubkey + ): + raise SignatureError("signature invalid for this content/deployment") + except SignatureError as exc: + audit( + sess, + actor=actor, + action="approve_content", + result="error", + detail={"reason": str(exc), "content_hash": chash}, + ) + return GATE_ERROR item.status = ContentStatus.APPROVED - item.approved_hash = _content_hash(item.caption) # binding: caption as approved + item.approved_hash = _content_hash(item.caption) + item.operator_signature = signature # FULL signature audit( sess, actor=actor, action="approve_content", - detail={"item": item.id, "channel": item.channel, "hash": item.approved_hash}, + detail={ + "item": item.id, + "channel": item.channel, + "hash": item.approved_hash, + "signature": signature, + }, ) return 0 +def cfg_from(sess): + """Config from env (publisher commands are single-invocation CLIs).""" + from cayceops.config import load + + return load() + + # --- image rendering ---------------------------------------------------------------- @@ -192,14 +259,41 @@ def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict: if item.scheduled_for and item.scheduled_for > now: continue # not due yet - # binding: caption must be exactly what was approved (audit finding) - if item.approved_hash and _content_hash(item.caption) != item.approved_hash: + # run-time signature gate (audit round 2, 1d): valid operator + # signature required for CURRENT channel/caption/link + deployment. + # Unsigned rows (raw SQL approval) or edited ones revert to DRAFT. + from cayceops.signing import verify_signature + + try: + deployment_id, pubkey = _signing(cfg) + except Exception as exc: # noqa: BLE001 — ConfigError audit( sess, actor="daemon:publisher", - action="publish_refused_hash_mismatch", + action="publish_refused", result="error", - detail={"item": item.id}, + detail={"item": item.id, "reason": str(exc)}, + ) + stats["failed"] += 1 + continue + if not verify_signature( + "content", + deployment_id, + item.id, + _content_binding(item), + item.operator_signature, + pubkey, + ): + item.status = ContentStatus.DRAFT # back to pre-approval + audit( + sess, + actor="daemon:publisher", + action="publish_refused", + result="error", + detail={ + "item": item.id, + "reason": "no valid operator signature for current content/deployment", + }, ) stats["tampered"] += 1 continue @@ -287,6 +381,13 @@ def publish_due(cfg: Config | None = None) -> int: run_row = DaemonRun(daemon="publisher") sess.add(run_row) sess.flush() + try: + _signing(cfg) # no degraded mode + except Exception as exc: # noqa: BLE001 — ConfigError + run_row.status = "error" + run_row.detail = {"refused": str(exc)[:200]} + run_row.finished_at = datetime.now(UTC) + return 2 stats = _publish_due(cfg, sess) run_row.status = "ok" run_row.detail = stats @@ -312,11 +413,11 @@ def make_image(item_id: int, cfg: Config | None = None) -> int: engine().dispose() -def approve(item_id: int, actor: str) -> int: +def approve(item_id: int, actor: str, signature: str | None = None) -> int: from cayceops.db import engine try: with session_scope() as sess: - return _approve(item_id, actor, sess) + return _approve(item_id, actor, sess, signature) finally: engine().dispose() diff --git a/cayceops/config.py b/cayceops/config.py index c58b158..9e68bbc 100644 --- a/cayceops/config.py +++ b/cayceops/config.py @@ -84,11 +84,15 @@ class Config: enable_http_actions: bool = field( default_factory=lambda: _bool("CAYCE_OPS_ENABLE_HTTP_ACTIONS", False) ) - # when set (64-char hex x-only pubkey), human approvals MUST carry a valid - # BIP-340 signature over "cayce-ops:approve:" made by the operator key. + # MANDATORY for approval-gated work: the operator x-only pubkey and the + # deployment identifier are part of every signed approval message. + # Without BOTH, daemons refuse to execute/send/publish — no degraded mode. operator_pubkey: str = field( default_factory=lambda: os.environ.get("CAYCE_OPS_OPERATOR_PUBKEY", "") ) + deployment_id: str = field( + default_factory=lambda: os.environ.get("CAYCE_OPS_DEPLOYMENT_ID", "") + ) # --- mail --- imap_host: str = field( @@ -167,6 +171,10 @@ class Config: ) +class ConfigError(RuntimeError): + """Raised when mandatory configuration is missing.""" + + def load() -> Config: """Build the process Config snapshot.""" return Config() diff --git a/cayceops/models.py b/cayceops/models.py index 65a6dad..fa2acec 100644 --- a/cayceops/models.py +++ b/cayceops/models.py @@ -21,9 +21,10 @@ from __future__ import annotations import enum import hashlib import json +import threading from datetime import UTC, datetime -from sqlalchemy import JSON, DateTime, Enum, Float, ForeignKey, Integer, String, Text +from sqlalchemy import JSON, DateTime, Enum, Float, ForeignKey, Integer, String, Text, text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column @@ -77,6 +78,8 @@ class Proposal(Base): dedupe_key: Mapped[str | None] = mapped_column(String(256), unique=True, nullable=True) approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True) approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + # full operator approval signature (128 hex chars) — never truncated + operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=utcnow, onupdate=utcnow @@ -103,6 +106,11 @@ class AuditEntry(Base): entry_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) +# reentrant: several audit() calls may share one transaction +_AUDIT_LOCK = threading.RLock() +_AUDIT_LOCK_PG_KEY = 7471186757 # pg_advisory_xact_lock key for cross-process + + def audit( sess, actor: str, @@ -113,13 +121,66 @@ def audit( ) -> AuditEntry: """Record one audited action, chained to the previous entry. - Chain: entry_hash = sha256(prev_hash | ts | actor | action | proposal_id | detail | result). - Every component writes through this helper. + Chain: entry_hash = sha256(prev_hash | ts_epoch | actor | action | + proposal_id | detail | result). Writes are SERIALIZED (audit round 2, + point 4): pg_advisory_xact_lock on Postgres, in-process lock on SQLite — + concurrent writers can no longer fork the chain. + + Legacy rows (entry_hash IS NULL, pre-0002) are ignored when picking + prev_hash: the chain starts GENESIS at the first hashed row. """ - prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() - prev_hash = prev.entry_hash if prev is not None else "GENESIS" + bind = sess.get_bind() + if bind.dialect.name == "postgresql": + # xact-scoped: PG releases the lock at commit/rollback automatically + sess.execute( + text("SELECT pg_advisory_xact_lock(:k)"), {"k": _AUDIT_LOCK_PG_KEY} + ) + prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() + prev_hash = prev.entry_hash if prev is not None else "GENESIS" + entry = _build_entry(prev_hash, actor, action, proposal_id, detail, result) + sess.add(entry) + return entry + + # SQLite path: hold the in-process lock until the session commits or + # rolls back, so concurrent writers cannot read the same prev_hash + # (round-2 point 4 — verified by test_concurrent_writers_keep_chain_verifiable) + _AUDIT_LOCK.acquire() + released = False + + def _release(*_args): + nonlocal released + if not released: + released = True + try: + _AUDIT_LOCK.release() + except RuntimeError: # pragma: no cover + pass + + from sqlalchemy import event + + event.listen(sess, "after_commit", _release) + event.listen(sess, "after_rollback", _release) + event.listen(sess, "after_soft_rollback", _release) + try: + prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() + prev_hash = prev.entry_hash if prev is not None else "GENESIS" + entry = _build_entry(prev_hash, actor, action, proposal_id, detail, result) + sess.add(entry) + return entry + except Exception: + _release() + raise + + +def _build_entry( + prev_hash: str, + actor: str, + action: str, + proposal_id: int | None, + detail: dict | None, + result: str, +) -> AuditEntry: ts = utcnow() - # epoch int: identical after a sqlite/pg round-trip (tz-aware isoformat is not) ts_int = int(ts.timestamp()) detail_json = json.dumps(detail or {}, sort_keys=True, ensure_ascii=False) payload = "|".join( @@ -133,31 +194,40 @@ def audit( result, ] ) - entry = AuditEntry( + return AuditEntry( ts=ts, actor=actor, action=action, proposal_id=proposal_id, detail=detail or {}, result=result, - prev_hash=prev_hash, + prev_hash=prev_hash or "GENESIS", entry_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(), ) - sess.add(entry) - return entry def verify_audit_chain(sess) -> tuple[bool, int | None]: - """Re-walk the chain. Returns (ok, first_broken_entry_id or None).""" - prev_hash = "GENESIS" + """Re-walk the chain. Returns (ok, first_broken_entry_id or None). + + Legacy rows (entry_hash IS NULL, written before migration 0002) are + skipped and documented: the chain begins GENESIS at the first hashed row, + so a migrated database verifies clean without any tampering. + """ + prev_hash: str | None = None # None = still in legacy prefix for entry in sess.query(AuditEntry).order_by(AuditEntry.id).all(): + if entry.entry_hash is None: + continue # legacy row: outside the chain by documented genesis + if prev_hash is None: + expected_prev = "GENESIS" # first hashed row anchors the chain + else: + expected_prev = prev_hash detail_json = json.dumps(entry.detail or {}, sort_keys=True, ensure_ascii=False) ts = entry.ts if ts.tzinfo is None: ts = ts.replace(tzinfo=UTC) payload = "|".join( [ - prev_hash, + expected_prev, str(int(ts.timestamp())), entry.actor, entry.action, @@ -167,7 +237,7 @@ def verify_audit_chain(sess) -> tuple[bool, int | None]: ] ) expected = hashlib.sha256(payload.encode("utf-8")).hexdigest() - if entry.prev_hash != prev_hash or entry.entry_hash != expected: + if entry.prev_hash != expected_prev or entry.entry_hash != expected: return False, entry.id prev_hash = entry.entry_hash return True, None @@ -222,6 +292,7 @@ class MailMessage(Base): # binding: hash of the draft text the human approved + who approved it approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True) + operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) @@ -254,6 +325,7 @@ class ContentItem(Base): attempts: Mapped[int] = mapped_column(Integer, default=0) # binding: hash of the caption the human approved approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) + operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) diff --git a/cayceops/signing.py b/cayceops/signing.py new file mode 100644 index 0000000..df25b06 --- /dev/null +++ b/cayceops/signing.py @@ -0,0 +1,103 @@ +"""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::::") + +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 diff --git a/tests/conftest.py b/tests/conftest.py index 1d663a4..4ba6864 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,8 +10,14 @@ from __future__ import annotations import os from pathlib import Path +from types import SimpleNamespace import pytest +from coincurve import PublicKey + +# test operator keypair (invented; allowlisted in .gitleaks.toml) +OPERATOR_SECRET = bytes.fromhex("22" * 32) +OPERATOR_PUBKEY = PublicKey.from_secret(OPERATOR_SECRET).format().hex()[2:] # Config must see the test DB before cayceops.config/db get imported. os.environ.setdefault("CAYCE_OPS_DATABASE_URL", "sqlite:///:memory:") @@ -54,6 +60,61 @@ def sess(db_engine): s.close() +@pytest.fixture() +def signer(): + """Signs approval messages with the test operator key (22*32, deployment + 'itest'). Mirrors exactly what the operator CLI does.""" + from coincurve import PrivateKey + + from cayceops.models import MailDirection + from cayceops.signing import ( + approval_message, + content_content_hash, + mail_content_hash, + proposal_content_hash, + ) + + secret = OPERATOR_SECRET + + def _sig(msg: bytes) -> str: + return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex() + + def proposal(p, deployment="itest"): + chash = proposal_content_hash(p.kind, p.title, p.url, p.detail) + return _sig(approval_message("proposal", deployment, p.id, chash)) + + def mail(record, deployment="itest"): + to = ( + record.from_addr + if record.direction == MailDirection.INBOUND + else record.to_addr + ) + chash = mail_content_hash(to, record.subject, record.draft_reply) + return _sig(approval_message("mail", deployment, record.id, chash)) + + def content(item, deployment="itest"): + chash = content_content_hash(item.channel, item.caption, item.link) + return _sig(approval_message("content", deployment, item.id, chash)) + + def raw(sig_bytes: bytes) -> str: + return _sig(sig_bytes) + + def msg(item_type: str, deployment: str, item_id: int, chash: str) -> bytes: + from cayceops.signing import approval_message + + return approval_message(item_type, deployment, item_id, chash) + + def content_hash(item, deployment="itest") -> str: + from cayceops.signing import content_content_hash + + return content_content_hash(item.channel, item.caption, item.link) + + return SimpleNamespace( + proposal=proposal, mail=mail, content=content, raw=raw, msg=msg, + content_hash=content_hash, + ) + + @pytest.fixture() def cfg(monkeypatch): """A Config snapshot with test defaults (no network, no autonomy).""" @@ -66,6 +127,10 @@ def cfg(monkeypatch): "CAYCE_OPS_ENABLE_INSTAGRAM": "false", "CAYCE_OPS_NOSTR_RELAYS": "", "CAYCE_OPS_MEDIA_DIR": "media", + # signing defaults (audit round 2): tests operate in a signed + # deployment unless they explicitly clear these + "CAYCE_OPS_DEPLOYMENT_ID": "itest", + "CAYCE_OPS_OPERATOR_PUBKEY": OPERATOR_PUBKEY, } def _apply(**overrides): diff --git a/tests/test_executor.py b/tests/test_executor.py index 0137e1e..5c8f382 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -1,4 +1,9 @@ -"""executor tests — human gate, autonomy bounds, retries, audit chain.""" +"""executor tests — human gate (signed), autonomy bounds, retries, audit chain. + +Since audit round 2, approvals REQUIRE an operator signature bound to the +proposal's current content + deployment id (tests sign with the conftest +operator key 22*32, deployment 'itest'). +""" from __future__ import annotations @@ -12,16 +17,22 @@ def _proposal(kind="content_gap", title="Test item", **kw): return Proposal(kind=kind, title=title, **kw) -def test_approve_requires_proposed_state(cfg, sess): +def test_approve_requires_proposed_state(cfg, sess, signer): p = _proposal() sess.add(p) sess.flush() - assert executor.approve(p.id, "gustavo", cfg(), sess) == 0 + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0 + ) # second approve must fail (state is now approved) - assert executor.approve(p.id, "gustavo", cfg(), sess) == 1 + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 1 + ) assert p.approved_by == "gustavo" - assert p.status == ProposalStatus.APPROVED + assert p.status is ProposalStatus.APPROVED + # FULL signature stored on the row + assert p.operator_signature and len(p.operator_signature) == 128 def test_reject_records_actor(cfg, sess): @@ -29,75 +40,26 @@ def test_reject_records_actor(cfg, sess): sess.add(p) sess.flush() assert executor.reject(p.id, "gustavo", cfg(), sess) == 0 - assert p.status == ProposalStatus.REJECTED + assert p.status is ProposalStatus.REJECTED -def test_signed_approval_required_when_operator_key_configured(cfg, sess): - """Critical finding 1: --actor was free text; now a key can be required.""" - from coincurve import PrivateKey, PublicKey - - from cayceops.components.executor import executor - - secret = bytes.fromhex("22" * 32) - pubkey = PublicKey.from_secret(secret).format().hex()[2:] - p = _proposal() - sess.add(p) - sess.flush() - - config = cfg(CAYCE_OPS_OPERATOR_PUBKEY=pubkey) - - # no signature -> refused - assert executor.approve(p.id, "gustavo", config, sess, signature=None) == 2 - # wrong key -> refused - bad = ( - PrivateKey(bytes.fromhex("33" * 32)) - .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32) - .hex() - ) - assert executor.approve(p.id, "gustavo", config, sess, signature=bad) == 2 - # correct key -> approved - good = ( - PrivateKey(secret) - .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32) - .hex() - ) - assert executor.approve(p.id, "gustavo", config, sess, signature=good) == 0 - assert p.status is ProposalStatus.APPROVED - entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one() - assert "signature" in entry.detail - - -def test_auto_approve_orders_by_relevance_desc(cfg, sess): - """Medium finding: ascending 'risk_score' approved noise first.""" - low = _proposal(kind="content_gap", title="barely relevant") - low.match_score = 0.1 - high = _proposal(kind="content_gap", title="highly relevant") - high.match_score = 0.9 - sess.add_all([low, high]) - sess.flush() - - config = cfg( - CAYCE_OPS_AUTONOMOUS_MODE="true", - CAYCE_OPS_AUTONOMOUS_KINDS="content_gap", - CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1", - ) - executor._auto_approve(config, sess) - assert high.status is ProposalStatus.APPROVED - assert low.status is ProposalStatus.PROPOSED - - -def test_run_executes_only_approved(cfg, sess): +def test_run_executes_only_approved(cfg, sess, signer): approved = _proposal(title="Approved item") proposed = _proposal(title="Proposed item") sess.add_all([approved, proposed]) sess.flush() - executor.approve(approved.id, "gustavo", cfg(), sess) + assert ( + executor.approve( + approved.id, "gustavo", cfg(), sess, signature=signer.proposal(approved) + ) + == 0 + ) config = cfg() # AUTONOMOUS_MODE=false assert executor.run(config, sess) == 0 - assert approved.status == ProposalStatus.DONE - assert proposed.status == ProposalStatus.PROPOSED # untouched + assert approved.status is ProposalStatus.DONE + assert proposed.status is ProposalStatus.PROPOSED # untouched acts = [a.action for a in sess.query(AuditEntry).order_by(AuditEntry.id).all()] assert "execute_start" in acts and "execute_done" in acts @@ -110,36 +72,127 @@ def test_autonomy_off_by_default(cfg, sess): sess.flush() config = cfg() # autonomous_mode False even with kinds configured executor._auto_approve(config, sess) - assert p.status == ProposalStatus.PROPOSED + assert p.status is ProposalStatus.PROPOSED -def test_autonomy_respects_kinds_and_daily_cap(cfg, sess): - p1 = _proposal(kind="content_gap", title="auto ok") - p2 = _proposal(kind="content_gap", title="auto over cap") - p3 = _proposal(kind="affiliate", title="kind not allowed") - sess.add_all([p1, p2, p3]) +def test_autonomy_cannot_approve_anymore(cfg, sess): + """Round-2 1a: with mandatory signatures there is NO non-human approval + path. AUTONOMOUS_MODE no longer approves anything (D13).""" + low = _proposal(kind="content_gap", title="barely relevant") + low.match_score = 0.1 + high = _proposal(kind="content_gap", title="highly relevant") + high.match_score = 0.9 + sess.add_all([low, high]) sess.flush() - config = cfg( CAYCE_OPS_AUTONOMOUS_MODE="true", CAYCE_OPS_AUTONOMOUS_KINDS="content_gap", CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1", ) - approved = executor._auto_approve(config, sess) - assert approved == 1 - assert p1.status == ProposalStatus.APPROVED - assert p2.status == ProposalStatus.PROPOSED # cap reached - assert p3.status == ProposalStatus.PROPOSED # kind not in policy - - actor = sess.query(AuditEntry).filter_by(action="approve").one().actor - assert actor == "auto:policy" + assert executor._auto_approve(config, sess) == 0 + assert low.status is ProposalStatus.PROPOSED + assert high.status is ProposalStatus.PROPOSED + entry = sess.query(AuditEntry).filter_by(action="autonomy_disabled").one() + assert entry.result == "error" -def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch): +def test_signed_approval_mandatory_and_content_bound(cfg, sess, signer): + """Round-2 1a/1b/1c: signature mandatory, bound to content+deployment, + stored COMPLETE on the row and in the audit log. Signatures for other + deployments or other content are refused.""" + from cayceops.signing import proposal_content_hash + + p = _proposal(kind="partnership", title="Acme partner", url="https://acme.example.org") + p.detail = {"region": "latam"} + sess.add(p) + sess.flush() + + chash = proposal_content_hash(p.kind, p.title, p.url, p.detail) + + # no signature -> refused + assert executor.approve(p.id, "gustavo", cfg(), sess, signature=None) == 2 + # signature for ANOTHER deployment -> refused (required test) + other_dep = signer.raw( + signer.msg("proposal", "other-deploy", p.id, chash) + ) + assert executor.approve(p.id, "gustavo", cfg(), sess, signature=other_dep) == 2 + # signature over OTHER CONTENT -> refused (required test) + other_content = signer.raw(signer.msg("proposal", "itest", p.id, "00" * 32)) + assert executor.approve(p.id, "gustavo", cfg(), sess, signature=other_content) == 2 + # valid signature -> approved + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0 + ) + assert p.status is ProposalStatus.APPROVED + entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one() + assert entry.detail["signature"] == p.operator_signature + + +def test_approve_fails_audited_from_cli_wrapper(cfg, sess): + """Round-2 1f: signature failures leave an audit record even when the + command runs through its CLI wrapper (the wrapper passes sess).""" + p = _proposal() + sess.add(p) + sess.flush() + + code = executor.approve(p.id, "gustavo", cfg(), sess, signature=None) + assert code == 2 + entry = sess.query(AuditEntry).filter_by(action="approve", result="error").one() + assert "signature required" in entry.detail["reason"] + + +def test_sql_approved_row_is_refused_at_execution(cfg, sess, monkeypatch): + """Round-2 1d (required test): a row put in `approved` by raw SQL without + a signature must not execute; it reverts to `proposed` with an error audit.""" + p = _proposal() + p.status = ProposalStatus.APPROVED # raw-SQL style: no signature + sess.add(p) + sess.flush() + + executed = [] + + def spy_plugin(proposal, cfg, sess): + executed.append(proposal.id) + return {} + + monkeypatch.setitem(plugins.REGISTRY, "content_gap", spy_plugin) + assert executor.run(cfg(), sess) == 0 + assert executed == [] + assert p.status is ProposalStatus.PROPOSED # reverted + refused = sess.query(AuditEntry).filter_by(action="execute_refused").one() + assert refused.result == "error" + + +def test_execution_verifies_signature_against_current_content(cfg, sess, signer, monkeypatch): + """Round-2 1d: signature over content that changed after approval is + refused at execution time.""" + p = _proposal(kind="partnership", title="original") + sess.add(p) + sess.flush() + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0 + ) + + p.title = "EDITED after approval" # content changed under the signature + + executed = [] + monkeypatch.setitem( + plugins.REGISTRY, + "partnership", + lambda proposal, cfg, sess: executed.append(proposal.id) or {}, + ) + assert executor.run(cfg(), sess) == 0 + assert executed == [] + assert p.status is ProposalStatus.PROPOSED + + +def test_failing_plugin_retries_then_fails(cfg, sess, signer, monkeypatch): p = _proposal(kind="affiliate", title="Will fail") sess.add(p) sess.flush() - executor.approve(p.id, "gustavo", cfg(), sess) + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0 + ) calls = {"n": 0} @@ -151,11 +204,11 @@ def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch): config = cfg(CAYCE_OPS_MAX_ATTEMPTS="2") executor.run(config, sess) # attempt 1 -> back to approved for retry - assert p.status == ProposalStatus.APPROVED + assert p.status is ProposalStatus.APPROVED assert p.attempts == 1 executor.run(config, sess) # attempt 2 -> failed for good - assert p.status == ProposalStatus.FAILED + assert p.status is ProposalStatus.FAILED assert p.attempts == 2 assert calls["n"] == 2 @@ -174,14 +227,16 @@ def test_http_post_blocked_when_disabled(cfg, sess): plugins.http_post(p, config, sess) -def test_unknown_kind_falls_back_to_log_action(cfg, sess): +def test_unknown_kind_falls_back_to_log_action(cfg, sess, signer): p = _proposal(kind="brand_new_kind", title="No plugin yet") sess.add(p) sess.flush() - executor.approve(p.id, "gustavo", cfg(), sess) + assert ( + executor.approve(p.id, "gustavo", cfg(), sess, signature=signer.proposal(p)) == 0 + ) assert executor.run(cfg(), sess) == 0 - assert p.status == ProposalStatus.DONE + assert p.status is ProposalStatus.DONE done = sess.query(AuditEntry).filter_by(action="execute_done").one() assert done.detail["outcome"]["plugin_fallback"] is True diff --git a/tests/test_maild.py b/tests/test_maild.py index 18a76be..d317073 100644 --- a/tests/test_maild.py +++ b/tests/test_maild.py @@ -2,7 +2,10 @@ from __future__ import annotations -from types import SimpleNamespace +import os + +from coincurve import PrivateKey +from coincurve import PublicKey as _PK from cayceops.components.maild import classify as clf from cayceops.components.maild import maild @@ -16,9 +19,20 @@ from cayceops.models import ( ProposalStatus, ) +# test operator keypair (matches conftest default pubkey; invented) +OPERATOR_SECRET = bytes.fromhex("22" * 32) +OPERATOR_PUBKEY = _PK.from_secret(OPERATOR_SECRET).format().hex()[2:] + OUTBOUND_MSGID = "" -STUB_CFG = SimpleNamespace(mail_from="agent@cayce-co.xyz") +# the signing gate reads env at runtime — make sure every path in this module +# (fixtures included) sees a signed deployment +os.environ.setdefault("CAYCE_OPS_DEPLOYMENT_ID", "itest") +os.environ.setdefault("CAYCE_OPS_OPERATOR_PUBKEY", OPERATOR_PUBKEY) +from cayceops.config import load as _load # noqa: E402 (env set above) + +STUB_CFG = _load() # a real, signed Config (replaces the old SimpleNamespace) + def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes: @@ -34,9 +48,24 @@ def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes: return "\r\n".join(lines).encode() +def _sign(msg: bytes, secret: bytes = OPERATOR_SECRET) -> str: + + return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex() + + +def _approve_signed(sess, mail_id: int) -> str: + """Approve with a valid operator signature over the row's current content.""" + from cayceops.signing import approval_message, mail_content_hash + + record = sess.get(MailMessage, mail_id) + to_addr = record.from_addr if record.direction == MailDirection.INBOUND else record.to_addr + chash = mail_content_hash(to_addr, record.subject, record.draft_reply) + return _sign(approval_message("mail", "itest", mail_id, chash)) + + def _proposal_with_sent_draft(sess) -> Proposal: """A proposal whose first-contact mail went through the REAL path: - compose -> approve -> send (creating a true OUTBOUND row).""" + compose -> approve (signed) -> send (creating a true OUTBOUND row).""" p = Proposal( kind="partnership", title="Acme hotel affiliate program", @@ -47,7 +76,10 @@ def _proposal_with_sent_draft(sess) -> Proposal: mail_id = maild._compose( p.id, "partners@acme.example.org", "[cayce] partnership inquiry", "Hello!", sess ) - assert maild._approve_draft(mail_id, "gustavo", sess) == 0 + assert ( + maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id)) + == 0 + ) assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent" out = ( sess.query(MailMessage) @@ -72,8 +104,15 @@ def test_rules_classification(cfg): clf.classify("Your application was approved, welcome to the program", config)[0] is MailClassification.APPROVAL ) + # negation present -> approval is vetoed to needs_review (round-2 rule; + # the human decides between rejection and other) assert ( clf.classify("Unfortunately we cannot accept new partners", config)[0] + is MailClassification.NEEDS_REVIEW + ) + # plain rejection (no approval word involved) still classifies directly + assert ( + clf.classify("We regret to inform you that your application was declined", config)[0] is MailClassification.REJECTION ) assert ( @@ -84,10 +123,16 @@ def test_rules_classification(cfg): assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW -def test_negated_approvals_are_rejections(cfg): - """Critical finding 4: 'cannot accept'/'not approved' read as approvals.""" +def test_negated_approvals_never_classify_approval(cfg): + """Round-2 point 2: any negation vetoes `approval` -> needs_review.""" config = cfg() cases = [ + "We cannot currently accept new affiliates", + "We will never accept crypto businesses", + "Approval was not granted for your site", + "We are no longer accepting applications", + "Your application hasn't been approved", + "We have decided not to move forward with your application", "We cannot accept your application", "We are not accepting new partners at this time", "Your application was not approved", @@ -95,9 +140,8 @@ def test_negated_approvals_are_rejections(cfg): "We do not accept external affiliates", ] for text in cases: - kind, confidence = clf.classify(text, config) - assert kind is MailClassification.REJECTION, f"{text!r} -> {kind}" - assert confidence >= 0.9 + kind, _ = clf.classify(text, config) + assert kind is not MailClassification.APPROVAL, f"{text!r} -> {kind}" def test_llm_prompt_format_survives_json_braces(): @@ -197,7 +241,12 @@ def test_send_requires_explicit_human_approval(cfg, sess): inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() original_id = inbound.message_id - assert maild._approve_draft(inbound.id, "gustavo", sess) == 0 + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) sent = [] config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz") @@ -237,7 +286,12 @@ def test_double_approve_is_rejected(cfg, sess): maild._poll(cfg(), sess, fetcher=_fetcher([raw])) inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() - assert maild._approve_draft(inbound.id, "gustavo", sess) == 0 + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) assert maild._approve_draft(inbound.id, "gustavo", sess) == 1 @@ -256,7 +310,10 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess): mail_id = maild._compose( p.id, f"desk{i}@acme.example.org", f"[cayce] inquiry {i}", "Hello!", sess ) - assert maild._approve_draft(mail_id, "gustavo", sess) == 0 + assert ( + maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id)) + == 0 + ) assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent" sess.flush() out = sess.get(MailMessage, mail_id) @@ -272,7 +329,12 @@ def test_send_failure_does_not_resurrect_sent_mails(cfg, sess): .filter_by(direction=MailDirection.INBOUND, message_id=f"") .one() ) - maild._approve_draft(inbound.id, "gustavo", sess) + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) sess.commit() # per-message factory sessions only see committed rows calls = {"n": 0} @@ -304,7 +366,12 @@ def test_edited_draft_refuses_to_send(cfg, sess): raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) maild._poll(cfg(), sess, fetcher=_fetcher([raw])) inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() - assert maild._approve_draft(inbound.id, "gustavo", sess) == 0 + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) inbound.draft_reply = "TOTALLY EDITED TEXT after approval" # tamper @@ -314,7 +381,7 @@ def test_edited_draft_refuses_to_send(cfg, sess): assert stats["sent"] == 0 and stats["hash_mismatch"] == 1 assert sent == [] assert inbound.draft_status is DraftStatus.DRAFTED # needs fresh approval - action = sess.query(AuditEntry).filter_by(action="send_refused_hash_mismatch").one() + action = sess.query(AuditEntry).filter_by(action="send_refused").one() assert action.result == "error" @@ -334,7 +401,10 @@ def test_compose_creates_first_contact_draft(cfg, sess): assert out.proposal_id == p.id # approve + send -> an inbound reply can now thread to this proposal - assert maild._approve_draft(mail_id, "gustavo", sess) == 0 + assert ( + maild._approve_draft(mail_id, "gustavo", sess, signature=_approve_signed(sess, mail_id)) + == 0 + ) stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None) assert stats["sent"] == 1 sess.flush() @@ -349,3 +419,112 @@ def test_mark_seen_returned_for_post_commit_processing(cfg, sess): stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw])) assert stats["processed"] == 1 assert nums == [b"0"] # caller marks seen only after commit + + +# --- round-2 signature gate (mail) ------------------------------------------------ + + +def _approved_reply(sess, cfg) -> MailMessage: + _proposal_with_sent_draft(sess) + raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + return sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + + +def test_unsigned_sql_approved_draft_is_refused(cfg, sess): + """Round-2 required test: raw-SQL approved row (no signature) not sent.""" + _proposal_with_sent_draft(sess) + raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + # raw-SQL style approval: status flips, no signature + sess.execute( + MailMessage.__table__.update() + .where(MailMessage.id == inbound.id) + .values(draft_status=DraftStatus.APPROVED) + ) + sess.expire(inbound) + + sent = [] + stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a)) + assert stats["sent"] == 0 and stats["hash_mismatch"] == 1 + assert sent == [] + assert inbound.draft_status is DraftStatus.DRAFTED # reverted + refused = sess.query(AuditEntry).filter_by(action="send_refused").one() + assert refused.result == "error" + + +def test_changed_to_addr_after_approval_blocks_send(cfg, sess): + """Round-2 required test: to_addr edited after approval -> refused.""" + _proposal_with_sent_draft(sess) + raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) + + inbound.from_addr = "attacker@evil.example.org" # redirect target + + sent = [] + stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: sent.append(a)) + assert stats["sent"] == 0 and stats["hash_mismatch"] == 1 + assert sent == [] + assert inbound.draft_status is DraftStatus.DRAFTED + + +def test_foreign_deployment_signature_refused(cfg, sess): + """Round-2 required test: a valid signature for another deployment_id (or + other content) is refused.""" + from cayceops.signing import approval_message, mail_content_hash + + _proposal_with_sent_draft(sess) + raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + chash = mail_content_hash(inbound.from_addr, inbound.subject, inbound.draft_reply) + + # signature valid for ANOTHER deployment + other_dep = _sign( + OPERATOR_SECRET, approval_message("mail", "other-deploy", inbound.id, chash) + ) + assert maild._approve_draft(inbound.id, "gustavo", sess, signature=other_dep) == 2 + assert inbound.draft_status is DraftStatus.DRAFTED + + # signature valid for OTHER CONTENT + other_content = _sign( + OPERATOR_SECRET, approval_message("mail", "itest", inbound.id, "00" * 32) + ) + assert maild._approve_draft(inbound.id, "gustavo", sess, signature=other_content) == 2 + assert inbound.draft_status is DraftStatus.DRAFTED + + +def test_send_failures_audited_and_exit_code(cfg, sess, monkeypatch): + """Round-2 point 5: failures audited + surfaced; non-zero exit handled by CLI.""" + import cayceops.db as appdb + + _proposal_with_sent_draft(sess) + raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + assert ( + maild._approve_draft( + inbound.id, "gustavo", sess, signature=_approve_signed(sess, inbound.id) + ) + == 0 + ) + + def boom(cfg, to, subject, body, message_id, in_reply_to=None): + raise OSError("relay down") + + + sess.commit() # per-message factory sessions only see committed rows + stats = maild._send_approved( + STUB_CFG, sess, sender=boom, factory=appdb.session_factory() + ) + assert stats["error"] == 1 and stats["sent"] == 0 + failed = sess.query(AuditEntry).filter_by(action="send_failed").one() + assert failed.result == "error" diff --git a/tests/test_models.py b/tests/test_models.py index 0b1fe79..41f7828 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -110,3 +110,35 @@ def test_audit_chain_and_verification(sess): first.actor = "tampered" ok, broken = verify_audit_chain(sess) assert not ok and broken == first.id + + +def test_concurrent_writers_keep_chain_verifiable(db_engine): + """Round-2 point 4: serialized writes -> no forked chains.""" + import threading + + import cayceops.db as appdb + + errors: list[Exception] = [] + + def writer(worker: int, n: int): + try: + for i in range(n): + with appdb.session_scope() as s: + audit(s, actor=f"w{worker}", action=f"step{i}", detail={"worker": worker}) + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + threads = [threading.Thread(target=writer, args=(w, 25)) for w in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, errors + s = appdb.session_factory()() + total = s.query(AuditEntry).count() + assert total == 100 + from cayceops.models import verify_audit_chain + + ok, broken = verify_audit_chain(s) + assert ok and broken is None diff --git a/tests/test_publisher.py b/tests/test_publisher.py index 2f7374a..d0d79f3 100644 --- a/tests/test_publisher.py +++ b/tests/test_publisher.py @@ -1,16 +1,30 @@ -"""publisher tests — calendar, human gate, nostr publish, disabled adapters.""" +"""publisher tests — calendar, human gate (signed), nostr publish, gates.""" from __future__ import annotations -import os +from datetime import UTC, datetime, timedelta import pytest from cayceops.components.publisher import adapters, nostr, publisher from cayceops.models import AuditEntry, ContentItem, ContentStatus -# invented test key (never a real nostr key — allowlisted in .gitleaks.toml) -TEST_KEY = "11" * 32 +# invented test keys (never real — allowlisted in .gitleaks.toml) +TEST_KEY = "11" * 32 # nostr signing key +OPERATOR_SECRET = bytes.fromhex("22" * 32) # approval signing key + + +def _sign(msg: bytes, secret: bytes = OPERATOR_SECRET) -> str: + from coincurve import PrivateKey + + return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex() + + +def _sign_content(item: ContentItem, deployment: str = "itest") -> str: + from cayceops.signing import approval_message, content_content_hash + + chash = content_content_hash(item.channel, item.caption, item.link) + return _sign(approval_message("content", deployment, item.id, chash)) def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED): @@ -24,6 +38,16 @@ def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatu ) +def _approved_item(cfg, sess, **kw) -> ContentItem: + """A content item properly approved with a signature (env set first).""" + cfg() # sets the signing env before any approval + item = _item(status=ContentStatus.DRAFT, **kw) + sess.add(item) + sess.flush() + assert publisher._approve(item.id, "gustavo", sess, signature=_sign_content(item)) == 0 + return item + + # --- nostr client ------------------------------------------------------------- @@ -65,10 +89,16 @@ def test_approve_gate(cfg, sess): item = _item(status=ContentStatus.DRAFT) sess.add(item) sess.flush() + cfg() # signing env first - assert publisher._approve(item.id, "gustavo", sess) == 0 + # unsigned approval refused (mandatory signature, no degraded mode) + assert publisher._approve(item.id, "gustavo", sess, signature=None) == 2 + # signed approval accepted, FULL signature stored + sig = _sign_content(item) + assert publisher._approve(item.id, "gustavo", sess, signature=sig) == 0 assert item.status is ContentStatus.APPROVED - assert publisher._approve(item.id, "gustavo", sess) == 1 # not draft anymore + assert item.operator_signature == sig and len(sig) == 128 + assert publisher._approve(item.id, "gustavo", sess, signature=sig) == 1 # not draft anymore actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor assert actor == "gustavo" @@ -78,9 +108,7 @@ def test_approve_gate(cfg, sess): def test_publish_due_nostr_happy_path(cfg, sess): - item = _item() # approved, scheduled None -> due now - sess.add(item) - sess.flush() + item = _approved_item(cfg, sess) relays_seen = [] @@ -99,25 +127,69 @@ def test_publish_due_nostr_happy_path(cfg, sess): assert item.status is ContentStatus.PUBLISHED assert item.post_ref and len(item.post_ref) == 64 assert item.published_at is not None - assert "wss://relay.example.org" in relays_seen[0] # noqa: E501 + assert "wss://relay.example.org" in relays_seen[0] def test_future_scheduled_item_not_published(cfg, sess): - from datetime import UTC, datetime, timedelta - - item = _item(scheduled=datetime.now(UTC) + timedelta(hours=2)) - sess.add(item) - sess.flush() - config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org") - stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: []) + item = _approved_item( + cfg, sess, scheduled=datetime.now(UTC) + timedelta(hours=2) + ) + config = cfg( + CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org" + ) + stats = publisher._publish_due( + config, sess, nostr_sender=lambda r, e, timeout=10: [] + ) assert stats["published"] == 0 assert item.status is ContentStatus.APPROVED -def test_nostr_without_config_fails_clean(cfg, sess): - item = _item() +def test_unsigned_sql_approved_item_not_published(cfg, sess): + """Round-2 required test: raw-SQL approved row (no signature) not published.""" + config = cfg( + CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org" + ) + item = _item() # status=APPROVED, no signature sess.add(item) sess.flush() + stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: []) + assert stats["published"] == 0 and stats["tampered"] == 1 + assert item.status is ContentStatus.DRAFT # reverted + refused = sess.query(AuditEntry).filter_by(action="publish_refused").one() + assert refused.result == "error" + + +def test_tampered_caption_refused_by_signature(cfg, sess): + """Round-2 required test: signature over other content refused at publish.""" + item = _approved_item(cfg, sess) + + item.caption = item.caption + " EDITED" # content no longer matches sig + nostr_config = cfg( + CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org" + ) + stats = publisher._publish_due( + nostr_config, sess, nostr_sender=lambda r, e, timeout=10: [] + ) + assert stats["published"] == 0 and stats["tampered"] == 1 + assert item.status is ContentStatus.DRAFT + + +def test_foreign_deployment_signature_refused(cfg, sess, signer): + """Round-2 required test: a valid signature for another deployment_id is + refused at approval time.""" + item = _item(status=ContentStatus.DRAFT) + sess.add(item) + sess.flush() + other_dep = signer.raw( + signer.msg("content", "other-deploy", item.id, signer.content_hash(item)) + ) + assert publisher._approve(item.id, "gustavo", sess, signature=other_dep) == 2 + assert item.status is ContentStatus.DRAFT + + +def test_nostr_without_config_fails_clean(cfg, sess): + """Signed item but no nostr credentials -> clean failure, item stays.""" + item = _approved_item(cfg, sess) config = cfg() # no key, no relays stats = publisher._publish_due(config, sess) assert stats["failed"] == 1 @@ -126,14 +198,15 @@ def test_nostr_without_config_fails_clean(cfg, sess): def test_non_nostr_channels_skipped_terminal(cfg, sess): - item = _item(channel="facebook", theme="lightning-abroad") - sess.add(item) - sess.flush() + item = _approved_item(cfg, sess, channel="facebook", theme="lightning-abroad") stats = publisher._publish_due(cfg(), sess) assert stats["skipped"] == 1 assert item.status is ContentStatus.SKIPPED # terminal (audit fix) - skip = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").one() - assert skip.detail["channel"] == "facebook" + + stats2 = publisher._publish_due(cfg(), sess) + assert stats2["skipped"] == 0 # never revisited + skips = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").count() + assert skips == 1 def test_meta_adapters_disabled_by_default(cfg): @@ -164,61 +237,27 @@ def test_unknown_brand_rejected(): images.render_brand_image(brand="xx", kicker=None, lines=["A"], out_path="/tmp/x.jpg") -# --- audit corrections ---------------------------------------------------------- - - -def test_publish_refuses_tampered_caption(cfg, sess): - """Medium finding: approved caption was not bound to content.""" - item = _item(status=ContentStatus.DRAFT) - sess.add(item) - sess.flush() - assert publisher._approve(item.id, "gustavo", sess) == 0 - - item.caption = item.caption + " TAMPERED" - config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org") - stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: []) - - assert stats["published"] == 0 and stats["tampered"] == 1 - assert item.status is ContentStatus.APPROVED # stays, needs re-approval - action = sess.query(AuditEntry).filter_by(action="publish_refused_hash_mismatch").one() - assert action.result == "error" - - -def test_disabled_channel_is_terminal(cfg, sess): - """Medium finding: disabled channels wrote skip audits every 5 minutes.""" - item = _item(channel="facebook", theme="lightning-abroad") - sess.add(item) - sess.flush() - config = cfg() - - stats1 = publisher._publish_due(config, sess) - assert stats1["skipped"] == 1 - assert item.status is ContentStatus.SKIPPED # terminal - - stats2 = publisher._publish_due(config, sess) - assert stats2["skipped"] == 0 # never revisited - skips = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").count() - assert skips == 1 +# --- round-2: attempts budget -------------------------------------------------- def test_publish_attempts_bounded(cfg, sess, monkeypatch): """Medium finding: nostr failures retried without limit.""" from cayceops.components.publisher import nostr as nostr_mod - item = _item() - sess.add(item) - sess.flush() - - def always_fail(relays, event, timeout=10.0): - raise RuntimeError("relay down") - - monkeypatch.setattr(nostr_mod, "publish", always_fail) config = cfg( CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org", CAYCE_OPS_MAX_ATTEMPTS="1", ) + item = _approved_item(cfg, sess) + def always_fail(relays, event, timeout=10.0): + raise RuntimeError("relay down") + + monkeypatch.setattr(nostr_mod, "publish", always_fail) publisher._publish_due(config, sess, nostr_sender=always_fail) assert item.attempts == 1 assert item.status is ContentStatus.SKIPPED # budget exhausted -> terminal + + +import os # noqa: E402 (used by image test)