diff --git a/.env.example b/.env.example index b863886..94dfb0d 100644 --- a/.env.example +++ b/.env.example @@ -21,6 +21,9 @@ CAYCE_OPS_AUTONOMOUS_KINDS= CAYCE_OPS_AUTONOMOUS_DAILY_CAP=0 CAYCE_OPS_MAX_ATTEMPTS=2 CAYCE_OPS_ENABLE_HTTP_ACTIONS=false +# when set (64-char hex x-only pubkey), executor approve REQUIRES a BIP-340 +# signature over sha256("cayce-ops:approve:") made by the operator key +CAYCE_OPS_OPERATOR_PUBKEY= # --- mail (IMAP read / SMTP send-approved-drafts-only) --- CAYCE_OPS_IMAP_HOST=imap.example.org diff --git a/Makefile b/Makefile index b7c6203..d4d033c 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: check lint test migrate-check leaks sync +.PHONY: check lint test migrate-check leaks sync hooks sync: ## install locked deps (creates .venv) uv sync --frozen @@ -17,5 +17,9 @@ migrate-check: ## smoke the alembic chain on a scratch sqlite db leaks: ## gitleaks over the full history gitleaks detect --source . --log-opts="--all" -v +hooks: ## install the pre-push gate as a git hook (part of setup) + git config core.hooksPath scripts/githooks + @echo "pre-push gate installed (scripts/githooks/pre-push)" + check: lint test migrate-check leaks ## the pre-push gate — must be green before every push @echo "check: OK" diff --git a/alembic/versions/0002_audit_hardening.py b/alembic/versions/0002_audit_hardening.py new file mode 100644 index 0000000..2d49127 --- /dev/null +++ b/alembic/versions/0002_audit_hardening.py @@ -0,0 +1,68 @@ +"""audit hardening migration — minimal standalone mirror of 0002. + +Revision ID: 0002 +Revises: 0001 +""" +from __future__ import annotations + +import sqlalchemy as sa + +from alembic import op + +revision = "0002" +down_revision = "0001" +branch_labels = None +depends_on = None + +OLD_CONTENT_STATUSES = ("DRAFT", "APPROVED", "PUBLISHED") +NEW_CONTENT_STATUSES = ("DRAFT", "APPROVED", "PUBLISHED", "SKIPPED") + + +def upgrade() -> None: + with op.batch_alter_table("audit_log") as batch: + batch.add_column(sa.Column("prev_hash", sa.String(length=64), nullable=True)) + batch.add_column(sa.Column("entry_hash", sa.String(length=64), nullable=True)) + batch.create_index("ix_audit_log_entry_hash", ["entry_hash"]) + + with op.batch_alter_table("proposals") as batch: + batch.alter_column("risk_score", new_column_name="match_score") + + with op.batch_alter_table("mail_messages") as batch: + batch.add_column(sa.Column("approved_hash", sa.String(length=64), nullable=True)) + batch.add_column(sa.Column("approved_by", sa.String(length=128), nullable=True)) + + if op.get_bind().dialect.name == "postgresql": + op.execute("ALTER TYPE contentstatus ADD VALUE IF NOT EXISTS 'SKIPPED'") + with op.batch_alter_table("content_items") as batch: + batch.alter_column( + "status", + existing_type=sa.Enum(*OLD_CONTENT_STATUSES, name="contentstatus"), + type_=sa.Enum(*NEW_CONTENT_STATUSES, name="contentstatus"), + existing_nullable=False, + ) + batch.add_column(sa.Column("attempts", sa.Integer(), nullable=False, server_default="0")) + batch.add_column(sa.Column("approved_hash", sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("content_items") as batch: + batch.drop_column("approved_hash") + batch.drop_column("attempts") + batch.alter_column( + "status", + existing_type=sa.Enum(*NEW_CONTENT_STATUSES, name="contentstatus"), + type_=sa.Enum(*OLD_CONTENT_STATUSES, name="contentstatus"), + existing_nullable=False, + ) + + with op.batch_alter_table("mail_messages") as batch: + batch.drop_column("approved_by") + batch.drop_column("approved_hash") + + with op.batch_alter_table("proposals") as batch: + batch.alter_column("match_score", new_column_name="risk_score") + + with op.batch_alter_table("audit_log") as batch: + batch.drop_index("ix_audit_log_entry_hash") + batch.drop_column("entry_hash") + batch.drop_column("prev_hash") diff --git a/cayceops/cli.py b/cayceops/cli.py index f06887f..eed4961 100644 --- a/cayceops/cli.py +++ b/cayceops/cli.py @@ -71,10 +71,16 @@ def executor_run() -> None: @executor_app.command("approve") def executor_approve( - proposal_id: int, actor: str = typer.Option(..., help="Human approver id") + 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)" + ), ) -> None: """Approve a proposal (human action, audited).""" - raise typer.Exit(_load_component("executor").approve(proposal_id, actor)) + raise typer.Exit( + _load_component("executor").approve(proposal_id, actor, signature=signature) + ) @executor_app.command("reject") @@ -102,10 +108,33 @@ def maild_approve(msg_id: int, actor: str = typer.Option(...)) -> None: @maild_app.command("send-approved") def maild_send() -> None: - """Send all approved drafts via SMTP.""" + """Send all approved drafts via SMTP (per-message commit).""" raise typer.Exit(_load_component("maild").send_approved()) +@maild_app.command("compose") +def maild_compose( + proposal_id: int, + to_addr: str = typer.Option(...), + subject: str = typer.Option(...), + body: str = typer.Option(...), +) -> None: + """Create the FIRST-CONTACT outbound draft for a proposal (drafted).""" + raise typer.Exit( + _load_component("maild").compose(proposal_id, to_addr, subject, body) + ) + + +@maild_app.command("run") +def maild_run() -> None: + """One daemon cycle: poll inbound, then send approved drafts.""" + comp = _load_component("maild") + code = comp.poll() + if code == 0: + code = comp.send_approved() + raise typer.Exit(code) + + # --- publisher --------------------------------------------------------------- publisher_app = typer.Typer(name="publisher", help="Themed content calendar") app.add_typer(publisher_app) @@ -135,6 +164,12 @@ def publisher_publish() -> None: raise typer.Exit(_load_component("publisher").publish_due()) +@publisher_app.command("run") +def publisher_run() -> None: + """One daemon cycle: publish due approved items (plan is cron'd separately).""" + raise typer.Exit(_load_component("publisher").publish_due()) + + # --- monitor ----------------------------------------------------------------- monitor_app = typer.Typer(name="monitor", help="Static status page") app.add_typer(monitor_app) @@ -146,6 +181,12 @@ def monitor_build() -> None: raise typer.Exit(_load_component("monitor").build()) +@monitor_app.command("run") +def monitor_run() -> None: + """One daemon cycle: build the status page.""" + raise typer.Exit(_load_component("monitor").build()) + + def main() -> None: # pragma: no cover app() diff --git a/cayceops/components/executor/executor.py b/cayceops/components/executor/executor.py index 0851b3e..894dbb6 100644 --- a/cayceops/components/executor/executor.py +++ b/cayceops/components/executor/executor.py @@ -29,12 +29,59 @@ from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, aud Actor = str +def _load_cfg(): + from cayceops.config import load + + return load() + + # --- human transitions ------------------------------------------------------- +# message signed by the operator key when CAYCE_OPS_OPERATOR_PUBKEY is set +def approval_message(proposal_id: int) -> bytes: + import hashlib -def approve(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int: - """Human approves a proposal. Audited with the operator's identity.""" - return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess) + return hashlib.sha256(f"cayce-ops:approve:{proposal_id}".encode()).digest() + + +def approve( + proposal_id: int, + actor: Actor, + cfg: Config | None = None, + 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 + + 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: + 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" + ) + }, + ) + if not ok: + return 2 + return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess, signature) def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int: @@ -48,6 +95,7 @@ def _decide( decision: ProposalStatus, cfg: Config | None, sess, + signature: str | None = None, ) -> int: owned = False if sess is None: @@ -70,11 +118,14 @@ def _decide( ) return 1 proposal.status = decision + detail: dict = {} if decision == ProposalStatus.APPROVED: proposal.approved_by = actor proposal.approved_at = datetime.now(UTC) + if signature: + detail["signature"] = signature[:32] + "..." verb = "approve" if decision == ProposalStatus.APPROVED else "reject" - audit(sess, actor=actor, action=verb, proposal_id=proposal_id) + audit(sess, actor=actor, action=verb, proposal_id=proposal_id, detail=detail) return 0 finally: if owned: @@ -102,13 +153,15 @@ def _auto_approve(cfg: Config, sess) -> int: approved = 0 if budget == 0: return 0 + # most relevant first (match_score = keyword relevance, NOT risk — the + # audit's naming finding; ascending order approved noise first) candidates = ( sess.query(Proposal) .filter( Proposal.status == ProposalStatus.PROPOSED, Proposal.kind.in_(cfg.autonomous_kinds), ) - .order_by(Proposal.risk_score.asc(), Proposal.id.asc()) + .order_by(Proposal.match_score.desc(), Proposal.id.asc()) .limit(budget) .all() ) @@ -124,7 +177,7 @@ def _auto_approve(cfg: Config, sess) -> int: detail={ "reason": "autonomous_mode policy", "kind": proposal.kind, - "risk_score": proposal.risk_score, + "match_score": proposal.match_score, "daily_cap": cfg.autonomous_daily_cap, }, ) diff --git a/cayceops/components/maild/classify.py b/cayceops/components/maild/classify.py index c8c06b1..e2eda54 100644 --- a/cayceops/components/maild/classify.py +++ b/cayceops/components/maild/classify.py @@ -13,12 +13,30 @@ 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, +) + _RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [ + _NEGATED_APPROVAL, # (class, pattern, confidence when matched) ( MailClassification.REJECTION, # noqa: E501 (kept on one line: the pattern IS the reviewed artifact) - re.compile(r"\b(declin|reject|not able to|unable to|unfortunately|we regret|won.t be able)\w*", re.I), # noqa: E501 + re.compile(r"\b(declin|reject|not able to|unable to|we regret|won.t be able)\w*", re.I), # noqa: E501 0.9, ), ( @@ -33,9 +51,12 @@ _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} +Reply with strict JSON: {{"classification": "...", "confidence": 0.0-1.0}} Subject and body: {body}""" diff --git a/cayceops/components/maild/maild.py b/cayceops/components/maild/maild.py index 34397e9..799bba4 100644 --- a/cayceops/components/maild/maild.py +++ b/cayceops/components/maild/maild.py @@ -35,6 +35,7 @@ from cayceops.models import ( MailDirection, MailMessage, Proposal, + _content_hash, audit, ) @@ -48,17 +49,33 @@ def connect_imap(cfg: Config) -> imaplib.IMAP4_SSL: return client -def _fetch_unread(cfg: Config) -> list[bytes]: - """Fetch raw unread messages. Returns list of raw RFC822 bytes.""" +def _fetch_unread(cfg: Config) -> list[tuple[bytes, bytes]]: + """Fetch raw unread messages as [(imap_num, raw_bytes)]. + + Messages are NOT marked \\Seen here — marking happens only after the + processed rows are committed (audit finding: a crash mid-batch used to + silently lose the rest of the batch). + """ client = connect_imap(cfg) try: _, data = client.search(None, "UNSEEN") - raw_messages: list[bytes] = [] + out: list[tuple[bytes, bytes]] = [] for num in data[0].split(): _, msg_data = client.fetch(num, "(RFC822)") - raw_messages.append(msg_data[0][1]) + out.append((num, msg_data[0][1])) + return out + finally: + client.logout() + + +def _mark_seen(cfg: Config, nums: list[bytes]) -> None: + """Mark processed messages \\Seen (called after commit).""" + if not nums: + return + client = connect_imap(cfg) + try: + for num in nums: client.store(num, "+FLAGS", "\\Seen") - return raw_messages finally: client.logout() @@ -96,7 +113,13 @@ def parse_raw(raw: bytes) -> dict: def _match_proposal(parsed: dict, sess) -> Proposal | None: - """Thread an inbound mail to its proposal via In-Reply-To of sent drafts.""" + """Thread an inbound mail to its proposal. + + 1. In-Reply-To matches a SENT OUTBOUND message (our replies carry their own + Message-ID and an outbound row, created at send time). + 2. Fallback: it matches ANY stored message (e.g. a second reply threaded to + our first inbound) that already carries a proposal_id. + """ irt = parsed.get("in_reply_to") if not irt: return None @@ -108,9 +131,16 @@ def _match_proposal(parsed: dict, sess) -> Proposal | None: ) .one_or_none() ) - if sent is None or sent.proposal_id is None: - return None - return sess.get(Proposal, sent.proposal_id) + if sent is not None and sent.proposal_id is not None: + return sess.get(Proposal, sent.proposal_id) + chain = ( + sess.query(MailMessage) + .filter(MailMessage.message_id == irt, MailMessage.proposal_id.isnot(None)) + .one_or_none() + ) + if chain is not None: + return sess.get(Proposal, chain.proposal_id) + return None # --- drafts ------------------------------------------------------------------- @@ -144,10 +174,15 @@ def _build_draft(parsed: dict, kind: MailClassification, proposal: Proposal | No # --- inbound pass --------------------------------------------------------------- -def _poll(cfg: Config, sess, fetcher=None) -> dict: - """One inbound pass: parse, classify, thread, draft. Returns stats.""" +def _poll(cfg: Config, sess, fetcher=None, marker=None) -> tuple[dict, list[bytes]]: + """One inbound pass: parse, classify, thread, draft. + + Returns (stats, processed_imap_nums). The caller marks \\Seen only after + its transaction commits. + """ stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0} - for raw in (fetcher or _fetch_unread)(cfg): + processed_nums: list[bytes] = [] + for num, raw in (fetcher or _fetch_unread)(cfg): parsed = parse_raw(raw) if ( sess.query(MailMessage) @@ -186,6 +221,7 @@ def _poll(cfg: Config, sess, fetcher=None) -> dict: detail={"classification": kind.value, "confidence": confidence}, ) stats["processed"] += 1 + processed_nums.append(num) # stored -> safe to mark seen after commit if kind == MailClassification.NEEDS_REVIEW: stats["needs_review"] += 1 @@ -210,18 +246,23 @@ def _poll(cfg: Config, sess, fetcher=None) -> dict: detail={"mail_id": record.id}, ) stats["drafted"] += 1 - return stats + return stats, processed_nums # --- human gate: approve draft ------------------------------------------------- def _approve_draft(mail_id: int, actor: str, sess) -> int: - """Human approves a drafted reply. Only then may SMTP send it.""" + """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). + """ record = sess.get(MailMessage, mail_id) - if record is None or record.direction != MailDirection.INBOUND: + if record is None: return 1 - if record.draft_status != DraftStatus.DRAFTED: + if record.draft_status != DraftStatus.DRAFTED or not record.draft_reply: audit( sess, actor=actor, @@ -232,12 +273,14 @@ def _approve_draft(mail_id: int, actor: str, sess) -> int: ) return 1 record.draft_status = DraftStatus.APPROVED + record.approved_hash = _content_hash(record.draft_reply) + record.approved_by = actor[:128] audit( sess, actor=actor, action="approve_draft", proposal_id=record.proposal_id, - detail={"mail_id": record.id}, + detail={"mail_id": record.id, "hash": record.approved_hash}, ) return 0 @@ -245,12 +288,21 @@ def _approve_draft(mail_id: int, actor: str, sess) -> int: # --- SMTP: only approved drafts ------------------------------------------------ -def _smtp_send(cfg: Config, to_addr: str, subject: str, body: str, message_id: str) -> None: +def _smtp_send( + cfg: Config, + to_addr: str, + subject: str, + body: str, + message_id: str, + in_reply_to: str | None = None, +) -> None: msg = EmailMessage() msg["From"] = cfg.mail_from msg["To"] = to_addr msg["Subject"] = subject msg["Message-ID"] = message_id + if in_reply_to: + msg["In-Reply-To"] = in_reply_to # keep merchant threads intact msg.set_content(body) with smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) as server: server.starttls() @@ -267,52 +319,172 @@ def _approved_by(sess, mail_id: int) -> str: return "unknown" -def _send_approved(cfg: Config, sess, sender=None) -> dict: - """Send every APPROVED draft. Anything else stays untouched.""" - stats = {"sent": 0, "blocked": 0} - pending = ( - sess.query(MailMessage) - .filter( - MailMessage.direction == MailDirection.INBOUND, - MailMessage.draft_status == DraftStatus.APPROVED, - ) - .all() - ) - for record in pending: +def _send_one(cfg: Config, sess, mail_id: int, sender=None) -> str: + """Send ONE approved draft. Returns: sent | hash_mismatch | missing | blocked. + + Creates the OUTBOUND row (its own Message-ID + In-Reply-To pointing at the + inbound message) so future replies thread correctly — the inbound row's + message_id is left untouched (audit finding: it used to be overwritten). + 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 + + record = sess.get(MailMessage, mail_id) + if record is None: + return "missing" + if record.draft_status != DraftStatus.APPROVED: + return "blocked" + + if record.direction == MailDirection.INBOUND: + # reply: goes back to the inbound sender, threads to the inbound mail, + # and gets its own OUTBOUND row to_addr = record.from_addr or "" - if not to_addr: - record.draft_status = DraftStatus.SKIPPED - stats["blocked"] += 1 - continue - message_id = ( - f"" + subject = f"Re: {record.subject or 'your message'}" + in_reply_to = record.message_id + outbound_msgid = ( + f"" ) - (sender or _smtp_send)( - cfg, - to_addr, - f"Re: {record.subject or 'your message'}", - record.draft_reply or "", - message_id, - ) - approved_by = _approved_by(sess, record.id) - record.draft_status = DraftStatus.SENT - # store the Message-ID so future replies thread to this draft - record.message_id = message_id + needs_row = True + else: + # first-contact draft: already an OUTBOUND row; send as-is + to_addr = record.to_addr or "" + subject = record.subject or "" + in_reply_to = None + outbound_msgid = record.message_id + needs_row = False + if not to_addr: + 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_draft", + action="send_refused_hash_mismatch", proposal_id=record.proposal_id, - detail={ - "mail_id": record.id, - "message_id": message_id, - "approved_by": approved_by, - }, + result="error", + detail={"mail_id": record.id}, ) - stats["sent"] += 1 + return "hash_mismatch" + + (sender or _smtp_send)( + cfg, + to_addr, + subject, + record.draft_reply or "", + outbound_msgid, + in_reply_to=in_reply_to, + ) + approved_by = _approved_by(sess, record.id) + if needs_row: + sess.add( + MailMessage( + direction=MailDirection.OUTBOUND, + message_id=outbound_msgid, + in_reply_to=record.message_id, + from_addr=cfg.mail_from, + to_addr=to_addr, + subject=subject, + body=record.draft_reply, + proposal_id=record.proposal_id, + draft_status=DraftStatus.SENT, + processed_at=datetime.now(UTC), + ) + ) + record.draft_status = DraftStatus.SENT + sess.flush() + audit( + sess, + actor="daemon:maild", + action="send_draft", + proposal_id=record.proposal_id, + detail={ + "mail_id": record.id, + "message_id": outbound_msgid, + "approved_by": approved_by, + }, + ) + return "sent" + + +def _send_approved(cfg: Config, sess, sender=None, factory=None) -> dict: + """Send every APPROVED draft, one transaction per message. + + `factory` is a SQLAlchemy sessionmaker bound to the shared engine; each + message is processed and committed in its own session (tests inject the + engine's maker; the CLI passes appdb.session_factory()). + """ + stats = {"sent": 0, "blocked": 0, "hash_mismatch": 0, "missing": 0} + ids = [ + row.id + for row in sess.query(MailMessage.id) + .filter(MailMessage.draft_status == DraftStatus.APPROVED) + .all() + ] + for mail_id in ids: + # one transaction per message — an SMTP failure on message N must not + # roll back (and later re-send) messages 1..N-1 (audit finding 3) + try: + if factory is not None: + with factory() as one_sess: # sessionmaker() -> Session + outcome = _send_one(cfg, one_sess, mail_id, sender) + one_sess.commit() + else: + 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) + stats[outcome] = stats.get(outcome, 0) + 1 return stats +def _approved_by(sess, mail_id: int) -> str: + from cayceops.models import AuditEntry + + for entry in sess.query(AuditEntry).all(): + if entry.action == "approve_draft" and entry.detail.get("mail_id") == mail_id: + return entry.actor + return "unknown" + + +def _compose(proposal_id: int, to_addr: str, subject: str, body: str, sess) -> int: + """Create the FIRST-CONTACT outbound draft for a proposal (state: drafted). + + This is the missing path the audit flagged: without it, threading only + worked for replies to mails we never actually sent. + """ + proposal = sess.get(Proposal, proposal_id) + if proposal is None or not to_addr: + return -1 + outbound = MailMessage( + direction=MailDirection.OUTBOUND, + message_id=f"", + from_addr=None, # filled from MAIL_FROM at send time + to_addr=to_addr[:320], + subject=subject[:998], + body=body, + draft_reply=body, # canonical approved-text field (same as reply drafts) + proposal_id=proposal.id, + draft_status=DraftStatus.DRAFTED, + ) + sess.add(outbound) + sess.flush() + audit( + sess, + actor="daemon:maild", + action="compose_draft", + proposal_id=proposal.id, + detail={"mail_id": outbound.id}, + ) + return outbound.id + + # --- CLI wrappers (session scope lives here) ------------------------------------ @@ -322,12 +494,13 @@ def poll(cfg: Config | None = None) -> int: cfg = cfg or load() try: + nums: list[bytes] = [] with session_scope() as sess: run_row = DaemonRun(daemon="maild") sess.add(run_row) sess.flush() try: - stats = _poll(cfg, sess) + stats, nums = _poll(cfg, sess) except Exception as exc: # noqa: BLE001 — daemon boundary run_row.status = "error" run_row.detail = {"error": str(exc)[:300]} @@ -336,6 +509,9 @@ def poll(cfg: Config | None = None) -> int: run_row.status = "ok" run_row.detail = stats run_row.finished_at = datetime.now(UTC) + # committed: now it is safe to mark the batch seen + if nums: + _mark_seen(cfg, nums) return 0 finally: engine().dispose() @@ -353,12 +529,38 @@ def approve_draft(mail_id: int, actor: str) -> int: def send_approved(cfg: Config | None = None) -> int: from cayceops.config import load - from cayceops.db import engine + 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: - _send_approved(cfg, 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: + 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 finally: engine().dispose() + + +def compose(proposal_id: int, to_addr: str, subject: str, body: str) -> int: + from cayceops.db import engine + + try: + with session_scope() as sess: + mail_id = _compose(proposal_id, to_addr, subject, body, sess) + return 0 if mail_id >= 0 else 1 + finally: + engine().dispose() diff --git a/cayceops/components/publisher/nostr.py b/cayceops/components/publisher/nostr.py index 2688257..8176545 100644 --- a/cayceops/components/publisher/nostr.py +++ b/cayceops/components/publisher/nostr.py @@ -148,6 +148,16 @@ def verify_event(event: dict[str, Any]) -> bool: ) +def verify_schnorr(sig_hex: str, message: bytes, pubkey_hex: str) -> bool: + """Verify a BIP-340 signature over an arbitrary 32-byte message.""" + try: + return _schnorr_verify( + bytes.fromhex(sig_hex), message, bytes.fromhex(pubkey_hex) + ) + except (ValueError, TypeError): + return False + + def relay_publish(relay_url: str, event: dict[str, Any], timeout: float = 10.0) -> str: """Publish one event to one relay; return the relay's OK message.""" from websockets.sync.client import connect diff --git a/cayceops/components/publisher/publisher.py b/cayceops/components/publisher/publisher.py index 218a8a7..872fb00 100644 --- a/cayceops/components/publisher/publisher.py +++ b/cayceops/components/publisher/publisher.py @@ -16,7 +16,13 @@ from datetime import UTC, datetime, timedelta from cayceops.components.publisher import images, nostr from cayceops.config import Config from cayceops.db import session_scope -from cayceops.models import ContentItem, ContentStatus, DaemonRun, audit +from cayceops.models import ( + ContentItem, + ContentStatus, + DaemonRun, + _content_hash, + audit, +) # rotation: one theme per slot; channel rotation nostr -> facebook -> instagram THEMES = [ @@ -120,11 +126,12 @@ def _approve(item_id: int, actor: str, sess) -> int: ) return 1 item.status = ContentStatus.APPROVED + item.approved_hash = _content_hash(item.caption) # binding: caption as approved audit( sess, actor=actor, action="approve_content", - detail={"item": item.id, "channel": item.channel}, + detail={"item": item.id, "channel": item.channel, "hash": item.approved_hash}, ) return 0 @@ -174,7 +181,7 @@ def _publish_nostr(item: ContentItem, cfg: Config, sess, sender=None) -> str: def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict: """Publish approved items that are due. Non-nostr channels are skipped unless their adapters are explicitly enabled (default: skipped, audited).""" - stats = {"published": 0, "skipped": 0, "failed": 0} + stats = {"published": 0, "skipped": 0, "failed": 0, "tampered": 0} due = ( sess.query(ContentItem) .filter(ContentItem.status == ContentStatus.APPROVED) @@ -184,31 +191,56 @@ def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict: for item in due: 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: + audit( + sess, + actor="daemon:publisher", + action="publish_refused_hash_mismatch", + result="error", + detail={"item": item.id}, + ) + stats["tampered"] += 1 + continue + try: if item.channel == "nostr": item.post_ref = _publish_nostr(item, cfg, sess, sender=nostr_sender) else: + # disabled channel is TERMINAL (audit finding: this used to + # write a skip audit row every 5 minutes, forever) + item.status = ContentStatus.SKIPPED + item.error = f"{item.channel} adapter disabled by default" audit( sess, actor="daemon:publisher", - action="publish_skipped", - detail={ - "item": item.id, - "channel": item.channel, - "reason": f"{item.channel} adapter disabled by default", - }, + action="publish_skipped_terminal", + detail={"item": item.id, "channel": item.channel}, ) stats["skipped"] += 1 continue except Exception as exc: # noqa: BLE001 — publish boundary: audit and continue + item.attempts += 1 item.error = str(exc)[:300] - audit( - sess, - actor="daemon:publisher", - action="publish_failed", - result="error", - detail={"item": item.id, "error": item.error}, - ) + if item.attempts >= cfg.max_attempts: + # attempt budget exhausted -> terminal (no infinite retries) + item.status = ContentStatus.SKIPPED + audit( + sess, + actor="daemon:publisher", + action="publish_abandoned", + result="error", + detail={"item": item.id, "attempts": item.attempts}, + ) + else: + audit( + sess, + actor="daemon:publisher", + action="publish_failed", + result="error", + detail={"item": item.id, "error": item.error, "attempt": item.attempts}, + ) stats["failed"] += 1 continue item.status = ContentStatus.PUBLISHED diff --git a/cayceops/components/scanner/scanner.py b/cayceops/components/scanner/scanner.py index 4b62580..432634b 100644 --- a/cayceops/components/scanner/scanner.py +++ b/cayceops/components/scanner/scanner.py @@ -165,7 +165,7 @@ def scan(cfg: Config, sess, fetch=_fetch) -> dict: "summary": desc, "scanned_at": datetime.now(UTC).isoformat(), }, - risk_score=score, + match_score=score, status=ProposalStatus.PROPOSED, dedupe_key=dedupe, ) diff --git a/cayceops/config.py b/cayceops/config.py index 9317e8f..c58b158 100644 --- a/cayceops/config.py +++ b/cayceops/config.py @@ -84,6 +84,11 @@ 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. + operator_pubkey: str = field( + default_factory=lambda: os.environ.get("CAYCE_OPS_OPERATOR_PUBKEY", "") + ) # --- mail --- imap_host: str = field( diff --git a/cayceops/models.py b/cayceops/models.py index ea2336e..65a6dad 100644 --- a/cayceops/models.py +++ b/cayceops/models.py @@ -5,13 +5,22 @@ Status machines proposal: proposed -> approved -> running -> done | failed proposed -> rejected mail: (inbound) classified: approval|rejection|question|other|needs_review - (outbound draft) none -> drafted -> approved -> sent | skipped -content: draft -> approved -> published + (any) draft: none -> drafted -> approved -> sent | skipped +content: draft -> approved -> published | skipped (terminal) + +Audit chain +----------- +audit_log is hash-chained: each entry stores sha256(prev_hash | fields). +``cayceops.models.verify_audit_chain`` re-walks it; the monitor surfaces breaks. +Not tamper-proof (DB writes can rewrite the chain) but tamper-EVIDENT against +selective edits. See AUDIT.md / audit response 2026-10-20. """ from __future__ import annotations import enum +import hashlib +import json from datetime import UTC, datetime from sqlalchemy import JSON, DateTime, Enum, Float, ForeignKey, Integer, String, Text @@ -22,6 +31,11 @@ def utcnow() -> datetime: return datetime.now(UTC) +def _content_hash(text: str) -> str: + """Binding hash for human-approved text (drafts, captions).""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + class Base(DeclarativeBase): pass @@ -54,7 +68,8 @@ class Proposal(Base): title: Mapped[str] = mapped_column(String(512)) url: Mapped[str | None] = mapped_column(String(1024), nullable=True) detail: Mapped[dict] = mapped_column(JSON, default=dict) - risk_score: Mapped[float] = mapped_column(Float, default=0.0) + # keyword-match relevance 0..1 — NOT risk. auto-approve takes highest first. + match_score: Mapped[float] = mapped_column(Float, default=0.0) status: Mapped[ProposalStatus] = mapped_column( Enum(ProposalStatus), default=ProposalStatus.PROPOSED, index=True ) @@ -68,7 +83,7 @@ class Proposal(Base): ) -# --- audit ------------------------------------------------------------------- +# --- audit (hash-chained) ---------------------------------------------------- class AuditEntry(Base): @@ -76,13 +91,16 @@ class AuditEntry(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True) ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True) - actor: Mapped[str] = mapped_column(String(128)) # human username | auto:policy | daemon: + actor: Mapped[str] = mapped_column(String(128)) # human | auto:policy | daemon: proposal_id: Mapped[int | None] = mapped_column( ForeignKey("proposals.id", ondelete="SET NULL"), nullable=True, index=True ) action: Mapped[str] = mapped_column(String(64)) # approve|reject|execute|classify|... detail: Mapped[dict] = mapped_column(JSON, default=dict) result: Mapped[str] = mapped_column(String(32), default="ok") # ok | error + # tamper-evident chain + prev_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) + entry_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) def audit( @@ -93,18 +111,68 @@ def audit( detail: dict | None = None, result: str = "ok", ) -> AuditEntry: - """Record one audited action. Every component writes through this helper.""" + """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. + """ + prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() + prev_hash = prev.entry_hash if prev is not None else "GENESIS" + 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( + [ + prev_hash or "GENESIS", + str(ts_int), + actor, + action, + str(proposal_id), + detail_json, + result, + ] + ) entry = AuditEntry( + ts=ts, actor=actor, action=action, proposal_id=proposal_id, detail=detail or {}, result=result, + prev_hash=prev_hash, + 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" + for entry in sess.query(AuditEntry).order_by(AuditEntry.id).all(): + 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, + str(int(ts.timestamp())), + entry.actor, + entry.action, + str(entry.proposal_id), + detail_json, + entry.result, + ] + ) + expected = hashlib.sha256(payload.encode("utf-8")).hexdigest() + if entry.prev_hash != prev_hash or entry.entry_hash != expected: + return False, entry.id + prev_hash = entry.entry_hash + return True, None + + # --- mail -------------------------------------------------------------------- @@ -151,6 +219,9 @@ class MailMessage(Base): Enum(DraftStatus), default=DraftStatus.NONE ) draft_reply: Mapped[str | None] = mapped_column(Text, nullable=True) + # 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) processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) @@ -161,6 +232,7 @@ class ContentStatus(enum.StrEnum): DRAFT = "draft" APPROVED = "approved" PUBLISHED = "published" + SKIPPED = "skipped" # terminal: disabled channel or attempt budget exhausted class ContentItem(Base): @@ -179,6 +251,9 @@ class ContentItem(Base): published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) post_ref: Mapped[str | None] = mapped_column(String(512), nullable=True) error: Mapped[str | None] = mapped_column(Text, nullable=True) + 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) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) diff --git a/deploy/roles.sql b/deploy/roles.sql new file mode 100644 index 0000000..2107940 --- /dev/null +++ b/deploy/roles.sql @@ -0,0 +1,22 @@ +-- 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'; + +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; + +-- sequences +GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor; diff --git a/deploy/systemd/README.md b/deploy/systemd/README.md index 6f5dde3..70f3658 100644 --- a/deploy/systemd/README.md +++ b/deploy/systemd/README.md @@ -9,9 +9,14 @@ sudo -u cayce uv sync --frozen sudo -u cayce cayceops db upgrade # alembic to head sudo cp deploy/systemd/cayce-ops@.service deploy/systemd/cayce-ops@.timer /etc/systemd/system/ +# Every daemon has a `run` subcommand now: +# scanner run · executor run · maild run (poll + send-approved) +# publisher run (publish-due) · monitor run (build) # cadence per daemon (edit OnCalendar in each timer copy): # scanner hourly executor */10 min maild */5 min # publisher */5 min monitor */1 min +# NOTE: `publisher plan` (fills the calendar) is a SEPARATE timer: +# OnCalendar=daily -> ExecStart=/srv/cayce-ops/.venv/bin/cayceops publisher plan for d in scanner executor maild publisher monitor; do sudo cp deploy/systemd/cayce-ops@.timer /etc/systemd/system/cayce-ops@$d.timer done diff --git a/docker-compose.yml b/docker-compose.yml index 6697411..ac0570f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,7 +6,7 @@ services: POSTGRES_PASSWORD: localdev-only-not-a-secret POSTGRES_DB: cayce_ops ports: - - "5432:5432" + - "127.0.0.1:5432:5432" # loopback only — Docker bypasses host firewalls volumes: - pgdata:/var/lib/postgresql/data healthcheck: diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 0a1e685..71ea27b 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -69,3 +69,29 @@ An action plugin is a Python function registered in `http_post` exists but is inert unless `CAYCE_OPS_ENABLE_HTTP_ACTIONS=true`. *Cost:* no sandboxing of plugins — plugins are repo code, reviewed like everything else. + +## D10 — Audit chain is hash-chained; approval can require operator signature (post-audit) +`audit_log` rows chain sha256(prev_hash|fields); `verify_audit_chain` walks it, +the monitor can surface breaks. With `CAYCE_OPS_OPERATOR_PUBKEY` set, +`executor approve` REQUIRES a BIP-340 signature over +sha256("cayce-ops:approve:") from the operator key — `--actor` becomes a +label, the key is the proof (the audit's "anyone can sign as anyone"). +*Cost:* without a pubkey configured the gate degrades to the old free-text +actor; separate DB roles per daemon remain a deployment TODO (SQL to tighten +ships in deploy/roles.sql). + +## D11 — Mail threading now has a real first-contact path (post-audit) +Sends create an OUTBOUND row (own Message-ID, In-Reply-To -> inbound); +inbound rows keep their original Message-ID. `maild compose` creates the +initial outreach draft for a proposal. Approval is bound to the draft text +via sha256 (`approved_hash`): an edited draft refuses to send until +re-approved. Each send commits in its own transaction — SMTP failure on +message N can no longer resurrect 1..N-1 as duplicates. + +## D12 — IMAP \\Seen only after commit; content gates tighten (post-audit) +Messages are fetched unmarked and flagged \\Seen after the processing +transaction commits (crash mid-batch no longer loses mail). Publisher: +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. diff --git a/scripts/githooks/pre-push b/scripts/githooks/pre-push new file mode 100755 index 0000000..14bbc49 --- /dev/null +++ b/scripts/githooks/pre-push @@ -0,0 +1,7 @@ +#!/bin/sh +# The gate, wired as a real hook (audit medium finding: it did not self-install). +# Runs the exact CI pipeline; refuses the push on any failure. +set -e +cd "$(git rev-parse --show-toplevel)" +make check +echo "pre-push gate passed — safe to push" diff --git a/tests/test_executor.py b/tests/test_executor.py index 52b9364..0137e1e 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -32,6 +32,60 @@ def test_reject_records_actor(cfg, sess): assert p.status == 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): approved = _proposal(title="Approved item") proposed = _proposal(title="Proposed item") diff --git a/tests/test_maild.py b/tests/test_maild.py index 68a20af..18a76be 100644 --- a/tests/test_maild.py +++ b/tests/test_maild.py @@ -2,6 +2,8 @@ from __future__ import annotations +from types import SimpleNamespace + from cayceops.components.maild import classify as clf from cayceops.components.maild import maild from cayceops.models import ( @@ -16,13 +18,10 @@ from cayceops.models import ( OUTBOUND_MSGID = "" +STUB_CFG = SimpleNamespace(mail_from="agent@cayce-co.xyz") -def _eml( - subject: str, - body: str, - in_reply_to: str | None = None, - msgid: str | None = None, -) -> bytes: + +def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes: lines = [ "From: Partner Desk ", "To: agent@cayce-co.xyz", @@ -36,6 +35,8 @@ def _eml( 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).""" p = Proposal( kind="partnership", title="Acme hotel affiliate program", @@ -43,35 +44,68 @@ def _proposal_with_sent_draft(sess) -> Proposal: ) sess.add(p) sess.flush() - sess.add( - MailMessage( - direction=MailDirection.OUTBOUND, - message_id=OUTBOUND_MSGID, - to_addr="partners@acme.example.org", - subject="[cayce] partnership inquiry", - proposal_id=p.id, - draft_status=DraftStatus.SENT, - ) + 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._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent" + out = ( + sess.query(MailMessage) + .filter_by(direction=MailDirection.OUTBOUND, proposal_id=p.id) + .one() + ) + out.message_id = OUTBOUND_MSGID # deterministic id for the fixtures sess.flush() return p +def _fetcher(raws: list[bytes]): + return lambda cfg: [(bytes(str(i), "ascii"), raw) for i, raw in enumerate(raws)] + + # --- classifier --------------------------------------------------------------- def test_rules_classification(cfg): config = cfg() - ok = clf.classify("Your application was approved, welcome to the program", config) - no = clf.classify("Unfortunately we cannot accept new partners", config) - assert ok[0] is MailClassification.APPROVAL - assert no[0] is MailClassification.REJECTION - q = clf.classify("Could you provide your traffic numbers?", config) - assert q[0] is MailClassification.QUESTION - kind, confidence = clf.classify("grocery list", config) + assert ( + clf.classify("Your application was approved, welcome to the program", config)[0] + is MailClassification.APPROVAL + ) + assert ( + clf.classify("Unfortunately we cannot accept new partners", config)[0] + is MailClassification.REJECTION + ) + assert ( + clf.classify("Could you provide your traffic numbers?", config)[0] + is MailClassification.QUESTION + ) + kind, _ = clf.classify("grocery list", config) 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.""" + config = cfg() + cases = [ + "We cannot accept your application", + "We are not accepting new partners at this time", + "Your application was not approved", + "We can't approve this request", + "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 + + +def test_llm_prompt_format_survives_json_braces(): + """Critical finding 4: .format() raised KeyError on the JSON sample.""" + formatted = clf._LLM_PROMPT.format(body="x") + assert '{"classification"' in formatted # literal braces survive + + def test_low_confidence_becomes_needs_review(cfg): config = cfg(CAYCE_OPS_LLM_REVIEW_THRESHOLD="0.95") kind, confidence = clf.classify("Your application was approved, welcome", config) @@ -90,8 +124,9 @@ def test_poll_threads_approval_and_drafts_reply(cfg, sess): in_reply_to=OUTBOUND_MSGID, ) - stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) + stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw])) assert stats["processed"] == 1 and stats["drafted"] == 1 + assert len(nums) == 1 inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() assert inbound.classification is MailClassification.APPROVAL @@ -107,16 +142,16 @@ def test_poll_idempotent_on_message_id(cfg, sess): _proposal_with_sent_draft(sess) raw = _eml("Re: hi", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) - def fetcher(cfg): - return [raw] - - assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1 - assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1 # noqa: E501 + first, second = maild._poll(cfg(), sess, fetcher=_fetcher([raw])), maild._poll( + cfg(), sess, fetcher=_fetcher([raw]) + ) + assert first[0]["processed"] == 1 + assert second[0]["skipped"] == 1 def test_unthreaded_approval_needs_review(cfg, sess): raw = _eml("Great news", "Approved! Welcome aboard.") # no In-Reply-To - stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) + stats, _ = maild._poll(cfg(), sess, fetcher=_fetcher([raw])) inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() assert inbound.classification is MailClassification.NEEDS_REVIEW assert stats["needs_review"] == 1 @@ -126,7 +161,7 @@ def test_unthreaded_approval_needs_review(cfg, sess): def test_question_gets_clarification_draft(cfg, sess): _proposal_with_sent_draft(sess) raw = _eml("Re: inquiry", "Could you provide your monthly traffic?", in_reply_to=OUTBOUND_MSGID) - maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() assert inbound.classification is MailClassification.QUESTION assert inbound.draft_status is DraftStatus.DRAFTED @@ -139,12 +174,14 @@ def test_question_gets_clarification_draft(cfg, sess): def test_send_blocked_without_approval(cfg, sess): _proposal_with_sent_draft(sess) raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) - maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) sent = [] config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz") stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a)) assert stats["sent"] == 0 and sent == [] + assert stats["blocked"] + stats["missing"] >= 0 # no crash paths + assert config.mail_from == "agent@cayce-co.xyz" row = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() assert row.draft_status is DraftStatus.DRAFTED @@ -156,27 +193,159 @@ def test_send_requires_explicit_human_approval(cfg, sess): "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID, ) - maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub) + maild._poll(cfg(), sess, fetcher=_fetcher([raw])) inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one() + original_id = inbound.message_id assert maild._approve_draft(inbound.id, "gustavo", sess) == 0 sent = [] config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz") - stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a)) + + def sender(cfg, to, subject, body, message_id, in_reply_to=None): + sent.append((to, subject, body, message_id, in_reply_to)) + + stats = maild._send_approved(config, sess, sender=sender) assert stats["sent"] == 1 assert inbound.draft_status is DraftStatus.SENT - assert sent[0][1] == "partners@acme.example.org" + to, subject, body, message_id, in_reply_to = sent[0] + assert to == "partners@acme.example.org" + assert in_reply_to == original_id # In-Reply-To threads to the inbound mail - row = sess.query(AuditEntry).filter_by(action="send_draft").one() - assert "gustavo" in str(row.detail) + # a real OUTBOUND row now exists (audit fix: they were never created) + out = ( + sess.query(MailMessage) + .filter_by(direction=MailDirection.OUTBOUND, message_id=message_id) + .one() + ) + assert out.in_reply_to == original_id + assert out.proposal_id == inbound.proposal_id + assert inbound.message_id == original_id # inbound id NOT overwritten + + row = ( + sess.query(AuditEntry) + .filter_by(action="send_draft") + .order_by(AuditEntry.id.desc()) + .first() + ) + assert row is not None and "gustavo" in str(row.detail) def test_double_approve_is_rejected(cfg, sess): _proposal_with_sent_draft(sess) - raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID) - maild._poll(cfg(), sess, fetcher=lambda cfg: [raw]) # noqa: E731 (test stub) + raw = _eml("Re: inquiry", "Approved, welcome.", 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) == 1 + + +# --- audit corrections ---------------------------------------------------------- + + +def test_send_failure_does_not_resurrect_sent_mails(cfg, sess): + """Critical finding 3: SMTP failure mid-batch rolled back the whole + transaction, re-sending already-delivered mails 5 minutes later.""" + config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz") + for i in range(3): + # unique first-contact thread per proposal (real compose -> send path) + p = Proposal(kind="partnership", title=f"Acme partner {i}") + sess.add(p) + sess.flush() + 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._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent" + sess.flush() + out = sess.get(MailMessage, mail_id) + raw = _eml( + f"Re: {i}", + "Approved, welcome to the program.", + in_reply_to=out.message_id, + msgid=f"", + ) + maild._poll(config, sess, fetcher=_fetcher([raw])) + inbound = ( + sess.query(MailMessage) + .filter_by(direction=MailDirection.INBOUND, message_id=f"") + .one() + ) + maild._approve_draft(inbound.id, "gustavo", sess) + sess.commit() # per-message factory sessions only see committed rows + + calls = {"n": 0} + + def flaky_smtp(cfg, to, subject, body, message_id, in_reply_to=None): + calls["n"] += 1 + if calls["n"] == 3: + raise OSError("smtp connection lost") + + import cayceops.db as appdb + + stats = maild._send_approved( + config, sess, sender=flaky_smtp, factory=appdb.session_factory() + ) + + assert calls["n"] == 3 + assert stats["sent"] == 2 and stats["error"] == 1 # no rollback of 1..2 + still_approved = ( + sess.query(MailMessage) + .filter_by(direction=MailDirection.INBOUND, draft_status=DraftStatus.APPROVED) + .count() + ) + assert still_approved == 1 # only the failed one stays queued + + +def test_edited_draft_refuses_to_send(cfg, sess): + """Medium finding: approved text was not bound to its content.""" + _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) == 0 + + inbound.draft_reply = "TOTALLY EDITED TEXT after approval" # tamper + + sent = [] + config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz") + stats = maild._send_approved(config, 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 # needs fresh approval + action = sess.query(AuditEntry).filter_by(action="send_refused_hash_mismatch").one() + assert action.result == "error" + + +def test_compose_creates_first_contact_draft(cfg, sess): + """Critical finding 2: there was no path for the first contact mail.""" + p = Proposal(kind="partnership", title="New hotel partner") + sess.add(p) + sess.flush() + + mail_id = maild._compose( + p.id, "desk@hotel.example.org", "Partnership", "We would like to...", sess + ) + assert mail_id > 0 + out = sess.get(MailMessage, mail_id) + assert out.direction is MailDirection.OUTBOUND + assert out.draft_status is DraftStatus.DRAFTED + 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 + stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None) + assert stats["sent"] == 1 + sess.flush() + + reply = _eml("Re: Partnership", "Approved, welcome!", in_reply_to=out.message_id) + assert maild._match_proposal(maild.parse_raw(reply), sess) is p + + +def test_mark_seen_returned_for_post_commit_processing(cfg, sess): + """Medium finding: \\\\Seen was set before processing (crash lost mail).""" + raw = _eml("Re: hi", "Approved, welcome to the program.") + stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw])) + assert stats["processed"] == 1 + assert nums == [b"0"] # caller marks seen only after commit diff --git a/tests/test_models.py b/tests/test_models.py index 65ebda4..0b1fe79 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -90,3 +90,23 @@ def test_mail_and_content_and_runs(sess): assert m.id and c.id and r.id assert m.draft_status == "drafted" assert c.status == "draft" + + +def test_audit_chain_and_verification(sess): + from cayceops.models import verify_audit_chain + + p = Proposal(kind="other", title="chain") + sess.add(p) + sess.flush() + for i in range(3): + audit(sess, actor=f"actor-{i}", action="step", proposal_id=p.id, detail={"i": i}) + sess.flush() + + ok, broken = verify_audit_chain(sess) + assert ok and broken is None + + # tamper with one entry -> chain breaks at that id + first = sess.query(AuditEntry).order_by(AuditEntry.id).first() + first.actor = "tampered" + ok, broken = verify_audit_chain(sess) + assert not ok and broken == first.id diff --git a/tests/test_publisher.py b/tests/test_publisher.py index fa9fc87..2f7374a 100644 --- a/tests/test_publisher.py +++ b/tests/test_publisher.py @@ -125,14 +125,14 @@ def test_nostr_without_config_fails_clean(cfg, sess): assert item.error and "nostr" in item.error -def test_non_nostr_channels_skipped_by_default(cfg, sess): +def test_non_nostr_channels_skipped_terminal(cfg, sess): item = _item(channel="facebook", theme="lightning-abroad") sess.add(item) sess.flush() stats = publisher._publish_due(cfg(), sess) assert stats["skipped"] == 1 - assert item.status is ContentStatus.APPROVED - skip = sess.query(AuditEntry).filter_by(action="publish_skipped").one() + 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" @@ -162,3 +162,63 @@ def test_unknown_brand_rejected(): with pytest.raises(ValueError): 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 + + +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", + ) + + publisher._publish_due(config, sess, nostr_sender=always_fail) + assert item.attempts == 1 + assert item.status is ContentStatus.SKIPPED # budget exhausted -> terminal diff --git a/tests/test_scanner.py b/tests/test_scanner.py index 54e87ba..d77d635 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -58,7 +58,7 @@ def test_scan_files_proposals_with_kind_and_score(cfg, sess): affiliate = by_title["Acme launches affiliate program for hotels"] assert affiliate.kind == "affiliate" assert affiliate.status == ProposalStatus.PROPOSED - assert affiliate.risk_score > 0 + assert affiliate.match_score > 0 assert affiliate.dedupe_key partner = by_title["Become a payment partner"]