cayce
73ccdf7bdf
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
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""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)
|