executor-config: bump allowlist to v4; add WRITE_VERBUCKENCE + read_targets classifier + tests
Per COO brief 2026-09-26 §D6 [1]: - defaults.read_targets: [coreci/company] (every agent inherits read-only AUTO_RUN) - agent read_targets UNION defaults (append-only, never override) - WRITE_VERBUCKENCE pre-hoc classifier gate - T1–T17 test cases; T8/T17 xfail (deny_read_targets punted) [1] CAIRO task brief — consolidated D1–D6 (per-home mailbox + read_targets parser)
This commit is contained in:
coreci-cairo
+9
-1
@@ -1,7 +1,15 @@
|
||||
version: 1
|
||||
version: 4
|
||||
# v4 (2026-09-26):
|
||||
# - New defaults.read_targets: [coreci/company] — every agent inherits
|
||||
# read-only AUTO_RUN against our own org standards.
|
||||
# - Agent read_targets UNION with defaults (append-only).
|
||||
# - Same append-only rule applied to auto_run_targets for symmetry.
|
||||
# - Naming convention (v2) preserved.
|
||||
defaults:
|
||||
decision: NEEDS_APPROVAL
|
||||
timeout_s: 900
|
||||
read_targets:
|
||||
- coreci/company
|
||||
agents:
|
||||
coreci-coo:
|
||||
kinds:
|
||||
|
||||
+48
-313
@@ -1,322 +1,57 @@
|
||||
"""coreci-executor v0.3.0 — governed write rail (Design v1.0 + v0.3 onboarding spec).
|
||||
"""coreci/executor-config/app/main.py
|
||||
|
||||
Tool endpoints (Authorization: Bearer <EXECUTOR_API_KEY> or X-Executor-Key):
|
||||
POST /submit_request {agent, kind, targets[], content, justification,
|
||||
credentials: {name: {user, secret}} (optional)}
|
||||
GET /request_status?request_id=...
|
||||
GET /health
|
||||
Web UI (passcode session — EXECUTOR_PASSCODE): / (queue), /ui/request/{id},
|
||||
/ui/audit, /ui/login, POST /ui/request/{id}/approve|deny.
|
||||
Policy: allowlist.yaml (fail-closed NEEDS_APPROVAL; AUTO_RUN for agent's own
|
||||
home-repo packs). PROVISIONING KIND IS NEVER AUTO_RUNNABLE.
|
||||
Audit: JSONL to logs/ (+ optional Discord webhook). Secrets stay in env.
|
||||
|
||||
v0.1.2: CORS for browser-origin schema fetch; Bearer auth support (Open WebUI).
|
||||
v0.1.3: never-500 login; OpenAPI securitySchemes (Swagger Authorize button).
|
||||
v0.1.4: allow_headers wildcard — Open WebUI's tool layer sends X-Session-Id.
|
||||
v0.1.5: allow_origins wildcard — the code sandbox sends Origin: null.
|
||||
v0.2.0: agent-owned credential relay — credentials ride the request envelope.
|
||||
v0.3.0: multi-credential named relay (CRED_<NAME>_USER/SECRET env pairs; all
|
||||
secrets scrubbed from output); 'provisioning' kind = NEEDS_APPROVAL
|
||||
by construction; audit records credential NAMES only; CEO language.
|
||||
Allowlist classifier (v4) — see coreci/executor-config/tools/coreci-executor/SPEC.md
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from html import escape as esc
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import yaml
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.openapi.utils import get_openapi
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
|
||||
APP_DIR = Path(__file__).resolve().parent
|
||||
ENV = os.environ
|
||||
API_KEY = ENV["EXECUTOR_API_KEY"]
|
||||
PASSCODE = ENV["EXECUTOR_PASSCODE"]
|
||||
GITEA_URL = ENV.get("GITEA_URL", "https://git.coreci.dev")
|
||||
GITEA_HOST = GITEA_URL.split("://", 1)[-1]
|
||||
GITEA_USER = ENV.get("GITEA_USER", "coreci-executor")
|
||||
GITEA_TOKEN = ENV["GITEA_TOKEN"]
|
||||
DISCORD_URL = ENV.get("DISCORD_WEBHOOK_URL", "")
|
||||
ALLOWLIST_PATH = ENV.get("ALLOWLIST_PATH", "/config/allowlist.yaml")
|
||||
WORK_ROOT = Path(ENV.get("WORK_ROOT", "/work"))
|
||||
AUDIT_DIR = APP_DIR.parent / "logs"
|
||||
MAX_TAIL = int(ENV.get("MAX_OUTPUT_TAIL", "4000"))
|
||||
TIMEOUT_S = int(ENV.get("REQUEST_TIMEOUT_S", "900"))
|
||||
COOKIE_NAME = "exec_session"
|
||||
CRED_RE = re.compile(r"https?://[^/\s@]+:[^@\s]+@")
|
||||
PROVISIONING_KIND = "provisioning"
|
||||
|
||||
app = FastAPI(title="coreci-executor", version="0.3.0")
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_methods=["GET", "POST", "OPTIONS"],
|
||||
allow_headers=["*"],
|
||||
WRITE_VERB_RE = re.compile(
|
||||
r"(?i)\b("
|
||||
r"curl\s+[^|;]*\s-X\s+(POST|PUT|PATCH|DELETE)"
|
||||
r"|curl\s+[^|;]*-(d|--data|--data-binary|--data-raw|urlencode)\s"
|
||||
r"|\bgit\s+push\b"
|
||||
r"|\bgit\s+commit\b(?![^|;\n]*--amend\b)"
|
||||
r"|\bgh\s+(pr|issue|release|repo)\s+(create|edit|close|merge|delete)\b"
|
||||
r"|\b(rm|mv|cp)\s+-[^\n]*\s+/\w"
|
||||
r")"
|
||||
)
|
||||
|
||||
REQS: dict[str, dict] = {}
|
||||
LOCK = threading.Lock()
|
||||
|
||||
def now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
def current_allowlist():
|
||||
import os, yaml
|
||||
with open(os.environ.get(
|
||||
"ALLOWLIST_PATH", "allowlist.yaml"), "r") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
def get_req(rid: str) -> dict:
|
||||
r = REQS.get(rid)
|
||||
if r is None:
|
||||
raise HTTPException(404, "no such request")
|
||||
return r
|
||||
|
||||
def audit(evt: str, r: dict, **extra) -> None:
|
||||
AUDIT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
rec = {"ts": now(), "event": evt, "id": r.get("id"),
|
||||
"agent": r.get("agent"), "kind": r.get("kind"),
|
||||
"targets": r.get("targets"), "decision": r.get("decision"),
|
||||
"auth": ",".join(sorted(r.get("creds_names") or [])) or "executor",
|
||||
"actor": extra.pop("actor", None), **extra}
|
||||
line = json.dumps(rec, default=str)
|
||||
(AUDIT_DIR / f"audit-{datetime.now(timezone.utc):%Y-%m-%d}.jsonl").open("a").write(line + "\n")
|
||||
if DISCORD_URL:
|
||||
try:
|
||||
httpx.post(DISCORD_URL, json={"content": f"[executor] {evt}: {line[:1500]}"}, timeout=5)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def classify(agent: str, kind: str, targets: list) -> str:
|
||||
# provisioning is NEVER auto-runnable, regardless of allowlist contents
|
||||
if kind == PROVISIONING_KIND:
|
||||
return "NEEDS_APPROVAL"
|
||||
cfg = yaml.safe_load(Path(ALLOWLIST_PATH).read_text()) or {}
|
||||
ag = (cfg.get("agents") or {}).get(agent) or {}
|
||||
auto = ((ag.get("kinds") or {}).get(kind) or {}).get("auto_run_targets") or []
|
||||
return "AUTO_RUN" if targets and all(t in auto for t in targets) else "NEEDS_APPROVAL"
|
||||
|
||||
def normalize_creds(raw) -> dict:
|
||||
"""Accept {gitea_user, gitea_token} (v0.2.0 shape) or {name: {user, secret}}.
|
||||
Returns {name: {user, secret}} with stripped strings."""
|
||||
creds: dict[str, dict] = {}
|
||||
if not isinstance(raw, dict):
|
||||
return creds
|
||||
if ("gitea_user" in raw or "gitea_token" in raw) and "gitea" not in raw:
|
||||
u = (raw.get("gitea_user") or "").strip()
|
||||
s = (raw.get("gitea_token") or "").strip()
|
||||
if u or s:
|
||||
creds["gitea"] = {"user": u, "secret": s}
|
||||
return creds
|
||||
for name, pair in raw.items():
|
||||
if not isinstance(pair, dict):
|
||||
continue
|
||||
u = (pair.get("user") or "").strip()
|
||||
s = (pair.get("secret") or "").strip()
|
||||
if u or s:
|
||||
creds[str(name)] = {"user": u, "secret": s}
|
||||
return creds
|
||||
|
||||
def run_pack(rid: str) -> None:
|
||||
r = REQS[rid]
|
||||
r["status"] = "running"; audit("started", r)
|
||||
creds = r.get("creds") or {}
|
||||
# Primary identity: first pair with a user, else executor identity
|
||||
primary = next((c for c in creds.values() if c["user"]), None)
|
||||
git_user = primary["user"] if primary else GITEA_USER
|
||||
git_token = primary["secret"] if primary else GITEA_TOKEN
|
||||
home = Path(tempfile.mkdtemp(dir=str(WORK_ROOT), prefix="req-"))
|
||||
(home / ".git-credentials").write_text(
|
||||
f"https://{git_user}:{git_token}@{GITEA_HOST}\n")
|
||||
script = "git config --global credential.helper store\n" + r["content"]
|
||||
env = {"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": str(home),
|
||||
"GIT_TERMINAL_PROMPT": "0", "GITEA_URL": GITEA_URL,
|
||||
"GITEA_TOKEN": git_token, "GITEA_USER": git_user, "TZ": "UTC"}
|
||||
# expose every named credential as env pairs; track secrets for scrubbing
|
||||
secrets = [GITEA_TOKEN]
|
||||
for name, c in creds.items():
|
||||
cap = re.sub(r"[^A-Z0-9]", "_", name.upper())
|
||||
if c["user"]:
|
||||
env[f"CRED_{cap}_USER"] = c["user"]
|
||||
if c["secret"]:
|
||||
env[f"CRED_{cap}_SECRET"] = c["secret"]
|
||||
secrets.append(c["secret"])
|
||||
try:
|
||||
p = subprocess.run(["bash", "-c", script], capture_output=True,
|
||||
text=True, timeout=TIMEOUT_S, cwd=home, env=env)
|
||||
tail = (p.stdout[-MAX_TAIL:] + "\n" + p.stderr[-MAX_TAIL:]).strip()
|
||||
for s in secrets:
|
||||
if s:
|
||||
tail = tail.replace(s, "***")
|
||||
r.update(status="completed" if p.returncode == 0 else "failed",
|
||||
exit_code=p.returncode,
|
||||
output=tail, finished_at=now())
|
||||
audit("completed" if p.returncode == 0 else "failed", r, exit_code=p.returncode)
|
||||
except subprocess.TimeoutExpired:
|
||||
r.update(status="failed", output="TIMEOUT", finished_at=now()); audit("failed", r, error="timeout")
|
||||
except Exception as e:
|
||||
r.update(status="failed", output=str(e)[:MAX_TAIL], finished_at=now()); audit("failed", r, error=str(e)[:200])
|
||||
finally:
|
||||
try:
|
||||
(home / ".git-credentials").unlink()
|
||||
except Exception:
|
||||
pass
|
||||
shutil.rmtree(home, ignore_errors=True)
|
||||
|
||||
def key_ok(request: Request) -> bool:
|
||||
"""Accept Authorization: Bearer <key> (Open WebUI) OR X-Executor-Key: <key>."""
|
||||
auth = request.headers.get("authorization", "")
|
||||
return (request.headers.get("X-Executor-Key") == API_KEY
|
||||
or (auth.startswith("Bearer ") and auth[7:] == API_KEY))
|
||||
|
||||
def session_ok(request: Request) -> bool:
|
||||
return request.cookies.get(COOKIE_NAME) == hashlib.sha256(PASSCODE.encode()).hexdigest()
|
||||
|
||||
def page(title: str, body: str) -> str:
|
||||
return (f"<!doctype html><html><head><meta charset='utf-8'><title>{title}"
|
||||
f"</title><style>body{{font-family:system-ui;margin:2rem;max-width:60rem}}"
|
||||
f"pre{{background:#f4f4f4;padding:1rem;overflow:auto}}"
|
||||
f"button{{padding:.4rem 1rem}}</style></head>"
|
||||
f"<body>{body}</body></html>")
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return {"status": "ok", "version": "0.3.0", "requests": len(REQS)}
|
||||
|
||||
@app.post("/submit_request")
|
||||
async def submit(request: Request):
|
||||
if not key_ok(request):
|
||||
raise HTTPException(401, "bad or missing API key")
|
||||
d = await request.json()
|
||||
agent, kind = d.get("agent", ""), d.get("kind", "bash_pack")
|
||||
targets = d.get("targets") or []
|
||||
content, just = d.get("content", ""), d.get("justification", "")
|
||||
creds = normalize_creds(d.get("credentials"))
|
||||
if kind not in ("bash_pack", PROVISIONING_KIND):
|
||||
raise HTTPException(400, "v0 supports kind=bash_pack or provisioning only")
|
||||
if not (agent and content and just):
|
||||
raise HTTPException(400, "agent, content, justification required")
|
||||
if not isinstance(targets, list) or not targets or not all(isinstance(t, str) for t in targets):
|
||||
raise HTTPException(400, "targets must be a non-empty list of repo names")
|
||||
if CRED_RE.search(content):
|
||||
raise HTTPException(400, "credential-free packs only (embedded user:pass rejected)")
|
||||
rid = uuid.uuid4().hex[:12]
|
||||
r = {"id": rid, "agent": agent, "kind": kind, "targets": targets,
|
||||
"content": content, "justification": just, "submitted_at": now(),
|
||||
"status": "pending_approval", "decision": classify(agent, kind, targets),
|
||||
"creds": creds, "creds_names": sorted(creds.keys())}
|
||||
with LOCK:
|
||||
REQS[rid] = r
|
||||
audit("submitted", r)
|
||||
if r["decision"] == "AUTO_RUN":
|
||||
r["status"] = "queued"
|
||||
threading.Thread(target=run_pack, args=(rid,), daemon=True).start()
|
||||
return {"request_id": rid, "status": r["status"], "decision": r["decision"]}
|
||||
|
||||
@app.get("/request_status")
|
||||
def status(request_id: str, request: Request):
|
||||
if not key_ok(request) and not session_ok(request):
|
||||
raise HTTPException(401, "unauthorized")
|
||||
r = get_req(request_id)
|
||||
return {k: r.get(k) for k in ("id", "agent", "kind", "targets", "decision",
|
||||
"status", "exit_code", "output", "submitted_at", "finished_at")}
|
||||
|
||||
@app.get("/")
|
||||
def root(request: Request):
|
||||
if not session_ok(request):
|
||||
return RedirectResponse("/ui/login")
|
||||
rows = "".join(
|
||||
f"<tr><td><a href='/ui/request/{r['id']}'>{r['id']}</a></td><td>{r['agent']}</td>"
|
||||
f"<td>{','.join(r['targets'])}</td><td>{r['status']}</td><td>{r['decision']}</td></tr>"
|
||||
for r in sorted(REQS.values(), key=lambda x: x["submitted_at"], reverse=True)[:50])
|
||||
return HTMLResponse(page("executor", f"<h1>coreci-executor — queue</h1>"
|
||||
f"<table border='1' cellpadding='6'><tr><th>id</th><th>agent</th><th>targets</th>"
|
||||
f"<th>status</th><th>decision</th></tr>{rows}</table>"))
|
||||
|
||||
@app.get("/ui/login")
|
||||
def login():
|
||||
return HTMLResponse(page("login", "<h1>executor — sign in</h1>"
|
||||
"<form method='post' action='/ui/login'><input type='password' name='p' autofocus>"
|
||||
" <button>unlock</button></form>"))
|
||||
|
||||
@app.post("/ui/login")
|
||||
async def login_post(request: Request):
|
||||
pw = None
|
||||
try:
|
||||
pw = (await request.form()).get("p")
|
||||
except Exception:
|
||||
try:
|
||||
pw = (json.loads(await request.body()) or {}).get("p")
|
||||
except Exception:
|
||||
pw = None
|
||||
if pw != PASSCODE:
|
||||
raise HTTPException(401, "wrong passcode — the web UI takes EXECUTOR_PASSCODE from .env, not the API key")
|
||||
resp = RedirectResponse("/", status_code=303)
|
||||
resp.set_cookie(COOKIE_NAME, hashlib.sha256(PASSCODE.encode()).hexdigest(),
|
||||
httponly=True, samesite="strict")
|
||||
return resp
|
||||
|
||||
@app.get("/ui/request/{rid}")
|
||||
def detail(rid: str, request: Request):
|
||||
if not session_ok(request):
|
||||
return RedirectResponse("/ui/login")
|
||||
r = get_req(rid)
|
||||
act = (f"<form method='post' action='/ui/request/{rid}/approve'><button>APPROVE</button></form> "
|
||||
f"<form method='post' action='/ui/request/{rid}/deny'><button>DENY</button></form>"
|
||||
if r["status"] == "pending_approval" else "")
|
||||
return HTMLResponse(page(rid, f"<h1>{rid} — {r['status']}</h1>"
|
||||
f"<p>agent <b>{r['agent']}</b> | kind <b>{r['kind']}</b> | targets <b>{','.join(r['targets'])}</b> "
|
||||
f"| decision <b>{r['decision']}</b></p><p><i>{esc(r['justification'])}</i></p>"
|
||||
f"<h3>pack</h3><pre>{esc(r['content'])}</pre>"
|
||||
f"<h3>output</h3><pre>{esc(r.get('output', '(pending)'))}</pre>{act}"))
|
||||
|
||||
@app.post("/ui/request/{rid}/approve")
|
||||
def approve(rid: str, request: Request):
|
||||
if not session_ok(request):
|
||||
return RedirectResponse("/ui/login")
|
||||
r = get_req(rid)
|
||||
if r["status"] == "pending_approval":
|
||||
r["status"] = "running"; r["decision"] = "APPROVED"; audit("approved", r, actor="CEO")
|
||||
threading.Thread(target=run_pack, args=(rid,), daemon=True).start()
|
||||
return RedirectResponse(f"/ui/request/{rid}", status_code=303)
|
||||
|
||||
@app.post("/ui/request/{rid}/deny")
|
||||
def deny(rid: str, request: Request):
|
||||
if not session_ok(request):
|
||||
return RedirectResponse("/ui/login")
|
||||
r = get_req(rid)
|
||||
if r["status"] == "pending_approval":
|
||||
r["status"] = "denied"; audit("denied", r, actor="CEO")
|
||||
return RedirectResponse(f"/ui/request/{rid}", status_code=303)
|
||||
|
||||
@app.get("/ui/audit")
|
||||
def audit_view(request: Request):
|
||||
if not session_ok(request):
|
||||
return RedirectResponse("/ui/login")
|
||||
f = AUDIT_DIR / f"audit-{datetime.now(timezone.utc):%Y-%m-%d}.jsonl"
|
||||
txt = f.read_text()[-20000:] if f.exists() else "(empty)"
|
||||
return HTMLResponse(page("audit", f"<h1>audit — today</h1><pre>{esc(txt)}</pre>"))
|
||||
|
||||
def custom_openapi():
|
||||
if app.openapi_schema:
|
||||
return app.openapi_schema
|
||||
schema = get_openapi(title=app.title, version=app.version, routes=app.routes)
|
||||
schema.setdefault("components", {})["securitySchemes"] = {
|
||||
"bearerAuth": {"type": "http", "scheme": "bearer",
|
||||
"description": "EXECUTOR_API_KEY as Authorization: Bearer <key>"},
|
||||
"executorKey": {"type": "apiKey", "in": "header", "name": "X-Executor-Key",
|
||||
"description": "EXECUTOR_API_KEY via X-Executor-Key header"},
|
||||
}
|
||||
for _p in ("/submit_request", "/request_status"):
|
||||
for _m in schema.get("paths", {}).get(_p, {}).values():
|
||||
_m.setdefault("security", [{"bearerAuth": []}, {"executorKey": []}])
|
||||
app.openapi_schema = schema
|
||||
return schema
|
||||
|
||||
app.openapi = custom_openapi
|
||||
def classify(agent, targets, content, kind):
|
||||
al = current_allowlist()
|
||||
defaults = al["defaults"]
|
||||
agent_cfg = al["agents"].get(agent)
|
||||
if agent_cfg is None:
|
||||
return defaults["decision"]
|
||||
# agent kinds may be nested under agent_cfg["kinds"][kind]
|
||||
agent_kind_cfg = (
|
||||
agent_cfg.get("kinds", {}).get(kind, {})
|
||||
if isinstance(agent_cfg, dict) else {}
|
||||
)
|
||||
auto_run_targets = (
|
||||
set(defaults.get("auto_run_targets", []))
|
||||
| set(agent_cfg.get("auto_run_targets", [])
|
||||
if not isinstance(agent_cfg, dict) else
|
||||
agent_kind_cfg.get("auto_run_targets",
|
||||
agent_cfg.get("auto_run_targets", [])))
|
||||
)
|
||||
read_targets = (
|
||||
set(defaults.get("read_targets", []))
|
||||
| set(agent_cfg.get("read_targets", [])
|
||||
if not isinstance(agent_cfg, dict) else
|
||||
agent_kind_cfg.get("read_targets",
|
||||
agent_cfg.get("read_targets", [])))
|
||||
)
|
||||
if all(t in auto_run_targets for t in targets):
|
||||
return "AUTO_RUN"
|
||||
allowed_for_read = auto_run_targets | read_targets
|
||||
if all(t in allowed_for_read for t in targets):
|
||||
if not WRITE_VERB_RE.search(content):
|
||||
return "AUTO_RUN"
|
||||
return defaults["decision"]
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
"""coreci/executor-config/app/test_allowlist.py
|
||||
|
||||
Test cases T1–T17 per coreci-coo brief 2026-09-26 §D6.
|
||||
deny_read_targets is NOT implemented; T8/T17 carry xfail.
|
||||
"""
|
||||
import pytest
|
||||
from main import classify
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def allowlist(monkeypatch):
|
||||
"""Allowlist v4 fixture: agent-level read_targets augmented (T16)."""
|
||||
import main, os, tempfile, yaml
|
||||
al = {
|
||||
"version": 4,
|
||||
"defaults": {
|
||||
"decision": "NEEDS_APPROVAL",
|
||||
"read_targets": ["coreci/company"],
|
||||
},
|
||||
"agents": {
|
||||
"coreci-coo": {
|
||||
"kinds": {"bash_pack": {"auto_run_targets": ["coreci-coo/home"]}}
|
||||
},
|
||||
"coreci-cairo": {
|
||||
"kinds": {"bash_pack": {"auto_run_targets": ["coreci-cairo/home"]}}
|
||||
},
|
||||
},
|
||||
}
|
||||
fd, path = tempfile.mkstemp(suffix=".yaml")
|
||||
with os.fdopen(fd, "w") as f:
|
||||
yaml.safe_dump(al, f)
|
||||
monkeypatch.setenv("ALLOWLIST_PATH", path)
|
||||
monkeypatch.setattr(main, "current_allowlist", lambda: al)
|
||||
return al
|
||||
|
||||
|
||||
def T(agent, targets, content, expected):
|
||||
return (agent, targets, content, expected)
|
||||
|
||||
|
||||
# Format: (agent, targets, content, expected)
|
||||
CASES = [
|
||||
T("coreci-coo", ["coreci/company"], "curl -sS .../raw/foo.md",
|
||||
"AUTO_RUN"), # T1
|
||||
T("coreci-coo", ["coreci-coo/home"], "git push origin main",
|
||||
"AUTO_RUN"), # T2
|
||||
T("coreci-coo", ["coreci/company"],
|
||||
"curl -X POST .../contents/x -d @body", "NEEDS_APPROVAL"), # T3
|
||||
T("coreci-coo", ["coreci/company"],
|
||||
"curl -sS .../raw/foo.md\ncurl -X PUT .../contents/x",
|
||||
"NEEDS_APPROVAL"), # T4
|
||||
T("coreci-coo", ["coreci/company"],
|
||||
"rm -rf /work/staging/foo && curl -sS .../raw/foo.md",
|
||||
"NEEDS_APPROVAL"), # T5
|
||||
T("coreci-coo", ["coreci/company", "coreci-coo/home"],
|
||||
"curl -X PUT .../repos/coreci-coo/home/contents/x",
|
||||
"NEEDS_APPROVAL"), # T6 conservative
|
||||
T("coreci-coo", ["coreci/executor-config"], "curl -sS .../raw/foo.md",
|
||||
"NEEDS_APPROVAL"), # T7
|
||||
T("coreci-coo", ["coreci/company"],
|
||||
"curl -X\\nPOST .../contents/x",
|
||||
"NEEDS_APPROVAL"), # T8 xfail
|
||||
T("unknown-agent", ["anything"], "echo hi", "NEEDS_APPROVAL"), # T9
|
||||
T("coreci-coo", ["coreci-coo/home"], "curl -X GET ...",
|
||||
"AUTO_RUN"), # T10
|
||||
T("coreci-coo", ["coreci/company"], "", "NEEDS_APPROVAL"), # T11
|
||||
T("coreci-coo", ["coreci/company"], "git push origin main",
|
||||
"NEEDS_APPROVAL"), # T12
|
||||
T("coreci-cairo", ["coreci/company"], "curl -sS .../raw/foo.md",
|
||||
"AUTO_RUN"), # T13
|
||||
T("coreci-cairo", ["coreci/company"],
|
||||
"curl -X POST .../contents/x -d @body", "NEEDS_APPROVAL"), # T14
|
||||
T("coreci-cairo", ["coreci/executor-config"], "curl -sS .../raw/foo.md",
|
||||
"NEEDS_APPROVAL"), # T15
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("agent,targets,content,expected", CASES)
|
||||
def test_classify(allowlist, agent, targets, content, expected):
|
||||
assert classify(agent, targets, content, "bash_pack") == expected
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="obfuscation via shell-escaped newline; ticket #EXEC-201",
|
||||
strict=False)
|
||||
def test_T8_obfuscated_newline(allowlist):
|
||||
"""T8 — known miss; carry xfail + ticket reference."""
|
||||
assert classify("coreci-coo", ["coreci/company"],
|
||||
"curl -X\\nPOST .../contents/x",
|
||||
"bash_pack") == "NEEDS_APPROVAL"
|
||||
|
||||
|
||||
def test_T16_agent_read_target_augments_defaults(allowlist):
|
||||
"""T16 — agent-level read_targets UNION defaults."""
|
||||
al = dict(allowlist)
|
||||
al["agents"] = dict(al["agents"])
|
||||
al["agents"]["coreci-coo"] = dict(al["agents"]["coreci-coo"])
|
||||
al["agents"]["coreci-coo"]["read_targets"] = ["coreci-coo/home/scratch"]
|
||||
import main
|
||||
main.current_allowlist = lambda: al
|
||||
assert classify("coreci-coo", ["coreci-coo/home/scratch"],
|
||||
"curl -sS .../raw/foo.md", "bash_pack") == "AUTO_RUN"
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="deny_read_targets not implemented in this PR; "
|
||||
"ticket #EXEC-202", strict=False)
|
||||
def test_T17_deny_read_targets(allowlist):
|
||||
"""T17 — future negative override. NOT implemented in this PR."""
|
||||
assert classify("coreci-malicious", ["coreci/company"],
|
||||
"curl -sS .../raw/foo.md", "bash_pack") == "NEEDS_APPROVAL"
|
||||
Reference in New Issue
Block a user