gitea-admin/CASAN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: chat + optmz control panel

Browse Source
This commit is contained in:
thanhnv
2026-07-10 16:26:30 +09:00
parent d882a9dc23
commit 7cea023dce
28 changed files with 1702 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/casan-harness/scripts/bash/model-call.py
+95 -6
View File
@@ -23,6 +23,8 @@ Usage:
import argparse
import json
import os
import ipaddress
from urllib.parse import urlparse
def _casan_app_root():
@@ -43,6 +45,7 @@ import urllib.request
from datetime import datetime, timezone
OLLAMA_HOST = "127.0.0.1:11434" # the only allowed ollama endpoint
DOCKER_OLLAMA_HOST = "host.docker.internal:11434"
ALLOWED_CLOUD = {"api.anthropic.com", "api.openai.com"}
REPO_ROOT = _casan_app_root()
PROVIDER_LOG = os.path.join(REPO_ROOT, ".specify/logs/level5/provider-usage.jsonl")
@@ -131,11 +134,20 @@ def extract_verdict(role, text):
return None, False
def ollama_host_allowed(host: str) -> bool:
return host == OLLAMA_HOST or (
os.environ.get("CASAN_ALLOW_DOCKER_HOST_OLLAMA") == "1"
and host == DOCKER_OLLAMA_HOST
)
def call_ollama(model_name, prompt, role):
# SSRF guard: hard-pinned loopback endpoint, no env override of host.
# SSRF guard: hard-pinned loopback endpoint by default. Docker Desktop has
# one explicit, opt-in bridge to the Mac host's Ollama daemon; arbitrary
# LAN, metadata and user-supplied hosts remain blocked.
host = os.environ.get("CASAN_OLLAMA_HOST", OLLAMA_HOST)
if host != OLLAMA_HOST:
fail(f"endpoint_not_allowed ollama host={host} (only {OLLAMA_HOST})")
if not ollama_host_allowed(host):
fail(f"endpoint_not_allowed ollama host={host} (only {OLLAMA_HOST}; Docker bridge requires explicit opt-in)")
digest_gate = os.path.join(os.path.dirname(__file__), "model-digest-check.sh")
if os.path.isfile(digest_gate):
check = subprocess.run(
@@ -211,6 +223,73 @@ def call_openai(model_name, prompt, role):
}
def openai_compatible_url():
"""Return a vetted OpenAI-compatible chat-completions endpoint.
This is opt-in, preserving the existing hard-pinned public OpenAI route.
The default allowlist permits only local development hosts; another host
must be explicitly approved by the operator.
"""
raw = os.environ.get("CASAN_OPENAI_COMPATIBLE_BASE_URL", "").strip()
if not raw:
fail("openai_compatible_base_url_unset")
parsed = urlparse(raw)
if parsed.scheme not in {"http", "https"} or not parsed.hostname \
or parsed.username or parsed.password or parsed.query or parsed.fragment:
fail("endpoint_not_allowed openai-compatible invalid_base_url")
configured = os.environ.get("CASAN_OPENAI_COMPATIBLE_ALLOWED_HOSTS", "")
allowed_hosts = {host.strip().lower() for host in configured.split(",") if host.strip()} \
or {"127.0.0.1", "localhost", "host.docker.internal"}
host = parsed.hostname.lower()
if host not in allowed_hosts:
fail(f"endpoint_not_allowed openai-compatible host={host}")
# HTTP is acceptable only for a loopback gateway or an explicitly named
# private-LAN IP. Public HTTP and arbitrary SSRF targets remain blocked.
if parsed.scheme == "http" and host not in {"127.0.0.1", "localhost", "host.docker.internal"}:
try:
is_private = ipaddress.ip_address(host).is_private
except ValueError:
is_private = False
if not is_private:
fail(f"endpoint_not_allowed openai-compatible insecure_host={host}")
base_path = parsed.path.rstrip("/")
if base_path not in {"", "/v1"}:
fail("endpoint_not_allowed openai-compatible base_path_must_be_v1")
return f"{parsed.scheme}://{parsed.netloc}{base_path}/chat/completions"
def call_openai_compatible(model_name, prompt, role):
"""Call an explicitly allowlisted OpenAI-compatible gateway (e.g. OmniRoute)."""
key = os.environ.get("CASAN_OPENAI_COMPATIBLE_API_KEY") or os.environ.get("OPENAI_API_KEY")
if not key:
fail("openai-compatible_backend_unavailable (API key unset)")
body = {
"model": model_name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0 if role in ("classify", "judge") else 0.2,
"max_tokens": 16 if role in ("classify", "judge") else 512,
}
data = json.dumps(body).encode()
req = urllib.request.Request(
openai_compatible_url(), data=data,
headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"},
)
t0 = time.time()
try:
with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
payload = json.loads(resp.read().decode())
except Exception as exc:
fail(f"backend_unreachable {type(exc).__name__}: {str(exc)[:120]}")
latency_ms = int((time.time() - t0) * 1000)
text, input_tokens, output_tokens = parse_openai_payload(payload)
return {
"text": text,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"latency_ms": latency_ms,
}
def parse_openai_payload(payload):
try:
text = (payload["choices"][0]["message"]["content"] or "").strip()
@@ -297,11 +376,18 @@ def main():
model_spec = args.model
if model_spec.startswith("ollama:"):
backend, model_name = "ollama", model_spec[len("ollama:"):]
elif model_spec.startswith(("anthropic:", "openai:")):
elif model_spec.startswith(("anthropic:", "openai:", "openai-compatible:")):
backend, model_name = model_spec.split(":", 1)
key = os.environ.get("ANTHROPIC_API_KEY" if backend == "anthropic" else "OPENAI_API_KEY", "")
if backend == "anthropic":
key = os.environ.get("ANTHROPIC_API_KEY", "")
elif backend == "openai-compatible":
key = os.environ.get("CASAN_OPENAI_COMPATIBLE_API_KEY") or os.environ.get("OPENAI_API_KEY", "")
else:
key = os.environ.get("OPENAI_API_KEY", "")
if not key:
# honest: cloud backend unavailable while key unset (do NOT fake)
if backend == "openai-compatible":
fail("openai-compatible_backend_unavailable (API key unset)")
# Preserve the existing public-cloud unavailable contract.
fail(f"cloud_backend_unavailable {backend} (API key unset)")
else:
fail(f"unknown_model_spec {model_spec}")
@@ -311,6 +397,8 @@ def main():
result = call_ollama(model_name, prompt, args.role)
elif backend == "openai":
result = call_openai(model_name, prompt, args.role)
elif backend == "openai-compatible":
result = call_openai_compatible(model_name, prompt, args.role)
else: # anthropic
result = call_anthropic(model_name, prompt, args.role)
@@ -343,6 +431,7 @@ def main():
"ollama": "ollama_local_real_tokens",
"openai": "openai_api_real_tokens",
"anthropic": "anthropic_api_real_tokens",
"openai-compatible": "openai_compatible_api_real_tokens",
}.get(backend, f"{backend}_real_tokens")
os.makedirs(os.path.dirname(PROVIDER_LOG), exist_ok=True)
usage = {