cayce
40453aadb4
- signing module: canonical content hashes + approval_message sha256(cayce-ops:<type>:<deployment_id>:<id>:<content_hash>) - approve/approve-draft/publisher-approve REQUIRE signature; without CAYCE_OPS_OPERATOR_PUBKEY+DEPLOYMENT_ID the daemons refuse to execute/send/publish (no degraded mode) - run-time re-verification at execute/send/publish: unsigned rows (raw SQL approval) or edited content revert to pre-approval state, audited as error - FULL signatures stored on rows (operator_signature, migration 0003) and in audit detail — never truncated - signature failures audited from CLI wrappers too (1f) - AUTONOMOUS_MODE approval path removed (D13): no non-human approvals tests: SQL-approved row refused at execute/send/publish; to_addr change blocks send; foreign deployment/content signatures refused; full-signature persistence. 65 tests green.
143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
"""Shared pytest fixtures.
|
|
|
|
Tests run against a throwaway SQLite database (fast, zero external services).
|
|
The production path is Postgres via Alembic; ``make migrate-check`` smoke-tests
|
|
the migration chain on SQLite in CI.
|
|
All fixture data is invented — never real merchant or account data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from coincurve import PublicKey
|
|
|
|
# test operator keypair (invented; allowlisted in .gitleaks.toml)
|
|
OPERATOR_SECRET = bytes.fromhex("22" * 32)
|
|
OPERATOR_PUBKEY = PublicKey.from_secret(OPERATOR_SECRET).format().hex()[2:]
|
|
|
|
# Config must see the test DB before cayceops.config/db get imported.
|
|
os.environ.setdefault("CAYCE_OPS_DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("CAYCE_OPS_AUTONOMOUS_MODE", "false")
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_engine(tmp_path: Path):
|
|
"""A fresh schema per test, on a file-backed SQLite DB (shared in-process)."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cayceops.db import reset_engine
|
|
from cayceops.models import Base
|
|
|
|
url = f"sqlite:///{tmp_path}/test.db"
|
|
eng = create_engine(url, future=True, connect_args={"check_same_thread": False})
|
|
Base.metadata.create_all(eng)
|
|
factory = sessionmaker(bind=eng, expire_on_commit=False, future=True)
|
|
|
|
# point the app at the same engine/session factory
|
|
reset_engine()
|
|
import cayceops.db as appdb
|
|
|
|
appdb._engine = eng
|
|
appdb._factory = factory
|
|
yield eng
|
|
eng.dispose()
|
|
reset_engine()
|
|
|
|
|
|
@pytest.fixture()
|
|
def sess(db_engine):
|
|
"""A session bound to the fresh schema (rolls back at test end)."""
|
|
import cayceops.db as appdb
|
|
|
|
s = appdb.session_factory()()
|
|
yield s
|
|
s.rollback()
|
|
s.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def signer():
|
|
"""Signs approval messages with the test operator key (22*32, deployment
|
|
'itest'). Mirrors exactly what the operator CLI does."""
|
|
from coincurve import PrivateKey
|
|
|
|
from cayceops.models import MailDirection
|
|
from cayceops.signing import (
|
|
approval_message,
|
|
content_content_hash,
|
|
mail_content_hash,
|
|
proposal_content_hash,
|
|
)
|
|
|
|
secret = OPERATOR_SECRET
|
|
|
|
def _sig(msg: bytes) -> str:
|
|
return PrivateKey(secret).sign_schnorr(msg, aux_randomness=b"\x00" * 32).hex()
|
|
|
|
def proposal(p, deployment="itest"):
|
|
chash = proposal_content_hash(p.kind, p.title, p.url, p.detail)
|
|
return _sig(approval_message("proposal", deployment, p.id, chash))
|
|
|
|
def mail(record, deployment="itest"):
|
|
to = (
|
|
record.from_addr
|
|
if record.direction == MailDirection.INBOUND
|
|
else record.to_addr
|
|
)
|
|
chash = mail_content_hash(to, record.subject, record.draft_reply)
|
|
return _sig(approval_message("mail", deployment, record.id, chash))
|
|
|
|
def content(item, deployment="itest"):
|
|
chash = content_content_hash(item.channel, item.caption, item.link)
|
|
return _sig(approval_message("content", deployment, item.id, chash))
|
|
|
|
def raw(sig_bytes: bytes) -> str:
|
|
return _sig(sig_bytes)
|
|
|
|
def msg(item_type: str, deployment: str, item_id: int, chash: str) -> bytes:
|
|
from cayceops.signing import approval_message
|
|
|
|
return approval_message(item_type, deployment, item_id, chash)
|
|
|
|
def content_hash(item, deployment="itest") -> str:
|
|
from cayceops.signing import content_content_hash
|
|
|
|
return content_content_hash(item.channel, item.caption, item.link)
|
|
|
|
return SimpleNamespace(
|
|
proposal=proposal, mail=mail, content=content, raw=raw, msg=msg,
|
|
content_hash=content_hash,
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def cfg(monkeypatch):
|
|
"""A Config snapshot with test defaults (no network, no autonomy)."""
|
|
from cayceops.config import Config
|
|
|
|
defaults = {
|
|
"CAYCE_OPS_DATABASE_URL": "sqlite:///:memory:",
|
|
"CAYCE_OPS_LLM_PROVIDER": "null",
|
|
"CAYCE_OPS_ENABLE_FACEBOOK": "false",
|
|
"CAYCE_OPS_ENABLE_INSTAGRAM": "false",
|
|
"CAYCE_OPS_NOSTR_RELAYS": "",
|
|
"CAYCE_OPS_MEDIA_DIR": "media",
|
|
# signing defaults (audit round 2): tests operate in a signed
|
|
# deployment unless they explicitly clear these
|
|
"CAYCE_OPS_DEPLOYMENT_ID": "itest",
|
|
"CAYCE_OPS_OPERATOR_PUBKEY": OPERATOR_PUBKEY,
|
|
}
|
|
|
|
def _apply(**overrides):
|
|
merged = {**defaults, **{k: str(v) for k, v in overrides.items()}}
|
|
for k, v in merged.items():
|
|
monkeypatch.setenv(k, v)
|
|
return Config()
|
|
|
|
return _apply
|