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

feat: harden control panel authentication

Browse Source
This commit is contained in:
thanhnv
2026-07-10 23:48:12 +09:00
parent 48b1186439
commit b56f7d357e
11 changed files with 571 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/casan-control-panel/scripts/provider-auth-bridge.py
+109 -45
View File
@@ -7,6 +7,7 @@ Only fixed commands are allowed and every request requires a generated bearer
token supplied to the backend container by local-full.sh.
"""
import argparse
import hashlib
import hmac
import json
import os
@@ -15,6 +16,8 @@ import subprocess
import tempfile
import threading
import time
from collections import deque
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Dict, Optional
@@ -35,7 +38,40 @@ PROVIDERS = {
}
RUNNING: Dict[str, subprocess.Popen] = {}
LOCK = threading.Lock()
MODEL_LOCKS = {provider: threading.Lock() for provider in PROVIDERS}
MODEL_GATES = {provider: threading.Semaphore(1) for provider in PROVIDERS}
MODEL_WINDOWS = {provider: deque() for provider in PROVIDERS}
RATE_LOCK = threading.Lock()
AUDIT_LOCK = threading.Lock()
def model_call_allowed(provider: str) -> bool:
timestamp = time.monotonic()
with RATE_LOCK:
window = MODEL_WINDOWS[provider]
while window and timestamp - window[0] > 600:
window.popleft()
if len(window) >= 10:
return False
window.append(timestamp)
return True
def audit_model_call(path: str, provider: str, status: str, prompt: str, latency_ms: int) -> None:
if not path:
return
record = {
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"provider": provider,
"status": status,
"prompt_hash": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
"prompt_characters": len(prompt),
"latency_ms": latency_ms,
}
os.makedirs(os.path.dirname(path), exist_ok=True)
with AUDIT_LOCK, open(path, "a", encoding="utf-8") as handle:
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
handle.flush()
os.chmod(path, 0o600)
def command_status(provider: str) -> dict:
@@ -112,56 +148,73 @@ def start_login(provider: str) -> dict:
return {"success": True, "reason": "browser_login_started", "provider": command_status(provider)}
def generate_with_account(provider: str, prompt: str) -> dict:
def run_account_command(provider: str, prompt: str, directory: str):
if provider == "codex":
output_path = os.path.join(directory, "last-message.txt")
command = [
"codex", "exec", "--ephemeral", "--ignore-user-config", "--ignore-rules",
"--skip-git-repo-check", "--sandbox", "read-only", "--color", "never",
"--cd", directory, "--output-last-message", output_path, "-",
]
result = subprocess.run(
command, input=prompt, capture_output=True, text=True, timeout=300,
env={**os.environ, "NO_COLOR": "1"},
)
text = ""
if result.returncode == 0 and os.path.isfile(output_path):
with open(output_path, encoding="utf-8") as handle:
text = handle.read().strip()
return result, text, {}, "codex-account-default"
command = [
"claude", "--print", "--output-format", "json", "--permission-mode", "plan",
"--tools", "", "--safe-mode", "--no-session-persistence",
]
result = subprocess.run(
command, input=prompt, capture_output=True, text=True, timeout=300, cwd=directory,
env={**os.environ, "NO_COLOR": "1"},
)
try:
payload = json.loads(result.stdout or "{}")
except ValueError:
payload = {}
text = str(payload.get("result") or "").strip()
raw_usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {}
usage = {
"input_tokens": int(raw_usage.get("input_tokens") or 0),
"output_tokens": int(raw_usage.get("output_tokens") or 0),
}
return result, text, usage, str(payload.get("model") or "claude-account-default")
def generate_with_account(provider: str, prompt: str, audit_path: str) -> dict:
started = time.monotonic()
status = command_status(provider)
if not status["available"] or not status["loggedIn"]:
audit_model_call(audit_path, provider, "provider_not_logged_in", prompt, 0)
return {"success": False, "reason": "provider_not_logged_in"}
if not prompt or len(prompt) > 24000:
audit_model_call(audit_path, provider, "prompt_length_invalid", prompt, 0)
return {"success": False, "reason": "prompt_length_invalid"}
started = time.monotonic()
with MODEL_LOCKS[provider], tempfile.TemporaryDirectory(prefix="casan-account-model-") as directory:
try:
if provider == "codex":
output_path = os.path.join(directory, "last-message.txt")
command = [
"codex", "exec", "--ephemeral", "--ignore-user-config", "--ignore-rules",
"--skip-git-repo-check", "--sandbox", "read-only", "--color", "never",
"--cd", directory, "--output-last-message", output_path, "-",
]
result = subprocess.run(
command, input=prompt, capture_output=True, text=True, timeout=300,
env={**os.environ, "NO_COLOR": "1"},
)
text = ""
if result.returncode == 0 and os.path.isfile(output_path):
with open(output_path, encoding="utf-8") as handle:
text = handle.read().strip()
usage = {}
model = "codex-account-default"
else:
command = [
"claude", "--print", "--output-format", "json", "--permission-mode", "plan",
"--tools", "", "--safe-mode", "--no-session-persistence",
]
result = subprocess.run(
command, input=prompt, capture_output=True, text=True, timeout=300, cwd=directory,
env={**os.environ, "NO_COLOR": "1"},
)
try:
payload = json.loads(result.stdout or "{}")
except ValueError:
payload = {}
text = str(payload.get("result") or "").strip()
raw_usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {}
usage = {
"input_tokens": int(raw_usage.get("input_tokens") or 0),
"output_tokens": int(raw_usage.get("output_tokens") or 0),
}
model = str(payload.get("model") or "claude-account-default")
except (OSError, subprocess.TimeoutExpired):
return {"success": False, "reason": "account_model_unreachable"}
if not model_call_allowed(provider):
audit_model_call(audit_path, provider, "rate_limited", prompt, 0)
return {"success": False, "reason": "account_model_rate_limited"}
gate = MODEL_GATES[provider]
if not gate.acquire(blocking=False):
audit_model_call(audit_path, provider, "busy", prompt, 0)
return {"success": False, "reason": "account_model_busy"}
try:
with tempfile.TemporaryDirectory(prefix="casan-account-model-") as directory:
result, text, usage, model = run_account_command(provider, prompt, directory)
except (OSError, subprocess.TimeoutExpired):
audit_model_call(audit_path, provider, "unreachable", prompt, int((time.monotonic() - started) * 1000))
return {"success": False, "reason": "account_model_unreachable"}
finally:
gate.release()
if result.returncode != 0 or not text:
audit_model_call(audit_path, provider, "failed", prompt, int((time.monotonic() - started) * 1000))
return {"success": False, "reason": "account_model_failed"}
audit_model_call(audit_path, provider, "success", prompt, int((time.monotonic() - started) * 1000))
return {
"success": True,
"provider": provider,
@@ -185,15 +238,23 @@ class BridgeHandler(BaseHTTPRequestHandler):
self.send_header("Content-Length", str(len(encoded)))
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("X-Frame-Options", "DENY")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
self.end_headers()
self.wfile.write(encoded)
def authorized(self) -> bool:
host = self.headers.get("Host", "").split(":", 1)[0].lower()
if host not in {"127.0.0.1", "localhost", "host.docker.internal"}:
return False
expected = self.server.bridge_token # type: ignore[attr-defined]
supplied = self.headers.get("X-CASAN-Bridge-Token", "")
return bool(expected) and hmac.compare_digest(expected, supplied)
def read_json(self) -> dict:
if self.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
return {}
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
@@ -229,7 +290,7 @@ class BridgeHandler(BaseHTTPRequestHandler):
return
if len(parts) == 4 and parts[:2] == ["v1", "models"] and parts[3] == "generate" and parts[2] in PROVIDERS:
body = self.read_json()
result = generate_with_account(parts[2], str(body.get("prompt") or ""))
result = generate_with_account(parts[2], str(body.get("prompt") or ""), self.server.audit_log) # type: ignore[attr-defined]
self.send_json(200 if result["success"] else 503, result)
return
self.send_json(404, {"success": False, "reason": "not_found"})
@@ -251,9 +312,12 @@ def main() -> int:
parser.add_argument("--bind", default="127.0.0.1")
parser.add_argument("--port", type=int, default=20130)
parser.add_argument("--token-file", required=True)
parser.add_argument("--audit-log", default="")
args = parser.parse_args()
server = ThreadingHTTPServer((args.bind, args.port), BridgeHandler)
server.bridge_token = read_token(args.token_file) # type: ignore[attr-defined]
server.audit_log = os.path.abspath(args.audit_log) if args.audit_log else "" # type: ignore[attr-defined]
server.daemon_threads = True
try:
server.serve_forever()
except KeyboardInterrupt: