65 lines
4.0 KiB
Python
65 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Governed model-provider selector for Chat.
|
|
|
|
The browser sends only a provider ID. This resolver reads the managed provider
|
|
policy, checks actor role, agent model role and credential availability, writes
|
|
an append-only selection audit, and returns no secret material.
|
|
"""
|
|
import argparse, hashlib, json, os, sys
|
|
from datetime import datetime, timezone
|
|
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))
|
|
CONFIG = os.path.join(ROOT, "packages", "casan-harness", "config", "model-providers.yaml")
|
|
|
|
def now(): return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
def sha(v): return hashlib.sha256(v.encode()).hexdigest()
|
|
def state_root(): return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
|
|
def audit_path(): return os.environ.get("CASAN_CHAT_MODEL_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "model-selection.jsonl")
|
|
def load():
|
|
try:
|
|
return json.load(open(os.environ.get("CASAN_MODEL_PROVIDERS_FILE") or CONFIG, encoding="utf-8"))
|
|
except Exception as exc:
|
|
return {"_error": f"providers_unreadable:{exc}"}
|
|
def visible(pid, p, role, model_role):
|
|
allowed_roles = p.get("roles_allowed", [])
|
|
allowed_models = p.get("model_roles_allowed", [])
|
|
enabled = not p.get("requires_key") or bool(os.environ.get(p.get("key_env", "")))
|
|
return {"id": pid, "model": p.get("model", ""), "class": p.get("class", "local"),
|
|
"requires_preflight": bool(p.get("requires_preflight")), "allowed": role in allowed_roles and model_role in allowed_models,
|
|
"configured": enabled, "reason": "" if enabled else "credential_not_configured"}
|
|
def append(rec):
|
|
path=audit_path(); os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
prev=""
|
|
if os.path.isfile(path):
|
|
for line in open(path, encoding="utf-8"):
|
|
if line.strip(): prev=json.loads(line).get("record_hash", prev)
|
|
core={**rec,"prev_hash":prev}; core["record_hash"]=sha(json.dumps(core,sort_keys=True,ensure_ascii=False))
|
|
with open(path,"a",encoding="utf-8") as f: f.write(json.dumps(core,ensure_ascii=False)+"\n")
|
|
return core
|
|
def list_models(args):
|
|
cfg=load()
|
|
if cfg.get("_error"): print(json.dumps({"success":False,"providers":[],"reason":cfg["_error"]})); return 2
|
|
rows=[visible(pid,p,args.role,args.model_role) for pid,p in cfg.get("providers",{}).items()]
|
|
print(json.dumps({"success":True,"model_role":args.model_role,"default":cfg.get("role_bindings",{}).get(args.model_role,""),"providers":rows}))
|
|
return 0
|
|
def bind(args):
|
|
cfg=load()
|
|
if cfg.get("_error"): print(json.dumps({"success":False,"decision":"DENIED","reason":cfg["_error"]})); return 2
|
|
pid=args.provider or cfg.get("role_bindings",{}).get(args.model_role,"")
|
|
p=cfg.get("providers",{}).get(pid)
|
|
if not p:
|
|
decision,reason="DENIED","unknown_provider"
|
|
else:
|
|
row=visible(pid,p,args.role,args.model_role)
|
|
decision,reason=("BOUND", "provider_bound") if row["allowed"] and row["configured"] else ("DENIED", row["reason"] or "provider_not_allowed")
|
|
rec=append({"timestamp":now(),"harness":"H5-chat-model-bind","actor":args.actor,"role":args.role,"model_role":args.model_role,"provider":pid,"decision":decision,"reason":reason})
|
|
out={"success":decision=="BOUND","decision":decision,"reason":reason,"provider":pid,"model":(p or {}).get("model"),"class":(p or {}).get("class"),"audit":{"seq":rec.get("record_hash","")[:16],"record_hash":rec["record_hash"]}}
|
|
print(json.dumps(out,ensure_ascii=False)); return 0 if decision=="BOUND" else 2
|
|
def main():
|
|
ap=argparse.ArgumentParser(); subs=ap.add_subparsers(dest="cmd",required=True)
|
|
for name in ("list","bind"):
|
|
p=subs.add_parser(name); p.add_argument("--role",required=True); p.add_argument("--model-role",required=True); p.add_argument("--actor",default="anonymous")
|
|
if name=="bind": p.add_argument("--provider",default="")
|
|
args=ap.parse_args(); return list_models(args) if args.cmd=="list" else bind(args)
|
|
if __name__=="__main__": raise SystemExit(main())
|