Claude
73a066631b
- Changing a custom endpoint's address or protocol without re-entering the key clears the saved key: the id is in every shared workflow, so it must not be enough to point a saved key at another server. The store file is created 0600 from the start. - CLI model names must match a strict pattern: on Windows an npm-installed CLI is a .cmd run through cmd.exe, where a crafted model name from a shared workflow could inject commands. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MmQWqMnuUXU4XwF6xyaDyM
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""Runtime-configured ("Custom Endpoint - WARNING") endpoints.
|
|
|
|
The address and key typed into the node's custom-endpoint panel are stored
|
|
here, on this machine, in the git-ignored `nodes/llm/CustomEndpoints.local.json`.
|
|
The workflow only holds a random id pointing at the entry, so the address and
|
|
key never end up in a saved workflow or an image's metadata.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
import threading
|
|
|
|
from .llm_endpoints import CUSTOM_ENDPOINT_NAME, LLM_DIR, Endpoint, normalize_url
|
|
|
|
CUSTOM_PROVIDERS = ("openai", "anthropic", "ollama")
|
|
STORE_FILE = os.path.join(LLM_DIR, "CustomEndpoints.local.json")
|
|
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def _load():
|
|
try:
|
|
with open(STORE_FILE, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, dict) else {}
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def _save(data):
|
|
tmp = STORE_FILE + ".tmp"
|
|
# Created private from the start (no window where others can read it).
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
os.replace(tmp, STORE_FILE)
|
|
|
|
|
|
def _valid_id(custom_id):
|
|
return isinstance(custom_id, str) and 16 <= len(custom_id) <= 64 and custom_id.replace("-", "").replace("_", "").isalnum()
|
|
|
|
|
|
def save_custom(custom_id, provider, base_url, api_key):
|
|
"""Create or update an entry. api_key None keeps the stored key; "" clears it.
|
|
|
|
Returns the entry's id.
|
|
"""
|
|
if provider not in CUSTOM_PROVIDERS:
|
|
raise ValueError(f"protocol must be one of {', '.join(CUSTOM_PROVIDERS)}")
|
|
base_url = normalize_url(base_url)
|
|
if not base_url.startswith(("http://", "https://")):
|
|
raise ValueError("the address must start with http:// or https://")
|
|
with _lock:
|
|
data = _load()
|
|
if not _valid_id(custom_id) or custom_id not in data:
|
|
custom_id = secrets.token_urlsafe(18)
|
|
data[custom_id] = {}
|
|
entry = data[custom_id]
|
|
if api_key is None and (entry.get("base_url") != base_url or entry.get("provider") != provider):
|
|
# The id travels with every workflow and image; without this,
|
|
# anyone who has it could point a saved key at their own server.
|
|
entry.pop("api_key", None)
|
|
entry["provider"] = provider
|
|
entry["base_url"] = base_url
|
|
if api_key is not None:
|
|
entry["api_key"] = api_key.strip()
|
|
_save(data)
|
|
return custom_id
|
|
|
|
|
|
def get_custom(custom_id):
|
|
if not _valid_id(custom_id):
|
|
return None
|
|
with _lock:
|
|
return _load().get(custom_id)
|
|
|
|
|
|
def custom_endpoint(custom_id):
|
|
"""An Endpoint for a stored entry, or None if the id is unknown here."""
|
|
entry = get_custom(custom_id)
|
|
if not entry:
|
|
return None
|
|
return Endpoint(
|
|
name=CUSTOM_ENDPOINT_NAME,
|
|
provider=entry.get("provider", "openai"),
|
|
base_url=entry.get("base_url", ""),
|
|
api_key=entry.get("api_key", ""),
|
|
api_key_optional=True,
|
|
# Treated like an address from .env: never shown back in full.
|
|
private_address=True,
|
|
description="Runtime-configured endpoint.",
|
|
)
|