# scikitplot/_externals/_sphinx_ext/_sphinx_ai_assistant/_hf_spaces_proxy/_utils/_shared_logic.py # # flake8: noqa: D213 # # Authors: The scikit-plots developers # SPDX-License-Identifier: BSD-3-Clause # _shared_logic.py v7.0.0 # # Single source of truth for shared constants, pure helper functions, and # type aliases used by the deployed proxy (_hf_spaces_proxy/app.py) and the # local development proxy (dev_proxy.py). # # Import discipline # ----------------- # Only the Python standard library is imported here. httpx, fastapi, and # torch are NOT imported so this module can be sourced by stdlib-only tools # (dev_proxy) and tested in isolation without any network or GPU environment. # # Routing paths (v6.0.0) # ---------------------- # Three ordered routing paths — each with its own configurable read timeout: # # Path 1 — BACKEND_URL set (explicit override) # Forward to BACKEND_URL. Only BACKEND_AUTH_TOKEN may be attached by callers. # Read timeout: proxy_timeout kwarg (env: PROXY_TIMEOUT, default 600 s). # # Path 2 — Model namespace in HF_SPACES_MODEL_NAMESPACES # Model owner (e.g. "scikit-plots") matches a custom namespace. # Forward to HF_SPACES_MODEL_URL (the ai-model HF Space, CPU inference). # These models have no HF Inference Provider → direct HF API returns 404/503. # Read timeout: path2_read_timeout kwarg (env: PATH2_TIMEOUT, default 600 s). # CPU inference on a 7B model takes 4-5 minutes; 600 s gives safe headroom. # # Path 3 — Standard HF Inference API (default) # Model has a registered HF Inference Provider (openai/*, Qwen/*, etc.). # Forward to HF_BASE/{model}/v1/chat/completions with HF_TOKEN. # Read timeout: path3_read_timeout kwarg (env: PATH3_TIMEOUT, default 120 s). # HF Serverless API (GPU-backed) normally responds within 30-90 s. # # Breaking changes v4.0.0 → v5.0.0 # ---------------------------------- # + DEFAULT_PROXY_TIMEOUT raised from 120 s to 600 s. # Root cause: 120 s was shorter than the 4-5 min CPU inference on the # ai-model HF Space, causing every request to return a network error. # + DEFAULT_PATH2_READ_TIMEOUT added (600 s) — ai-model space per-path timeout. # + DEFAULT_PATH3_READ_TIMEOUT added (120 s) — HF API per-path timeout. # + _resolve_upstream_url now accepts path2_read_timeout, path3_read_timeout, # and proxy_timeout keyword-only parameters. # + _resolve_upstream_url return type changed from tuple[str, dict] to # tuple[str, dict, float] — the third element is the per-path read timeout. # Callers must unpack all three values. # + load_proxy_env extended with path2_read_timeout and path3_read_timeout. # # Breaking changes v5.0.0 → v6.0.0 # ---------------------------------- # + DEFAULT_HF_BASE changed from ``https://api-inference.huggingface.co/models`` # to ``https://router.huggingface.co``. # Root cause: api-inference.huggingface.co was DNS-unresolvable ([Errno -5] # EAI_NODATA / EAI_NONAME) from within HF Docker Spaces. # router.huggingface.co is the current HF Inference Providers endpoint and # resolves correctly in all deployment environments. # Callers who hard-code ``HF_BASE`` to the old hostname must migrate to # the new router URL. # # New in v6.1.0 — Three-type HF token system # ------------------------------------------- # + ``HFTokenType`` literal type alias added: ``"fine-grained" | "read" | # ``"write" | "unknown"``. Maps directly to the three token types exposed # in HF Settings → Tokens. # + ``HF_TOKEN_TYPE_*`` string constants and ``HF_INFERENCE_TOKEN_TYPES`` / # ``HF_WRITE_TOKEN_TYPES`` frozensets added for type-safe comparisons. # + ``_classify_token_type()`` — classify a token by explicit env-var # declaration (``HF_TOKEN_TYPE``, ``HF_WRITE_TOKEN_TYPE``) with a length- # based heuristic fallback. # + ``_token_suitable_for_inference()`` / ``_token_suitable_for_writes()`` # predicates for principle-of-least-privilege validation. # + ``_validate_token_config()`` — returns actionable WARNING / ERROR strings # for token-type mismatches detected at startup. # + ``_token_log_fragment()`` gains an optional ``token_type`` parameter so # log lines include the token type (e.g. ``hf_abcde...1234 (read)``). # + ``load_proxy_env()`` extended with ``hf_token_type`` and # ``hf_write_token_type`` keys read from the matching env vars. # + ``_safe_float`` added to ``__all__`` (was importable but unadvertised). """ Shared utilities for the sphinx-ai-assistant proxy solutions. This module provides pure, stateless helper functions and typed constants that are common to all server-side proxy implementations. It has **no** runtime dependencies beyond the Python standard library. Public API: PROXY_VERSION : str Proxy release version string. DEFAULT_HF_BASE : str HuggingFace Serverless Inference API base URL. DEFAULT_MODEL : str Fallback model ID when the request body omits ``model``. DEFAULT_PROXY_TIMEOUT : int Global upstream read timeout in seconds (Path 1 / backward-compat). DEFAULT_PATH2_READ_TIMEOUT : float Per-path read timeout for Path 2 (ai-model space, CPU inference). DEFAULT_PATH3_READ_TIMEOUT : float Per-path read timeout for Path 3 (HF Serverless Inference API). DEFAULT_MAX_BODY_BYTES : int Maximum accepted request body size. DEFAULT_HF_SPACES_MODEL_URL : str Default URL for the custom ai-model HF Space (Path 2). DEFAULT_HF_SPACES_MODEL_NAMESPACES : tuple[str, ...] Default model owner namespaces routed to the model Space (Path 2). _safe_int : callable Parse an integer environment variable with a safe fallback. _parse_model : callable Extract the ``model`` field from a raw JSON request body. _is_custom_model_namespace : callable Return True when a model's owner namespace is in the custom list. _build_cors_headers : callable Return the CORS response-header mapping. _token_log_fragment : callable Produce a safely-truncated token string for log output. _resolve_upstream_url : callable Centralised three-path routing: choose upstream URL, auth headers, and per-path read timeout. _validate_env : callable Fail-fast startup check with actionable error messages. load_proxy_env : callable Read all proxy-relevant environment variables into a typed dict. Notes ----- **Developer note** — All functions are pure (no side effects, no I/O). Tests can import this module without a running event loop or any network. The proxy (FastAPI / asyncio) and dev_proxy (stdlib HTTPServer) both import from here so that routing and CORS logic are *never* duplicated. **Breaking change v5.0.0** — ``_resolve_upstream_url`` now returns a 3-tuple ``(url, headers, read_timeout_s: float)`` instead of the previous 2-tuple ``(url, headers)``. All callers must unpack the third element or the per-path timeout falls through to the old flat-timeout behaviour. **Breaking change v6.0.0** — :data:`DEFAULT_HF_BASE` migrated from ``https://api-inference.huggingface.co/models`` to ``https://router.huggingface.co``. The old hostname was DNS-unresolvable ([Errno -5] EAI_NONAME) from within HF Docker Spaces. Deployments that override ``HF_BASE`` to the legacy hostname must update their configuration. **Security note** — :func:`_token_log_fragment` ensures the full API token never appears in log output. Never widen the exposed fragment beyond the current 8+4 character window without reviewing log-aggregation policy first. **Versioning note** — Bump :data:`PROXY_VERSION` on every breaking change so deployed Spaces and log aggregators can correlate errors to a specific release. """ from __future__ import annotations import ipaddress import json import logging import os import re from typing import Any, Literal from urllib.parse import urlsplit try: from ._telemetry import sanitize_log_text except ImportError: # standalone HF Space deployment from _utils._telemetry import sanitize_log_text logger = logging.getLogger(__name__) __all__ = [ # noqa: RUF022 # Version "PROXY_VERSION", # Constants — routing / timeout "DEFAULT_HF_BASE", "DEFAULT_HF_PROVIDER_MODELS", "DEFAULT_HF_SPACES_MODEL_NAMESPACES", "DEFAULT_HF_SPACES_MODEL_URL", "DEFAULT_MAX_BODY_BYTES", "DEFAULT_MODEL", "DEFAULT_PATH2_READ_TIMEOUT", "DEFAULT_PATH3_READ_TIMEOUT", "DEFAULT_PROXY_TIMEOUT", # Constants — token type system (v6.1.0) "HFTokenType", "HF_TOKEN_TYPE_FINE_GRAINED", "HF_TOKEN_TYPE_READ", "HF_TOKEN_TYPE_WRITE", "HF_TOKEN_TYPE_UNKNOWN", "HF_INFERENCE_TOKEN_TYPES", "HF_WRITE_TOKEN_TYPES", # Helpers — general "_build_cors_headers", "_is_custom_model_namespace", "_parse_model", "_safe_float", "_safe_int", "_token_log_fragment", # Privacy / log-redaction (v6.2.0) "_REDACT_PATTERNS", "_RedactingFilter", "_mask_ip", # Helpers — token type system (v6.1.0) "_classify_token_type", "_token_suitable_for_inference", "_token_suitable_for_writes", "_validate_token_config", # Helpers — routing / env "_resolve_upstream_url", "_validate_credential_destination", "_validate_env", "load_proxy_env", ] # ───────────────────────────────────────────────────────────────────────────── # Module-level constants # ───────────────────────────────────────────────────────────────────────────── #: Proxy release version — bump on every breaking change. PROXY_VERSION: str = "7.4.0" #: HuggingFace Inference Providers router base URL (no trailing slash). #: Only used for Path 3 (standard provider models) when ``BACKEND_URL`` is #: empty and the model namespace is not in ``HF_SPACES_MODEL_NAMESPACES``. #: #: Migrated from ``https://api-inference.huggingface.co/models`` (v5.0.0) to #: ``https://router.huggingface.co`` (v6.0.0). #: Root cause: api-inference.huggingface.co was DNS-unresolvable ([Errno -5] #: EAI_NODATA / EAI_NONAME) from within HF Docker Spaces; the router hostname #: resolves correctly and is the current HF Inference Providers endpoint. DEFAULT_HF_BASE: str = "https://router.huggingface.co" #: Public Hugging Face Inference Provider models advertised by the bundled #: example configuration. Keep this default synchronized with the Cloudflare #: Worker so both bundled proxies accept the same public model choices. #: Operators can replace the exact set with ``ALLOWED_MODELS``. DEFAULT_HF_PROVIDER_MODELS: tuple[str, ...] = ( "Qwen/Qwen2.5-Coder-7B-Instruct", "Qwen/Qwen2.5-Coder-32B-Instruct", "openai/gpt-oss-20b", ) #: Fallback model ID when the request body omits the ``model`` field. #: Must have a registered HF Inference Provider for Path 3. DEFAULT_MODEL: str = "scikit-plots/Qwen2.5-Coder-7B-Instruct" #: Global upstream read timeout in seconds (used for Path 1 / backward compat). #: #: Raised from 120 s (v4.0.0) to 600 s (v5.0.0). #: #: Root cause of the increase: the ai-model HF Space runs a 7B model on CPU #: basic hardware. Cold-start inference (model loading + generation) takes #: 4-5 minutes. The 120 s ceiling caused every request to the ai-model Space #: to return ``httpx.ReadTimeout``, which the browser reported as #: "Sorry, something went wrong: network error". DEFAULT_PROXY_TIMEOUT: int = 600 #: Per-path read timeout for Path 2 (ai-model HF Space, CPU inference). #: #: CPU inference on a 7B model takes 4-5 minutes. 600 s gives 1 minute of #: additional headroom for cold-start model loading (~50 s tokenizer + #: ~50 s model load + ~4.5 min generation on the first request). DEFAULT_PATH2_READ_TIMEOUT: float = 600.0 #: Per-path read timeout for Path 3 (HF Serverless Inference API). #: #: The HF Serverless API runs inference on GPU hardware. Most responses #: arrive within 30-90 s. 120 s gives a comfortable margin. DEFAULT_PATH3_READ_TIMEOUT: float = 120.0 #: Maximum accepted request body size in bytes (10 MiB). #: Prevents memory exhaustion from maliciously oversized POST bodies. DEFAULT_MAX_BODY_BYTES: int = 10 * 1024 * 1024 # 10 MiB #: Default URL for the custom ai-model HF Space (Path 2). #: Requests for models whose namespace is in ``DEFAULT_HF_SPACES_MODEL_NAMESPACES`` #: are forwarded here instead of the HF Serverless Inference API. #: Overridable via the ``HF_SPACES_MODEL_URL`` environment variable. DEFAULT_HF_SPACES_MODEL_URL: str = ( "https://scikit-plots-ai-model.hf.space/v1/chat/completions" ) #: Default model owner namespaces routed to :data:`DEFAULT_HF_SPACES_MODEL_URL`. #: Models whose owner (the part before ``/``) matches any entry in this tuple #: are routed to the ai-model Space (Path 2) rather than the HF API (Path 3). #: Overridable via the ``HF_SPACES_MODEL_NAMESPACES`` environment variable. DEFAULT_HF_SPACES_MODEL_NAMESPACES: tuple[str, ...] = ("scikit-plots",) # ───────────────────────────────────────────────────────────────────────────── # HuggingFace token type system (v6.1.0) # ───────────────────────────────────────────────────────────────────────────── # # HuggingFace exposes exactly three token types in # https://huggingface.co/settings/tokens: # # ① Fine-grained — New-style token. Permissions set at creation time: # choose any combination of per-repo access levels and # API capabilities. Recommended for production because # each token carries only the minimum required scope. # # ② Read (classic) — Legacy read-only token. Grants read access to all # public repos and any private repos you can access. # Always includes the Serverless Inference API capability. # Cannot push commits or create repos. # # ③ Write (classic)— Legacy read+write token. All read permissions plus # the ability to push commits, create repos, manage # members, etc. Over-privileged for inference-only use. # # Mapping to proxy env vars # ───────────────────────── # HF_TOKEN — inference token (Path 2 private Space + Path 3 HF API). # Best practice: fine-grained with inference-api scope only, # OR classic read. Never use a write token here. # # HF_DATASET_TOKEN — preferred dataset-persistence token. Best practice: # fine-grained scoped to ONE dataset repo. Classic Write # also works; classic Read never does. # HF_WRITE_TOKEN — historical alias for HF_DATASET_TOKEN. # # Optional type-declaration env vars (Space → Settings → Repository secrets): # HF_TOKEN_TYPE = fine-grained | read | write (default: auto-detect) # HF_DATASET_TOKEN_TYPE = fine-grained | read | write (preferred) # HF_WRITE_TOKEN_TYPE = fine-grained | read | write (legacy alias) # # ───────────────────────────────────────────────────────────────────────────── #: Literal type for HuggingFace token type labels. #: Use as type annotation and for exhaustive ``isinstance``-free comparisons. HFTokenType = Literal["fine-grained", "read", "write", "unknown"] #: New-style fine-grained HF token. Permissions defined at creation time. #: Declare via env var: ``HF_TOKEN_TYPE=fine-grained``. HF_TOKEN_TYPE_FINE_GRAINED: str = "fine-grained" # noqa: S105 #: Classic HF read token. Read + Inference API; no write capability. #: Declare via env var: ``HF_TOKEN_TYPE=read``. HF_TOKEN_TYPE_READ: str = "read" # noqa: S105 #: Classic HF write token. All read permissions + repo push capability. #: Declare via env var: ``HF_TOKEN_TYPE=write`` or ``HF_WRITE_TOKEN_TYPE=write``. HF_TOKEN_TYPE_WRITE: str = "write" # noqa: S105 #: Sentinel: token type not declared and could not be inferred. #: Runtime operations are not blocked, but :func:`_validate_token_config` omits #: least-privilege warnings because the type is unknown. HF_TOKEN_TYPE_UNKNOWN: str = "unknown" # noqa: S105 #: Token types that are appropriate for HF Serverless Inference API calls #: (Path 3) and private HF Space access (Path 2). #: #: Classic write tokens ARE technically capable of inference (write ⊇ read), #: but are excluded from this set so :func:`_validate_token_config` can emit #: a startup warning when a write token is used where a read / fine-grained #: token is the correct choice. The ``"unknown"`` sentinel is included so #: that un-declared tokens do not trigger false-positive warnings. HF_INFERENCE_TOKEN_TYPES: frozenset[str] = frozenset( { HF_TOKEN_TYPE_FINE_GRAINED, HF_TOKEN_TYPE_READ, HF_TOKEN_TYPE_UNKNOWN, } ) #: Token types that can push commits to HuggingFace repos and datasets. #: #: Classic read tokens **cannot** write — any ``HfApi.create_commit`` call #: returns HTTP 403 / 401. ``"unknown"`` is excluded so that #: :func:`_validate_token_config` can flag a read token configured as the write #: token as a hard error rather than silently failing at request time. HF_WRITE_TOKEN_TYPES: frozenset[str] = frozenset( { HF_TOKEN_TYPE_FINE_GRAINED, HF_TOKEN_TYPE_WRITE, } ) # ───────────────────────────────────────────────────────────────────────────── # Pure helper functions # ───────────────────────────────────────────────────────────────────────────── def _safe_int(value: str | None, default: int) -> int: """ Parse *value* as an integer, returning *default* on any failure. Parameters ---------- value : str or None String to parse. Typically the raw value of an environment variable (may be ``None`` when the variable is absent). default : int Returned when *value* is ``None``, empty, or cannot be converted. Returns ------- int Parsed integer, or *default* on any ``ValueError`` / ``TypeError``. Notes ----- **Developer note** — This function is intentionally never-raise. A misconfigured ``PROXY_TIMEOUT`` or ``MAX_BODY_BYTES`` must not prevent the proxy from starting — the safe default is better than a crash. Examples -------- >>> _safe_int("120", 60) 120 >>> _safe_int("not-a-number", 60) 60 >>> _safe_int(None, 60) 60 >>> _safe_int("", 60) 60 """ if value is None: return default try: return int(value) except (ValueError, TypeError): return default def _safe_float(value: str | None, default: float) -> float: """ Parse *value* as a float, returning *default* on any failure. Parameters ---------- value : str or None String to parse. Typically the raw value of an environment variable. default : float Returned when *value* is ``None``, empty, or cannot be converted. Returns ------- float Parsed float, or *default* on any ``ValueError`` / ``TypeError``. Notes ----- **Developer note** — Like :func:`_safe_int`, this is intentionally never-raise. A misconfigured ``PATH2_TIMEOUT`` or ``PATH3_TIMEOUT`` must not crash the proxy at startup. Examples -------- >>> _safe_float("600.0", 120.0) 600.0 >>> _safe_float("bad", 120.0) 120.0 >>> _safe_float(None, 120.0) 120.0 """ if value is None: return default try: return float(value) except (ValueError, TypeError): return default def _parse_model(body: bytes, default: str = DEFAULT_MODEL) -> str: """ Extract the ``model`` field from a raw JSON request body. Parameters ---------- body : bytes Raw HTTP request body forwarded from the browser. Expected to be valid JSON but the function never raises on malformed input. default : str, optional Fallback model ID when the field is absent or the body cannot be decoded. Defaults to :data:`DEFAULT_MODEL`. Returns ------- str The ``model`` value from the body, or *default* if the field is absent, empty, or the body is not valid JSON. Notes ----- **Developer note** — This function is intentionally never-raise. A malformed body must not crash the proxy; the upstream model backend will return a meaningful error that the browser can display. Examples -------- >>> _parse_model(b'{"model": "Qwen/Qwen2.5-Coder-7B-Instruct"}') 'Qwen/Qwen2.5-Coder-7B-Instruct' >>> _parse_model(b"{}") 'scikit-plots/Qwen2.5-Coder-7B-Instruct' >>> _parse_model(b"not-json") 'scikit-plots/Qwen2.5-Coder-7B-Instruct' >>> _parse_model(b'{"model": " "}') 'scikit-plots/Qwen2.5-Coder-7B-Instruct' """ try: data: Any = json.loads(body) candidate = str(data.get("model", "")).strip() return candidate or default except (json.JSONDecodeError, ValueError, AttributeError, TypeError): return default def _is_custom_model_namespace( model: str, namespaces: tuple[str, ...] | list[str], ) -> bool: """ Return ``True`` when the model owner namespace is in *namespaces*. The owner is the portion of the model ID before the first ``/``. An optional HF Router variant suffix (e.g. ``:fastest``) is stripped before comparison so ``"scikit-plots/Qwen2.5-Coder-7B-Instruct:fastest"`` is correctly identified as belonging to the ``"scikit-plots"`` namespace. Parameters ---------- model : str Model ID string, e.g. ``"scikit-plots/Qwen2.5-Coder-7B-Instruct"`` or ``"openai/gpt-oss-20b:fastest"``. namespaces : tuple[str, ...] or list[str] Iterable of owner namespace strings to match against (case-insensitive). Typically :data:`DEFAULT_HF_SPACES_MODEL_NAMESPACES` or parsed from the ``HF_SPACES_MODEL_NAMESPACES`` environment variable. Returns ------- bool ``True`` when the model owner is in *namespaces*, ``False`` otherwise. Notes ----- **Developer note** — Comparison is case-insensitive and strips leading / trailing whitespace from both the model owner and each namespace entry. A model string without a ``/`` separator (i.e. no namespace component) always returns ``False``; such IDs are routed to Path 3 (HF Inference API). Examples -------- >>> _is_custom_model_namespace( ... "scikit-plots/Qwen2.5-Coder-7B-Instruct", ... ("scikit-plots",), ... ) True >>> _is_custom_model_namespace( ... "scikit-plots/Qwen2.5-Coder-7B-Instruct:fastest", ... ("scikit-plots",), ... ) True >>> _is_custom_model_namespace("openai/gpt-oss-20b", ("scikit-plots",)) False >>> _is_custom_model_namespace("no-slash-model", ("scikit-plots",)) False """ base = model.split(":", maxsplit=1)[0].strip() if not base or "/" not in base: return False owner = base.split("/", 1)[0].lower().strip() normalised = {ns.lower().strip() for ns in namespaces if ns.strip()} return owner in normalised def _build_cors_headers(allowed_origin: str = "*") -> dict[str, str]: """ Return the standard CORS response-header mapping. Parameters ---------- allowed_origin : str, optional Value for the ``Access-Control-Allow-Origin`` header. Defaults to ``"*"`` (allow all origins). Returns ------- dict[str, str] CORS response headers. Examples -------- >>> headers = _build_cors_headers() >>> headers["Access-Control-Allow-Origin"] '*' """ return { "Access-Control-Allow-Origin": allowed_origin, "Access-Control-Allow-Methods": "POST, OPTIONS", "Access-Control-Allow-Headers": "Content-Type", } def _token_log_fragment(token: str, token_type: str = "") -> str: """Return non-secret token configuration state for legacy log call sites. The historical implementation exposed an 8+4 character credential fragment. Run 5 deliberately removes that behavior: partial credentials are still credentials and may become identifying/correlatable in retained logs. Keep the helper name for source compatibility, but return only presence and optional type metadata. """ if not token: return "" label = str(token_type or "").strip().lower() return f" ({label})" if label and label != "unknown" else "" # ───────────────────────────────────────────────────────────────────────────── # Privacy / log-redaction helpers (v6.2.0) # ───────────────────────────────────────────────────────────────────────────── # # Design rationale # ---------------- # Two complementary layers protect PII in log output: # # Layer 1 — call-site masking via :func:`_mask_ip` # Every ``json.dumps({..., "ip": ...})`` call in ``app.py`` passes # ``client_ip`` through :func:`_mask_ip` before it is serialised. # This is the PRIMARY control: the raw IP never enters the log string. # # Layer 2 — defence-in-depth via :class:`_RedactingFilter` # Attached to the root logging handler. Applies :data:`_REDACT_PATTERNS` # to the fully formatted message BEFORE it is emitted. Catches: # • HF token strings leaked via exception messages from # ``huggingface_hub`` (e.g. ``snapshot_download`` auth failures). # • IPv4 addresses emitted by third-party library loggers (httpx, # uvicorn) that bypass the call-site masking. # • Any future code that forgets to call :func:`_mask_ip` first. # # IPv6 is handled exclusively at Layer 1 (:func:`_mask_ip`). A generic # IPv6 regex in Layer 2 has unacceptable false-positive rates (e.g. it # would match ``12:34:56:78`` in log timestamps or MAC addresses). # ───────────────────────────────────────────────────────────────────────────── def _mask_ip(ip: str) -> str: """Mask a client IP address for privacy-safe log output. Preserves enough network context for rate-limit and abuse analysis while zeroing the host portion that identifies the individual user. * **IPv4** — zero the last octet, retaining the /24 subnet. ``"192.168.1.100"`` → ``"192.168.1.0"`` * **IPv6** — zero the interface identifier (last 64 bits), retaining the /64 prefix. ``"2001:db8:85a3::8a2e:370:7334"`` → ``"2001:db8:85a3::"`` * **IPv6 scope suffix** (e.g. ``"fe80::1%eth0"``) — stripped before parsing (Python's :mod:`ipaddress` does not accept scope identifiers). * **Non-IP strings** — returned as ``""``. * **Sentinel** ``"unknown"`` — returned unchanged (already non-identifying). Parameters ---------- ip : str Client IP string extracted from the HTTP request headers. May be ``"unknown"`` when the proxy header is absent. Returns ------- str Masked IP suitable for structured log output. This function is intentionally never-raise — any :exc:`ValueError` from :mod:`ipaddress` is caught and replaced by the safe fallback. Notes ----- **Security note** — This is the canonical privacy gate for all IP values written to log records in ``app.py``. Every ``json.dumps({..., "ip": …})`` call must pass ``client_ip`` through :func:`_mask_ip` before serialising. Callers must **not** write raw ``client_ip`` values to any log record. **Developer note** — Uses :mod:`ipaddress` from the Python standard library; no third-party dependencies are introduced. Examples -------- >>> _mask_ip("192.168.1.100") '192.168.1.0' >>> _mask_ip("10.0.0.255") '10.0.0.0' >>> _mask_ip("2001:db8:85a3::8a2e:370:7334") '2001:db8:85a3::' >>> _mask_ip("fe80::1%eth0") 'fe80::' >>> _mask_ip("unknown") 'unknown' >>> _mask_ip("not-an-ip") '' """ if ip in ("unknown", ""): return ip try: # Strip IPv6 zone/scope identifier (e.g. "%eth0") — ipaddress rejects it. clean: str = ip.split("%", 1)[0].strip() addr = ipaddress.ip_address(clean) if isinstance(addr, ipaddress.IPv4Address): # Retain /24 (first three octets); zero the host octet. return str(ipaddress.ip_network(f"{addr}/24", strict=False).network_address) # IPv6: retain /64 prefix; zero the 64-bit interface identifier. return str(ipaddress.ip_network(f"{addr}/64", strict=False).network_address) except ValueError: return "" #: Ordered list of ``(compiled_pattern, replacement)`` tuples applied by #: :class:`_RedactingFilter` to every log record before emission. #: #: **Pattern order matters** — patterns are applied left-to-right; more #: specific patterns must precede catch-all patterns. There is no overlap #: between the current patterns, but this convention must be maintained when #: extending this list. #: #: IPv6 addresses are intentionally **absent** — they are handled at the #: call-site by :func:`_mask_ip` (Layer 1). A generic IPv6 regex in a #: global filter produces too many false positives (hex timestamps, MAC #: addresses, Docker overlay IDs) to be safe in a production log stream. _REDACT_PATTERNS: list[tuple[re.Pattern[str], str]] = [ # HuggingFace API tokens — ``hf_`` prefix followed by ≥ 4 alphanumeric # characters. Classic tokens are ~34 chars; fine-grained tokens are ≥ 52. # The {4,} lower bound avoids matching ``hf_`` in legitimate identifiers # (e.g. Python identifiers that start with ``hf_``) while still catching # any partial token fragment that huggingface_hub may embed in an error # message. (re.compile(r"\bhf_[a-zA-Z0-9]{4,}\b"), ""), # IPv4 addresses — strict dotted-decimal notation with per-octet range # validation (0-255). Word boundaries prevent partial matches inside # longer numeric strings. This pattern catches IPv4 strings emitted by # third-party loggers (httpx, uvicorn) that bypass :func:`_mask_ip`. ( re.compile( r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}" r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b" ), "", ), ] class _RedactingFilter(logging.Filter): """Scrub sensitive values from log records before emission. Applies the regex patterns in :data:`_REDACT_PATTERNS` to the fully formatted log message, replacing HuggingFace API tokens and raw IPv4 addresses with opaque placeholders. This class is the **defence-in-depth layer** (Layer 2). The primary control is :func:`_mask_ip` at each call site (Layer 1). The filter catches values that slip through Layer 1 — most importantly, HF token strings embedded in exception messages from ``huggingface_hub``. Parameters ---------- name : str, optional Filter name forwarded to :class:`logging.Filter`. Default ``""``. Notes ----- **Security note** — This filter materialises the fully formatted message via :meth:`logging.LogRecord.getMessage`, applies every pattern in :data:`_REDACT_PATTERNS`, then replaces :attr:`~logging.LogRecord.msg` with the scrubbed result and clears :attr:`~logging.LogRecord.args`. Clearing ``args`` prevents downstream handlers from re-applying ``%`` formatting to a string that no longer contains positional placeholders. **Developer note** — Attach to the root handler immediately after construction so **every** handler in the process benefits:: handler = logging.StreamHandler() handler.addFilter(_RedactingFilter()) logging.root.handlers = [handler] To extend the redaction vocabulary, append a ``(pattern, replacement)`` tuple to :data:`_REDACT_PATTERNS`. Examples -------- >>> import logging >>> f = _RedactingFilter() >>> rec = logging.makeLogRecord( ... {"msg": "token=hf_abc1234defg5678 ip=10.0.1.99", "args": ()} ... ) >>> f.filter(rec) True >>> rec.msg 'token= ip=' """ def filter(self, record: logging.LogRecord) -> bool: # noqa: A003 """Redact sensitive patterns from *record*'s formatted message. Parameters ---------- record : logging.LogRecord Log record to inspect and mutate in-place. Returns ------- bool Always ``True`` — this filter never suppresses records, only scrubs their message content. """ # Materialise the full %-formatted string first, then scrub it. msg: str = sanitize_log_text(record.getMessage()) # Write the scrubbed text back and clear args so that any subsequent # call to getMessage() returns the already-scrubbed string without # attempting to re-apply % formatting. record.msg = msg record.args = () return True def _classify_token_type( token: str, declared_type: str | None = None, ) -> HFTokenType: """ Classify a HuggingFace token by its declared type or format heuristics. Token type classification is used at startup by :func:`_validate_token_config` to enforce the principle of least privilege before any requests arrive. Parameters ---------- token : str The HuggingFace API token string. declared_type : str or None, optional Explicitly declared type from an environment variable (``HF_TOKEN_TYPE`` or ``HF_WRITE_TOKEN_TYPE``). Accepted values: ``"fine-grained"``, ``"read"``, ``"write"`` (and minor formatting variants: ``"finegrained"``, ``"fine_grained"``). When provided and recognized, it takes precedence over all heuristics. Returns ------- HFTokenType One of ``"fine-grained"``, ``"read"``, ``"write"``, or ``"unknown"``. Notes ----- **Security note** — Token type cannot be verified without an authenticated call to the HF API (``GET https://huggingface.co/api/whoami-v2``). This function applies lightweight format heuristics only. For production deployments, always declare the type explicitly via ``HF_TOKEN_TYPE`` / ``HF_WRITE_TOKEN_TYPE`` so :func:`_validate_token_config` can enforce least-privilege at startup without any network calls. **Developer note** — As of 2025, classic HF tokens are approximately 34 characters total (``hf_`` prefix + 30 alphanumeric chars). Fine-grained tokens are substantially longer (≥ 52 characters total as of the HF 2025 token format). This length heuristic is imprecise and subject to silent change by HF; explicit declaration via env vars is always preferred. Examples -------- Explicit declaration takes precedence over heuristics: >>> _classify_token_type("hf_" + "a" * 30, declared_type="read") 'read' >>> _classify_token_type("hf_" + "a" * 30, declared_type="write") 'write' Heuristic: token ≥ 52 chars → fine-grained: >>> _classify_token_type("hf_" + "a" * 50) 'fine-grained' Short classic token without declaration → unknown: >>> _classify_token_type("hf_" + "a" * 28) 'unknown' Empty or malformed token → unknown: >>> _classify_token_type("") 'unknown' """ # Normalise accepted declared-type values (tolerate minor formatting variants). _declared_map: dict[str, HFTokenType] = { "fine-grained": "fine-grained", "finegrained": "fine-grained", "fine_grained": "fine-grained", "read": "read", "write": "write", } if declared_type: normalised = _declared_map.get(declared_type.lower().strip()) if normalised is not None: return normalised # Validate basic token format — all HF tokens start with "hf_". if not token or not token.startswith("hf_") or len(token) < 10: # noqa: PLR2004 return "unknown" # Heuristic: fine-grained tokens are substantially longer than classic tokens. # Classic tokens: ~34 chars total. Fine-grained tokens: ≥ 52 chars (HF 2025). # Best-effort only; explicit declaration via env vars is always preferred. if len(token) >= 52: # noqa: PLR2004 return "fine-grained" # Cannot distinguish classic read vs write by token string alone. return "unknown" def _token_suitable_for_inference(token_type: str) -> bool: """ Return ``True`` when *token_type* is appropriate for HF Inference API calls. This predicate guards inference paths (Path 2 private Space access and Path 3 HF Serverless API). Returning ``False`` for a classic write token does not block the token at runtime — it causes :func:`_validate_token_config` to emit a startup ``WARNING`` so the operator knows they are running with more permission than necessary. Parameters ---------- token_type : str One of the ``HF_TOKEN_TYPE_*`` constants or a free-form string parsed from an environment variable. Returns ------- bool ``True`` for ``"fine-grained"``, ``"read"``, and ``"unknown"``. ``False`` for ``"write"`` (classic write token — over-privileged). Notes ----- The recommended configuration is a fine-grained token scoped exclusively to ``Make calls to the serverless Inference API``, or a classic read token. Classic write tokens carry unnecessary repo-write permission and violate the principle of least privilege. Examples -------- >>> _token_suitable_for_inference("read") True >>> _token_suitable_for_inference("fine-grained") True >>> _token_suitable_for_inference("write") False >>> _token_suitable_for_inference("unknown") True """ return token_type in HF_INFERENCE_TOKEN_TYPES def _token_suitable_for_writes(token_type: str) -> bool: """ Return ``True`` when *token_type* can authorize HuggingFace write operations. This predicate guards the ``/v1/contribute`` endpoint. Returning ``False`` for a classic read or unknown token causes :func:`_validate_token_config` to emit a startup ``ERROR`` string because the token WILL fail at ``HfApi.create_commit`` time (HTTP 403 / 401 from HF). Parameters ---------- token_type : str One of the ``HF_TOKEN_TYPE_*`` constants or a free-form string parsed from an environment variable. Returns ------- bool ``True`` for ``"fine-grained"`` and ``"write"``. ``False`` for ``"read"`` and ``"unknown"``. Notes ----- Fine-grained tokens can write **only if** write permission was granted to the target repo at token-creation time. A fine-grained token created with only inference-API scope will also fail on write operations, but the proxy cannot verify fine-grained permissions without an authenticated API call. Fine-grained tokens are therefore accepted here and any permission failures surface at operation time with a clear HTTP 503 error. Examples -------- >>> _token_suitable_for_writes("write") True >>> _token_suitable_for_writes("fine-grained") True >>> _token_suitable_for_writes("read") False >>> _token_suitable_for_writes("unknown") False """ return token_type in HF_WRITE_TOKEN_TYPES def _validate_token_config( hf_token: str, hf_write_token: str, training_dataset_repo: str = "", *, hf_token_type: str = HF_TOKEN_TYPE_UNKNOWN, hf_write_token_type: str = HF_TOKEN_TYPE_UNKNOWN, ) -> list[str]: """ Validate token types and return actionable warning / error strings. Enforces the principle of least privilege and detects token-type misconfigurations that would cause silent failures at request time. Returns a list of strings rather than raising exceptions so the proxy can start in degraded mode and surface issues through structured logs. Call this at startup **after** :func:`_validate_env` so routing is confirmed viable before type checks are run. Parameters ---------- hf_token : str HuggingFace token used for inference (``HF_TOKEN`` env var). hf_write_token : str HuggingFace token used for dataset persistence. New deployments pass the effective ``HF_DATASET_TOKEN``; legacy callers may still pass ``HF_WRITE_TOKEN``. Pass empty string when not configured. training_dataset_repo : str, optional HuggingFace Dataset repo ID (``TRAINING_DATASET_REPO`` env var). Pass empty string when ``/v1/contribute`` is not enabled. hf_token_type : str, optional Classified type for *hf_token* (from :func:`_classify_token_type`). Defaults to ``"unknown"``. hf_write_token_type : str, optional Classified type for *hf_write_token*. Defaults to ``"unknown"``. Returns ------- list[str] Zero or more diagnostic strings. Each message is prefixed with ``"WARNING:"`` or ``"ERROR:"`` so callers can log at the correct level. An empty list means the configuration passes all checks. Notes ----- **Security note** — ``"write"`` token used for inference is a WARNING (not an error) because it functions correctly at runtime. The warning exists to prompt the operator to apply least-privilege. **Security note** — ``"read"`` token used for writes is a hard ERROR: the token WILL fail on every ``HfApi.create_commit`` call. The proxy can still start (useful for operators who only need inference), but ``/v1/contribute`` will be permanently non-functional until the token is replaced. Examples -------- Clean configuration — no messages: >>> _validate_token_config("hf_readtok", "", hf_token_type="read") [] Write token for inference (overprivileged) → WARNING: >>> msgs = _validate_token_config("hf_writetok", "", hf_token_type="write") >>> any("WARNING" in m for m in msgs) True Read token for writes → ERROR: >>> msgs = _validate_token_config( ... "hf_tok", ... "hf_readtok", ... training_dataset_repo="org/dataset", ... hf_write_token_type="read", ... ) >>> any("ERROR" in m for m in msgs) True """ messages: list[str] = [] # ── Inference token (HF_TOKEN) type check ──────────────────────────────── if hf_token and not _token_suitable_for_inference(hf_token_type): messages.append( f"WARNING: HF_TOKEN type is {hf_token_type!r} (classic write token). " "Write tokens carry unnecessary repo-push permission and violate the " "principle of least privilege for inference. " "Replace HF_TOKEN with: (a) a fine-grained token scoped to " "'Make calls to the serverless Inference API' only, or " "(b) a classic read token. " "See HF Settings → Tokens → New token → Fine-grained. " "Set HF_TOKEN_TYPE=read or HF_TOKEN_TYPE=fine-grained after replacing." ) # ── Dataset-persistence token type check ───────────────────────────────── if hf_write_token and not _token_suitable_for_writes(hf_write_token_type): messages.append( f"ERROR: dataset persistence token type is {hf_write_token_type!r}. " "Read tokens cannot push commits to Hugging Face repositories. " "Use HF_DATASET_TOKEN with a fine-grained token scoped to write the " "target dataset repo (preferred), or a classic Write token. " "Legacy HF_WRITE_TOKEN remains supported as an alias." ) # ── Training repo + effective write token consistency ──────────────────── if training_dataset_repo: # Effective write token is HF_WRITE_TOKEN when set; else falls back to # HF_TOKEN. Check that the effective token type can authorize writes. effective_token = hf_write_token or hf_token effective_type = hf_write_token_type if hf_write_token else hf_token_type if effective_token and not _token_suitable_for_writes(effective_type): messages.append( "ERROR: TRAINING_DATASET_REPO is configured but the effective " "write token type " f"({effective_type!r}) cannot push to HuggingFace repositories. " "POST /v1/contribute will always fail with HTTP 503. " "Set HF_DATASET_TOKEN to a write-capable token (fine-grained with " "write access to the dataset repo, or a classic Write token). " f"Set HF_DATASET_TOKEN_TYPE accordingly." ) return messages def _resolve_upstream_url( body: bytes, *, backend_url: str, hf_token: str, backend_auth_token: str = "", hf_spaces_auth_token: str = "", hf_base: str = DEFAULT_HF_BASE, default_model: str = DEFAULT_MODEL, hf_spaces_model_url: str = DEFAULT_HF_SPACES_MODEL_URL, hf_spaces_model_namespaces: ( tuple[str, ...] | list[str] ) = DEFAULT_HF_SPACES_MODEL_NAMESPACES, proxy_timeout: float = float(DEFAULT_PROXY_TIMEOUT), path2_read_timeout: float = DEFAULT_PATH2_READ_TIMEOUT, path3_read_timeout: float = DEFAULT_PATH3_READ_TIMEOUT, ) -> tuple[str, dict[str, str], float]: """ Centralised three-path routing — choose upstream endpoint, auth headers, and per-path read timeout. Priority -------- 1. *backend_url* is non-empty → **Path 1**: explicit custom backend. Forward to *backend_url* (Docker Model Runner, Ollama, any backend). *backend_auth_token* is injected only when explicitly configured. Read timeout: *proxy_timeout* (env ``PROXY_TIMEOUT``, default 600 s). 2. Model namespace is in *hf_spaces_model_namespaces* → **Path 2**: HF model Space. Forward to *hf_spaces_model_url* (the ``scikit-plots/ai-model`` Space). CPU inference on a 7B model takes 4-5 minutes; *path2_read_timeout* (env ``PATH2_TIMEOUT``, default 600 s) prevents premature timeout. *hf_spaces_auth_token* is injected only when explicitly configured. 3. Otherwise → **Path 3**: HF Serverless Inference API (default). Build ``{hf_base}/{model}/v1/chat/completions`` and inject *hf_token* (always required for the HF API). *path3_read_timeout* (env ``PATH3_TIMEOUT``, default 120 s) is appropriate for GPU-backed HF API inference. Parameters ---------- body : bytes Raw JSON request body. Used to extract the ``model`` field for Paths 2 and 3. backend_url : str Value of the ``BACKEND_URL`` environment variable. Non-empty string triggers Path 1; empty string means "proceed to Path 2 / 3". hf_token : str HuggingFace inference token. Used only for Path 3. backend_auth_token : str, optional Dedicated bearer capability bound to Path 1 ``backend_url``. hf_spaces_auth_token : str, optional Dedicated bearer capability bound to Path 2 ``hf_spaces_model_url``. hf_base : str, optional HF Serverless Inference API base URL (no trailing slash). default_model : str, optional Fallback model ID when the body omits the ``model`` field. hf_spaces_model_url : str, optional URL of the custom ai-model HF Space (Path 2 target). hf_spaces_model_namespaces : tuple[str, ...] or list[str], optional Model owner namespaces routed to *hf_spaces_model_url*. proxy_timeout : float, optional Read timeout (seconds) for Path 1. Default: 600 s. path2_read_timeout : float, optional Read timeout (seconds) for Path 2 (ai-model Space). Default: 600 s. path3_read_timeout : float, optional Read timeout (seconds) for Path 3 (HF Serverless API). Default: 120 s. Returns ------- url : str Fully-qualified upstream endpoint URL. headers : dict[str, str] HTTP headers for the upstream POST request. read_timeout_s : float Per-path read timeout in seconds. Pass to ``httpx.Timeout(read=...)``. Notes ----- **Breaking change v5.0.0** — Return type changed from ``tuple[str, dict]`` to ``tuple[str, dict, float]``. All callers must unpack the third element. **Breaking change v6.0.0** — :data:`DEFAULT_HF_BASE` changed from ``https://api-inference.huggingface.co/models`` to ``https://router.huggingface.co``. The old hostname was DNS-unresolvable from HF Docker Spaces ([Errno -5] EAI_NONAME). **Developer note** — All routing logic lives here. To add a new backend type, add a new branch in this function. Callers (``app.py``, ``dev_proxy.py``) remain unchanged when they already unpack 3 values. Examples -------- Path 2 — scikit-plots namespace → ai-model Space: >>> url, hdrs, t = _resolve_upstream_url( ... b'{"model":"scikit-plots/Qwen2.5-Coder-7B-Instruct","messages":[]}', ... backend_url="", ... hf_token="", ... ) >>> "scikit-plots-ai-model.hf.space" in url True >>> t 600.0 Path 3 — standard HF Inference API: >>> url, hdrs, t = _resolve_upstream_url( ... b'{"model":"openai/gpt-oss-20b","messages":[]}', ... backend_url="", ... hf_token="hf_test_token_abc123", ... ) >>> "router.huggingface.co" in url True >>> t 120.0 Path 1 — explicit BACKEND_URL: >>> url, hdrs, t = _resolve_upstream_url( ... b"{}", ... backend_url="https://my-model.hf.space/v1/chat/completions", ... hf_token="", ... ) >>> url 'https://my-model.hf.space/v1/chat/completions' >>> t 600.0 """ # noqa: D205 headers: dict[str, str] = {"Content-Type": "application/json"} # ── Path 1: explicit custom backend override ────────────────────────────── if backend_url: if backend_auth_token: headers["Authorization"] = f"Bearer {backend_auth_token}" return backend_url, headers, proxy_timeout # Extract model ID from request body (needed for Paths 2 and 3). model: str = _parse_model(body, default=default_model) # ── Path 2: custom model namespace → HF Spaces model backend ───────────── if hf_spaces_model_url and _is_custom_model_namespace( model, hf_spaces_model_namespaces ): if hf_spaces_auth_token: headers["Authorization"] = f"Bearer {hf_spaces_auth_token}" return hf_spaces_model_url, headers, path2_read_timeout # ── Path 3: HF Serverless Inference API (provider models) ───────────────── # router.huggingface.co is a flat OpenAI-compatible endpoint. # The model is supplied in the request body (already present in `body`), # NOT embedded in the URL path. The old api-inference.huggingface.co/models # API DID embed the model in the path as /{model}/v1/chat/completions, but # router.huggingface.co uses a single endpoint for all models: # POST https://router.huggingface.co/v1/chat/completions # body: {"model": "Qwen/Qwen2.5-Coder-7B-Instruct:nscale", ...} # Embedding the model ID in the path produces a 404/422 with no log entry # because _forward passes non-2xx upstream responses through transparently. url = f"{hf_base.rstrip('/')}/v1/chat/completions" # Do not manufacture an empty ``Authorization: Bearer `` header. Besides # being useless, malformed/whitespace-only auth values may be rejected at # the local HTTP protocol layer before a request ever reaches Hugging Face. # When the token is absent, send no Authorization header and let the caller # or upstream return a normal authentication/configuration error. if hf_token: headers["Authorization"] = f"Bearer {hf_token}" return url, headers, path3_read_timeout def _validate_credential_destination( url: str, *, credential_kind: str, allow_local_http: bool = False, ) -> None: """Fail closed when a server credential could be sent to an unsafe URL. ``credential_kind`` is descriptive and never contains the credential. HF inference tokens are bound to official Hugging Face HTTPS origins; custom backend/Space tokens are separately configured and therefore bind to the exact operator-selected destination rather than reusing ``HF_TOKEN``. """ if not url: raise RuntimeError(f"{credential_kind} is configured without a destination URL") try: parts = urlsplit(url) host = (parts.hostname or "").lower().rstrip(".") port = parts.port except (TypeError, ValueError) as exc: raise RuntimeError( f"unsafe destination for {credential_kind}: malformed URL" ) from exc if parts.username or parts.password or parts.query or parts.fragment: raise RuntimeError( f"unsafe destination for {credential_kind}: userinfo/query/fragment is not allowed" ) is_local = host in {"localhost", "127.0.0.1", "::1"} if parts.scheme != "https" and not ( allow_local_http and parts.scheme == "http" and is_local ): raise RuntimeError( f"unsafe destination for {credential_kind}: HTTPS is required" ) if credential_kind == "HF_TOKEN": if host != "router.huggingface.co" and not host.endswith(".huggingface.co"): raise RuntimeError( "unsafe destination for HF_TOKEN: token is bound to official Hugging Face origins" ) if port not in (None, 443): raise RuntimeError("unsafe destination for HF_TOKEN: non-standard port") def _validate_env( backend_url: str, hf_token: str, hf_spaces_model_url: str = DEFAULT_HF_SPACES_MODEL_URL, ) -> None: """ Validate the minimum required environment at proxy startup. At least one of the three routing paths must be viable: * **Path 1** — *backend_url* is non-empty. * **Path 2** — *hf_spaces_model_url* is non-empty (serves custom namespace models). * **Path 3** — *hf_token* is non-empty (HF Inference API for provider models). Parameters ---------- backend_url : str Value of the ``BACKEND_URL`` environment variable (may be empty). hf_token : str Value of the ``HF_TOKEN`` environment variable (may be empty). hf_spaces_model_url : str, optional Value of the ``HF_SPACES_MODEL_URL`` environment variable. Raises ------ RuntimeError When all three routing paths are disabled (all parameters are empty). Examples -------- >>> _validate_env("https://my-model.hf.space/v1/chat/completions", "", "") >>> _validate_env("", "hf_mytoken", "") >>> _validate_env( ... "", "", "https://scikit-plots-ai-model.hf.space/v1/chat/completions" ... ) >>> import pytest >>> with pytest.raises(RuntimeError, match="no viable routing path"): ... _validate_env("", "", "") """ if not backend_url and not hf_token and not hf_spaces_model_url: raise RuntimeError( "Proxy configuration error: no viable routing path configured.\n\n" "Set at least ONE of the following in Space → Settings → Repository secrets:\n\n" " Option 1 — HF Inference API (standard provider models):\n" " HF_TOKEN = hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxx\n" " DEFAULT_MODEL = openai/gpt-oss-20b\n\n" " Option 2 — Custom ai-model Space (scikit-plots/* models):\n" " HF_SPACES_MODEL_URL = " "https://scikit-plots-ai-model.hf.space/v1/chat/completions\n\n" " Option 3 — Explicit custom backend (DMR, Ollama, or any backend):\n" " BACKEND_URL = http://localhost:12434/engines/llama.cpp/v1/chat/completions\n\n" "See FREE_PROXY_SOLUTIONS.md for the full path decision tree." ) def load_proxy_env() -> dict[str, Any]: """ Read all proxy-relevant environment variables and return a typed dict. Returns ------- dict[str, Any] Keys and types: ``backend_url`` : str ``hf_token`` : str ``hf_base`` : str ``default_model`` : str ``hf_spaces_model_url`` : str ``hf_spaces_model_namespaces`` : tuple[str, ...] ``proxy_timeout`` : int Global / Path 1 read timeout (env ``PROXY_TIMEOUT``). ``path2_read_timeout`` : float Path 2 read timeout (env ``PATH2_TIMEOUT``). ``path3_read_timeout`` : float Path 3 read timeout (env ``PATH3_TIMEOUT``). ``max_body_bytes`` : int ``allowed_origins`` : str ``allowed_origins_mode`` : str Raw deployment composition mode (``additive`` or ``replace``). ``hf_token_type`` : str Classified token type for *hf_token* (env ``HF_TOKEN_TYPE``). One of ``"fine-grained"``, ``"read"``, ``"write"``, ``"unknown"``. ``hf_write_token_type`` : str Classified type for the legacy ``HF_WRITE_TOKEN`` alias. ``hf_dataset_token_type`` : str Classified type for the effective dataset-persistence token. One of ``"fine-grained"``, ``"read"``, ``"write"``, ``"unknown"``. Examples -------- >>> import os >>> os.environ["PROXY_TIMEOUT"] = "600" >>> cfg = load_proxy_env() >>> cfg["proxy_timeout"] 600 >>> os.environ["PATH2_TIMEOUT"] = "900" >>> cfg = load_proxy_env() >>> cfg["path2_read_timeout"] 900.0 """ _raw_namespaces: str = os.environ.get( "HF_SPACES_MODEL_NAMESPACES", ",".join(DEFAULT_HF_SPACES_MODEL_NAMESPACES), ) _parsed_namespaces: tuple[str, ...] = ( tuple(ns.strip() for ns in _raw_namespaces.split(",") if ns.strip()) or DEFAULT_HF_SPACES_MODEL_NAMESPACES ) _hf_token: str = os.environ.get("HF_TOKEN", "").strip() _hf_dataset_token_explicit: str = os.environ.get("HF_DATASET_TOKEN", "").strip() _hf_write_token: str = os.environ.get("HF_WRITE_TOKEN", "").strip() # Classify token types from explicit declarations (preferred) or heuristics. # Explicit: set HF_TOKEN_TYPE=read|write|fine-grained in Space secrets. # Heuristic: length-based guess (fine-grained tokens are ≥ 52 chars). _hf_token_type: str = _classify_token_type( _hf_token, declared_type=os.environ.get("HF_TOKEN_TYPE"), ) _hf_write_token_type: str = _classify_token_type( _hf_write_token, declared_type=os.environ.get("HF_WRITE_TOKEN_TYPE"), ) _hf_dataset_token: str = _hf_dataset_token_explicit or _hf_write_token or _hf_token _hf_dataset_token_type: str = ( _classify_token_type( _hf_dataset_token_explicit, declared_type=os.environ.get("HF_DATASET_TOKEN_TYPE"), ) if _hf_dataset_token_explicit else (_hf_write_token_type if _hf_write_token else _hf_token_type) ) return { "backend_url": os.environ.get("BACKEND_URL", "").strip(), "hf_token": _hf_token, # Preferred dataset token + legacy alias. Never forward the effective # dataset token to model backends. "hf_write_token": _hf_write_token, "hf_dataset_token": _hf_dataset_token, # Token type metadata — used by startup validation and discovery. "hf_token_type": _hf_token_type, "hf_write_token_type": _hf_write_token_type, "hf_dataset_token_type": _hf_dataset_token_type, "hf_base": os.environ.get("HF_BASE", DEFAULT_HF_BASE).rstrip("/"), "default_model": ( os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL).strip() or DEFAULT_MODEL ), "hf_spaces_model_url": ( os.environ.get("HF_SPACES_MODEL_URL", DEFAULT_HF_SPACES_MODEL_URL).strip() ), "hf_spaces_model_namespaces": _parsed_namespaces, "proxy_timeout": _safe_int( os.environ.get("PROXY_TIMEOUT"), DEFAULT_PROXY_TIMEOUT, ), "path2_read_timeout": _safe_float( os.environ.get("PATH2_TIMEOUT"), DEFAULT_PATH2_READ_TIMEOUT, ), "path3_read_timeout": _safe_float( os.environ.get("PATH3_TIMEOUT"), DEFAULT_PATH3_READ_TIMEOUT, ), "max_body_bytes": _safe_int( os.environ.get("MAX_BODY_BYTES"), DEFAULT_MAX_BODY_BYTES, ), "allowed_origins": os.environ.get("ALLOWED_ORIGINS", "").strip(), "allowed_origins_mode": ( os.environ.get("ALLOWED_ORIGINS_MODE", "additive").strip().lower() or "additive" ), }