From 40453aadb4fb897fee00f276c9b2ffea10cb84fe Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:20:09 +0000 Subject: [PATCH 1/7] audit-1 (critical 1): mandatory operator signatures bound to content+deployment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - signing module: canonical content hashes + approval_message sha256(cayce-ops::::) - approve/approve-draft/publisher-approve REQUIRE signature; without CAYCE_OPS_OPERATOR_PUBKEY+DEPLOYMENT_ID the daemons refuse to execute/send/publish (no degraded mode) - run-time re-verification at execute/send/publish: unsigned rows (raw SQL approval) or edited content revert to pre-approval state, audited as error - FULL signatures stored on rows (operator_signature, migration 0003) and in audit detail — never truncated - signature failures audited from CLI wrappers too (1f) - AUTONOMOUS_MODE approval path removed (D13): no non-human approvals tests: SQL-approved row refused at execute/send/publish; to_addr change blocks send; foreign deployment/content signatures refused; full-signature persistence. 65 tests green. --- alembic/versions/0003_operator_signatures.py | 42 ++++ cayceops/cli.py | 38 +++- cayceops/components/executor/executor.py | 163 ++++++++++--- cayceops/components/maild/maild.py | 197 ++++++++++++---- cayceops/components/publisher/publisher.py | 119 +++++++++- cayceops/config.py | 12 +- cayceops/models.py | 100 ++++++-- cayceops/signing.py | 103 +++++++++ tests/conftest.py | 65 ++++++ tests/test_executor.py | 227 ++++++++++++------- tests/test_maild.py | 211 +++++++++++++++-- tests/test_models.py | 32 +++ tests/test_publisher.py | 173 ++++++++------ 13 files changed, 1209 insertions(+), 273 deletions(-) create mode 100644 alembic/versions/0003_operator_signatures.py create mode 100644 cayceops/signing.py 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) From 903c591a066c4870196ba9f6b6f111464dd1813c Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:20:26 +0000 Subject: [PATCH 2/7] =?UTF-8?q?audit-2=20(critical=204):=20approval-veto?= =?UTF-8?q?=20classifier=20=E2=80=94=20any=20negation=20vetoes=20approval?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces per-pattern negation lists: if ANY negation token (not, no, never, cannot, cannot, will not, has not, is not, no longer, unable, decline, regret, unfortunately) appears anywhere in the mail, an approval classification downgrades to needs_review. Approval requires a mail with zero negations. The LLM prompt braces stay escaped (round-1 fix retained). tests: the six mandated sentences all classify non-approval. --- cayceops/components/maild/classify.py | 49 ++++++++++++++------------- 1 file changed, 26 insertions(+), 23 deletions(-) diff --git a/cayceops/components/maild/classify.py b/cayceops/components/maild/classify.py index e2eda54..3cfa658 100644 --- a/cayceops/components/maild/classify.py +++ b/cayceops/components/maild/classify.py @@ -1,8 +1,12 @@ -"""maild classification: rules first, LLM optional, conservative threshold. +"""maild classification: rules + sentence-level negation, LLM optional. -The LLM never decides alone: any result below ``CAYCE_OPS_LLM_REVIEW_THRESHOLD`` -confidence becomes ``needs_review``. With the ``null`` provider (default) only -deterministic rules run. +Approval rule (audit round 2, point 2): a mail may only classify `approval` +when NO sentence contains a negation. Any negation token anywhere in the mail +(conservative: even outside the approval sentence) downgrades an approval to +`needs_review`. Rejections/questions/other are unaffected. + +The LLM never decides alone: below-threshold confidence → `needs_review`. +With the `null` provider (default) only deterministic rules run. """ from __future__ import annotations @@ -13,25 +17,16 @@ import re from cayceops.config import Config from cayceops.models import MailClassification -# NEGATED approval forms are rejections and must be checked FIRST: -# "We cannot accept your application", "not accepting new partners", -# "Your application was not approved" used to match the bare approval -# keywords (accept/approv) and read as approvals — audit finding 4. -_NEGATED_APPROVAL = ( - MailClassification.REJECTION, - re.compile( - r"\b(cannot|can.?t|will not|won.?t|do not|don.?t|does not|doesn.?t)\s+" - r"(be\s+able\s+to\s+)?(accept|approve|proceed|offer|welcome)" - r"|\bnot\s+(been\s+)?approved\b" - r"|\bnot\s+accepting\b" - r"|\bunfortunately\b", - re.I, - ), - 0.95, +_APPROVAL_WORDS = re.compile(r"\b(approv|accept|welcome)", re.I) + +# any of these tokens anywhere in the mail vetoes an `approval` result +_NEGATIONS = re.compile( + r"\b(not|no|never|cannot|can'?t|won'?t|hasn'?t|isn'?t|" + r"no\s+longer|unable|decline\w*|regret\w*|unfortunately)\b", + re.I, ) _RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [ - _NEGATED_APPROVAL, # (class, pattern, confidence when matched) ( MailClassification.REJECTION, @@ -51,9 +46,6 @@ _RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [ ), ] -# braces are DOUBLED so .format() does not die on the literal JSON sample -# (the old prompt raised KeyError on every call and the generic except -# silently degraded to rules — audit finding 4) _LLM_PROMPT = """Classify this merchant email into exactly one of: approval, rejection, question, other. Reply with strict JSON: {{"classification": "...", "confidence": 0.0-1.0}} @@ -69,6 +61,10 @@ def _rules(text: str) -> tuple[MailClassification, float]: return MailClassification.OTHER, 0.3 +def _has_negation(text: str) -> bool: + return bool(_NEGATIONS.search(text)) + + def _llm_classify(text: str, cfg: Config) -> tuple[MailClassification, float] | None: """Ask the configured LLM. Returns None on any failure (rules take over).""" from cayceops.llm import get_provider @@ -100,6 +96,13 @@ def classify(text: str, cfg: Config) -> tuple[MailClassification, float]: else: kind, confidence = _rules(text) + # APPROVAL VETO (audit round 2): any negation anywhere in the mail + # downgrades an approval to needs_review. Approval requires a mail with + # zero negations — "cannot currently accept", "never accept", + # "hasn't been approved", "no longer accepting" all land here. + if kind is MailClassification.APPROVAL and _has_negation(text): + return MailClassification.NEEDS_REVIEW, min(confidence, 0.6) + if confidence < cfg.llm_review_threshold: return MailClassification.NEEDS_REVIEW, confidence return kind, confidence From d77409148c3913b245b4b62f6e83cd9b9be9261c Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:20:26 +0000 Subject: [PATCH 3/7] audit-3 (critical 1 deploy half): roles.sql rewritten + PG integration test - one least-privilege role per daemon; cayce_approver is the ONLY writer of status=APPROVED / operator_signature / approved columns - per-table BEFORE triggers reject approval writes from daemon roles - no literal passwords: refuses to run without the six -v password vars (psql exits 3 via missing-table SELECT guard) - idempotent: revokes + drops trigger/function before recreating - tests/integration/test_pg_roles.py: applies roles.sql via psql -v, runs a full cycle per daemon role (propose -> approve -> execute; compose -> approve -> send; plan -> approve -> publish; anchor) and asserts the denials; runs the alembic upgrade->downgrade->upgrade round-trip on Postgres (point 6). Skips cleanly without a reachable PG. --- deploy/roles.sql | 198 ++++++++++++++++++-- tests/integration/__init__.py | 0 tests/integration/test_pg_roles.py | 289 +++++++++++++++++++++++++++++ 3 files changed, 467 insertions(+), 20 deletions(-) create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/test_pg_roles.py diff --git a/deploy/roles.sql b/deploy/roles.sql index 2107940..760f0f0 100644 --- a/deploy/roles.sql +++ b/deploy/roles.sql @@ -1,22 +1,180 @@ --- Per-daemon least-privilege roles (audit critical finding 1, deployment half). --- The app-level chain + signatures are in code; this locks the DB layer. --- Apply on the production Postgres; adjust the connection limits. -CREATE ROLE cayce_scanner LOGIN PASSWORD 'set-from-env'; -CREATE ROLE cayce_executor LOGIN PASSWORD 'set-from-env'; -CREATE ROLE cayce_maild LOGIN PASSWORD 'set-from-env'; -CREATE ROLE cayce_publisher LOGIN PASSWORD 'set-from-env'; -CREATE ROLE cayce_monitor LOGIN PASSWORD 'set-from-env'; +-- cayce-ops — least-privilege per-daemon roles (audit round 2, point 3) +-- +-- Apply with psql and MANDATORY password variables; the script refuses to run +-- without them (no literal credentials): +-- +-- psql -h 127.0.0.1 -U postgres -d cayce_ops -f deploy/roles.sql \ +-- -v scanner_password='…' -v executor_password='…' \ +-- -v maild_password='…' -v publisher_password='…' \ +-- -v monitor_password='…' -v approver_password='…' +-- +-- Model: +-- cayce_scanner : INSERT proposals only (+ INSERT audit_log, daemon_runs) +-- cayce_executor : UPDATE proposals WITHOUT approved/signature columns +-- cayce_maild : UPDATE mail_messages WITHOUT approved/signature columns +-- cayce_publisher : INSERT content_items, UPDATE WITHOUT approved/signature +-- cayce_monitor : SELECT everything (+ INSERT audit_log for anchors) +-- cayce_approver : THE ONLY role that can write status='approved', +-- approved_hash, operator_signature, approved_by/at +-- +-- A BEFORE trigger rejects status='approved' writes from any role except +-- cayce_approver — column grants alone cannot express that condition. +-- +-- NOTE: do NOT apply on production yourself; the operator does. -GRANT SELECT ON proposals, audit_log, mail_messages, content_items, daemon_runs TO cayce_monitor; -GRANT SELECT, INSERT, UPDATE ON proposals TO cayce_scanner; -- no DELETE -GRANT INSERT ON audit_log TO cayce_scanner; -- append-only -GRANT SELECT, UPDATE ON proposals TO cayce_executor; -- status transitions -GRANT INSERT ON audit_log TO cayce_executor; -- append-only -GRANT SELECT, INSERT, UPDATE ON mail_messages, proposals TO cayce_maild; -GRANT INSERT ON audit_log TO cayce_maild; -GRANT SELECT, UPDATE ON content_items TO cayce_publisher; -GRANT INSERT ON audit_log TO cayce_publisher; -GRANT SELECT ON daemon_runs TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher; +\set ON_ERROR_STOP on --- sequences -GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor; +\if :{?scanner_password} +\else +\echo 'ERROR: missing variables. Pass -v scanner_password=… executor_password=… maild_password=… publisher_password=… monitor_password=… approver_password=…' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif +\if :{?executor_password} +\else +\echo 'ERROR: missing executor_password' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif +\if :{?maild_password} +\else +\echo 'ERROR: missing maild_password' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif +\if :{?publisher_password} +\else +\echo 'ERROR: missing publisher_password' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif +\if :{?monitor_password} +\else +\echo 'ERROR: missing monitor_password' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif +\if :{?approver_password} +\else +\echo 'ERROR: missing approver_password' +SELECT * FROM _roles_sql_missing_required_password_variables; +\endif + +-- idempotency: detach old incarnations before dropping +DROP FUNCTION IF EXISTS cayce_ops_approval_gate() CASCADE; +REVOKE ALL ON SCHEMA public FROM + cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; +REVOKE ALL ON ALL TABLES IN SCHEMA public FROM + cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; +REVOKE ALL ON ALL SEQUENCES IN SCHEMA public FROM + cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; + +DROP ROLE IF EXISTS cayce_scanner; +DROP ROLE IF EXISTS cayce_executor; +DROP ROLE IF EXISTS cayce_maild; +DROP ROLE IF EXISTS cayce_publisher; +DROP ROLE IF EXISTS cayce_monitor; +DROP ROLE IF EXISTS cayce_approver; + +CREATE ROLE cayce_scanner LOGIN PASSWORD :'scanner_password'; +CREATE ROLE cayce_executor LOGIN PASSWORD :'executor_password'; +CREATE ROLE cayce_maild LOGIN PASSWORD :'maild_password'; +CREATE ROLE cayce_publisher LOGIN PASSWORD :'publisher_password'; +CREATE ROLE cayce_monitor LOGIN PASSWORD :'monitor_password'; +CREATE ROLE cayce_approver LOGIN PASSWORD :'approver_password'; + +GRANT USAGE ON SCHEMA public TO cayce_scanner, cayce_executor, cayce_maild, + cayce_publisher, cayce_monitor, cayce_approver; + +-- sequences everybody needs +GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public + TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; + +-- common reads +GRANT SELECT ON proposals, audit_log, mail_messages, content_items, daemon_runs + TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; + +-- every daemon appends to audit_log and daemon_runs (never UPDATE/DELETE) +GRANT INSERT ON audit_log + TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor, cayce_approver; +GRANT INSERT ON daemon_runs + TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor; + +-- scanner: propose only +GRANT INSERT ON proposals TO cayce_scanner; + +-- executor: transition proposals WITHOUT touching approval columns +GRANT UPDATE (status, attempts, updated_at) ON proposals TO cayce_executor; +-- (there is no error column; adjust if added) — see trigger below for status + +-- maild: classify + drafts + send bookkeeping WITHOUT approval columns +GRANT INSERT, UPDATE (classification, confidence, draft_status, draft_reply, + in_reply_to, message_id, from_addr, to_addr, subject, body, processed_at) + ON mail_messages TO cayce_maild; + +-- publisher: plan drafts and record publishes WITHOUT approval columns +GRANT INSERT ON content_items TO cayce_publisher; +GRANT UPDATE (status, attempts, error, media_path, scheduled_for, published_at, post_ref) + ON content_items TO cayce_publisher; + +-- approver: the ONLY writer of approvals and signatures +GRANT SELECT, UPDATE ON proposals TO cayce_approver; +GRANT SELECT, UPDATE ON mail_messages TO cayce_approver; +GRANT SELECT, UPDATE ON content_items TO cayce_approver; +GRANT INSERT ON audit_log TO cayce_approver; + +-- --------------------------------------------------------------------------- +-- approval gate: status/draft_status='approved' and signature columns may +-- only be written by cayce_approver (one trigger per table — column sets differ) +-- --------------------------------------------------------------------------- +CREATE OR REPLACE FUNCTION cayce_ops_gate_proposals() RETURNS trigger AS $$ +BEGIN + IF current_user IN ('cayce_approver', 'postgres') THEN + RETURN NEW; + END IF; + IF NEW.status::text = 'APPROVED' THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may set proposals.status=approved (user=%)', current_user; + END IF; + IF TG_OP = 'UPDATE' AND NEW.operator_signature IS DISTINCT FROM OLD.operator_signature THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may write operator_signature (user=%)', current_user; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION cayce_ops_gate_mail() RETURNS trigger AS $$ +BEGIN + IF current_user IN ('cayce_approver', 'postgres') THEN + RETURN NEW; + END IF; + IF NEW.draft_status::text = 'APPROVED' THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may set mail draft_status=approved (user=%)', current_user; + END IF; + IF TG_OP = 'UPDATE' AND NEW.operator_signature IS DISTINCT FROM OLD.operator_signature THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may write operator_signature (user=%)', current_user; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION cayce_ops_gate_content() RETURNS trigger AS $$ +BEGIN + IF current_user IN ('cayce_approver', 'postgres') THEN + RETURN NEW; + END IF; + IF NEW.status::text = 'APPROVED' THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may set content status=approved (user=%)', current_user; + END IF; + IF TG_OP = 'UPDATE' AND NEW.operator_signature IS DISTINCT FROM OLD.operator_signature THEN + RAISE EXCEPTION 'cayce-ops: only cayce_approver may write operator_signature (user=%)', current_user; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS proposals_approval_gate ON proposals; +CREATE TRIGGER proposals_approval_gate BEFORE INSERT OR UPDATE ON proposals + FOR EACH ROW EXECUTE FUNCTION cayce_ops_gate_proposals(); + +DROP TRIGGER IF EXISTS mail_approval_gate ON mail_messages; +CREATE TRIGGER mail_approval_gate BEFORE INSERT OR UPDATE ON mail_messages + FOR EACH ROW EXECUTE FUNCTION cayce_ops_gate_mail(); + +DROP TRIGGER IF EXISTS content_approval_gate ON content_items; +CREATE TRIGGER content_approval_gate BEFORE INSERT OR UPDATE ON content_items + FOR EACH ROW EXECUTE FUNCTION cayce_ops_gate_content(); diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_pg_roles.py b/tests/integration/test_pg_roles.py new file mode 100644 index 0000000..398ca91 --- /dev/null +++ b/tests/integration/test_pg_roles.py @@ -0,0 +1,289 @@ +"""Integration tests against Postgres 16: roles.sql + one full daemon cycle +per role (audit round 2, point 3). + +Skipped unless a Postgres is reachable at TEST_PG_DSN (default: the +docker-compose service). Roles are applied by piping deploy/roles.sql +through psql WITH the mandatory -v password variables — exercising the same +path an operator uses; the test skips if the psql client is not installed. + +Run: uv run pytest -m integration -v +""" + +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.integration + +PG_DSN = os.environ.get( + "TEST_PG_DSN", + "postgresql://cayce:ci-only-not-a-secret@127.0.0.1:5432/cayce_ops", +) +ROLES_SQL = Path(__file__).resolve().parents[2] / "deploy" / "roles.sql" + +ROLE_PASSWORDS = { + "scanner_password": "itest-scanner-pw", + "executor_password": "itest-executor-pw", + "maild_password": "itest-maild-pw", + "publisher_password": "itest-publisher-pw", + "monitor_password": "itest-monitor-pw", + "approver_password": "itest-approver-pw", +} + + +def _pg_available() -> bool: + try: + import psycopg + + with psycopg.connect(PG_DSN, connect_timeout=3): + return True + except Exception: # noqa: BLE001 + return False + + +def _psql_available() -> bool: + try: + return ( + subprocess.run( + ["psql", "--version"], capture_output=True, timeout=10 + ).returncode + == 0 + ) + except (OSError, subprocess.SubprocessError): + return False + + +@pytest.fixture(scope="module") +def roles_applied(): + if not _pg_available(): + pytest.skip("Postgres not reachable (set TEST_PG_DSN)") + if not _psql_available(): + pytest.skip("psql client not installed") + + # fresh slate on an admin connection + import psycopg + + admin = psycopg.connect(PG_DSN, autocommit=True) + with admin.cursor() as cur: + cur.execute( + """ + DROP TABLE IF EXISTS audit_log, mail_messages, content_items, + daemon_runs, proposals, alembic_version CASCADE; + DROP TYPE IF EXISTS proposalsource, proposalstatus, maildirection, + mailclassification, draftstatus, contentstatus CASCADE; + DROP FUNCTION IF EXISTS cayce_ops_approval_gate() CASCADE; + """ + ) + admin.close() + + # apply alembic chain (imports app; config must point at PG first) + os.environ["CAYCE_OPS_DATABASE_URL"] = PG_DSN.replace( + "postgresql://", "postgresql+psycopg://" + ) + import subprocess as sp + + env = {**os.environ, "CAYCE_OPS_DATABASE_URL": os.environ["CAYCE_OPS_DATABASE_URL"]} + # upgrade -> downgrade -> upgrade round-trip on Postgres (audit round 2, + # point 6): exercises 0002.downgrade and 0003.downgrade for real + sp.run(["uv", "run", "alembic", "upgrade", "head"], check=True, env=env) + sp.run(["uv", "run", "alembic", "downgrade", "0001"], check=True, env=env) + sp.run(["uv", "run", "alembic", "upgrade", "head"], check=True, env=env) + + # apply roles.sql the operator way: psql -v + cmd = ["psql", PG_DSN, "-v", "ON_ERROR_STOP=1", "-f", str(ROLES_SQL)] + for var, value in ROLE_PASSWORDS.items(): + cmd += ["-v", f"{var}={value}"] + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + assert result.returncode == 0, f"roles.sql failed:\n{result.stderr}" + + # re-apply WITHOUT variables must fail (audit requirement: no silent run) + bad = subprocess.run( + ["psql", PG_DSN, "-v", "ON_ERROR_STOP=1", "-f", str(ROLES_SQL)], + capture_output=True, + text=True, + timeout=60, + ) + assert bad.returncode != 0, "roles.sql must refuse to run without vars" + + yield + + +def _connect_as(role: str): + """Connect with the role's credentials on the same DB as PG_DSN.""" + import psycopg + + key = f"{role.split('_', 1)[1]}_password" + after_scheme = PG_DSN.split("://", 1)[1] + userinfo, hostport_db = after_scheme.split("@", 1) + hostport, _db = hostport_db.split("/", 1) + dsn = f"postgresql://{role}:{ROLE_PASSWORDS[key]}@{hostport}/{_db}" + return psycopg.connect(dsn, autocommit=True) + + +def _must_fail(conn, sql: str, because: str): + import psycopg + + try: + with conn.cursor() as cur: + cur.execute(sql) + except psycopg.Error as exc: + assert because.lower() in str(exc).lower(), f"{because} — got: {exc}" + conn.rollback() + return + pytest.fail(f"expected failure: {because}") + + +def test_roles_full_daemon_cycle(roles_applied): + + # --- scanner role: proposes (INSERT proposals) --- + with _connect_as("cayce_scanner") as conn, conn.cursor() as cur: + cur.execute( + """INSERT INTO proposals (source, kind, title, url, detail, + match_score, status, attempts, created_at, updated_at) + VALUES ('SCANNER', 'partnership', 'Acme partner', NULL, '{}', + 0.9, 'PROPOSED', 0, now(), now()) RETURNING id""" + ) + proposal_id = cur.fetchone()[0] + cur.execute( + "INSERT INTO audit_log (ts, actor, action, detail, result) " + "VALUES (now(), 'daemon:scanner', 'propose', '{}', 'ok')" + ) + cur.execute( + "INSERT INTO daemon_runs (daemon, started_at, status, detail) " + "VALUES ('scanner', now(), 'ok', '{}')" + ) + + # scanner cannot approve: no UPDATE grant at all, and the trigger would + # still block status=APPROVED for non-approver roles + scanner = _connect_as("cayce_scanner") + _must_fail( + scanner, + f"UPDATE proposals SET status='APPROVED' WHERE id={proposal_id}", + because="permission denied", + ) + scanner.close() + + # executor cannot approve either (trigger), and cannot write signatures + executor = _connect_as("cayce_executor") + # the signature column is outside the executor's column grant: the whole + # UPDATE is denied before the trigger even runs + fake_sig = "ab" * 64 # invented 128-hex signature + _must_fail( + executor, + f"UPDATE proposals SET status='APPROVED', operator_signature='{fake_sig}' " + f"WHERE id={proposal_id}", + because="permission denied", + ) + # executor CAN run its cycle once the approver approved + with executor.cursor() as cur: + cur.execute( + f"UPDATE proposals SET status='RUNNING', attempts=attempts+1, " + f"updated_at=now() WHERE id={proposal_id}" + ) + cur.execute( + f"UPDATE proposals SET status='DONE', updated_at=now() WHERE id={proposal_id}" + ) + cur.execute( + "INSERT INTO daemon_runs (daemon, started_at, status, detail) " + "VALUES ('executor', now(), 'ok', '{}')" + ) + executor.close() + + # --- approver: the only role that can approve + sign --- + sig = "c9" * 64 # invented 128-hex signature for the permission test + with _connect_as("cayce_approver") as conn, conn.cursor() as cur: + cur.execute( + f"UPDATE proposals SET status='APPROVED', approved_by='gustavo', " + f"approved_at=now(), operator_signature='{sig}' WHERE id={proposal_id}" + ) + + # --- maild role: drafts + sends, cannot approve --- + with _connect_as("cayce_maild") as conn, conn.cursor() as cur: + cur.execute( + """INSERT INTO mail_messages (direction, message_id, from_addr, + subject, body, draft_reply, draft_status, confidence) + VALUES ('OUTBOUND', '', 'desk@acme.example.org', 'Hello', + 'Body', 'Body', 'DRAFTED', 0) RETURNING id""" + ) + mail_id = cur.fetchone()[0] + maild = _connect_as("cayce_maild") + _must_fail( + maild, + f"UPDATE mail_messages SET draft_status='APPROVED' WHERE id={mail_id}", + because="only cayce_approver may set mail draft_status=approved", + ) + _must_fail( + maild, + f"UPDATE mail_messages SET operator_signature='{sig}' WHERE id={mail_id}", + because="permission denied", # signature column outside maild's grant + ) + # the approver approves; then maild can mark it sent + with _connect_as("cayce_approver") as conn, conn.cursor() as cur: + cur.execute( + "UPDATE mail_messages SET draft_status='APPROVED', " + "approved_by='gustavo', approved_hash='" + + "a" * 64 + + f"', operator_signature='{sig}' WHERE id={mail_id}" + ) + with _connect_as("cayce_maild") as conn, conn.cursor() as cur: + cur.execute( + f"UPDATE mail_messages SET draft_status='SENT' WHERE id={mail_id}" + ) + maild.close() + + # --- publisher role: plans content, cannot approve --- + with _connect_as("cayce_publisher") as conn, conn.cursor() as cur: + cur.execute( + """INSERT INTO content_items (theme, channel, caption, status, + attempts, created_at) + VALUES ('atm-fees', 'nostr', 'caption', 'DRAFT', 0, now()) + RETURNING id""" + ) + item_id = cur.fetchone()[0] + publisher = _connect_as("cayce_publisher") + _must_fail( + publisher, + f"UPDATE content_items SET status='APPROVED' WHERE id={item_id}", + because="only cayce_approver may set content status=approved", + ) + # approver approves; publisher publishes + with _connect_as("cayce_approver") as conn, conn.cursor() as cur: + cur.execute( + "UPDATE content_items SET status='APPROVED', approved_hash='" + + "b" * 64 + + f"', operator_signature='{sig}' WHERE id={item_id}" + ) + with _connect_as("cayce_publisher") as conn, conn.cursor() as cur: + cur.execute( + f"UPDATE content_items SET status='PUBLISHED', published_at=now() " + f"WHERE id={item_id}" + ) + publisher.close() + + # --- monitor role: read-only + audit anchors --- + monitor = _connect_as("cayce_monitor") + with monitor.cursor() as cur: + cur.execute("SELECT count(*) FROM proposals") + cur.fetchone() + cur.execute( + "INSERT INTO audit_log (ts, actor, action, detail, result) " + "VALUES (now(), 'daemon:monitor', 'anchor', '{}', 'ok')" + ) + # monitor CAN book its runs, but cannot touch proposals + with monitor.cursor() as cur: + cur.execute( + "INSERT INTO daemon_runs (daemon, started_at, status, detail) " + "VALUES ('monitor', now(), 'ok', '{}')" + ) + _must_fail( + monitor, + "INSERT INTO proposals (source, kind, title, detail, match_score, status, " + "attempts, created_at, updated_at) " + "VALUES ('SCANNER', 'other', 'x', '{}', 0, 'PROPOSED', 0, now(), now())", + because="permission denied", + ) + monitor.close() From ffe425320d367b2952ab18bd03eddb8ea98bc226 Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:20:56 +0000 Subject: [PATCH 4/7] audit-4: serialized audit chain + legacy genesis + daily Nostr anchoring - audit() writes are serialized: pg_advisory_xact_lock on Postgres; in-process RLock held until session commit/rollback on SQLite (test: 4 concurrent writers x 25 entries -> chain verifies, 100 rows) - legacy rows (entry_hash NULL, pre-0002) are skipped: the chain begins GENESIS at the first hashed row; migrated DBs verify clean by design - monitor anchor: publishes the last chain hash as a signed NIP-01 note once per day, logs it in the audit log (self-verified before publish) tests: concurrent writers; legacy-genesis verification; anchor signs, logs once per day. --- cayceops/components/monitor/monitor.py | 79 ++++++++++++++++++++++++++ tests/test_monitor.py | 39 +++++++++++++ 2 files changed, 118 insertions(+) diff --git a/cayceops/components/monitor/monitor.py b/cayceops/components/monitor/monitor.py index 2f2115d..cd61b5a 100644 --- a/cayceops/components/monitor/monitor.py +++ b/cayceops/components/monitor/monitor.py @@ -147,6 +147,85 @@ def _build(cfg: Config, sess) -> str: return cfg.monitor_output +def _anchor(cfg: Config, sess) -> int: + """Anchor core (injectable session for tests). Returns 0 ok / 0 skip.""" + from cayceops.components.publisher import nostr + from cayceops.config import ConfigError + from cayceops.models import AuditEntry, audit + + if not cfg.nostr_private_key or not cfg.nostr_relays: + raise ConfigError( + "anchor requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS" + ) + if not cfg.deployment_id: + raise ConfigError("anchor requires CAYCE_OPS_DEPLOYMENT_ID") + + last = ( + sess.query(AuditEntry) + .filter(AuditEntry.entry_hash.isnot(None)) + .order_by(AuditEntry.id.desc()) + .first() + ) + if last is None: + print("nothing to anchor: chain empty") + return 0 + today = datetime.now(UTC).date() + last_anchor = ( + sess.query(AuditEntry) + .filter(AuditEntry.action == "anchor") + .order_by(AuditEntry.id.desc()) + .first() + ) + if last_anchor is not None and last_anchor.ts.date() == today: + print("already anchored today") + return 0 + + secret = nostr.normalize_key(cfg.nostr_private_key) + event = nostr.build_event( + secret, + f"cayce-ops audit anchor deployment={cfg.deployment_id} " + f"last_hash={last.entry_hash} entries={last.id}", + tags=[["t", "cayce-ops"], ["t", "audit-anchor"]], + ) + if not nostr.verify_event(event): + raise RuntimeError("anchor event failed self-verification") + results = nostr.publish(cfg.nostr_relays, event) + audit( + sess, + actor="daemon:monitor", + action="anchor", + detail={ + "last_hash": last.entry_hash, + "last_entry": last.id, + "event_id": event["id"], + "relays": results, + "deployment": cfg.deployment_id, + }, + ) + print(f"anchored: {event['id']}") + return 0 + + +def anchor(cfg: Config | None = None) -> int: + """CLI: anchor once per day (external anchoring, audit round 2 point 4).""" + from cayceops.config import ConfigError, load + from cayceops.db import engine + + cfg = cfg or load() + try: + with session_scope() as sess: + try: + return _anchor(cfg, sess) + except ConfigError as exc: + print(f"anchor refused: {exc}") + return 2 + except Exception as exc: # noqa: BLE001 — CLI boundary + print(f"anchor failed: {str(exc)[:200]}") + return 1 + finally: + engine().dispose() + + def build(cfg: Config | None = None) -> int: """CLI entry: render the status page inside a DaemonRun bookkeeping wrap.""" from cayceops.config import load diff --git a/tests/test_monitor.py b/tests/test_monitor.py index a6e2222..5b8fa91 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -67,3 +67,42 @@ def test_build_escapes_injected_content(cfg, sess, tmp_path): page = out.read_text() assert "" not in page assert "<script>x</script>" in page + + +def test_anchor_signs_and_logs_once_per_day(cfg, sess, monkeypatch): + from cayceops.components.monitor import monitor + from cayceops.models import AuditEntry + + # seed the chain + from cayceops.models import audit as audit_fn + + audit_fn(sess, actor="daemon:executor", action="execute_done", detail={}) + sess.flush() + + published = [] + + def fake_publish(relays, event, timeout=10.0): + published.append(event) + return [{"relay": r, "ok": "true"} for r in relays] + + import cayceops.components.publisher.nostr as nostr_mod + + monkeypatch.setattr(nostr_mod, "publish", fake_publish) + + config = cfg( + CAYCE_OPS_NOSTR_PRIVATE_KEY="11" * 32, + CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org", + CAYCE_OPS_DEPLOYMENT_ID="itest", + ) + code = monitor._anchor(config, sess) + assert code == 0 + assert len(published) == 1 + content = published[0]["content"] + assert "audit anchor" in content and "itest" in content + entry = sess.query(AuditEntry).filter_by(action="anchor").one() + assert entry.detail["last_hash"] + + # second call same day -> skipped, no new nostr event + code2 = monitor._anchor(config, sess) + assert code2 == 0 + assert len(published) == 1 From 56b35d3a639b93804965a4798080e8d985c3bac6 Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:21:36 +0000 Subject: [PATCH 5/7] audit-5: CLI send path uses the tested code, failures audited, non-zero exit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Note: the send-path refactor (send_approved CLI -> _send_approved with factory, send_failed audits in their own transaction, DaemonRun record, non-zero exit on any failure class) landed within the audit-1 commit — maild.py was staged whole. This empty commit marks the point in the review trail; the behavior is covered by test_send_failures_audited_and_ exit_code. From c855f01da5ad716dc4eb25fc0b3db71555cbf6be Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:21:36 +0000 Subject: [PATCH 6/7] audit-6: CI runs migrations + integration against Postgres 16 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ci.yml gains a postgres:16 service; alembic upgrade/downgrade/upgrade round-trip runs on PG, unit tests run -m "not integration", and the roles.sql integration suite runs with a postgresql-client install. (Gate executed locally against Postgres 18.6 — see PR description.) --- .gitea/workflows/ci.yml | 34 +++++++++++++++++++++++++++++++--- .github/workflows/ci.yml | 34 +++++++++++++++++++++++++++++++--- 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index d400fef..f5524d0 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -8,6 +8,21 @@ on: jobs: lint-test-leaks: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: cayce + POSTGRES_PASSWORD: ci-only-not-a-secret + POSTGRES_DB: cayce_ops + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U cayce -d cayce_ops" + --health-interval 5s --health-timeout 3s --health-retries 10 + env: + CAYCE_OPS_DATABASE_URL: postgresql+psycopg://cayce:ci-only-not-a-secret@127.0.0.1:5432/cayce_ops + PGPASSWORD: ci-only-not-a-secret steps: - uses: actions/checkout@v4 with: @@ -23,11 +38,24 @@ jobs: - name: Lint (ruff) run: uv run ruff check . - - name: Migrations smoke test (SQLite) + - name: Migrations upgrade/downgrade round-trip (Postgres 16) + run: | + uv run alembic upgrade head + uv run alembic downgrade 0001 + uv run alembic upgrade head + uv run alembic downgrade base + uv run alembic upgrade head + + - name: Alembic smoke (SQLite) run: uv run alembic upgrade head - - name: Tests (pytest) - run: uv run pytest + - name: Unit tests (pytest) + run: uv run pytest -m "not integration" + + - name: Integration tests — roles.sql + daemon cycles (Postgres 16) + run: | + sudo apt-get update && sudo apt-get install -y postgresql-client + uv run pytest -m integration -v - name: Secret scan (gitleaks, full history) run: | diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d400fef..f5524d0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,6 +8,21 @@ on: jobs: lint-test-leaks: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: cayce + POSTGRES_PASSWORD: ci-only-not-a-secret + POSTGRES_DB: cayce_ops + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U cayce -d cayce_ops" + --health-interval 5s --health-timeout 3s --health-retries 10 + env: + CAYCE_OPS_DATABASE_URL: postgresql+psycopg://cayce:ci-only-not-a-secret@127.0.0.1:5432/cayce_ops + PGPASSWORD: ci-only-not-a-secret steps: - uses: actions/checkout@v4 with: @@ -23,11 +38,24 @@ jobs: - name: Lint (ruff) run: uv run ruff check . - - name: Migrations smoke test (SQLite) + - name: Migrations upgrade/downgrade round-trip (Postgres 16) + run: | + uv run alembic upgrade head + uv run alembic downgrade 0001 + uv run alembic upgrade head + uv run alembic downgrade base + uv run alembic upgrade head + + - name: Alembic smoke (SQLite) run: uv run alembic upgrade head - - name: Tests (pytest) - run: uv run pytest + - name: Unit tests (pytest) + run: uv run pytest -m "not integration" + + - name: Integration tests — roles.sql + daemon cycles (Postgres 16) + run: | + sudo apt-get update && sudo apt-get install -y postgresql-client + uv run pytest -m integration -v - name: Secret scan (gitleaks, full history) run: | From 3bde7852613a3768842d111519307cf4aec328f1 Mon Sep 17 00:00:00 2001 From: Cayce Date: Tue, 20 Oct 2026 02:21:36 +0000 Subject: [PATCH 7/7] audit-7 + docs: plan systemd units, D13-D15 - cayce-ops-plan.service/.timer (daily): the @ template cannot express publisher plan, so it gets its own unit - DECISIONS: D13 no non-human approvals; D14 signatures bind content + deployment; D15 negation veto classifier --- deploy/systemd/cayce-ops-plan.service | 15 +++++++++++++++ deploy/systemd/cayce-ops-plan.timer | 10 ++++++++++ docs/DECISIONS.md | 22 ++++++++++++++++++++++ 3 files changed, 47 insertions(+) create mode 100644 deploy/systemd/cayce-ops-plan.service create mode 100644 deploy/systemd/cayce-ops-plan.timer diff --git a/deploy/systemd/cayce-ops-plan.service b/deploy/systemd/cayce-ops-plan.service new file mode 100644 index 0000000..e779dc4 --- /dev/null +++ b/deploy/systemd/cayce-ops-plan.service @@ -0,0 +1,15 @@ +[Unit] +Description=cayce-ops publisher plan (fills the content calendar, daily) +After=network-online.target postgresql.service +Wants=network-online.target + +[Service] +Type=oneshot +User=cayce +WorkingDirectory=/srv/cayce-ops +EnvironmentFile=/srv/cayce-ops/.env +ExecStart=/srv/cayce-ops/.venv/bin/cayceops publisher plan +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=yes +ReadWritePaths=/srv/cayce-ops diff --git a/deploy/systemd/cayce-ops-plan.timer b/deploy/systemd/cayce-ops-plan.timer new file mode 100644 index 0000000..2684063 --- /dev/null +++ b/deploy/systemd/cayce-ops-plan.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Run cayce-ops publisher plan daily + +[Timer] +OnCalendar=*-*-* 06:00:00 +RandomizedDelaySec=300 +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 71ea27b..8380ee1 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -95,3 +95,25 @@ caption hash-bound at approval, disabled channels are terminal (`SKIPPED`), publish failures burn a bounded attempt budget before abandonment. `risk_score` renamed `match_score` (keyword relevance) and auto-approval takes the HIGHEST score first. + +## D13 — No non-human approval path (round 2, supersedes D5) +Operator signatures are mandatory for approvals AND re-verified at +execution/send/publish time against current content + deployment id. With +that gate, the AUTONOMOUS_MODE auto-approval path cannot exist: it would +produce unsigned approvals the gate refuses. `_auto_approve` is inert and +audits `autonomy_disabled` when invoked. *Cost:* AUTONOMOUS_MODE is now +inert by design. + +## D14 — Signatures bind content + deployment (round 2) +message = sha256("cayce-ops::::"). +content_hash is canonical-json of the fields that matter per type. Full +signatures live on the row (operator_signature) and in the audit detail. +Verification happens again at run time: unsigned rows (raw SQL approval) +revert to their pre-approval state with an error audit. *Cost:* a deployment +id env var is mandatory; signatures must be recomputed if content changes. + +## D15 — Approval classifier veto (round 2) +Any negation token anywhere in a mail vetoes an `approval` classification → +needs_review. Simpler and strictly conservative vs. per-pattern negation +lists. *Cost:* genuinely-positive approvals containing stray negations +(e.g. "we have no reason not to welcome you") need human triage.