tools/llm_api_check.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""BYOK provider endpoints on a real panel: encrypted, redacted, scoped.
Spins its own throwaway panel (its own home, port and tmux socket — it never
touches a panel you are using) and drives the provider API the way the UI will:
create a provider, list it, test it, update it, delete it. What it proves is the
security contract, not just the happy path:
- the plaintext key is never returned by any endpoint and never appears in
state.json on disk — only its ciphertext (``gcm1:``) does;
- providers ride no read poll: /api/state carries neither the key nor the record;
- a read-only token may list providers but not create one; the attention-scoped
hook token is refused the whole surface;
- bad input (unknown kind, missing key, non-http URL) is a clean 4xx, and a
test against an unreachable provider is a structured failure, not a 500.
python3 tools/llm_api_check.py
Exit status is 0 on pass, 1 on fail.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PASSWORD = "llm-api-check" # noqa: S105 — a throwaway panel on loopback
PORT = 3282
BASE = f"http://127.0.0.1:{PORT}"
HOME = Path("/tmp/clique-llm-api-home")
SOCKET = "clique-llm-api-check"
SECRET = "sk-PLAINTEXT-DO-NOT-LEAK-9f3a" # noqa: S105 — the value we assert never leaks
def _mint(env, *args) -> str:
out = subprocess.run(
[sys.executable, "-m", "clique", "token", "create", *args],
env=env,
cwd=str(ROOT),
capture_output=True,
text=True,
).stdout
return next((ln.strip() for ln in out.splitlines() if ln.strip().startswith("mxp_")), "")
def _panel() -> tuple[subprocess.Popen, str, str]:
shutil.rmtree(HOME, ignore_errors=True)
HOME.mkdir(parents=True)
env = dict(os.environ, CLIQUE_HOME=str(HOME), CLIQUE_TMUX_SOCKET=SOCKET)
subprocess.run(
[sys.executable, "-m", "clique", "password"],
input=f"{PASSWORD}\n{PASSWORD}\n",
text=True,
env=env,
cwd=str(ROOT),
capture_output=True,
)
admin = _mint(env, "admin")
readonly = _mint(env, "watcher", "--read-only")
proc = subprocess.Popen(
[
sys.executable,
"-m",
"clique",
"--host",
"127.0.0.1",
"--port",
str(PORT),
"--state",
str(HOME / "state.json"),
],
env=env,
cwd=str(ROOT),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
for _ in range(80):
try:
urllib.request.urlopen(BASE + "/healthz", timeout=2).read()
return proc, admin, readonly
except (urllib.error.URLError, OSError):
time.sleep(0.25)
proc.kill()
raise SystemExit(f"the check's own panel never came up on {PORT}")
def req(token: str, method: str, path: str, body=None) -> tuple[int, str]:
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(
BASE + path,
data=data,
method=method,
headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(r, timeout=20) as resp:
return resp.status, resp.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return e.code, e.read().decode("utf-8", "replace")
def main() -> int:
proc, admin, readonly = _panel()
res: dict[str, object] = {}
try:
# A session's own attention token, read from its pane environment, to
# prove that token is shut out of the provider surface too.
attn = ""
tsid = json.loads(
req(admin, "POST", "/api/sessions", {"cli": "shell", "cwd": "/tmp", "name": "t"})[1]
)["id"]
tmux_target = "sm-" + tsid.replace("-", "")[:8]
for _ in range(40):
out = subprocess.run(
["tmux", "-L", SOCKET, "show-environment", "-t", tmux_target, "CLIQUE_TOKEN"],
capture_output=True,
text=True,
).stdout.strip()
if out.startswith("CLIQUE_TOKEN="):
attn = out.split("=", 1)[1].strip()
if attn:
break
time.sleep(0.25)
# Create a provider with a real-looking secret.
status, raw = req(
admin,
"POST",
"/api/llm/providers",
{
"label": "Test OR",
"kind": "openai",
"base_url": "http://127.0.0.1:9/v1",
"model": "test/model",
"key": SECRET,
},
)
created = json.loads(raw)
pid = created.get("id", "")
res["create_201"] = status == 201 and pid.startswith("lp-")
res["create_key_set"] = created.get("key_set") is True
res["create_hides_key"] = (
"key" not in created and "key_enc" not in created and SECRET not in raw
)
# List: redacted, and still no key material.
listed = json.loads(req(admin, "GET", "/api/llm/providers")[1])
res["list_shows_provider"] = any(p["id"] == pid for p in listed["providers"])
res["list_encryption_flag"] = listed.get("encryption") is True
res["list_hides_key"] = SECRET not in json.dumps(listed) and all(
"key_enc" not in p for p in listed["providers"]
)
# On disk: ciphertext, never the plaintext.
disk = (HOME / "state.json").read_text()
res["disk_has_ciphertext"] = "gcm1:" in disk
res["disk_no_plaintext"] = SECRET not in disk
# The poll never carries the key or the provider record.
state = req(admin, "GET", "/api/state")[1]
res["state_no_key"] = SECRET not in state and "key_enc" not in state
# Scope: a read token lists but cannot create; attention is shut out.
res["readonly_lists"] = req(readonly, "GET", "/api/llm/providers")[0] == 200
res["readonly_cannot_create"] = (
req(
readonly,
"POST",
"/api/llm/providers",
{
"kind": "openai",
"base_url": "https://x/v1",
"model": "m",
"key": "sk-x",
},
)[0]
== 403
)
res["attn_get_401"] = req(attn, "GET", "/api/llm/providers")[0] == 401
res["attn_post_refused"] = req(
attn,
"POST",
"/api/llm/providers",
{
"kind": "openai",
"base_url": "https://x/v1",
"model": "m",
"key": "sk-x",
},
)[0] in (401, 403)
# Update: change the model, blank key keeps the stored one.
upd = json.loads(req(admin, "POST", f"/api/llm/providers/{pid}", {"model": "test/v2"})[1])
res["update_model"] = upd.get("model") == "test/v2"
res["update_keeps_key"] = upd.get("key_set") is True
# Bad input is a clean 4xx, not a 500.
res["bad_kind_400"] = (
req(
admin,
"POST",
"/api/llm/providers",
{"kind": "wat", "base_url": "https://x", "model": "m", "key": "k"},
)[0]
== 400
)
res["no_key_400"] = (
req(
admin,
"POST",
"/api/llm/providers",
{"kind": "openai", "base_url": "https://x/v1", "model": "m"},
)[0]
== 400
)
res["bad_url_400"] = (
req(
admin,
"POST",
"/api/llm/providers",
{"kind": "openai", "base_url": "ftp://x", "model": "m", "key": "k"},
)[0]
== 400
)
# Routes: point the inbox feature at this provider, and guard the edges.
state2 = json.loads(req(admin, "GET", "/api/llm/providers")[1])
res["routes_feature_listed"] = "inbox" in state2.get("features", [])
setr = req(admin, "POST", "/api/llm/routes", {"feature": "inbox", "provider_id": pid})
res["route_set"] = setr[0] == 200 and json.loads(setr[1])["routes"].get("inbox") == pid
res["route_in_list"] = (
json.loads(req(admin, "GET", "/api/llm/providers")[1])["routes"].get("inbox") == pid
)
res["route_unknown_feature_400"] = (
req(admin, "POST", "/api/llm/routes", {"feature": "nope", "provider_id": pid})[0] == 400
)
res["route_unknown_provider_400"] = (
req(admin, "POST", "/api/llm/routes", {"feature": "inbox", "provider_id": "lp-nope"})[0]
== 400
)
res["route_readonly_refused"] = (
req(readonly, "POST", "/api/llm/routes", {"feature": "inbox", "provider_id": pid})[0]
== 403
)
# Test an unreachable provider → structured failure, HTTP 200.
tstatus, traw = req(admin, "POST", f"/api/llm/providers/{pid}/test")
tbody = json.loads(traw)
res["test_structured_failure"] = (
tstatus == 200 and tbody.get("ok") is False and bool(tbody.get("error"))
)
# Delete, then it is gone and un-testable.
res["delete_ok"] = (
json.loads(req(admin, "POST", f"/api/llm/providers/{pid}/delete")[1]).get("ok") is True
)
res["gone_after_delete"] = not any(
p["id"] == pid
for p in json.loads(req(admin, "GET", "/api/llm/providers")[1])["providers"]
)
res["test_missing_404"] = req(admin, "POST", f"/api/llm/providers/{pid}/test")[0] == 404
# Deleting the provider drops the route that pointed at it — no dangle.
res["route_dropped_on_delete"] = "inbox" not in json.loads(
req(admin, "GET", "/api/llm/providers")[1]
).get("routes", {})
finally:
proc.terminate()
subprocess.run(
["tmux", "-L", SOCKET, "kill-server"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
shutil.rmtree(HOME, ignore_errors=True)
ok = all(res.values())
for key, value in res.items():
print(f" {'ok ' if value else 'FAIL'} {key}: {value}")
print("llm_api_check:", "PASS" if ok else "FAIL")
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())