Compare commits

...
18 Commits
Author SHA1 Message Date
cayce 61d5cfd67d audit: response section — all findings addressed at c9bd441
ci / lint-test-leaks (push) Canceled after 0s
2026-10-19 23:07:17 +00:00
cayce c9bd4411e8 Merge PR #9: audit fixes — criticals 1-5 + mediums (56 tests, gate green)
ci / lint-test-leaks (push) Canceled after 0s
2026-10-19 23:06:30 +00:00
cayce 90e6cfcc19 fix: address external audit — criticals 1-5 + mediums
Critical 1 (approval forgery): audit_log is now hash-chained
(prev_hash/entry_hash, verify_audit_chain); executor approve REQUIRES a
BIP-340 signature over sha256(cayce-ops:approve:<id>) when
CAYCE_OPS_OPERATOR_PUBKEY is set; deploy/roles.sql ships least-privilege
per-daemon DB roles.

Critical 2 (threading dead): sends create real OUTBOUND rows (own
Message-ID, In-Reply-To -> inbound); inbound message_id never overwritten;
maild compose adds the first-contact path; _match_proposal falls back to
inbound chains.

Critical 3 (duplicate sends): per-message transactions in send path —
SMTP failure on message N no longer resurrects 1..N-1; failures audited
(send_failed) and the row stays queued.

Critical 4 (classifier): negated approvals (cannot accept / not approved /
not accepting) classify REJECTION at 0.95 before the bare-keyword rule;
LLM prompt braces escaped (.format no longer KeyError into silent fallback).

Critical 5 (systemd): every daemon has a run subcommand
(maild run = poll+send, publisher run = publish-due, monitor run = build);
deploy README documents the separate daily publisher plan timer.

Mediums: approved_hash binds drafts/captions to the approved text (edited
text refuses to send/publish until re-approved); risk_score renamed
match_score (migration 0002) and auto-approve orders HIGHEST relevance
first; compose binds postgres to 127.0.0.1; IMAP marks \\Seen only after
commit (fetch returns (num, raw), mark post-commit); disabled channels are
terminal (ContentStatus.SKIPPED) and publish failures burn a bounded
attempt budget before abandonment; make hooks installs the pre-push gate.

tests: 12 new — 56 total, all green; ruff 0 findings; gitleaks clean.
2026-10-19 23:06:29 +00:00
cayce 2da5f3ff24 Merge PR #8: audit update — audited commit pinned, gap disclosed
ci / lint-test-leaks (push) Canceled after 0s
2026-10-19 22:04:35 +00:00
cayce 2423cd10b8 audit: pin audited commit d500d88 + disclose lint-gate gap and fix
ci / lint-test-leaks (pull_request) Canceled after 0s
2026-10-19 22:04:34 +00:00
cayce d500d88870 Merge PR #7: lint-fix — make check fully green (E501/F841/F821/E731)
ci / lint-test-leaks (push) Canceled after 0s
Closes the gate gap flagged during final verification: prior per-branch
runs read ruff tail lines incorrectly. This commit is the audited one.
2026-10-19 22:03:55 +00:00
cayce a282aedb94 lint: wrap long lines (E501), fix F841/F821/E731 — make check fully green
All 32 findings resolved: regex literals kept single-line with noqa
(pattern IS the reviewed artifact), HTML rows string-concatenated for
line length, walrus tuple in BIP-340 verify split into named generator,
unused pytest import dropped. No behavior changes; 44/44 tests pass.
Full-gate discipline restored: ruff exit 0.
2026-10-19 22:03:53 +00:00
cayce abbaf7e4e2 Merge PR #6: audit — AUDIT.md for operator review
ci / lint-test-leaks (push) Canceled after 0s
2026-10-19 21:58:47 +00:00
cayce 6d77b3eff6 audit: AUDIT.md for operator review
ci / lint-test-leaks (pull_request) Canceled after 0s
Component map + data flow, every external call (where/what/trigger),
human-gate vs AUTONOMOUS_MODE matrix, env+permissions per component,
pinned deps with purpose, known risks (8, each with mitigation),
test+stack instructions. Audited-commit hash lands in the issue.
2026-10-19 21:58:46 +00:00
cayce 3f07362b7a Merge PR #5: monitor
ci / lint-test-leaks (push) Canceled after 0s
Static status page with privacy assertions and full escaping. 44 tests green.
2026-10-19 21:57:32 +00:00
cayce f61372e28e monitor: static status page — health, queue, gates, audit tail
ci / lint-test-leaks (pull_request) Canceled after 0s
- self-contained HTML (no external assets/JS), everything html-escaped
- daemon health from daemon_runs (last status + last run per daemon)
- proposal queue counts by state; human-gate counters (mail drafts awaiting
  approval, content draft/approved)
- audit tail: actor/action/result only — no detail payloads, no addresses,
  no subjects, no content bodies (tests assert the negatives)
- DaemonRun bookkeeping like every other daemon

tests: 3 new (sections render, leak-negatives, XSS escape) — 44 total
2026-10-19 21:57:31 +00:00
cayce 25c7db3a31 Merge PR #4: publisher
ci / lint-test-leaks (push) Canceled after 0s
Themed calendar + NIP-01 nostr (deterministic schnorr, self-verify),
FB/IG adapters disabled by default, hard human gate. 41 tests green.
2026-10-19 21:54:48 +00:00
cayce aabd6bafc7 publisher: themed calendar, NIP-01 nostr adapter, gated brand images
ci / lint-test-leaks (pull_request) Canceled after 0s
- calendar: one draft per theme/channel (idempotent), rotated schedule
- human gate: approve() audited; AUTONOMOUS_MODE has no effect here by design
- nostr (primary): in-repo NIP-01 client (D4) — canonical serialization,
  sha256 id, BIP-340 schnorr sign via coincurve with zero aux randomness
  (deterministic), pure-python reference verification before publish,
  per-relay publish with isolated errors, event id stored as post_ref
- facebook/instagram adapters: exist but refuse to run unless explicitly
  enabled via env (default deployment never touches Meta)
- images: Pillow brand templates (bh gold / hc orange), 1080x1350,
  template-driven typeset design (not AI-generated media)
- failure isolation: config errors audited, item stays approved

tests: 12 new (event roundtrip+tamper, determinism, key validation,
calendar idempotence, approve gate, nostr happy path, future schedule,
clean config failure, disabled meta adapters, image render) — 41 total
2026-10-19 21:54:39 +00:00
cayce 15072a0cd2 Merge PR #3: mail-negotiator
ci / lint-test-leaks (push) Canceled after 0s
IMAP classification (rules + optional LLM), conservative needs_review,
reply drafts gated on explicit human approval, SMTP approved-only,
message-id threading, full audit chain. 29 tests green.
2026-10-19 21:47:29 +00:00
cayce 5b259bc663 mail-negotiator: classify merchant replies, draft gated, audited sends
ci / lint-test-leaks (pull_request) Canceled after 0s
- IMAP poll (UNSEEN, stdlib imaplib) -> parse (bare address via email.utils)
- threading: In-Reply-To -> outbound sent-draft message-id -> proposal
- classify: deterministic rules; optional LLM behind provider interface;
  below-threshold confidence -> needs_review (never guessed)
- unthreadable approvals flagged needs_review (suspicious by definition)
- merchant_status recorded on proposal detail; proposal lifecycle untouched
- reply drafts generated from templates (drafted), NEVER auto-sent
- SMTP sends only draft_status=approved (human approve_draft, audited);
  sent drafts store their Message-ID for future threading
- idempotent on Message-ID; DaemonRun wrapper for monitor

tests: 9 new (classification rules, threshold->review, threading, idempotence,
blocked send without approval, approve->send->audit chain, double approve)
— 29 total passing
2026-10-19 21:47:18 +00:00
cayce 1b2c3cc1cf Merge PR #2: scanner + executor
ci / lint-test-leaks (push) Canceled after 0s
Discovery (read-only) + approval-gated execution with plugin registry,
audited autonomy policy, retry-then-fail. Gate: ruff clean, 20 tests green,
migration smoke ok, no leaks (full history).
2026-10-19 21:42:41 +00:00
cayce 73ccdf7bdf scanner + executor: discovery files proposals; execution gated on approval
ci / lint-test-leaks (pull_request) Canceled after 0s
scanner (read-only):
- sources from CAYCE_OPS_SCANNER_SOURCES JSON (rss + page types), stdlib only
- keyword scoring -> risk_score; kind via heuristics (affiliate/partnership/content_gap)
- idempotent: sha1 dedupe_key; repeated scans skip existing
- every proposal audited (actor daemon:scanner); source errors recorded, never fatal

executor (human gate):
- approve/reject audited with operator identity; only proposed -> approved
- AUTONOMOUS_MODE=true adds exactly one path: policy auto-approval
  (kinds allowlist + daily cap, actor auto:policy) — DECISIONS D5
- plugin registry by kind; unknown kinds fall back to log_action (D9)
- retries with max_attempts, then failed; every transition audited
- http_post plugin inert unless CAYCE_OPS_ENABLE_HTTP_ACTIONS=true

tests: 13 new (dedupe, keyword filter, autonomy bounds, retry-then-fail,
blocked http_post, fallback plugin) — 20 total passing
2026-10-19 21:42:34 +00:00
cayce 00fa3777ab Merge PR #1: skeleton — structure, schema, CI, deploy
ci / lint-test-leaks (push) Canceled after 0s
Component wiring, ORM schema + state machines, env-only config, alembic 0001,
Typer CLI, CI (ruff/pytest/alembic/gitleaks), compose + systemd, DECISIONS D1-D9.
Gate: make check green (ruff 0 errors, 7 tests pass, migration smoke ok, no leaks).
2026-10-19 21:38:06 +00:00
33 changed files with 3494 additions and 11 deletions
+3
View File
@@ -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:<id>") made by the operator key
CAYCE_OPS_OPERATOR_PUBKEY=
# --- mail (IMAP read / SMTP send-approved-drafts-only) ---
CAYCE_OPS_IMAP_HOST=imap.example.org
+177
View File
@@ -0,0 +1,177 @@
# AUDIT — cayce-ops
Audited commit: **d500d88** — merge of PR #7 on `main`.
> Note for the auditor: PRs #1-#6 merged with E501 lint findings present
> (per-branch gate runs were misread). PR #7 is a lint-only cleanup
> (line wrapping, F841/F821/E731 fixes, no behavior change) that restores
> a fully green gate. The honest trail is kept here on purpose.
> Re-run `make check` at d500d88 to verify: ruff 0 · 44 tests · no leaks.
Gate that commit passed: `make check` — ruff (0 findings), 44 tests green,
alembic smoke on sqlite, gitleaks full history (no leaks).
## 1. Component map and data flow
```
CAYCE_OPS_SCANNER_SOURCES (external URLs)
│ read-only fetch
▼
┌──────────┐ proposals(status=proposed) ┌──────────┐
│ scanner │─────────────────────────────▶│ DB │
└──────────┘ │ (shared) │
│ │ audit_log (every write)
┌──────────┐ approved → running → │ │◀────── actor / action / detail /
│ executor │◀──── done|failed ───────────│ │ result on every transition
└──────────┘ plugins by kind └──────────┘
│
┌──────────┐ IMAP inbox ───────────────▶│ mail_messages,
│ maild │ drafts (gated) │ proposals(detail)
└──────────┘ SMTP only draft=approved │
│
┌──────────┐ content_items ────────────▶│ │──▶ relay (nostr) / Meta (opt-in)
│publisher │ draft→approved→published └──────────┘
└──────────┘
┌──────────┐ reads daemon_runs, audit_log, counts
│ monitor │────────────────────────────────────▶ status.html (static)
└──────────┘
```
Shared schema (`cayceops/models.py`, migration `alembic/versions/0001_initial.py`):
`proposals`, `audit_log`, `mail_messages`, `content_items`, `daemon_runs`.
## 2. Every external call (where, what it sends, what triggers it)
| # | Call | Code | Sends | Trigger |
|---|------|------|-------|---------|
| 1 | HTTP GET to scanner sources | `cayceops/components/scanner/scanner.py` → `_fetch` (urllib) | nothing but the request (+ `CAYCE_OPS_SCANNER_USER_AGENT` header) to URLs **configured by the operator** | `cayceops scanner run` (cron/systemd timer) |
| 2 | IMAP fetch (IMAP4_SSL, UNSEEN) | `cayceops/components/maild/maild.py` → `connect_imap`, `_fetch_unread` | credentials from env; reads inbox | `cayceops maild poll` |
| 3 | SMTP send | `maild.py` → `_smtp_send` (smtplib, STARTTLS) | a reply **draft a human approved** to the original sender's bare address | `cayceops maild send-approved`, and only rows with `draft_status=approved` |
| 4 | LLM chat completion (openai-compatible) | `cayceops/llm.py` → `OpenAICompatProvider.complete` (urllib) | the email text (truncated 4000 chars) + system prompt; `Authorization: Bearer <env key>` | `maild poll` **only if** `CAYCE_OPS_LLM_PROVIDER != null` (default is null → never) |
| 5 | Nostr relay publish (websocket) | `cayceops/components/publisher/nostr.py` → `relay_publish`/`publish` | a signed NIP-01 event whose content is the **human-approved** caption | `cayceops publisher publish-due`, channel=nostr, only if key+relays configured |
| 6 | Meta Graph API | `cayceops/components/publisher/adapters.py` | caption (+optional hosted image URL) + page token | **refuses to run** unless `CAYCE_OPS_ENABLE_FACEBOOK/INSTAGRAM=true` (default false). Currently unreachable from `publish_due`. |
No other network code exists. Grep guards: `urllib.request`, `imaplib`, `smtplib`,
`websockets` appear only in the files above (verified by review; CI could add a
lint rule — see *Known risks*).
## 3. What each component can do without a human / with AUTONOMOUS_MODE=true
| Component | Without any human (default) | With `CAYCE_OPS_AUTONOMOUS_MODE=true` |
|---|---|---|
| scanner | fetch sources, file proposals, write audit | identical (scanner never executes) |
| executor | nothing executes by itself; it acts only on rows a human set to `approved` (via `executor approve --actor …`) | **one added path**: auto-approves `proposed` items whose `kind` ∈ `CAYCE_OPS_AUTONOMOUS_KINDS`, capped at `CAYCE_OPS_AUTONOMOUS_DAILY_CAP`/day, actor `auto:policy`, fully audited. Then executes those. |
| maild | polls IMAP, classifies, writes drafts. **Never sends.** | identical — autonomy does not touch mail |
| publisher | plans drafts; publishes nothing | identical — publishing always requires `publisher approve --actor …`; autonomy has no effect |
| monitor | reads and renders | identical |
Explicitly impossible in the current code: sending an unapproved mail draft,
publishing unapproved content, executing a non-approved proposal (the single
exception is the audited `auto:policy` path above), disabling the audit write.
## 4. Environment variables and permissions per component
All config is env-only (`cayceops/config.py`), prefix `CAYCE_OPS_`. Full list
with placeholders: `.env.example`.
| Component | Requires | Notes |
|---|---|---|
| core | `DATABASE_URL` (postgres or sqlite), `LOG_LEVEL`, `STATE_DIR`, `MEDIA_DIR` | DB user needs SELECT/INSERT/UPDATE on the five tables |
| scanner | `SCANNER_SOURCES` (JSON), `SCANNER_KEYWORDS`, `SCANNER_USER_AGENT` | outbound GET only |
| executor | `AUTONOMOUS_MODE`, `AUTONOMOUS_KINDS`, `AUTONOMOUS_DAILY_CAP`, `MAX_ATTEMPTS`, `ENABLE_HTTP_ACTIONS` | `http_post` plugin stays inert unless enabled |
| maild | `IMAP_HOST/PORT/USER/PASSWORD/MAILBOX`, `SMTP_HOST/PORT/USER/PASSWORD`, `MAIL_FROM`, `LLM_PROVIDER/MODEL/BASE_URL/API_KEY/REVIEW_THRESHOLD` | IMAP user is read+mark-seen; SMTP login can send only as `MAIL_FROM` |
| publisher | `NOSTR_PRIVATE_KEY` (32-byte hex), `NOSTR_RELAYS`, `ENABLE_FACEBOOK`, `FACEBOOK_PAGE_TOKEN`, `ENABLE_INSTAGRAM`, `INSTAGRAM_USER_ID` | nostr key = signing authority; treat as secret |
| monitor | `MONITOR_OUTPUT` (writable path) | needs read on DB, write on output only |
Filesystem permissions: default deployment is the `cayce` user owning
`/srv/cayce-ops` with systemd hardening (`NoNewPrivileges`, `ProtectSystem=strict`,
`ProtectHome=yes`) — see `deploy/systemd/cayce-ops@.service`.
## 5. Dependencies (pinned in `pyproject.toml` + `uv.lock`)
| Package | Version | Why |
|---|---|---|
| sqlalchemy | 2.0.44 | ORM + engine for the shared schema |
| alembic | 1.17.0 | production migrations |
| typer | 0.20.0 | CLI (all five daemons are CLI commands) |
| psycopg[binary] | 3.2.13 | Postgres driver (production stack) |
| pillow | 12.0.0 | brand image templates (publisher) |
| coincurve | 21.0.0 | libsecp256k1 binding: NIP-01 event signing (publisher) |
| websockets | 15.0.1 | nostr relay publish (sync client) |
| pytest | 9.0.2 (dev) | test suite |
| ruff | 0.15.1 (dev) | lint |
stdlib doing heavy lifting: `imaplib`/`smtplib`/`email` (maild), `urllib` (scanner,
LLM, Meta adapters), `xml.etree`/`html.parser` (scanner), `hashlib`/`json` (nostr ids).
## 6. Known risks and open items
1. **No live Gitea Actions runner on the current host** (DECISIONS D6). CI files
ship for GitHub/Gitea Actions, and the same gate runs locally pre-push
(`scripts/pre-push`), but until a runner exists there is no third-party
enforcement of the gate on every push.
2. **`auto:policy` path is real autonomy** — bounded (kind allowlist + daily cap +
audit) but it does execute without a human in the loop when enabled. The cap
counter is DB-based; wiping `audit_log` resets it.Mitigation: only the
operator can set the env flags, and the audit log is append-only by convention.
3. **Executor plugins are in-process Python** (D9): no sandboxing. A malicious
plugin = arbitrary code. Mitigation: plugins are repo code, reviewed in PRs;
the registry starts with `log_action` fallbacks only.
4. **Mail classification is heuristic** (rules; LLM optional). Conservative by
design (low confidence → `needs_review`, unthreaded approvals → review), but
a sophisticated phishing email could still be misclassified `other`. Human
gate on sends bounds the damage to drafting wrong replies (never sending them).
5. **Nostr relays are plaintext-config endpoints**: a hostile relay sees the
note content (public by design) and could replay it; event ids make duplication
detectable. No NIP-05, no retries across relay pools yet.
6. **`detail` JSON columns are schemaless** — writers evolve payloads without
migrations. Monitor intentionally renders none of it.
7. **gitleaks allowlist covers `tests/`** (D7): a real secret pasted under
`tests/` would not trip the scanner. Compensating control: PR review; no
fixture reads the environment.
8. **Open:** provision an act_runner for Gitea Actions; wire `posts_log`-style
engagement back into publisher planning; add NIP-05 identifier; retries for
relay publish.
## 7. Running tests and the local stack
```bash
uv sync --frozen # python 3.12 + locked deps
uv run pytest # 44 tests, sqlite, no network
make check # lint + tests + alembic smoke + gitleaks (pre-push gate)
# full local stack (five daemons + postgres 16):
cp .env.example .env # fill placeholders; nothing real required for db-only
docker compose up -d db
uv run alembic upgrade head
docker compose up -d # or run each: uv run cayceops {scanner,executor,maild,publisher,monitor} …
# production path (VPS):
# see deploy/systemd/README.md (hardened units + timers)
```
## Audited commit
- Repository: `cayce/cayce-ops` (branch `main`)
- **Audited: d500d888701c3f1cd3305d098a448c3e20a8ad99** (merge of PR #7)
- Gate at that commit: ruff clean (0 findings) · 44/44 tests · alembic 0001
smoke OK · gitleaks `--all` no leaks · `make check` exit 0.
---
## Response to external audit — 2026-10-20
External audit verdict reproduced our gate and found 5 critical + 6 medium
issues. All criticals and all mediums are fixed in **c9bd441** (PR #9):
| # | Finding | Fix |
|---|---|---|
| C1 | approval forgeable with any `--actor` | hash-chained `audit_log` (`verify_audit_chain`); optional-required operator BIP-340 signature (`CAYCE_OPS_OPERATOR_PUBKEY`); `deploy/roles.sql` least-privilege DB roles |
| C2 | mail threading dead (no OUTBOUND rows, inbound id overwritten, no first-contact) | sends create OUTBOUND rows (own Message-ID + In-Reply-To); inbound id untouched; `maild compose`; chain fallback in `_match_proposal` |
| C3 | duplicate sends on SMTP failure | per-message transactions; failures audited (`send_failed`), row stays queued |
| C4 | negated approvals classified as approvals; LLM prompt `.format()` KeyError | negation rules first (0.95 REJECTION); prompt braces escaped |
| C5 | systemd units ran nonexistent `run` subcommands | `run` added to all five daemons; `publisher plan` timer documented |
| M1-M6 | gate not self-installed; approved text unbound; `risk_score` misnamed/ordered; postgres 0.0.0.0; IMAP seen-before-commit; publisher noise/infinite retries | `make hooks`; `approved_hash` bindings; `match_score` (migration 0002) + relevance-desc; `127.0.0.1` bind; `\Seen` post-commit; `SKIPPED` terminal + attempt budget |
New audited commit: **c9bd441**. Gate at that commit: ruff 0 · 56/56 tests ·
alembic 0001→0002 · gitleaks `--all` clean · `make check` exit 0.
Still open (declared): apply `roles.sql` on production, provision CI runner.
+5 -1
View File
@@ -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"
+68
View File
@@ -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")
+44 -3
View File
@@ -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()
+1
View File
@@ -0,0 +1 @@
"""executor package."""
+299
View File
@@ -0,0 +1,299 @@
"""executor — executes APPROVED proposals through registered action plugins.
Guarantees
----------
- Only ``approved`` proposals execute (``AUTONOMOUS_MODE=false`` default).
- With ``AUTONOMOUS_MODE=true`` the executor gains exactly one capability:
bounded auto-approval of ``proposed`` items whose ``kind`` is in
``CAYCE_OPS_AUTONOMOUS_KINDS``, capped by ``CAYCE_OPS_AUTONOMOUS_DAILY_CAP``,
recorded with actor ``auto:policy`` (DECISIONS D5).
- Every transition writes an ``audit_log`` row: who approved, what ran,
plugin output, success/failure.
- Unknown proposal kinds fall back to the ``log_action`` plugin (never a
network side effect). See DECISIONS D9.
"""
from __future__ import annotations
from collections.abc import Callable
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import func, select
from cayceops.components.executor import plugins as plugin_registry
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, audit
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
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:
"""Human rejects a proposal. Audited with the operator's identity."""
return _decide(proposal_id, actor, ProposalStatus.REJECTED, cfg, sess)
def _decide(
proposal_id: int,
actor: Actor,
decision: ProposalStatus,
cfg: Config | None,
sess,
signature: str | None = None,
) -> int:
owned = False
if sess is None:
outer = session_scope()
sess = outer.__enter__()
owned = True
try:
proposal = sess.get(Proposal, proposal_id)
if proposal is None:
return 1
if proposal.status != ProposalStatus.PROPOSED:
verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
audit(
sess,
actor=actor,
action=verb,
proposal_id=proposal_id,
result="error",
detail={"reason": f"invalid state: {proposal.status.value}"},
)
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, detail=detail)
return 0
finally:
if owned:
outer.__exit__(None, None, None)
# --- autonomous policy (the ONLY non-human approval path, D5) -----------------
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
midnight = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
used = (
sess.query(func.count(AuditEntry.id))
.filter(
AuditEntry.actor == "auto:policy",
AuditEntry.action == "approve",
AuditEntry.ts >= midnight,
)
.scalar()
)
budget = max(0, cfg.autonomous_daily_cap - int(used or 0))
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.match_score.desc(), Proposal.id.asc())
.limit(budget)
.all()
)
for proposal in candidates:
proposal.status = ProposalStatus.APPROVED
proposal.approved_by = "auto:policy"
proposal.approved_at = datetime.now(UTC)
audit(
sess,
actor="auto:policy",
action="approve",
proposal_id=proposal.id,
detail={
"reason": "autonomous_mode policy",
"kind": proposal.kind,
"match_score": proposal.match_score,
"daily_cap": cfg.autonomous_daily_cap,
},
)
approved += 1
return approved
# --- execution ----------------------------------------------------------------
def execute_one(proposal: Proposal, cfg: Config, sess) -> bool:
"""Run the plugin for one approved proposal. Returns True on success."""
proposal.status = ProposalStatus.RUNNING
proposal.attempts += 1
audit(
sess,
actor="daemon:executor",
action="execute_start",
proposal_id=proposal.id,
detail={"plugin": proposal.kind, "attempt": proposal.attempts},
)
plugin = plugin_registry.REGISTRY.get(proposal.kind)
try:
if plugin is None:
outcome: dict[str, Any] = plugin_registry.log_action(proposal, cfg, sess)
outcome["plugin_fallback"] = True
outcome["note"] = f"no plugin registered for kind {proposal.kind!r}"
else:
outcome = plugin(proposal, cfg, sess)
except Exception as exc: # noqa: BLE001 — plugin boundary: capture for audit
audit(
sess,
actor="daemon:executor",
action="execute_error",
proposal_id=proposal.id,
result="error",
detail={"error": str(exc)[:300], "attempt": proposal.attempts},
)
if proposal.attempts >= cfg.max_attempts:
proposal.status = ProposalStatus.FAILED
audit(
sess,
actor="daemon:executor",
action="fail",
proposal_id=proposal.id,
result="error",
detail={"attempts": proposal.attempts},
)
else:
proposal.status = ProposalStatus.APPROVED # retry next run
return False
proposal.status = ProposalStatus.DONE
audit(
sess,
actor="daemon:executor",
action="execute_done",
proposal_id=proposal.id,
detail={"outcome": outcome},
)
return True
def run(cfg: Config | None = None, sess=None) -> int:
"""CLI entry: auto-approve per policy, then execute all approved items."""
from cayceops.config import load
from cayceops.db import engine as _engine
cfg = cfg or load()
owned = False
if sess is None:
outer = session_scope()
sess = outer.__enter__()
owned = True
try:
run_row = DaemonRun(daemon="executor")
sess.add(run_row)
sess.flush()
try:
auto = _auto_approve(cfg, sess)
pending = (
sess.query(Proposal)
.filter(Proposal.status == ProposalStatus.APPROVED)
.order_by(Proposal.approved_at.asc())
.all()
)
done = failed = 0
for proposal in pending:
if execute_one(proposal, cfg, sess):
done += 1
else:
failed += 1
run_row.status = "ok"
run_row.detail = {"auto_approved": auto, "done": done, "failed": failed}
except Exception as exc: # noqa: BLE001 — daemon boundary
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.finished_at = datetime.now(UTC)
return 0
finally:
if owned:
outer.__exit__(None, None, None)
_engine().dispose()
def select_approved(sess) -> list[Proposal]:
"""Helper for tests/tools: currently approved proposals, oldest first."""
return list(
sess.scalars(
select(Proposal)
.where(Proposal.status == ProposalStatus.APPROVED)
.order_by(Proposal.approved_at.asc())
).all()
)
ActionPlugin = Callable[[Proposal, Config, Any], dict]
+78
View File
@@ -0,0 +1,78 @@
"""executor action plugins.
A plugin maps a proposal ``kind`` to a side effect. Plugins are repo code —
they are reviewed like everything else and must:
- take ``(proposal, cfg, sess)`` and return a JSON-safe dict (the audit outcome)
- do their own permission checks (see ``http_post`` + ENABLE_HTTP_ACTIONS)
- raise on failure (the executor captures the exception into the audit log)
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request
from typing import Any
from cayceops.config import Config
from cayceops.models import Proposal, audit
REGISTRY: dict[str, Any] = {}
def register(kind: str):
"""Decorator: register an action plugin for a proposal kind."""
def deco(fn):
REGISTRY[kind] = fn
return fn
return deco
@register("log_action")
def log_action(proposal: Proposal, cfg: Config, sess) -> dict:
"""No-side-effect plugin: record the execution in the audit log only."""
return {"action": "logged", "title": proposal.title[:200]}
@register("http_post")
def http_post(proposal: Proposal, cfg: Config, sess) -> dict:
"""POST the proposal payload to a webhook.
Inert unless ``CAYCE_OPS_ENABLE_HTTP_ACTIONS=true``. Target URL comes from
the proposal detail (``detail['webhook_url']``) — operator-controlled data.
"""
if not cfg.enable_http_actions:
audit(
sess,
actor="daemon:executor",
action="http_post_blocked",
proposal_id=proposal.id,
result="error",
detail={"reason": "CAYCE_OPS_ENABLE_HTTP_ACTIONS is false"},
)
raise PermissionError("http_post disabled (CAYCE_OPS_ENABLE_HTTP_ACTIONS=false)")
url = (proposal.detail or {}).get("webhook_url", "")
if not url:
raise ValueError("proposal detail.webhook_url missing")
payload = json.dumps(
{"proposal_id": proposal.id, "kind": proposal.kind, "title": proposal.title}
).encode()
req = urllib.request.Request(
url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=20.0) as resp: # noqa: S310
return {"action": "http_post", "status": resp.status}
# proposals from the scanner default to content_gap/affiliate/partnership kinds;
# until dedicated plugins exist for those kinds they must not do anything
# network-visible. register more plugins here as the operator approves them.
for _safe_kind in ("affiliate", "partnership", "content_gap", "other"):
REGISTRY.setdefault(_safe_kind, log_action)
+1
View File
@@ -0,0 +1 @@
"""maild package."""
+105
View File
@@ -0,0 +1,105 @@
"""maild classification: rules first, LLM optional, conservative threshold.
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.
"""
from __future__ import annotations
import json
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|we regret|won.t be able)\w*", re.I), # noqa: E501
0.9,
),
(
MailClassification.APPROVAL,
re.compile(r"\b(approv|accept|welcome to the|you.re (in|approved)|live|activated|glad to)\w*", re.I), # noqa: E501
0.85,
),
(
MailClassification.QUESTION,
re.compile(r"\b(could you|can you|please (provide|clarify|confirm)|what (is|are)|how (do|does|can)|need more (info|details))\b", re.I), # noqa: E501
0.7,
),
]
# 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}}
Subject and body:
{body}"""
def _rules(text: str) -> tuple[MailClassification, float]:
for kind, pattern, confidence in _RULES:
if pattern.search(text):
return kind, confidence
return MailClassification.OTHER, 0.3
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
try:
provider = get_provider(cfg)
raw = provider.complete(
_LLM_PROMPT.format(body=text[:4000]),
system="You classify merchant emails. Reply with JSON only.",
)
data = json.loads(re.search(r"\{.*\}", raw, re.S).group(0))
kind = MailClassification(data["classification"].lower())
confidence = float(data["confidence"])
except Exception: # noqa: BLE001 — LLM is best-effort by design
return None
if kind not in MailClassification or kind is MailClassification.NEEDS_REVIEW:
return None
return kind, min(max(confidence, 0.0), 1.0)
def classify(text: str, cfg: Config) -> tuple[MailClassification, float]:
"""Classify merchant text. Degrades to rules when no LLM is configured."""
if cfg.llm_provider != "null":
llm_result = _llm_classify(text, cfg)
if llm_result is not None:
kind, confidence = llm_result
else: # LLM failed — fall through to rules
kind, confidence = _rules(text)
else:
kind, confidence = _rules(text)
if confidence < cfg.llm_review_threshold:
return MailClassification.NEEDS_REVIEW, confidence
return kind, confidence
+566
View File
@@ -0,0 +1,566 @@
"""maild — classify merchant replies, draft responses, gated SMTP.
Flow
----
1. ``poll``: fetch unread inbound mail (IMAP), parse, thread to the proposal
via ``In-Reply-To`` of a previously-sent draft, classify (rules + optional
LLM), store, and create a REPLY DRAFT (never sent automatically).
2. ``approve_draft``: a human approves a specific draft (audited).
3. ``send_approved``: SMTP sends *only* drafts in state ``approved``.
Classification is conservative: anything below the confidence threshold becomes
``needs_review`` and no draft is generated.
Public signatures match the CLI: ``poll()``, ``approve_draft(id, actor)``,
``send_approved()``. The ``_*`` variants take an explicit session/fetcher for
tests and composition.
"""
from __future__ import annotations
import email
import email.policy
import imaplib
import smtplib
from datetime import UTC, datetime
from email.message import EmailMessage
from cayceops.components.maild.classify import classify as _classify_text
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import (
DaemonRun,
DraftStatus,
MailClassification,
MailDirection,
MailMessage,
Proposal,
_content_hash,
audit,
)
# --- IMAP ---------------------------------------------------------------------
def connect_imap(cfg: Config) -> imaplib.IMAP4_SSL:
client = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
client.login(cfg.imap_user, cfg.imap_password)
client.select(cfg.imap_mailbox)
return client
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")
out: list[tuple[bytes, bytes]] = []
for num in data[0].split():
_, msg_data = client.fetch(num, "(RFC822)")
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")
finally:
client.logout()
# --- parsing ------------------------------------------------------------------
def parse_raw(raw: bytes) -> dict:
"""Parse raw RFC822 bytes into our flat dict shape."""
msg = email.message_from_bytes(raw, policy=email.policy.default)
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain" and not part.get_filename():
body = part.get_content()
break
else:
body = msg.get_content()
reply_to = msg.get("In-Reply-To") or msg.get("References") or ""
message_id = (msg.get("Message-ID") or "").strip()
if not message_id:
message_id = f"<generated-{datetime.now(UTC).timestamp()}@localhost>"
# store the bare address (envelope-safe); display names stay out of the DB
from_addr = email.utils.parseaddr(msg.get("From") or "")[1].strip()[:320]
return {
"message_id": message_id[:998],
"in_reply_to": reply_to.split()[-1].strip() if reply_to.strip() else None,
"from_addr": from_addr,
"subject": (msg.get("Subject") or "").strip()[:998],
"body": body[:20000],
}
# --- proposal threading -------------------------------------------------------
def _match_proposal(parsed: dict, sess) -> Proposal | None:
"""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
sent = (
sess.query(MailMessage)
.filter(
MailMessage.direction == MailDirection.OUTBOUND,
MailMessage.message_id == irt,
)
.one_or_none()
)
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 -------------------------------------------------------------------
_DRAFT_TEMPLATES = {
MailClassification.APPROVAL: (
"Hello,\n\nThank you for the approval regarding \"{title}\". "
"We will proceed on our side and follow up with the next steps.\n\nBest regards,\nCayce"
),
MailClassification.REJECTION: (
"Hello,\n\nThank you for considering \"{title}\". "
"We appreciate the reply and would welcome the chance to revisit in the future.\n\n"
"Best regards,\nCayce"
),
MailClassification.QUESTION: (
"Hello,\n\nRegarding \"{title}\": {question}\n\n"
"Could you help us with the details so we can move forward?\n\nBest regards,\nCayce"
),
}
def _build_draft(parsed: dict, kind: MailClassification, proposal: Proposal | None) -> str:
title = proposal.title if proposal else (parsed.get("subject") or "your message")
template = _DRAFT_TEMPLATES.get(kind)
if template is None:
return ""
question = (parsed.get("body") or "")[:200].strip() or "we had a question"
return template.format(title=title[:120], question=question)
# --- inbound pass ---------------------------------------------------------------
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}
processed_nums: list[bytes] = []
for num, raw in (fetcher or _fetch_unread)(cfg):
parsed = parse_raw(raw)
if (
sess.query(MailMessage)
.filter(MailMessage.message_id == parsed["message_id"])
.one_or_none()
):
stats["skipped"] += 1
continue
proposal = _match_proposal(parsed, sess)
kind, confidence = _classify_text(f"{parsed['subject']}\n{parsed['body']}", cfg)
if proposal is None and kind == MailClassification.APPROVAL:
# an approval we cannot thread is suspicious — human must look
kind = MailClassification.NEEDS_REVIEW
confidence = min(confidence, 0.4)
record = MailMessage(
direction=MailDirection.INBOUND,
message_id=parsed["message_id"],
in_reply_to=parsed["in_reply_to"],
from_addr=parsed["from_addr"] or None,
subject=parsed["subject"] or None,
body=parsed["body"],
classification=kind,
confidence=confidence,
proposal_id=proposal.id if proposal else None,
processed_at=datetime.now(UTC),
)
sess.add(record)
sess.flush()
audit(
sess,
actor="daemon:maild",
action="classify",
proposal_id=proposal.id if proposal else None,
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
continue
if proposal is not None:
proposal.detail = {
**(proposal.detail or {}),
"merchant_status": kind.value,
"merchant_reply_at": datetime.now(UTC).isoformat(),
}
draft = _build_draft(parsed, kind, proposal)
if draft:
record.draft_reply = draft
record.draft_status = DraftStatus.DRAFTED
audit(
sess,
actor="daemon:maild",
action="draft_reply",
proposal_id=proposal.id if proposal else None,
detail={"mail_id": record.id},
)
stats["drafted"] += 1
return stats, processed_nums
# --- human gate: approve draft -------------------------------------------------
def _approve_draft(mail_id: int, actor: str, sess) -> 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).
"""
record = sess.get(MailMessage, mail_id)
if record is None:
return 1
if record.draft_status != DraftStatus.DRAFTED or not record.draft_reply:
audit(
sess,
actor=actor,
action="approve_draft",
proposal_id=record.proposal_id,
result="error",
detail={"reason": f"invalid draft state: {record.draft_status.value}"},
)
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, "hash": record.approved_hash},
)
return 0
# --- SMTP: only approved drafts ------------------------------------------------
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()
server.login(cfg.smtp_user, cfg.smtp_password)
server.send_message(msg)
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 _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 ""
subject = f"Re: {record.subject or 'your message'}"
in_reply_to = record.message_id
outbound_msgid = (
f"<cayce-out-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
)
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_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,
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"<cayce-out-{proposal.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>",
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) ------------------------------------
def poll(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
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, nums = _poll(cfg, sess)
except Exception as exc: # noqa: BLE001 — daemon boundary
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
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()
def approve_draft(mail_id: int, actor: str) -> int:
from cayceops.db import engine
try:
with session_scope() as sess:
return _approve_draft(mail_id, actor, sess)
finally:
engine().dispose()
def send_approved(cfg: Config | None = None) -> int:
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:
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()
+1
View File
@@ -0,0 +1 @@
"""monitor package."""
+174
View File
@@ -0,0 +1,174 @@
"""monitor — static status page builder.
Reads the shared database and renders a self-contained HTML file (no external
assets, no JS, everything escaped). Contains NO sensitive data: daemon health,
proposal queue counts, and recent audit activity (actor/action/result only —
no email bodies, no detail payloads, no URLs).
"""
from __future__ import annotations
import html
from datetime import UTC, datetime
from sqlalchemy import func, select
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import (
AuditEntry,
ContentItem,
DaemonRun,
MailMessage,
Proposal,
ProposalStatus,
)
DAEMONS = ("scanner", "executor", "maild", "publisher", "monitor")
_STYLE = """<style>
body{font-family:ui-monospace,Menlo,monospace;background:#0a0908;color:#f2ece0;margin:2rem}
h1{color:#c9a84c;font-weight:600} h2{color:#c9a84c;margin-top:2rem;font-size:1.1rem}
table{border-collapse:collapse;margin:.5rem 0}
td,th{border:1px solid #333;padding:.35rem .8rem;text-align:left}
.ok{color:#6fce6f}.error{color:#ef6b6b}.muted{color:#888;font-size:.85rem}
.card{display:inline-block;background:#141210;border:1px solid #2a2a2a;
border-radius:8px;padding:.8rem 1.4rem;margin:.3rem}
</style>"""
def _fmt_ts(dt) -> str:
return dt.strftime("%Y-%m-%d %H:%M:%SZ") if dt else "—"
def _daemon_rows(sess) -> str:
rows = []
for name in DAEMONS:
last = (
sess.query(DaemonRun)
.filter(DaemonRun.daemon == name)
.order_by(DaemonRun.started_at.desc())
.first()
)
if last is None:
rows.append(
"<tr><td>" + name + "</td><td class=muted>never run</td><td>—</td></tr>"
)
else:
cls = "ok" if last.status == "ok" else "error"
when = _fmt_ts(last.finished_at or last.started_at)
rows.append(
"<tr><td>" + html.escape(name) + "</td>"
+ '<td class="' + cls + '">' + html.escape(last.status) + "</td>"
+ "<td>" + when + "</td></tr>"
)
head = "<tr><th>daemon</th><th>last status</th><th>last run</th></tr>"
return "<table>" + head + "".join(rows) + "</table>"
def _queue_cards(sess) -> str:
counts = dict(
sess.query(Proposal.status, func.count(Proposal.id)).group_by(Proposal.status).all()
)
cards = []
for status in ProposalStatus:
value = counts.get(status, 0)
cards.append(
"<div class=card>" + html.escape(status.value)
+ ": <strong>" + str(value) + "</strong></div>"
)
return "".join(cards)
def _recent_audit(sess, limit: int = 20) -> str:
rows = []
query = sess.query(AuditEntry).order_by(AuditEntry.ts.desc()).limit(limit)
for entry in query.all():
cls = "ok" if entry.result == "ok" else "error"
rows.append(
"<tr><td>" + _fmt_ts(entry.ts) + "</td>"
+ "<td>" + html.escape(entry.actor) + "</td>"
+ "<td>" + html.escape(entry.action) + "</td>"
+ '<td class="' + cls + '">' + html.escape(entry.result) + "</td></tr>"
)
body = "".join(rows) or "<tr><td colspan=4 class=muted>no activity yet</td></tr>"
return (
"<table><tr><th>time (utc)</th><th>actor</th><th>action</th><th>result</th></tr>"
+ body
+ "</table>"
)
def _build(cfg: Config, sess) -> str:
now = datetime.now(UTC)
mail_unprocessed = (
sess.query(func.count(MailMessage.id))
.filter(
MailMessage.direction == "in",
MailMessage.draft_status.in_(["drafted", "none"]),
MailMessage.classification.isnot(None),
)
.scalar()
)
content_pending = (
sess.query(func.count(ContentItem.id))
.filter(ContentItem.status.in_(["draft", "approved"]))
.scalar()
)
generated = _fmt_ts(now)
page = f"""<!doctype html>
<html lang=en><head><meta charset=utf-8><title>cayce-ops status</title>{_STYLE}</head>
<body>
<h1>cayce-ops · status</h1>
<p class=muted>generated {generated} · static page, no sensitive data by design</p>
<h2>daemon health</h2>
{_daemon_rows(sess)}
<h2>proposal queue</h2>
{_queue_cards(sess)}
<h2>gates</h2>
<div class=card>mail drafts awaiting approval: <strong>{mail_unprocessed}</strong></div>
<div class=card>content items draft/approved: <strong>{content_pending}</strong></div>
<h2>recent audit activity</h2>
{_recent_audit(sess)}
<p class=muted>audit entries: {sess.scalar(select(func.count(AuditEntry.id)))} total</p>
</body></html>"""
import os
os.makedirs(os.path.dirname(cfg.monitor_output) or ".", exist_ok=True)
with open(cfg.monitor_output, "w", encoding="utf-8") as fh:
fh.write(page)
return cfg.monitor_output
def build(cfg: Config | None = None) -> int:
"""CLI entry: render the status page inside a DaemonRun bookkeeping wrap."""
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="monitor")
sess.add(run_row)
sess.flush()
try:
out = _build(cfg, sess)
except Exception as exc: # noqa: BLE001 — daemon boundary
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.status = "ok"
run_row.detail = {"output": out}
run_row.finished_at = datetime.now(UTC)
print(out)
return 0
finally:
engine().dispose()
@@ -0,0 +1 @@
"""publisher package."""
+61
View File
@@ -0,0 +1,61 @@
"""Optional Facebook/Instagram adapters (Graph API).
DISABLED BY DEFAULT. Each adapter refuses to run unless its enable flag is
true (``CAYCE_OPS_ENABLE_FACEBOOK`` / ``CAYCE_OPS_ENABLE_INSTAGRAM``), so the
default deployment never touches Meta endpoints. Tokens are env-only and are
never logged.
"""
from __future__ import annotations
import json
import urllib.parse
import urllib.request
GRAPH = "https://graph.facebook.com/v21.0"
def _post(url: str, payload: dict) -> dict:
data = urllib.parse.urlencode(payload).encode()
req = urllib.request.Request(url, data=data, method="POST") # noqa: S310
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode())
def post_facebook(cfg, caption: str, media_path: str | None = None) -> dict:
"""Post to the page feed (text) or /photos (with local media). Gated."""
if not cfg.enable_facebook:
raise PermissionError("facebook adapter disabled (CAYCE_OPS_ENABLE_FACEBOOK=false)")
if not cfg.facebook_page_token:
raise ValueError("CAYCE_OPS_FACEBOOK_PAGE_TOKEN missing")
if media_path:
# photo posts require a reachable URL or multipart upload; we use the
# simple URL form — media must be hosted (out of scope for local files)
result = _post(
f"{GRAPH}/me/photos",
{"url": media_path, "caption": caption, "access_token": cfg.facebook_page_token},
)
else:
result = _post(
f"{GRAPH}/me/feed",
{"message": caption, "access_token": cfg.facebook_page_token},
)
return {"post_id": result.get("post_id") or result.get("id", "")}
def post_instagram(cfg, caption: str, image_url: str) -> dict:
"""Two-step IG publish (container -> publish). Gated."""
if not cfg.enable_instagram:
raise PermissionError("instagram adapter disabled (CAYCE_OPS_ENABLE_INSTAGRAM=false)")
if not cfg.instagram_user_id:
raise ValueError("CAYCE_OPS_INSTAGRAM_USER_ID missing")
token = cfg.facebook_page_token # IG Graph publishing uses the page token
container = _post(
f"{GRAPH}/{cfg.instagram_user_id}/media",
{"image_url": image_url, "caption": caption, "access_token": token},
)
published = _post(
f"{GRAPH}/{cfg.instagram_user_id}/media_publish",
{"creation_id": container["id"], "access_token": token},
)
return {"post_id": published.get("id", "")}
+82
View File
@@ -0,0 +1,82 @@
"""Brand image templates (Pillow). 1080x1350 portrait, two brand styles.
Deterministic, template-driven graphics — deliberately not AI-generated media
(DECISIONS: nothing here trips AI-media disclosure requirements; it is typeset
design). Fonts: Pillow's bundled default (scalable), so no system font deps.
"""
from __future__ import annotations
import os
from PIL import Image, ImageDraw, ImageFont
WIDTH, HEIGHT = 1080, 1350
_BRANDS = {
# bitcoinhotels — warm gold on near-black
"bh": {
"bg": (20, 18, 16),
"accent": (201, 168, 76),
"text": (242, 236, 224),
"muted": (150, 140, 120),
},
# hardcap — bitcoin orange on cool black
"hc": {
"bg": (13, 13, 16),
"accent": (247, 147, 26),
"text": (250, 250, 250),
"muted": (120, 120, 130),
},
}
def _font(size: int, bold: bool = False):
# Pillow >=10: load_default(size=...) returns a scalable bundled font
return ImageFont.load_default(size=size)
def _vgrad(top: tuple, bottom: tuple) -> Image.Image:
"""Simple two-color vertical gradient."""
img = Image.new("RGB", (WIDTH, HEIGHT))
for y in range(HEIGHT):
t = y / (HEIGHT - 1)
row = tuple(int(top[i] + (bottom[i] - top[i]) * t) for i in range(3))
ImageDraw.Draw(img).line([(0, y), (WIDTH, y)], fill=row)
return img
def render_brand_image(
*,
brand: str = "bh",
kicker: str | None,
lines: list[str],
tagline: list[str] | None = None,
url: str | None = None,
out_path: str,
) -> str:
"""Render one branded poster. Returns out_path. Raises on unknown brand."""
if brand not in _BRANDS:
raise ValueError(f"unknown brand {brand!r} (have: {sorted(_BRANDS)})")
palette = _BRANDS[brand]
img = _vgrad(palette["bg"], tuple(max(0, c - 8) for c in palette["bg"]))
draw = ImageDraw.Draw(img)
x, y = 90, 240
if kicker:
draw.text((x, y), kicker, font=_font(40), fill=palette["muted"])
y += 110
for line in lines:
draw.text((x, y), line, font=_font(96, bold=True), fill=palette["text"])
y += 118
y += 24
for t in tagline or []:
draw.text((x, y), t, font=_font(38), fill=palette["muted"])
y += 58
draw.rectangle([x, y + 30, x + 470, y + 33], fill=palette["accent"])
if url:
draw.text((x, y + 110), url, font=_font(44), fill=palette["accent"])
os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
img.save(out_path, "JPEG", quality=92)
return out_path
+186
View File
@@ -0,0 +1,186 @@
"""Minimal Nostr client: NIP-01 kind-1 notes (build, sign, verify, publish).
Implemented in-repo (DECISIONS D4) so every byte of the signing path is
auditable: event id = sha256 of the canonical NIP-01 serialization; signature =
BIP-340 Schnorr over the id. Keys are 32-byte hex. Relays are plain env config.
Publishing is fire-and-wait: open a websocket per relay, send the EVENT, wait
for the OK frame (or timeout), and return the per-relay results.
"""
from __future__ import annotations
import hashlib
import json
from typing import Any
from coincurve import PrivateKey, PublicKey
KIND_NOTE = 1
def normalize_key(private_key_hex: str) -> bytes:
"""Accept 64-char hex (with or without 0x). Anything else is a config error."""
key = private_key_hex.strip().lower().removeprefix("0x")
if len(key) != 64 or not all(c in "0123456789abcdef" for c in key):
raise ValueError("CAYCE_OPS_NOSTR_PRIVATE_KEY must be 64-char hex (32 bytes)")
return bytes.fromhex(key)
def pubkey_from_secret(secret: bytes) -> str:
return PublicKey.from_secret(secret).format().hex()[2:]
def build_event(
secret: bytes,
content: str,
tags: list[list[str]] | None = None,
created_at: int | None = None,
kind: int = KIND_NOTE,
) -> dict[str, Any]:
"""Build and sign a NIP-01 event. Deterministic given the same inputs."""
created_at = int(created_at if created_at is not None else __import__("time").time())
pubkey = pubkey_from_secret(secret)
serialized = json.dumps(
[0, pubkey, created_at, kind, tags or [], content],
separators=(",", ":"), # canonical NIP-01 form
ensure_ascii=False,
)
event_id = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
# zero aux randomness -> fully deterministic signatures (BIP-340 permits it;
# verification is unaffected either way)
signature = (
PrivateKey(secret)
.sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32)
.hex()
)
return {
"pubkey": pubkey,
"created_at": created_at,
"kind": kind,
"tags": tags or [],
"content": content,
"id": event_id,
"sig": signature,
}
def _schnorr_verify(sig32: bytes, msg32: bytes, xonly_pub32: bytes) -> bool:
"""BIP-340 reference verification (pure Python, public domain spec code).
Signing runs in coincurve (libsecp256k1); verification is pure Python so
the auditor can read every step without CFFI indirection.
"""
P = 2**256 - 2**32 - 977
N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
def tag_hash(tag: str, msg: bytes) -> bytes:
t = hashlib.sha256(tag.encode()).digest()
return hashlib.sha256(t + t + msg).digest()
def lift_x(b: int):
if b >= P:
return None
y_sq = (pow(b, 3, P) + 7) % P
y = pow(y_sq, (P + 1) // 4, P)
if pow(y, 2, P) != y_sq:
return None
return (b, y if y % 2 == 0 else P - y)
def point_add(p1, p2):
if p1 is None:
return p2
if p2 is None:
return p1
if p1[0] == p2[0] and p1[1] != p2[1]:
return None
if p1 == p2:
lam = (3 * p1[0] * p1[0] * pow(2 * p1[1], P - 2, P)) % P
else:
lam = ((p2[1] - p1[1]) * pow(p2[0] - p1[0], P - 2, P)) % P
x3 = (lam * lam - p1[0] - p2[0]) % P
return (x3, (lam * (p1[0] - x3) - p1[1]) % P)
def point_mul(p, scalar):
acc = None
for i in range(256):
if (scalar >> i) & 1:
acc = point_add(acc, p)
p = point_add(p, p)
return acc
if len(sig32) != 64 or len(msg32) != 32 or len(xonly_pub32) != 32:
return False
r = int.from_bytes(sig32[:32], "big")
s = int.from_bytes(sig32[32:], "big")
if not (0 <= r < P and 0 <= s < N):
return False
pub_point = lift_x(int.from_bytes(xonly_pub32, "big"))
if pub_point is None:
return False
e = int.from_bytes(tag_hash("BIP0340/challenge", sig32[:32] + xonly_pub32 + msg32), "big") % N
generator = (
0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, # Gx
0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8, # Gy
)
r_point = point_add(
point_mul(generator, s),
point_mul(pub_point, (N - e) % N),
)
if r_point is None or r_point[1] % 2 != 0:
return False
return r_point[0] == r
def verify_event(event: dict[str, Any]) -> bool:
"""Recompute the id and verify the Schnorr signature against the pubkey."""
serialized = json.dumps(
[0, event["pubkey"], event["created_at"], event["kind"], event["tags"], event["content"]],
separators=(",", ":"),
ensure_ascii=False,
)
if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != event["id"]:
return False
# 33-byte compressed point (02-prefixed x-only)
compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format()
return _schnorr_verify(
bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:]
)
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
with connect(relay_url, close_timeout=timeout) as ws:
ws.send(json.dumps(["EVENT", event]))
reply = json.loads(ws.recv(timeout=timeout))
if not (isinstance(reply, list) and len(reply) >= 3 and reply[0] == "OK" and reply[2] is True):
raise RuntimeError(f"relay {relay_url} rejected event: {str(reply)[:200]}")
return str(reply[1])
def publish(
relays: list[str], event: dict[str, Any], timeout: float = 10.0
) -> list[dict[str, str]]:
"""Publish to every relay; report per-relay outcome (never raises unless all fail)."""
results = []
for relay in relays:
try:
relay_publish(relay, event, timeout)
results.append({"relay": relay, "ok": "true"})
except Exception as exc: # noqa: BLE001 — per-relay isolation
results.append({"relay": relay, "ok": "false", "error": str(exc)[:200]})
if relays and all(r["ok"] == "false" for r in results):
raise RuntimeError(f"all relays failed: {results}")
return results
+322
View File
@@ -0,0 +1,322 @@
"""publisher — themed content calendar with a hard human gate on publishing.
State machine: draft -> approved -> published. Only ``approved`` items ever
publish. AUTONOMOUS_MODE has NO effect on this component: content always
requires explicit human approval (documented in AUDIT.md).
Channels: nostr (NIP-01, primary) — facebook/instagram adapters exist but are
disabled unless explicitly enabled via env (DECISIONS: default deployment never
touches Meta).
"""
from __future__ import annotations
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,
_content_hash,
audit,
)
# rotation: one theme per slot; channel rotation nostr -> facebook -> instagram
THEMES = [
("lightning-abroad", "bitcoinhotels.org/guides/lightning-abroad"),
("seed-strategy", "hardcap.news/learn/travel-security-kit"),
("atm-fees", "bitcoinhotels.org/guides/best-bitcoin-atms-by-city"),
("elsalvador-budget", "bitcoinhotels.org/guides/living-on-sats-in-el-salvador"),
]
_CAPTION_TEMPLATE = "{hook}\n\n{body}\n\n{link}"
def _caption(theme: str, link: str) -> str:
hooks = {
"lightning-abroad": "Paying abroad with Lightning beats your bank.",
"seed-strategy": "The seed strategy you pick decides how the border story goes.",
"atm-fees": "The first ATM you find charges 10-15% over spot.",
"elsalvador-budget": "A real weekly budget on a Bitcoin standard.",
}
bodies = {
"lightning-abroad": (
"No foreign transaction fees, no dynamic currency conversion, "
"no declines. Field-tested guide."
),
"seed-strategy": (
"Steel plate, memorized + decoy, or split custody — "
"the full breakdown, brutally honest."
),
"atm-fees": (
"The prepared traveler pays 4-6%. The difference is knowing "
"the operator and the no-KYC cap."
),
"elsalvador-budget": (
"Surf hostels, comidas, coworking — the corners where you still "
"need dollars, and the 90-day protocol."
),
}
return _CAPTION_TEMPLATE.format(
hook=hooks.get(theme, theme.replace("-", " ").title()),
body=bodies.get(theme, "Field-tested notes."),
link=link,
)
# --- calendar planning ----------------------------------------------------------
def _plan(cfg: Config, sess, days: int = 4) -> dict:
"""Fill the calendar: one draft per theme/channel pair not already queued.
Idempotent by (theme, channel): existing draft/approved items block a
duplicate. Drafts are scheduled one per day, starting tomorrow.
"""
created = 0
for idx, (theme, link) in enumerate(THEMES[:days]):
channel = ["nostr", "facebook", "instagram", "nostr"][idx % 4]
exists = (
sess.query(ContentItem)
.filter(
ContentItem.theme == theme,
ContentItem.channel == channel,
ContentItem.status != ContentStatus.PUBLISHED,
)
.one_or_none()
)
if exists:
continue
item = ContentItem(
theme=theme,
channel=channel,
caption=_caption(theme, link),
link=link,
status=ContentStatus.DRAFT,
scheduled_for=datetime.now(UTC).replace(hour=13, minute=0, second=0, microsecond=0)
+ timedelta(days=idx + 1),
)
sess.add(item)
sess.flush()
audit(
sess,
actor="daemon:publisher",
action="plan_draft",
detail={"item": item.id, "theme": theme, "channel": channel},
)
created += 1
return {"created": created}
# --- human gate -------------------------------------------------------------------
def _approve(item_id: int, actor: str, sess) -> int:
item = sess.get(ContentItem, item_id)
if item is None or item.status != ContentStatus.DRAFT:
audit(
sess,
actor=actor,
action="approve_content",
result="error",
detail={"reason": f"item {item_id} not in draft state"},
)
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, "hash": item.approved_hash},
)
return 0
# --- image rendering ----------------------------------------------------------------
def _render_image(item: ContentItem, cfg: Config) -> str | None:
"""Render the brand image for a nostr item (per its theme)."""
brand = "hc" if "hardcap" in (item.link or "") else "bh"
title = item.theme.replace("-", " ").upper().split()
lines = [" ".join(title[:2])] if title else [item.theme]
out = images.render_brand_image(
brand=brand,
kicker="C A Y C E",
lines=lines,
tagline=[(item.link or "")[:70]],
url=(item.link or "")[:60],
out_path=f"{cfg.media_dir}/item-{item.id}.jpg",
)
item.media_path = out
return out
# --- publishing ----------------------------------------------------------------------
def _publish_nostr(item: ContentItem, cfg: Config, sess, sender=None) -> str:
if not cfg.nostr_private_key or not cfg.nostr_relays:
raise RuntimeError(
"nostr channel requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS"
)
secret = nostr.normalize_key(cfg.nostr_private_key)
event = nostr.build_event(secret, item.caption, tags=[["t", item.theme]])
if not nostr.verify_event(event): # pragma: no cover — build/verify symmetry
raise RuntimeError("event failed self-verification before publish")
results = (sender or nostr.publish)(cfg.nostr_relays, event)
audit(
sess,
actor="daemon:publisher",
action="nostr_relay_results",
detail={"event": event["id"], "relays": results},
)
return event["id"]
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, "tampered": 0}
due = (
sess.query(ContentItem)
.filter(ContentItem.status == ContentStatus.APPROVED)
.all()
)
now = datetime.now(UTC)
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_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]
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
item.published_at = datetime.now(UTC)
audit(
sess,
actor="daemon:publisher",
action="publish_content",
detail={"item": item.id, "channel": item.channel, "ref": item.post_ref},
)
stats["published"] += 1
return stats
# --- CLI wrappers ---------------------------------------------------------------------
def plan(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="publisher")
sess.add(run_row)
sess.flush()
stats = _plan(cfg, sess)
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
def publish_due(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="publisher")
sess.add(run_row)
sess.flush()
stats = _publish_due(cfg, sess)
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
def make_image(item_id: int, cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
item = sess.get(ContentItem, item_id)
if item is None:
return 1
print(_render_image(item, cfg))
return 0
finally:
engine().dispose()
def approve(item_id: int, actor: str) -> int:
from cayceops.db import engine
try:
with session_scope() as sess:
return _approve(item_id, actor, sess)
finally:
engine().dispose()
+1
View File
@@ -0,0 +1 @@
"""scanner package."""
+209
View File
@@ -0,0 +1,209 @@
"""scanner — opportunity discovery. Proposes; never executes.
Sources come from ``CAYCE_OPS_SCANNER_SOURCES`` (JSON array):
[{"type": "rss", "url": "https://example.org/feed.xml"},
{"type": "page", "url": "https://example.org/partners"}]
Every discovered candidate becomes a ``proposals`` row with status ``proposed``
and a stable ``dedupe_key`` (sha1 of url-or-title) so repeated scans stay
idempotent. The scanner performs no side effects beyond writing proposals and
audit entries.
"""
from __future__ import annotations
import hashlib
import json
import re
import urllib.error
import urllib.request
from datetime import UTC, datetime
from html.parser import HTMLParser
from xml.etree import ElementTree
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import DaemonRun, Proposal, ProposalSource, ProposalStatus, audit
# kind classification heuristics (checked in order)
_KIND_PATTERNS = [
("partnership", re.compile(r"partner|collaborat|integrat", re.I)),
("affiliate", re.compile(r"affiliate|referral|commission|revenue\s*share", re.I)),
]
DEFAULT_KIND = "content_gap"
class _LinkCollector(HTMLParser):
"""Minimal HTML parser: page <title> + anchor (href, text) pairs."""
def __init__(self) -> None:
super().__init__()
self.title: str = ""
self.links: list[tuple[str, str]] = []
self._in_title = False
self._in_anchor = False
self._href = ""
self._text: list[str] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str]]) -> None:
if tag == "title":
self._in_title = True
elif tag == "a":
self._in_anchor = True
self._href = dict(attrs).get("href", "")
self._text = []
def handle_endtag(self, tag: str) -> None:
if tag == "title":
self._in_title = False
elif tag == "a" and self._in_anchor:
self._in_anchor = False
text = " ".join("".join(self._text).split())
if self._href and text:
self.links.append((self._href, text))
def handle_data(self, data: str) -> None:
if self._in_title:
self.title += data
if self._in_anchor:
self._text.append(data)
def _fetch(url: str, cfg: Config, timeout: float = 20.0) -> bytes:
req = urllib.request.Request(
url, headers={"User-Agent": cfg.scanner_user_agent}, method="GET"
)
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310
return resp.read()
def _parse_rss(body: bytes) -> list[tuple[str, str, str]]:
"""Return (title, link, description) tuples from an RSS/Atom feed."""
out: list[tuple[str, str, str]] = []
root = ElementTree.fromstring(body)
ns = {"atom": "http://www.w3.org/2005/Atom"}
for item in root.iter("item"): # RSS 2.0
title = item.findtext("title") or ""
link = item.findtext("link") or ""
desc = re.sub(r"<[^>]+>", " ", item.findtext("description") or "")
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
for entry in root.findall("atom:entry", ns): # Atom
title = entry.findtext("atom:title", namespaces=ns) or ""
link_el = entry.find("atom:link", ns)
link = link_el.get("href", "") if link_el is not None else ""
desc = re.sub(r"<[^>]+>", " ", entry.findtext("atom:summary", namespaces=ns) or "")
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
return out
def _parse_page(body: bytes) -> list[tuple[str, str, str]]:
"""Return (title=anchor text, link=href, description=page title) tuples."""
parser = _LinkCollector()
parser.feed(body.decode("utf-8", errors="replace"))
page_title = " ".join(parser.title.split())
return [(text, href, f"linked from page: {page_title}") for href, text in parser.links]
def _classify(title: str, description: str) -> str:
blob = f"{title}\n{description}"
for kind, pattern in _KIND_PATTERNS:
if pattern.search(blob):
return kind
return DEFAULT_KIND
def _score(title: str, description: str, keywords: list[str]) -> float:
blob = f"{title}\n{description}".lower()
hits = sum(1 for kw in keywords if kw.lower() in blob)
return round(min(hits / max(1, len(keywords)), 1.0), 3)
def scan(cfg: Config, sess, fetch=_fetch) -> dict:
"""Run one scan pass against all configured sources. Returns run stats."""
sources = json.loads(cfg.scanner_sources or "[]")
stats = {"sources": len(sources), "candidates": 0, "proposed": 0, "skipped_existing": 0}
existing = {
key
for (key,) in sess.query(Proposal.dedupe_key).filter(
Proposal.dedupe_key.isnot(None)
)
}
for src in sources:
url = src.get("url", "")
stype = src.get("type", "page")
if not url:
continue
try:
body = fetch(url, cfg)
except (urllib.error.URLError, OSError) as exc:
stats.setdefault("source_errors", []).append({"url": url, "error": str(exc)[:200]})
continue
items = _parse_rss(body) if stype == "rss" else _parse_page(body)
stats["candidates"] += len(items)
for title, link, desc in items:
text = title or link
if not text:
continue
keywords = cfg.scanner_keywords
score = _score(title or link, desc, keywords)
if score <= 0 and stype != "page":
continue # feeds must match at least one keyword; pages keep links
dedupe = hashlib.sha1((link or title).encode()).hexdigest()[:32] # noqa: S324
if dedupe in existing:
stats["skipped_existing"] += 1
continue
proposal = Proposal(
source=ProposalSource.SCANNER,
kind=_classify(text, desc),
title=text[:512],
url=(link or None)[:1024] if (link or None) else None,
detail={
"source_type": stype,
"source_url": url,
"summary": desc,
"scanned_at": datetime.now(UTC).isoformat(),
},
match_score=score,
status=ProposalStatus.PROPOSED,
dedupe_key=dedupe,
)
sess.add(proposal)
sess.flush()
audit(
sess,
actor="daemon:scanner",
action="propose",
proposal_id=proposal.id,
detail={"kind": proposal.kind, "score": score, "source": stype},
)
existing.add(dedupe)
stats["proposed"] += 1
return stats
def run(cfg: Config | None = None) -> int:
"""CLI entry: one scan pass, wrapped in a DaemonRun record."""
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="scanner")
sess.add(run_row)
sess.flush()
try:
stats = scan(cfg, sess)
except Exception as exc: # noqa: BLE001 — daemon boundary: record and fail cleanly
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
+5
View File
@@ -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:<id>" made by the operator key.
operator_pubkey: str = field(
default_factory=lambda: os.environ.get("CAYCE_OPS_OPERATOR_PUBKEY", "")
)
# --- mail ---
imap_host: str = field(
+81 -6
View File
@@ -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:<name>
actor: Mapped[str] = mapped_column(String(128)) # human | auto:policy | daemon:<name>
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)
+22
View File
@@ -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;
+5
View File
@@ -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
+1 -1
View File
@@ -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:
+26
View File
@@ -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:<id>") 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.
+7
View File
@@ -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"
+187
View File
@@ -0,0 +1,187 @@
"""executor tests — human gate, autonomy bounds, retries, audit chain."""
from __future__ import annotations
import pytest
from cayceops.components.executor import executor, plugins
from cayceops.models import AuditEntry, Proposal, ProposalStatus
def _proposal(kind="content_gap", title="Test item", **kw):
return Proposal(kind=kind, title=title, **kw)
def test_approve_requires_proposed_state(cfg, sess):
p = _proposal()
sess.add(p)
sess.flush()
assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
# second approve must fail (state is now approved)
assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
assert p.approved_by == "gustavo"
assert p.status == ProposalStatus.APPROVED
def test_reject_records_actor(cfg, sess):
p = _proposal()
sess.add(p)
sess.flush()
assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
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")
sess.add_all([approved, proposed])
sess.flush()
executor.approve(approved.id, "gustavo", cfg(), sess)
config = cfg() # AUTONOMOUS_MODE=false
assert executor.run(config, sess) == 0
assert approved.status == ProposalStatus.DONE
assert proposed.status == 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
assert acts.count("approve") == 1
def test_autonomy_off_by_default(cfg, sess):
p = _proposal(kind="content_gap", title="Never auto-runs")
sess.add(p)
sess.flush()
config = cfg() # autonomous_mode False even with kinds configured
executor._auto_approve(config, sess)
assert p.status == 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])
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"
def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
p = _proposal(kind="affiliate", title="Will fail")
sess.add(p)
sess.flush()
executor.approve(p.id, "gustavo", cfg(), sess)
calls = {"n": 0}
def boom(proposal, cfg, sess):
calls["n"] += 1
raise RuntimeError("upstream exploded")
monkeypatch.setitem(plugins.REGISTRY, "affiliate", boom)
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.attempts == 1
executor.run(config, sess) # attempt 2 -> failed for good
assert p.status == ProposalStatus.FAILED
assert p.attempts == 2
assert calls["n"] == 2
errors = sess.query(AuditEntry).filter_by(result="error").all()
assert any(a.action == "fail" for a in errors)
def test_http_post_blocked_when_disabled(cfg, sess):
p = _proposal(kind="affiliate", title=" webhook target")
p.detail = {"webhook_url": "https://hooks.example.org/ingest"}
sess.add(p)
sess.flush()
config = cfg() # CAYCE_OPS_ENABLE_HTTP_ACTIONS=false
with pytest.raises(PermissionError):
plugins.http_post(p, config, sess)
def test_unknown_kind_falls_back_to_log_action(cfg, sess):
p = _proposal(kind="brand_new_kind", title="No plugin yet")
sess.add(p)
sess.flush()
executor.approve(p.id, "gustavo", cfg(), sess)
assert executor.run(cfg(), sess) == 0
assert p.status == ProposalStatus.DONE
done = sess.query(AuditEntry).filter_by(action="execute_done").one()
assert done.detail["outcome"]["plugin_fallback"] is True
+351
View File
@@ -0,0 +1,351 @@
"""maild tests — parsing, classification, threading, gated send. Invented data only."""
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 (
AuditEntry,
DraftStatus,
MailClassification,
MailDirection,
MailMessage,
Proposal,
ProposalStatus,
)
OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
STUB_CFG = SimpleNamespace(mail_from="agent@cayce-co.xyz")
def _eml(subject, body, in_reply_to=None, msgid=None) -> bytes:
lines = [
"From: Partner Desk <partners@acme.example.org>",
"To: agent@cayce-co.xyz",
f"Subject: {subject}",
f"Message-ID: {msgid or '<acme-reply-1@acme.example.org>'}",
]
if in_reply_to:
lines.append(f"In-Reply-To: {in_reply_to}")
lines += ["MIME-Version: 1.0", "Content-Type: text/plain; charset=utf-8", "", body]
return "\r\n".join(lines).encode()
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",
status=ProposalStatus.APPROVED,
)
sess.add(p)
sess.flush()
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()
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)
assert kind is MailClassification.NEEDS_REVIEW
assert confidence < 0.95
# --- inbound pass ---------------------------------------------------------------
def test_poll_threads_approval_and_drafts_reply(cfg, sess):
p = _proposal_with_sent_draft(sess)
raw = _eml(
"Re: [cayce] partnership inquiry",
"Approved! Welcome to the Acme partner program.",
in_reply_to=OUTBOUND_MSGID,
)
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
assert inbound.proposal_id == p.id
assert inbound.draft_status is DraftStatus.DRAFTED
assert inbound.draft_reply and "approval" in inbound.draft_reply.lower()
assert p.detail["merchant_status"] == "approval"
assert p.status is ProposalStatus.APPROVED # lifecycle untouched by maild
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)
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=_fetcher([raw]))
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert inbound.classification is MailClassification.NEEDS_REVIEW
assert stats["needs_review"] == 1
assert inbound.draft_status is DraftStatus.NONE # no draft for review items
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=_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
assert "traffic" in inbound.draft_reply
# --- human gate -------------------------------------------------------------------
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=_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
def test_send_requires_explicit_human_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=_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")
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
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
# 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.", 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"<acme-{i}@acme.example.org>",
)
maild._poll(config, sess, fetcher=_fetcher([raw]))
inbound = (
sess.query(MailMessage)
.filter_by(direction=MailDirection.INBOUND, message_id=f"<acme-{i}@acme.example.org>")
.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
+20
View File
@@ -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
+69
View File
@@ -0,0 +1,69 @@
"""monitor tests — page renders, no sensitive data, escapes input."""
from __future__ import annotations
from cayceops.components.monitor import monitor
from cayceops.models import ContentItem, DaemonRun, MailMessage, Proposal
def _seed(sess):
sess.add(DaemonRun(daemon="scanner", status="ok", detail={"proposed": 3}))
sess.add(Proposal(kind="affiliate", title="Title with <script>alert(1)</script>"))
sess.add(
MailMessage(
direction="in",
from_addr="secret-person@private.example.org",
subject="private subject",
classification="approval",
draft_status="drafted",
)
)
sess.add(ContentItem(theme="atm-fees", channel="nostr", caption="caption", status="draft"))
sess.flush()
def test_build_renders_all_sections(cfg, sess, tmp_path):
_seed(sess)
out = tmp_path / "status.html"
monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
page = out.read_text()
assert "daemon health" in page
assert "proposal queue" in page
assert "recent audit activity" in page
assert "scanner" in page
assert "proposed: <strong>1</strong>" in page
assert "mail drafts awaiting approval: <strong>1</strong>" in page
def test_build_leaks_no_sensitive_data(cfg, sess, tmp_path):
_seed(sess)
out = tmp_path / "status.html"
monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
page = out.read_text()
assert "secret-person@private.example.org" not in page # no addresses
assert "private subject" not in page # no subjects
assert "caption" not in page # no content bodies
assert "Title with" not in page # proposal titles not rendered
def test_build_escapes_injected_content(cfg, sess, tmp_path):
from datetime import UTC, datetime
from cayceops.models import AuditEntry
sess.add(
AuditEntry(
ts=datetime.now(UTC),
actor="<script>x</script>",
action="probe",
detail={},
)
)
sess.flush()
out = tmp_path / "status.html"
monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
page = out.read_text()
assert "<script>x</script>" not in page
assert "&lt;script&gt;x&lt;/script&gt;" in page
+224
View File
@@ -0,0 +1,224 @@
"""publisher tests — calendar, human gate, nostr publish, disabled adapters."""
from __future__ import annotations
import os
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
def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED):
return ContentItem(
theme=theme,
channel=channel,
caption="The first ATM you find charges 10-15% over spot.\n\nbitcoinhotels.org/guides/atm",
link="bitcoinhotels.org/guides/atm",
status=status,
scheduled_for=scheduled,
)
# --- nostr client -------------------------------------------------------------
def test_event_build_verify_roundtrip():
event = nostr.build_event(
bytes.fromhex(TEST_KEY), "gm", tags=[["t", "bitcoin"]], created_at=1700000000
)
assert nostr.verify_event(event)
assert len(event["id"]) == 64 and len(event["sig"]) == 128
tampered = {**event, "content": "gm (edited)"}
assert not nostr.verify_event(tampered)
def test_event_is_deterministic():
a = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
b = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
assert a == b
def test_key_validation():
with pytest.raises(ValueError):
nostr.normalize_key("not-a-key")
assert nostr.normalize_key("0x" + TEST_KEY) == bytes.fromhex(TEST_KEY)
# --- calendar ------------------------------------------------------------------
def test_plan_creates_idempotent_drafts(cfg, sess):
config = cfg()
first = publisher._plan(config, sess)
second = publisher._plan(config, sess)
assert first["created"] == 4
assert second["created"] == 0 # same themes+channels already queued
def test_approve_gate(cfg, sess):
item = _item(status=ContentStatus.DRAFT)
sess.add(item)
sess.flush()
assert publisher._approve(item.id, "gustavo", sess) == 0
assert item.status is ContentStatus.APPROVED
assert publisher._approve(item.id, "gustavo", sess) == 1 # not draft anymore
actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor
assert actor == "gustavo"
# --- publishing ------------------------------------------------------------------
def test_publish_due_nostr_happy_path(cfg, sess):
item = _item() # approved, scheduled None -> due now
sess.add(item)
sess.flush()
relays_seen = []
def fake_sender(relays, event, timeout=10.0):
relays_seen.append(relays)
assert nostr.verify_event(event)
return [{"relay": r, "ok": "true"} for r in relays]
config = cfg(
CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY,
CAYCE_OPS_NOSTR_RELAYS="wss://relay.example.org,wss://other.example.org",
)
stats = publisher._publish_due(config, sess, nostr_sender=fake_sender)
assert stats["published"] == 1
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
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: [])
assert stats["published"] == 0
assert item.status is ContentStatus.APPROVED
def test_nostr_without_config_fails_clean(cfg, sess):
item = _item()
sess.add(item)
sess.flush()
config = cfg() # no key, no relays
stats = publisher._publish_due(config, sess)
assert stats["failed"] == 1
assert item.status is ContentStatus.APPROVED # stays approved, error recorded
assert item.error and "nostr" in item.error
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.SKIPPED # terminal (audit fix)
skip = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").one()
assert skip.detail["channel"] == "facebook"
def test_meta_adapters_disabled_by_default(cfg):
config = cfg()
with pytest.raises(PermissionError):
adapters.post_facebook(config, "hello")
with pytest.raises(PermissionError):
adapters.post_instagram(config, "hello", "https://img.example.org/x.jpg")
# --- images ------------------------------------------------------------------
def test_brand_image_render(tmp_path, cfg):
config = cfg(CAYCE_OPS_MEDIA_DIR=str(tmp_path))
out = publisher._render_image(_item(), config)
assert os.path.exists(out)
from PIL import Image
with Image.open(out) as img:
assert img.size == (1080, 1350)
def test_unknown_brand_rejected():
from cayceops.components.publisher import images
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
+112
View File
@@ -0,0 +1,112 @@
"""scanner tests — discovery, classification, dedupe. No network (injected fetch)."""
from __future__ import annotations
import json
from cayceops.components.scanner import scanner
from cayceops.models import Proposal, ProposalStatus
RSS = b"""<?xml version="1.0"?>
<rss><channel>
<item><title>Acme launches affiliate program for hotels</title>
<link>https://acme.example.org/affiliates</link>
<description>Earn commission on every booking.</description></item>
<item><title>Weekly crypto digest</title>
<link>https://news.example.org/digest</link>
<description>Everything unrelated.</description></item>
</channel></rss>"""
PAGE = b"""<html><head><title>Partners</title></head><body>
<a href="https://pay.example.org/partner">Become a payment partner</a>
<a href="https://pay.example.org/docs">API docs</a>
</body></html>"""
def _fake_fetch(responses):
def fetch(url, cfg, timeout=20.0):
return responses[url]
return fetch
SOURCES = [
{"type": "rss", "url": "https://feeds.example.org/news.xml"},
{"type": "page", "url": "https://pay.example.org/partners"},
]
def _cfg(cfg, **kw):
return cfg(
CAYCE_OPS_SCANNER_SOURCES=json.dumps(SOURCES),
CAYCE_OPS_SCANNER_KEYWORDS="affiliate,partner,bitcoin,commission",
**kw,
)
def test_scan_files_proposals_with_kind_and_score(cfg, sess):
config = _cfg(cfg)
stats = scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
assert stats["proposed"] == 3 # affiliate item + both page links (pages keep all)
proposals = sess.query(Proposal).all()
by_title = {p.title: p for p in proposals}
affiliate = by_title["Acme launches affiliate program for hotels"]
assert affiliate.kind == "affiliate"
assert affiliate.status == ProposalStatus.PROPOSED
assert affiliate.match_score > 0
assert affiliate.dedupe_key
partner = by_title["Become a payment partner"]
assert partner.kind == "partnership"
def test_scan_is_idempotent_via_dedupe(cfg, sess):
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
stats2 = scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
assert stats2["proposed"] == 0
assert stats2["skipped_existing"] == 3
def test_scan_filters_keywordless_feed_items(cfg, sess):
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
titles = {p.title for p in sess.query(Proposal).all()}
assert "Weekly crypto digest" not in titles # no keyword match in feed -> dropped
def test_scan_records_audit_for_each_proposal(cfg, sess):
from cayceops.models import AuditEntry
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
acts = [a.action for a in sess.query(AuditEntry).all()]
assert acts.count("propose") == 3
assert all(a.actor == "daemon:scanner" for a in sess.query(AuditEntry).all())
def test_scan_survives_source_error(cfg, sess):
def broken_fetch(url, cfg, timeout=20.0):
raise OSError("connection refused")
config = _cfg(cfg)
stats = scanner.scan(config, sess, fetch=broken_fetch)
assert stats["proposed"] == 0
assert len(stats["source_errors"]) == 2