From 328bbbc397c3f102808c063bce59d35be10d110e Mon Sep 17 00:00:00 2001 From: 1bcMax Date: Tue, 21 Jul 2026 00:01:43 -0500 Subject: [PATCH 1/2] fix(validation): stop capping max_tokens below what models actually serve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `validate_max_tokens` rejected anything over 100000 client-side. That number was not a model limit and not the gateway's — it was a hardcoded sanity check with no comment, and it silently became the binding constraint on every SDK caller. Verified against the live gateway 2026-07-21 by bypassing the guard: 19 models advertise a ceiling above 100000 and all 19 accepted it. zai/glm-5.2 262144 accepted claude-opus-4.8 / sonnet-5 / fable-5 / opus-4.7 / sonnet-4.6 128000 accepted gpt-5.6-sol / -terra / -luna, gpt-5.5, gpt-5.4, gpt-5.4-pro, gpt-5.4-mini, gpt-5.2, gpt-5.2-pro, gpt-5.3-codex 128000 accepted glm-5 / glm-5.1 / glm-5-turbo 128000 accepted Zero rejections. A caller asking glm-5.2 for the 262144 it advertises got a ValueError instead of a request. The message made it worse: "max_tokens too large (maximum: 100000)" reads like a provider response. It was read as exactly that during an investigation and recorded as an upstream ceiling in a downstream token table, on the strength of 19 identical "rejections" that never left the process. Now MAX_TOKENS_SANITY_LIMIT = 1_000_000, documented as a typo guard, with a message that says the number is the SDK's and the gateway owns the real limit. Keeping a bound is still right — a byte count or stray 1e9 should fail locally rather than become a payment quote — it just must never be reachable by a real request. Tests: rejection case moved to 2_000_000; added a regression asserting 128000 and 262144 pass. Suite 372 pass. --- blockrun_llm/validation.py | 984 +++++++++++++++++----------------- tests/unit/test_client.py | 355 ++++++------ tests/unit/test_validation.py | 708 ++++++++++++------------ 3 files changed, 1045 insertions(+), 1002 deletions(-) diff --git a/blockrun_llm/validation.py b/blockrun_llm/validation.py index 8957621..9c798eb 100644 --- a/blockrun_llm/validation.py +++ b/blockrun_llm/validation.py @@ -1,481 +1,503 @@ -""" -Input validation and security utilities for BlockRun LLM SDK. - -This module provides validation functions to ensure: -- Private keys are properly formatted -- API URLs use HTTPS -- Parameters are within valid ranges -- Server responses don't leak sensitive information -- Resource URLs match expected domains -""" - -import re -from typing import Optional, Dict, Any, TYPE_CHECKING -from urllib.parse import urlparse - -if TYPE_CHECKING: - from .types import PaymentError - - -# Localhost domains that are allowed to use HTTP -LOCALHOST_DOMAINS = {"localhost", "127.0.0.1"} - -# Known LLM providers (for optional validation) -KNOWN_PROVIDERS = { - "openai", - "anthropic", - "google", - "deepseek", - "mistralai", - "meta-llama", - "together", - "xai", - "moonshot", - "nvidia", - "minimax", - "zai", -} - -# Seed modes a caller may assert via `input_type` on /v1/videos/generations. -# Mirrors the gateway enum; the gateway stays the authority on whether the -# declared mode matches the seed fields actually sent. -VIDEO_INPUT_TYPES = ("text", "image", "first_last_frame", "reference") - -# Latency/fidelity levels for `quality` on Solana image generation + editing. -# Mirrors the gateway enum, which accepts the field for openai/gpt-image-* only. -IMAGE_QUALITY_LEVELS = ("low", "medium", "high", "auto") - - -# Base58 alphabet characters that never appear in a hex string. Their presence -# is a strong signal that a key is a base58-encoded Solana key, not an EVM key. -_BASE58_ONLY_CHARS = frozenset("GHJKLMNPQRSTUVWXYZghijkmnopqrstuvwxyz") - - -def _looks_like_solana_key(key: str) -> bool: - """ - Heuristically detect a base58-encoded Solana secret key. - - Solana secret keys are base58, not hex: a 32-byte seed is ~43-44 chars and a - 64-byte keypair is ~87-88 chars. An EVM key is exactly 64 hex chars (sans the - ``0x`` prefix). We treat a key as Solana when it contains a base58-only - character (one absent from the hex alphabet) and its length is outside the - EVM 64-char range — so a malformed 64-char hex key still routes to the - regular hex error rather than the Solana hint. - """ - candidate = key[2:] if key.startswith("0x") else key - if len(candidate) == 64 or not (40 <= len(candidate) <= 90): - return False - return any(c in _BASE58_ONLY_CHARS for c in candidate) - - -def validate_private_key(key: str) -> None: - """ - Validate that a private key is properly formatted. - - Args: - key: The private key to validate - - Raises: - ValueError: If the key format is invalid - - Example: - >>> validate_private_key("0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80") - """ - if not isinstance(key, str): - raise ValueError("Private key must be a string") - - # Detect a base58 Solana key fed into the EVM (Base) client and point the - # user at the right entry point instead of the cryptic "66 characters" error. - if _looks_like_solana_key(key): - raise ValueError( - "This looks like a Solana (base58) private key, but this client uses " - "the Base (EVM) chain. Use the Solana client instead:\n" - " from blockrun_llm import SolanaLLMClient\n" - ' client = SolanaLLMClient(private_key="")\n' - "Or for agent use:\n" - " from blockrun_llm import setup_agent_solana_wallet\n" - " client = setup_agent_solana_wallet()\n" - 'Install Solana support first: pip install "blockrun-llm[solana]"' - ) - - # Must start with 0x - if not key.startswith("0x"): - raise ValueError("Private key must start with 0x") - - # Must be exactly 66 characters (0x + 64 hex chars) - if len(key) != 66: - raise ValueError("Private key must be 66 characters (0x + 64 hexadecimal characters)") - - # Must contain only valid hexadecimal characters - if not re.match(r"^0x[0-9a-fA-F]{64}$", key): - raise ValueError("Private key must contain only hexadecimal characters (0-9, a-f, A-F)") - - -def validate_eth_address(address: str) -> None: - """ - Validate that a value is a well-formed Ethereum / Base address. - - Args: - address: The 0x-prefixed 20-byte address to validate - - Raises: - ValueError: If the address format is invalid - - Example: - >>> validate_eth_address("0x036CbD53842c5426634e7929541eC2318f3dCF7e") - """ - if not isinstance(address, str): - raise ValueError("Address must be a string") - - # Must be a 0x-prefixed 40-character hexadecimal string - if not re.match(r"^0x[0-9a-fA-F]{40}$", address): - raise ValueError("Address must be a 0x-prefixed 40-character hexadecimal string") - - -def validate_model(model: str) -> None: - """ - Validate model ID format. - - Args: - model: The model ID (e.g., "openai/gpt-5.2", "anthropic/claude-sonnet-4.5") - - Raises: - ValueError: If model is invalid - - Example: - >>> validate_model("openai/gpt-5.2") - """ - if not model or not isinstance(model, str): - raise ValueError("Model must be a non-empty string") - - # Optionally validate provider (just a warning, don't fail) - if "/" in model: - provider = model.split("/", 1)[0] - if provider not in KNOWN_PROVIDERS: - # Just log, don't fail (allows new providers) - pass - - -def validate_video_input_type(input_type: Optional[str]) -> None: - """ - Validate the optional `input_type` seed-mode assertion on video generation. - - Only the spelling is checked. Whether the declared mode agrees with the - seed fields actually sent is the gateway's call — it infers the mode and - rejects with 400 *before* charging, so re-deriving that inference here - would add a second copy to keep in sync for no benefit. - - Args: - input_type: One of VIDEO_INPUT_TYPES, or None to leave it unset. - - Raises: - ValueError: If input_type is not one of the accepted values. - - Example: - >>> validate_video_input_type("first_last_frame") - """ - if input_type is None: - return - if input_type not in VIDEO_INPUT_TYPES: - raise ValueError( - f"input_type must be one of {', '.join(VIDEO_INPUT_TYPES)}; got {input_type!r}." - ) - - -def validate_image_quality(quality: Optional[str]) -> None: - """ - Validate the optional `quality` knob on Solana image generation/editing. - - Model compatibility is left to the gateway, which accepts `quality` only - for openai/gpt-image-* and returns a clear error otherwise — encoding that - model list here would go stale every time the catalog changes. - - Args: - quality: One of IMAGE_QUALITY_LEVELS, or None to leave it unset. - - Raises: - ValueError: If quality is not one of the accepted values. - - Example: - >>> validate_image_quality("low") - """ - if quality is None: - return - if quality not in IMAGE_QUALITY_LEVELS: - raise ValueError( - f"quality must be one of {', '.join(IMAGE_QUALITY_LEVELS)}; got {quality!r}." - ) - - -def validate_max_tokens(max_tokens: Optional[int]) -> None: - """ - Validate max_tokens parameter. - - Args: - max_tokens: Maximum number of tokens to generate - - Raises: - ValueError: If max_tokens is invalid - - Example: - >>> validate_max_tokens(1000) - """ - if max_tokens is None: - return - - if not isinstance(max_tokens, int): - raise ValueError("max_tokens must be an integer") - - if max_tokens < 1: - raise ValueError("max_tokens must be positive (minimum: 1)") - - if max_tokens > 100000: - raise ValueError("max_tokens too large (maximum: 100000)") - - -def validate_temperature(temperature: Optional[float]) -> None: - """ - Validate temperature parameter. - - Args: - temperature: Sampling temperature (0-2) - - Raises: - ValueError: If temperature is invalid - - Example: - >>> validate_temperature(0.7) - """ - if temperature is None: - return - - if not isinstance(temperature, (int, float)): - raise ValueError("temperature must be a number") - - if temperature < 0 or temperature > 2: - raise ValueError("temperature must be between 0 and 2") - - -def validate_top_p(top_p: Optional[float]) -> None: - """ - Validate top_p parameter (nucleus sampling). - - Args: - top_p: Top-p sampling parameter (0-1) - - Raises: - ValueError: If top_p is invalid - - Example: - >>> validate_top_p(0.9) - """ - if top_p is None: - return - - if not isinstance(top_p, (int, float)): - raise ValueError("top_p must be a number") - - if top_p < 0 or top_p > 1: - raise ValueError("top_p must be between 0 and 1") - - -def validate_api_url(url: str) -> None: - """ - Validate that an API URL is secure and properly formatted. - - Args: - url: The API URL to validate - - Raises: - ValueError: If the URL is invalid or insecure - - Example: - >>> validate_api_url("https://blockrun.ai/api") - >>> validate_api_url("http://localhost:3000") # OK for development - """ - try: - parsed = urlparse(url) - except Exception as e: - raise ValueError(f"Invalid API URL: {e}") - - if not parsed.scheme: - raise ValueError("API URL must include scheme (http:// or https://)") - - if not parsed.netloc: - raise ValueError("API URL must include domain") - - # Require HTTPS for non-localhost URLs - is_localhost = parsed.netloc.split(":")[0] in LOCALHOST_DOMAINS - - if parsed.scheme != "https" and not is_localhost: - raise ValueError( - "API URL must use HTTPS for non-localhost endpoints. " - f"Use https:// instead of {parsed.scheme}://" - ) - - -def build_payment_rejected_error(response: Any) -> "PaymentError": - """Translate a 402 retry response into a :class:`PaymentError` that - preserves the gateway's original failure reason. - - Without this helper, clients used to throw a generic - ``"Payment rejected. Check your wallet balance."`` and the real - facilitator reason (e.g. ``transaction_simulation_failed``, - ``insufficient_funds``) was lost. - - The gateway's ``details`` field on a 402 settlement-failed response - is the x402 facilitator's well-defined error enum — safe to surface - verbatim. We bound the length defensively in case a future server - bug widens the field. - - Args: - response: An ``httpx.Response`` with status 402 from a paid - retry. Anything with a ``.json()`` method works for tests. - - Returns: - A :class:`PaymentError` carrying ``status_code=402`` and a - ``response`` dict that includes the gateway's ``details``. - """ - # Local import to avoid a circular module dependency at import time. - from .types import PaymentError - - try: - body = response.json() - except Exception: - body = {} - if not isinstance(body, dict): - body = {} - sanitized = dict(sanitize_error_response(body)) - raw_details = body.get("details") - if isinstance(raw_details, str) and 0 < len(raw_details) < 256: - sanitized["details"] = raw_details - # The x402 facilitator's `invalidMessage` — the simulation-level cause that - # the coarse `invalidReason` enum collapses away (an unfunded wallet and a - # stale blockhash both arrive as transaction_simulation_failed). Same - # provenance and safety rationale as `details` above: a facilitator error - # string, not upstream text, so it's safe to surface verbatim — bounded - # defensively all the same. Folded into the message because the retry - # classifiers in solana_client only ever see `str(exc)`. - raw_invalid_message = body.get("invalidMessage") - if isinstance(raw_invalid_message, str) and 0 < len(raw_invalid_message) < 256: - sanitized["invalidMessage"] = raw_invalid_message - detail_part = sanitized.get("details") or sanitized.get("message") or "" - invalid_message = sanitized.get("invalidMessage") - if invalid_message: - detail_part = f"{detail_part} ({invalid_message})" if detail_part else invalid_message - msg = ( - f"Payment rejected by gateway: {detail_part}" - if detail_part - else "Payment rejected by gateway" - ) - return PaymentError(msg, status_code=402, response=sanitized) - - -def sanitize_error_response(error_body: Any) -> Dict[str, Any]: - """ - Sanitize API error responses to prevent information leakage. - - Only exposes safe error fields to the caller, filtering out: - - Internal stack traces - - Server-side paths - - API keys or tokens - - Debugging information - - Args: - error_body: The raw error response from the API - - Returns: - Sanitized error dict with only safe fields - - Example: - >>> sanitize_error_response({ - ... "error": "Invalid model", - ... "internal_stack": "/var/app/handler.py:123", - ... "api_key": "secret" - ... }) - {'message': 'Invalid model', 'code': None} - """ - if not isinstance(error_body, dict): - return {"message": "API request failed", "code": None} - - # The gateway returns OpenAI-compatible *nested* errors: - # {"error": {"message", "type", "code", "param"}, "message", "code", "debug"} - # while older endpoints (and the SDK's own fallbacks) still use the *flat* shape: - # {"error": "Request failed", "code": "..."} - # Pass the real message/code through for either shape. Never surface `debug` - # (raw upstream error text — may leak internal paths/keys). - nested = error_body.get("error") - - if isinstance(nested, dict): - message = nested.get("message") - code = nested.get("code") or error_body.get("code") - result: Dict[str, Any] = { - "message": message if isinstance(message, str) else "API request failed", - "code": code if isinstance(code, str) else None, - } - # Pass through OpenAI error metadata when present. - if isinstance(nested.get("type"), str): - result["type"] = nested["type"] - if isinstance(nested.get("param"), str): - result["param"] = nested["param"] - return result - - # Flat shape: `error` is the human-readable title; fall back to top-level `message`. - if isinstance(nested, str): - message = nested - elif isinstance(error_body.get("message"), str): - message = error_body["message"] - else: - message = "API request failed" - - return { - "message": message, - "code": (error_body.get("code") if isinstance(error_body.get("code"), str) else None), - } - - -def validate_resource_url(url: str, base_url: str) -> str: - """ - Validate a resource URL from the server to prevent redirection attacks. - - Ensures that the resource URL's hostname matches the API's hostname. - If domains don't match, returns a safe default URL instead. - - Args: - url: The resource URL provided by the server - base_url: The base API URL (trusted) - - Returns: - The validated URL or a safe default - - Example: - >>> validate_resource_url( - ... "https://blockrun.ai/api/v1/chat", - ... "https://blockrun.ai/api" - ... ) - 'https://blockrun.ai/api/v1/chat' - - >>> validate_resource_url( - ... "https://malicious.com/steal", - ... "https://blockrun.ai/api" - ... ) - 'https://blockrun.ai/api/v1/chat/completions' - """ - try: - parsed = urlparse(url) - base_parsed = urlparse(base_url) - - # Resource URL hostname must match API hostname - if parsed.netloc != base_parsed.netloc: - # Return safe default - return f"{base_url}/v1/chat/completions" - - # Ensure resource uses same protocol as base - if parsed.scheme != base_parsed.scheme: - return f"{base_url}/v1/chat/completions" - - return url - - except Exception: - # Invalid URL format, return safe default - return f"{base_url}/v1/chat/completions" +""" +Input validation and security utilities for BlockRun LLM SDK. + +This module provides validation functions to ensure: +- Private keys are properly formatted +- API URLs use HTTPS +- Parameters are within valid ranges +- Server responses don't leak sensitive information +- Resource URLs match expected domains +""" + +import re +from typing import Optional, Dict, Any, TYPE_CHECKING +from urllib.parse import urlparse + +if TYPE_CHECKING: + from .types import PaymentError + + +# Localhost domains that are allowed to use HTTP +LOCALHOST_DOMAINS = {"localhost", "127.0.0.1"} + +# Known LLM providers (for optional validation) +KNOWN_PROVIDERS = { + "openai", + "anthropic", + "google", + "deepseek", + "mistralai", + "meta-llama", + "together", + "xai", + "moonshot", + "nvidia", + "minimax", + "zai", +} + +# Seed modes a caller may assert via `input_type` on /v1/videos/generations. +# Mirrors the gateway enum; the gateway stays the authority on whether the +# declared mode matches the seed fields actually sent. +VIDEO_INPUT_TYPES = ("text", "image", "first_last_frame", "reference") + +# Latency/fidelity levels for `quality` on Solana image generation + editing. +# Mirrors the gateway enum, which accepts the field for openai/gpt-image-* only. +IMAGE_QUALITY_LEVELS = ("low", "medium", "high", "auto") + + +# Base58 alphabet characters that never appear in a hex string. Their presence +# is a strong signal that a key is a base58-encoded Solana key, not an EVM key. +_BASE58_ONLY_CHARS = frozenset("GHJKLMNPQRSTUVWXYZghijkmnopqrstuvwxyz") + + +def _looks_like_solana_key(key: str) -> bool: + """ + Heuristically detect a base58-encoded Solana secret key. + + Solana secret keys are base58, not hex: a 32-byte seed is ~43-44 chars and a + 64-byte keypair is ~87-88 chars. An EVM key is exactly 64 hex chars (sans the + ``0x`` prefix). We treat a key as Solana when it contains a base58-only + character (one absent from the hex alphabet) and its length is outside the + EVM 64-char range — so a malformed 64-char hex key still routes to the + regular hex error rather than the Solana hint. + """ + candidate = key[2:] if key.startswith("0x") else key + if len(candidate) == 64 or not (40 <= len(candidate) <= 90): + return False + return any(c in _BASE58_ONLY_CHARS for c in candidate) + + +def validate_private_key(key: str) -> None: + """ + Validate that a private key is properly formatted. + + Args: + key: The private key to validate + + Raises: + ValueError: If the key format is invalid + + Example: + >>> validate_private_key("0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80") + """ + if not isinstance(key, str): + raise ValueError("Private key must be a string") + + # Detect a base58 Solana key fed into the EVM (Base) client and point the + # user at the right entry point instead of the cryptic "66 characters" error. + if _looks_like_solana_key(key): + raise ValueError( + "This looks like a Solana (base58) private key, but this client uses " + "the Base (EVM) chain. Use the Solana client instead:\n" + " from blockrun_llm import SolanaLLMClient\n" + ' client = SolanaLLMClient(private_key="")\n' + "Or for agent use:\n" + " from blockrun_llm import setup_agent_solana_wallet\n" + " client = setup_agent_solana_wallet()\n" + 'Install Solana support first: pip install "blockrun-llm[solana]"' + ) + + # Must start with 0x + if not key.startswith("0x"): + raise ValueError("Private key must start with 0x") + + # Must be exactly 66 characters (0x + 64 hex chars) + if len(key) != 66: + raise ValueError("Private key must be 66 characters (0x + 64 hexadecimal characters)") + + # Must contain only valid hexadecimal characters + if not re.match(r"^0x[0-9a-fA-F]{64}$", key): + raise ValueError("Private key must contain only hexadecimal characters (0-9, a-f, A-F)") + + +def validate_eth_address(address: str) -> None: + """ + Validate that a value is a well-formed Ethereum / Base address. + + Args: + address: The 0x-prefixed 20-byte address to validate + + Raises: + ValueError: If the address format is invalid + + Example: + >>> validate_eth_address("0x036CbD53842c5426634e7929541eC2318f3dCF7e") + """ + if not isinstance(address, str): + raise ValueError("Address must be a string") + + # Must be a 0x-prefixed 40-character hexadecimal string + if not re.match(r"^0x[0-9a-fA-F]{40}$", address): + raise ValueError("Address must be a 0x-prefixed 40-character hexadecimal string") + + +def validate_model(model: str) -> None: + """ + Validate model ID format. + + Args: + model: The model ID (e.g., "openai/gpt-5.2", "anthropic/claude-sonnet-4.5") + + Raises: + ValueError: If model is invalid + + Example: + >>> validate_model("openai/gpt-5.2") + """ + if not model or not isinstance(model, str): + raise ValueError("Model must be a non-empty string") + + # Optionally validate provider (just a warning, don't fail) + if "/" in model: + provider = model.split("/", 1)[0] + if provider not in KNOWN_PROVIDERS: + # Just log, don't fail (allows new providers) + pass + + +def validate_video_input_type(input_type: Optional[str]) -> None: + """ + Validate the optional `input_type` seed-mode assertion on video generation. + + Only the spelling is checked. Whether the declared mode agrees with the + seed fields actually sent is the gateway's call — it infers the mode and + rejects with 400 *before* charging, so re-deriving that inference here + would add a second copy to keep in sync for no benefit. + + Args: + input_type: One of VIDEO_INPUT_TYPES, or None to leave it unset. + + Raises: + ValueError: If input_type is not one of the accepted values. + + Example: + >>> validate_video_input_type("first_last_frame") + """ + if input_type is None: + return + if input_type not in VIDEO_INPUT_TYPES: + raise ValueError( + f"input_type must be one of {', '.join(VIDEO_INPUT_TYPES)}; got {input_type!r}." + ) + + +def validate_image_quality(quality: Optional[str]) -> None: + """ + Validate the optional `quality` knob on Solana image generation/editing. + + Model compatibility is left to the gateway, which accepts `quality` only + for openai/gpt-image-* and returns a clear error otherwise — encoding that + model list here would go stale every time the catalog changes. + + Args: + quality: One of IMAGE_QUALITY_LEVELS, or None to leave it unset. + + Raises: + ValueError: If quality is not one of the accepted values. + + Example: + >>> validate_image_quality("low") + """ + if quality is None: + return + if quality not in IMAGE_QUALITY_LEVELS: + raise ValueError( + f"quality must be one of {', '.join(IMAGE_QUALITY_LEVELS)}; got {quality!r}." + ) + + +# Client-side typo guard, NOT a model limit. The gateway already enforces the +# real per-model ceiling and rejects with that model's own number, so anything +# the SDK hardcodes here can only be wrong in one direction: too low. +# +# This was 100000, and it silently capped every SDK caller below what models +# actually support. Verified against the live gateway 2026-07-21 with the guard +# bypassed: zai/glm-5.2 accepts 262144, and the entire 128000 class accepts +# 128000 (claude-opus-4.8, claude-sonnet-5, claude-fable-5, gpt-5.6-sol/terra/ +# luna, gpt-5.5, gpt-5.4, gpt-5.3-codex, glm-5/5.1/5-turbo — 19 models, 19 +# accepted, zero rejections). Callers asking for those ceilings got a +# ValueError that never reached the network and named a limit no provider set. +# +# Keep a bound so an obvious mistake (1e9, a byte count, a timestamp) fails +# fast locally instead of becoming a payment quote. Set it far above any real +# model so it can never be the binding constraint again. +MAX_TOKENS_SANITY_LIMIT = 1_000_000 + + +def validate_max_tokens(max_tokens: Optional[int]) -> None: + """ + Validate max_tokens parameter. + + Args: + max_tokens: Maximum number of tokens to generate + + Raises: + ValueError: If max_tokens is invalid + + Example: + >>> validate_max_tokens(1000) + """ + if max_tokens is None: + return + + if not isinstance(max_tokens, int): + raise ValueError("max_tokens must be an integer") + + if max_tokens < 1: + raise ValueError("max_tokens must be positive (minimum: 1)") + + if max_tokens > MAX_TOKENS_SANITY_LIMIT: + raise ValueError( + f"max_tokens implausibly large (client-side sanity limit: " + f"{MAX_TOKENS_SANITY_LIMIT}). This is not a model limit — the " + f"gateway enforces the real per-model ceiling and reports it." + ) + + +def validate_temperature(temperature: Optional[float]) -> None: + """ + Validate temperature parameter. + + Args: + temperature: Sampling temperature (0-2) + + Raises: + ValueError: If temperature is invalid + + Example: + >>> validate_temperature(0.7) + """ + if temperature is None: + return + + if not isinstance(temperature, (int, float)): + raise ValueError("temperature must be a number") + + if temperature < 0 or temperature > 2: + raise ValueError("temperature must be between 0 and 2") + + +def validate_top_p(top_p: Optional[float]) -> None: + """ + Validate top_p parameter (nucleus sampling). + + Args: + top_p: Top-p sampling parameter (0-1) + + Raises: + ValueError: If top_p is invalid + + Example: + >>> validate_top_p(0.9) + """ + if top_p is None: + return + + if not isinstance(top_p, (int, float)): + raise ValueError("top_p must be a number") + + if top_p < 0 or top_p > 1: + raise ValueError("top_p must be between 0 and 1") + + +def validate_api_url(url: str) -> None: + """ + Validate that an API URL is secure and properly formatted. + + Args: + url: The API URL to validate + + Raises: + ValueError: If the URL is invalid or insecure + + Example: + >>> validate_api_url("https://blockrun.ai/api") + >>> validate_api_url("http://localhost:3000") # OK for development + """ + try: + parsed = urlparse(url) + except Exception as e: + raise ValueError(f"Invalid API URL: {e}") + + if not parsed.scheme: + raise ValueError("API URL must include scheme (http:// or https://)") + + if not parsed.netloc: + raise ValueError("API URL must include domain") + + # Require HTTPS for non-localhost URLs + is_localhost = parsed.netloc.split(":")[0] in LOCALHOST_DOMAINS + + if parsed.scheme != "https" and not is_localhost: + raise ValueError( + "API URL must use HTTPS for non-localhost endpoints. " + f"Use https:// instead of {parsed.scheme}://" + ) + + +def build_payment_rejected_error(response: Any) -> "PaymentError": + """Translate a 402 retry response into a :class:`PaymentError` that + preserves the gateway's original failure reason. + + Without this helper, clients used to throw a generic + ``"Payment rejected. Check your wallet balance."`` and the real + facilitator reason (e.g. ``transaction_simulation_failed``, + ``insufficient_funds``) was lost. + + The gateway's ``details`` field on a 402 settlement-failed response + is the x402 facilitator's well-defined error enum — safe to surface + verbatim. We bound the length defensively in case a future server + bug widens the field. + + Args: + response: An ``httpx.Response`` with status 402 from a paid + retry. Anything with a ``.json()`` method works for tests. + + Returns: + A :class:`PaymentError` carrying ``status_code=402`` and a + ``response`` dict that includes the gateway's ``details``. + """ + # Local import to avoid a circular module dependency at import time. + from .types import PaymentError + + try: + body = response.json() + except Exception: + body = {} + if not isinstance(body, dict): + body = {} + sanitized = dict(sanitize_error_response(body)) + raw_details = body.get("details") + if isinstance(raw_details, str) and 0 < len(raw_details) < 256: + sanitized["details"] = raw_details + # The x402 facilitator's `invalidMessage` — the simulation-level cause that + # the coarse `invalidReason` enum collapses away (an unfunded wallet and a + # stale blockhash both arrive as transaction_simulation_failed). Same + # provenance and safety rationale as `details` above: a facilitator error + # string, not upstream text, so it's safe to surface verbatim — bounded + # defensively all the same. Folded into the message because the retry + # classifiers in solana_client only ever see `str(exc)`. + raw_invalid_message = body.get("invalidMessage") + if isinstance(raw_invalid_message, str) and 0 < len(raw_invalid_message) < 256: + sanitized["invalidMessage"] = raw_invalid_message + detail_part = sanitized.get("details") or sanitized.get("message") or "" + invalid_message = sanitized.get("invalidMessage") + if invalid_message: + detail_part = f"{detail_part} ({invalid_message})" if detail_part else invalid_message + msg = ( + f"Payment rejected by gateway: {detail_part}" + if detail_part + else "Payment rejected by gateway" + ) + return PaymentError(msg, status_code=402, response=sanitized) + + +def sanitize_error_response(error_body: Any) -> Dict[str, Any]: + """ + Sanitize API error responses to prevent information leakage. + + Only exposes safe error fields to the caller, filtering out: + - Internal stack traces + - Server-side paths + - API keys or tokens + - Debugging information + + Args: + error_body: The raw error response from the API + + Returns: + Sanitized error dict with only safe fields + + Example: + >>> sanitize_error_response({ + ... "error": "Invalid model", + ... "internal_stack": "/var/app/handler.py:123", + ... "api_key": "secret" + ... }) + {'message': 'Invalid model', 'code': None} + """ + if not isinstance(error_body, dict): + return {"message": "API request failed", "code": None} + + # The gateway returns OpenAI-compatible *nested* errors: + # {"error": {"message", "type", "code", "param"}, "message", "code", "debug"} + # while older endpoints (and the SDK's own fallbacks) still use the *flat* shape: + # {"error": "Request failed", "code": "..."} + # Pass the real message/code through for either shape. Never surface `debug` + # (raw upstream error text — may leak internal paths/keys). + nested = error_body.get("error") + + if isinstance(nested, dict): + message = nested.get("message") + code = nested.get("code") or error_body.get("code") + result: Dict[str, Any] = { + "message": message if isinstance(message, str) else "API request failed", + "code": code if isinstance(code, str) else None, + } + # Pass through OpenAI error metadata when present. + if isinstance(nested.get("type"), str): + result["type"] = nested["type"] + if isinstance(nested.get("param"), str): + result["param"] = nested["param"] + return result + + # Flat shape: `error` is the human-readable title; fall back to top-level `message`. + if isinstance(nested, str): + message = nested + elif isinstance(error_body.get("message"), str): + message = error_body["message"] + else: + message = "API request failed" + + return { + "message": message, + "code": (error_body.get("code") if isinstance(error_body.get("code"), str) else None), + } + + +def validate_resource_url(url: str, base_url: str) -> str: + """ + Validate a resource URL from the server to prevent redirection attacks. + + Ensures that the resource URL's hostname matches the API's hostname. + If domains don't match, returns a safe default URL instead. + + Args: + url: The resource URL provided by the server + base_url: The base API URL (trusted) + + Returns: + The validated URL or a safe default + + Example: + >>> validate_resource_url( + ... "https://blockrun.ai/api/v1/chat", + ... "https://blockrun.ai/api" + ... ) + 'https://blockrun.ai/api/v1/chat' + + >>> validate_resource_url( + ... "https://malicious.com/steal", + ... "https://blockrun.ai/api" + ... ) + 'https://blockrun.ai/api/v1/chat/completions' + """ + try: + parsed = urlparse(url) + base_parsed = urlparse(base_url) + + # Resource URL hostname must match API hostname + if parsed.netloc != base_parsed.netloc: + # Return safe default + return f"{base_url}/v1/chat/completions" + + # Ensure resource uses same protocol as base + if parsed.scheme != base_parsed.scheme: + return f"{base_url}/v1/chat/completions" + + return url + + except Exception: + # Invalid URL format, return safe default + return f"{base_url}/v1/chat/completions" diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 8483682..883399c 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,175 +1,180 @@ -"""Unit tests for LLMClient.""" - -import pytest -from unittest.mock import Mock, patch -from blockrun_llm import LLMClient, APIError -from ..helpers import ( - TEST_PRIVATE_KEY, - build_error_response, - build_models_response, - MockResponse, -) - - -class TestLLMClientInit: - def test_init_with_valid_key(self): - """Should create client with valid private key.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - assert client is not None - assert client.get_wallet_address().startswith("0x") - - def test_init_missing_key_raises_error(self, monkeypatch, tmp_path): - """Should raise ValueError when no wallet configured.""" - monkeypatch.delenv("BLOCKRUN_WALLET_KEY", raising=False) - monkeypatch.delenv("BASE_CHAIN_WALLET_KEY", raising=False) - # Mock load_wallet to return None (no session file) - monkeypatch.setattr("blockrun_llm.wallet.load_wallet", lambda: None) - # Should raise ValueError with helpful message - with pytest.raises(ValueError, match="No wallet configured"): - LLMClient(private_key=None) - - def test_init_invalid_key_format(self): - """Should raise ValueError for invalid key format (after 0x normalization).""" - # "invalid" becomes "0xinvalid" after normalization, which is too short - with pytest.raises(ValueError, match="66 characters"): - LLMClient(private_key="invalid") - - def test_init_short_key(self): - """Should raise ValueError for short key.""" - with pytest.raises(ValueError, match="66 characters"): - LLMClient(private_key="0x123") - - def test_init_non_hex_key(self): - """Should raise ValueError for non-hex key.""" - with pytest.raises(ValueError, match="hexadecimal"): - LLMClient( - private_key="0xGGGG74bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" - ) - - def test_default_api_url(self): - """Should use default API URL.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - assert client.api_url == "https://blockrun.ai/api" - - def test_custom_api_url(self): - """Should accept custom API URL.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY, api_url="https://custom.example.com") - assert client.api_url == "https://custom.example.com" - - def test_invalid_api_url_http(self): - """Should reject HTTP for non-localhost.""" - with pytest.raises(ValueError, match="HTTPS"): - LLMClient(private_key=TEST_PRIVATE_KEY, api_url="http://insecure.com") - - def test_allow_localhost_http(self): - """Should allow HTTP for localhost.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY, api_url="http://localhost:3000") - assert client.api_url == "http://localhost:3000" - - -class TestLLMClientMethods: - def test_get_wallet_address(self): - """Should return valid Ethereum address.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - address = client.get_wallet_address() - - assert address == "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266" - assert address.startswith("0x") - assert len(address) == 42 - - @patch("blockrun_llm.client.httpx.Client") - def test_list_models(self, mock_client_class): - """Should list available models.""" - mock_client = Mock() - mock_client_class.return_value = mock_client - - mock_response = MockResponse(200, build_models_response()) - mock_client.get.return_value = mock_response - - client = LLMClient(private_key=TEST_PRIVATE_KEY) - models = client.list_models() - - assert len(models) == 3 - assert models[0]["id"] == "openai/gpt-5.2" - assert models[0]["provider"] == "openai" - - @patch("blockrun_llm.client.httpx.Client") - def test_list_models_error(self, mock_client_class): - """Should raise APIError on failure.""" - mock_client = Mock() - mock_client_class.return_value = mock_client - - mock_response = MockResponse(500) - mock_client.get.return_value = mock_response - - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - with pytest.raises(APIError): - client.list_models() - - -class TestErrorSanitization: - @patch("blockrun_llm.client.httpx.Client") - def test_sanitize_error_responses(self, mock_client_class): - """Should sanitize error responses.""" - mock_client = Mock() - mock_client_class.return_value = mock_client - - raw_error = build_error_response(error="Invalid model", include_sensitive=True) - mock_response = MockResponse(400, raw_error) - mock_client.get.return_value = mock_response - - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - try: - client.list_models() - pytest.fail("Should have raised APIError") - except APIError as e: - # Should only contain safe fields - assert e.response == {"message": "Invalid model", "code": "test_error"} - - # Should NOT contain sensitive fields - assert "internal_stack" not in e.response - assert "api_key" not in e.response - assert "database_url" not in e.response - - -class TestInputValidation: - @patch("blockrun_llm.client.httpx.Client") - def test_validate_model_parameter(self, mock_client_class): - """Should validate model parameter.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - with pytest.raises(ValueError, match="non-empty string"): - client.chat_completion("", [{"role": "user", "content": "test"}]) - - @patch("blockrun_llm.client.httpx.Client") - def test_validate_max_tokens(self, mock_client_class): - """Should validate max_tokens parameter.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - with pytest.raises(ValueError, match="positive"): - client.chat_completion("gpt-5.2", [{"role": "user", "content": "test"}], max_tokens=-1) - - with pytest.raises(ValueError, match="too large"): - client.chat_completion( - "gpt-5.2", [{"role": "user", "content": "test"}], max_tokens=200000 - ) - - @patch("blockrun_llm.client.httpx.Client") - def test_validate_temperature(self, mock_client_class): - """Should validate temperature parameter.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - with pytest.raises(ValueError, match="between 0 and 2"): - client.chat_completion( - "gpt-5.2", [{"role": "user", "content": "test"}], temperature=3.0 - ) - - @patch("blockrun_llm.client.httpx.Client") - def test_validate_top_p(self, mock_client_class): - """Should validate top_p parameter.""" - client = LLMClient(private_key=TEST_PRIVATE_KEY) - - with pytest.raises(ValueError, match="between 0 and 1"): - client.chat_completion("gpt-5.2", [{"role": "user", "content": "test"}], top_p=1.5) +"""Unit tests for LLMClient.""" + +import pytest +from unittest.mock import Mock, patch +from blockrun_llm import LLMClient, APIError +from ..helpers import ( + TEST_PRIVATE_KEY, + build_error_response, + build_models_response, + MockResponse, +) + + +class TestLLMClientInit: + def test_init_with_valid_key(self): + """Should create client with valid private key.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + assert client is not None + assert client.get_wallet_address().startswith("0x") + + def test_init_missing_key_raises_error(self, monkeypatch, tmp_path): + """Should raise ValueError when no wallet configured.""" + monkeypatch.delenv("BLOCKRUN_WALLET_KEY", raising=False) + monkeypatch.delenv("BASE_CHAIN_WALLET_KEY", raising=False) + # Mock load_wallet to return None (no session file) + monkeypatch.setattr("blockrun_llm.wallet.load_wallet", lambda: None) + # Should raise ValueError with helpful message + with pytest.raises(ValueError, match="No wallet configured"): + LLMClient(private_key=None) + + def test_init_invalid_key_format(self): + """Should raise ValueError for invalid key format (after 0x normalization).""" + # "invalid" becomes "0xinvalid" after normalization, which is too short + with pytest.raises(ValueError, match="66 characters"): + LLMClient(private_key="invalid") + + def test_init_short_key(self): + """Should raise ValueError for short key.""" + with pytest.raises(ValueError, match="66 characters"): + LLMClient(private_key="0x123") + + def test_init_non_hex_key(self): + """Should raise ValueError for non-hex key.""" + with pytest.raises(ValueError, match="hexadecimal"): + LLMClient( + private_key="0xGGGG74bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" + ) + + def test_default_api_url(self): + """Should use default API URL.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + assert client.api_url == "https://blockrun.ai/api" + + def test_custom_api_url(self): + """Should accept custom API URL.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY, api_url="https://custom.example.com") + assert client.api_url == "https://custom.example.com" + + def test_invalid_api_url_http(self): + """Should reject HTTP for non-localhost.""" + with pytest.raises(ValueError, match="HTTPS"): + LLMClient(private_key=TEST_PRIVATE_KEY, api_url="http://insecure.com") + + def test_allow_localhost_http(self): + """Should allow HTTP for localhost.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY, api_url="http://localhost:3000") + assert client.api_url == "http://localhost:3000" + + +class TestLLMClientMethods: + def test_get_wallet_address(self): + """Should return valid Ethereum address.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + address = client.get_wallet_address() + + assert address == "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266" + assert address.startswith("0x") + assert len(address) == 42 + + @patch("blockrun_llm.client.httpx.Client") + def test_list_models(self, mock_client_class): + """Should list available models.""" + mock_client = Mock() + mock_client_class.return_value = mock_client + + mock_response = MockResponse(200, build_models_response()) + mock_client.get.return_value = mock_response + + client = LLMClient(private_key=TEST_PRIVATE_KEY) + models = client.list_models() + + assert len(models) == 3 + assert models[0]["id"] == "openai/gpt-5.2" + assert models[0]["provider"] == "openai" + + @patch("blockrun_llm.client.httpx.Client") + def test_list_models_error(self, mock_client_class): + """Should raise APIError on failure.""" + mock_client = Mock() + mock_client_class.return_value = mock_client + + mock_response = MockResponse(500) + mock_client.get.return_value = mock_response + + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + with pytest.raises(APIError): + client.list_models() + + +class TestErrorSanitization: + @patch("blockrun_llm.client.httpx.Client") + def test_sanitize_error_responses(self, mock_client_class): + """Should sanitize error responses.""" + mock_client = Mock() + mock_client_class.return_value = mock_client + + raw_error = build_error_response(error="Invalid model", include_sensitive=True) + mock_response = MockResponse(400, raw_error) + mock_client.get.return_value = mock_response + + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + try: + client.list_models() + pytest.fail("Should have raised APIError") + except APIError as e: + # Should only contain safe fields + assert e.response == {"message": "Invalid model", "code": "test_error"} + + # Should NOT contain sensitive fields + assert "internal_stack" not in e.response + assert "api_key" not in e.response + assert "database_url" not in e.response + + +class TestInputValidation: + @patch("blockrun_llm.client.httpx.Client") + def test_validate_model_parameter(self, mock_client_class): + """Should validate model parameter.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + with pytest.raises(ValueError, match="non-empty string"): + client.chat_completion("", [{"role": "user", "content": "test"}]) + + @patch("blockrun_llm.client.httpx.Client") + def test_validate_max_tokens(self, mock_client_class): + """Should validate max_tokens parameter.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + with pytest.raises(ValueError, match="positive"): + client.chat_completion("gpt-5.2", [{"role": "user", "content": "test"}], max_tokens=-1) + + # 200000 used to be rejected here. It no longer is, and must not be: + # gpt-5.2 advertises 128000 and zai/glm-5.2 serves 262144, so a bound + # below those made the SDK the binding constraint instead of the model. + # Only implausible values fail locally now; real ceilings go to the + # gateway, which rejects with the model's own number. + with pytest.raises(ValueError, match="implausibly large"): + client.chat_completion( + "gpt-5.2", [{"role": "user", "content": "test"}], max_tokens=2_000_000 + ) + + @patch("blockrun_llm.client.httpx.Client") + def test_validate_temperature(self, mock_client_class): + """Should validate temperature parameter.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + with pytest.raises(ValueError, match="between 0 and 2"): + client.chat_completion( + "gpt-5.2", [{"role": "user", "content": "test"}], temperature=3.0 + ) + + @patch("blockrun_llm.client.httpx.Client") + def test_validate_top_p(self, mock_client_class): + """Should validate top_p parameter.""" + client = LLMClient(private_key=TEST_PRIVATE_KEY) + + with pytest.raises(ValueError, match="between 0 and 1"): + client.chat_completion("gpt-5.2", [{"role": "user", "content": "test"}], top_p=1.5) diff --git a/tests/unit/test_validation.py b/tests/unit/test_validation.py index 7074b4f..967a9ac 100644 --- a/tests/unit/test_validation.py +++ b/tests/unit/test_validation.py @@ -1,346 +1,362 @@ -"""Unit tests for validation module.""" - -import pytest -from blockrun_llm.validation import ( - validate_private_key, - validate_api_url, - validate_model, - validate_max_tokens, - validate_temperature, - validate_top_p, - sanitize_error_response, - validate_resource_url, -) - - -class TestValidatePrivateKey: - def test_valid_private_key(self): - """Should accept valid private key.""" - key = "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" - validate_private_key(key) # Should not raise - - def test_reject_non_string(self): - """Should reject non-string input.""" - with pytest.raises(ValueError, match="must be a string"): - validate_private_key(123) # type: ignore - - def test_reject_no_prefix(self): - """Should reject key without 0x prefix.""" - with pytest.raises(ValueError, match="must start with 0x"): - validate_private_key("ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80") - - def test_reject_short_key(self): - """Should reject short key.""" - with pytest.raises(ValueError, match="66 characters"): - validate_private_key("0x123") - - def test_reject_long_key(self): - """Should reject long key.""" - with pytest.raises(ValueError, match="66 characters"): - validate_private_key( - "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80123" - ) - - def test_reject_non_hex(self): - """Should reject non-hexadecimal characters.""" - with pytest.raises(ValueError, match="hexadecimal"): - validate_private_key( - "0xGGGG74bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" - ) - - def test_accept_uppercase(self): - """Should accept uppercase hex.""" - key = "0xAC0974BEC39A17E36BA4A6B4D238FF944BACB478CBED5EFCAE784D7BF4F2FF80" - validate_private_key(key) # Should not raise - - def test_accept_mixed_case(self): - """Should accept mixed case hex.""" - key = "0xAc0974Bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" - validate_private_key(key) # Should not raise - - def test_reject_solana_base58_keypair_with_helpful_message(self): - """A 64-byte base58 Solana keypair should point users to SolanaLLMClient.""" - key = "3zZXZ37shyzxw7ZUePxwvJk8wkab8vPjHY6AWwE7CTJzZSP6zp8hnYNSsL6U4FgkacrbMhq2c1BZwgoKu17tdUa8" - with pytest.raises(ValueError, match="SolanaLLMClient"): - validate_private_key(key) - - def test_reject_solana_base58_seed_with_helpful_message(self): - """A 32-byte base58 Solana seed should point users to SolanaLLMClient.""" - key = "B5Fx69Nhu21vhFotKkFsURy554TqSo5ESN7ew4M6yjvH" - with pytest.raises(ValueError, match="SolanaLLMClient"): - validate_private_key(key) - - def test_reject_solana_base58_keypair_with_0x_prefix(self): - """Even after a caller prepends 0x, a Solana key should be detected.""" - key = "0x3zZXZ37shyzxw7ZUePxwvJk8wkab8vPjHY6AWwE7CTJzZSP6zp8hnYNSsL6U4FgkacrbMhq2c1BZwgoKu17tdUa8" - with pytest.raises(ValueError, match="Solana"): - validate_private_key(key) - - -class TestValidateApiUrl: - def test_accept_https(self): - """Should accept HTTPS URLs.""" - validate_api_url("https://api.blockrun.ai") - validate_api_url("https://example.com:8443") - - def test_accept_localhost_http(self): - """Should accept localhost HTTP.""" - validate_api_url("http://localhost") - validate_api_url("http://localhost:3000") - validate_api_url("http://127.0.0.1") - validate_api_url("http://127.0.0.1:8080") - - def test_reject_http_production(self): - """Should reject HTTP for non-localhost.""" - with pytest.raises(ValueError, match="HTTPS"): - validate_api_url("http://api.example.com") - with pytest.raises(ValueError, match="HTTPS"): - validate_api_url("http://192.168.1.1") - - def test_reject_invalid_url(self): - """Should reject invalid URL format.""" - with pytest.raises(ValueError, match="scheme"): - validate_api_url("not-a-url") - with pytest.raises(ValueError, match="scheme"): - validate_api_url("") - - -class TestValidateModel: - def test_accept_valid_model(self): - """Should accept valid model IDs.""" - validate_model("openai/gpt-5.2") - validate_model("anthropic/claude-sonnet-4.5") - validate_model("google/gemini-2.5-flash") - - def test_reject_empty_string(self): - """Should reject empty string.""" - with pytest.raises(ValueError, match="non-empty string"): - validate_model("") - - def test_reject_non_string(self): - """Should reject non-string.""" - with pytest.raises(ValueError, match="non-empty string"): - validate_model(None) # type: ignore - - -class TestValidateMaxTokens: - def test_accept_valid_values(self): - """Should accept valid max_tokens.""" - validate_max_tokens(1) - validate_max_tokens(100) - validate_max_tokens(1000) - validate_max_tokens(100000) - - def test_accept_none(self): - """Should accept None.""" - validate_max_tokens(None) - - def test_reject_negative(self): - """Should reject negative values.""" - with pytest.raises(ValueError, match="positive"): - validate_max_tokens(-1) - - def test_reject_zero(self): - """Should reject zero.""" - with pytest.raises(ValueError, match="positive"): - validate_max_tokens(0) - - def test_reject_too_large(self): - """Should reject values too large.""" - with pytest.raises(ValueError, match="too large"): - validate_max_tokens(200000) - - def test_reject_non_integer(self): - """Should reject non-integer.""" - with pytest.raises(ValueError, match="integer"): - validate_max_tokens(100.5) # type: ignore - - -class TestValidateTemperature: - def test_accept_valid_values(self): - """Should accept valid temperature.""" - validate_temperature(0.0) - validate_temperature(0.7) - validate_temperature(1.0) - validate_temperature(2.0) - - def test_accept_none(self): - """Should accept None.""" - validate_temperature(None) - - def test_reject_negative(self): - """Should reject negative values.""" - with pytest.raises(ValueError, match="between 0 and 2"): - validate_temperature(-0.1) - - def test_reject_too_large(self): - """Should reject values > 2.""" - with pytest.raises(ValueError, match="between 0 and 2"): - validate_temperature(2.1) - - def test_reject_non_number(self): - """Should reject non-numeric.""" - with pytest.raises(ValueError, match="number"): - validate_temperature("0.7") # type: ignore - - -class TestValidateTopP: - def test_accept_valid_values(self): - """Should accept valid top_p.""" - validate_top_p(0.0) - validate_top_p(0.5) - validate_top_p(0.9) - validate_top_p(1.0) - - def test_accept_none(self): - """Should accept None.""" - validate_top_p(None) - - def test_reject_negative(self): - """Should reject negative values.""" - with pytest.raises(ValueError, match="between 0 and 1"): - validate_top_p(-0.1) - - def test_reject_too_large(self): - """Should reject values > 1.""" - with pytest.raises(ValueError, match="between 0 and 1"): - validate_top_p(1.1) - - def test_reject_non_number(self): - """Should reject non-numeric.""" - with pytest.raises(ValueError, match="number"): - validate_top_p("0.9") # type: ignore - - -class TestSanitizeErrorResponse: - def test_extract_safe_fields(self): - """Should extract only safe error fields.""" - result = sanitize_error_response( - { - "error": "User-facing error", - "internal_stack": "/var/app/sensitive.py:123", - "api_key": "sk-secret", - "database_url": "postgres://user:pass@host/db", - } - ) - assert result == {"message": "User-facing error", "code": None} - - def test_include_code_if_present(self): - """Should include code if present.""" - result = sanitize_error_response( - {"error": "Invalid request", "code": "invalid_request_error"} - ) - assert result == {"message": "Invalid request", "code": "invalid_request_error"} - - def test_handle_non_dict(self): - """Should handle non-dict input.""" - assert sanitize_error_response("error") == { - "message": "API request failed", - "code": None, - } - assert sanitize_error_response(None) == { - "message": "API request failed", - "code": None, - } - assert sanitize_error_response(123) == { - "message": "API request failed", - "code": None, - } - - def test_handle_missing_error_field(self): - """Should handle missing error field.""" - result = sanitize_error_response({"something": "else"}) - assert result == {"message": "API request failed", "code": None} - - def test_nested_openai_error_shape(self): - """Should pass through the gateway's OpenAI-compatible nested error.""" - result = sanitize_error_response( - { - "error": { - "message": "Conversation too long — Message @bc1max on Telegram", - "type": "invalid_request_error", - "code": "CONTEXT_LENGTH_EXCEEDED", - "param": None, - }, - "message": "Message @bc1max on Telegram", - "code": "CONTEXT_LENGTH_EXCEEDED", - "debug": "/var/app/handler.py:123 SECRET_KEY=xyz", - } - ) - assert result["message"] == "Conversation too long — Message @bc1max on Telegram" - assert result["code"] == "CONTEXT_LENGTH_EXCEEDED" - assert result["type"] == "invalid_request_error" - # Raw upstream debug text must never be surfaced. - assert "debug" not in result - - def test_nested_error_falls_back_to_top_level_code(self): - """Should use top-level code when the nested object omits it.""" - result = sanitize_error_response( - { - "error": {"message": "Rate limited", "type": "rate_limit_error"}, - "code": "RATE_LIMITED", - } - ) - assert result == { - "message": "Rate limited", - "code": "RATE_LIMITED", - "type": "rate_limit_error", - } - - def test_nested_error_passes_param(self): - """Should pass through the OpenAI `param` field when present.""" - result = sanitize_error_response( - { - "error": { - "message": "Set stream: false", - "type": "invalid_request_error", - "code": "STREAM_UNSUPPORTED", - "param": "stream", - } - } - ) - assert result["param"] == "stream" - - def test_flat_string_error_still_supported(self): - """Should keep supporting the legacy flat string `error` shape.""" - result = sanitize_error_response({"error": "Unknown model: foo. Available models: gpt-5.2"}) - assert result == { - "message": "Unknown model: foo. Available models: gpt-5.2", - "code": None, - } - - -class TestValidateResourceUrl: - def test_allow_matching_domain(self): - """Should allow matching domain.""" - result = validate_resource_url("https://api.blockrun.ai/v1/chat", "https://api.blockrun.ai") - assert result == "https://api.blockrun.ai/v1/chat" - - def test_allow_different_path(self): - """Should allow different path on same domain.""" - result = validate_resource_url( - "https://api.blockrun.ai/v2/models", "https://api.blockrun.ai" - ) - assert result == "https://api.blockrun.ai/v2/models" - - def test_reject_different_domain(self): - """Should reject different domain.""" - result = validate_resource_url("https://malicious.com/steal", "https://api.blockrun.ai") - assert result == "https://api.blockrun.ai/v1/chat/completions" - - def test_reject_different_protocol(self): - """Should reject different protocol.""" - result = validate_resource_url("http://api.blockrun.ai/v1/chat", "https://api.blockrun.ai") - assert result == "https://api.blockrun.ai/v1/chat/completions" - - def test_handle_invalid_url(self): - """Should handle invalid URL format.""" - result = validate_resource_url("not-a-url", "https://api.blockrun.ai") - assert result == "https://api.blockrun.ai/v1/chat/completions" - - def test_reject_subdomain_difference(self): - """Should reject subdomain differences.""" - result = validate_resource_url( - "https://evil.api.blockrun.ai/v1/chat", "https://api.blockrun.ai" - ) - assert result == "https://api.blockrun.ai/v1/chat/completions" +"""Unit tests for validation module.""" + +import pytest +from blockrun_llm.validation import ( + validate_private_key, + validate_api_url, + validate_model, + validate_max_tokens, + validate_temperature, + validate_top_p, + sanitize_error_response, + validate_resource_url, +) + + +class TestValidatePrivateKey: + def test_valid_private_key(self): + """Should accept valid private key.""" + key = "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" + validate_private_key(key) # Should not raise + + def test_reject_non_string(self): + """Should reject non-string input.""" + with pytest.raises(ValueError, match="must be a string"): + validate_private_key(123) # type: ignore + + def test_reject_no_prefix(self): + """Should reject key without 0x prefix.""" + with pytest.raises(ValueError, match="must start with 0x"): + validate_private_key("ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80") + + def test_reject_short_key(self): + """Should reject short key.""" + with pytest.raises(ValueError, match="66 characters"): + validate_private_key("0x123") + + def test_reject_long_key(self): + """Should reject long key.""" + with pytest.raises(ValueError, match="66 characters"): + validate_private_key( + "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80123" + ) + + def test_reject_non_hex(self): + """Should reject non-hexadecimal characters.""" + with pytest.raises(ValueError, match="hexadecimal"): + validate_private_key( + "0xGGGG74bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" + ) + + def test_accept_uppercase(self): + """Should accept uppercase hex.""" + key = "0xAC0974BEC39A17E36BA4A6B4D238FF944BACB478CBED5EFCAE784D7BF4F2FF80" + validate_private_key(key) # Should not raise + + def test_accept_mixed_case(self): + """Should accept mixed case hex.""" + key = "0xAc0974Bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" + validate_private_key(key) # Should not raise + + def test_reject_solana_base58_keypair_with_helpful_message(self): + """A 64-byte base58 Solana keypair should point users to SolanaLLMClient.""" + key = "3zZXZ37shyzxw7ZUePxwvJk8wkab8vPjHY6AWwE7CTJzZSP6zp8hnYNSsL6U4FgkacrbMhq2c1BZwgoKu17tdUa8" + with pytest.raises(ValueError, match="SolanaLLMClient"): + validate_private_key(key) + + def test_reject_solana_base58_seed_with_helpful_message(self): + """A 32-byte base58 Solana seed should point users to SolanaLLMClient.""" + key = "B5Fx69Nhu21vhFotKkFsURy554TqSo5ESN7ew4M6yjvH" + with pytest.raises(ValueError, match="SolanaLLMClient"): + validate_private_key(key) + + def test_reject_solana_base58_keypair_with_0x_prefix(self): + """Even after a caller prepends 0x, a Solana key should be detected.""" + key = "0x3zZXZ37shyzxw7ZUePxwvJk8wkab8vPjHY6AWwE7CTJzZSP6zp8hnYNSsL6U4FgkacrbMhq2c1BZwgoKu17tdUa8" + with pytest.raises(ValueError, match="Solana"): + validate_private_key(key) + + +class TestValidateApiUrl: + def test_accept_https(self): + """Should accept HTTPS URLs.""" + validate_api_url("https://api.blockrun.ai") + validate_api_url("https://example.com:8443") + + def test_accept_localhost_http(self): + """Should accept localhost HTTP.""" + validate_api_url("http://localhost") + validate_api_url("http://localhost:3000") + validate_api_url("http://127.0.0.1") + validate_api_url("http://127.0.0.1:8080") + + def test_reject_http_production(self): + """Should reject HTTP for non-localhost.""" + with pytest.raises(ValueError, match="HTTPS"): + validate_api_url("http://api.example.com") + with pytest.raises(ValueError, match="HTTPS"): + validate_api_url("http://192.168.1.1") + + def test_reject_invalid_url(self): + """Should reject invalid URL format.""" + with pytest.raises(ValueError, match="scheme"): + validate_api_url("not-a-url") + with pytest.raises(ValueError, match="scheme"): + validate_api_url("") + + +class TestValidateModel: + def test_accept_valid_model(self): + """Should accept valid model IDs.""" + validate_model("openai/gpt-5.2") + validate_model("anthropic/claude-sonnet-4.5") + validate_model("google/gemini-2.5-flash") + + def test_reject_empty_string(self): + """Should reject empty string.""" + with pytest.raises(ValueError, match="non-empty string"): + validate_model("") + + def test_reject_non_string(self): + """Should reject non-string.""" + with pytest.raises(ValueError, match="non-empty string"): + validate_model(None) # type: ignore + + +class TestValidateMaxTokens: + def test_accept_valid_values(self): + """Should accept valid max_tokens.""" + validate_max_tokens(1) + validate_max_tokens(100) + validate_max_tokens(1000) + validate_max_tokens(100000) + # Real ceilings the gateway serves — these were rejected before the + # sanity bound was raised, despite every provider accepting them. + validate_max_tokens(128000) # opus-4.8 / sonnet-5 / gpt-5.6 / glm-5 + validate_max_tokens(262144) # zai/glm-5.2 + + def test_accept_none(self): + """Should accept None.""" + validate_max_tokens(None) + + def test_reject_negative(self): + """Should reject negative values.""" + with pytest.raises(ValueError, match="positive"): + validate_max_tokens(-1) + + def test_reject_zero(self): + """Should reject zero.""" + with pytest.raises(ValueError, match="positive"): + validate_max_tokens(0) + + def test_reject_implausible(self): + """Should reject values no model could mean (typo guard, not a limit).""" + with pytest.raises(ValueError, match="implausibly large"): + validate_max_tokens(2_000_000) + + def test_does_not_cap_below_real_ceilings(self): + """The bound must never be the binding constraint on a real request. + + Regression: this was 100000, which rejected every ceiling above it + client-side — the caller saw a ValueError naming a limit no provider + had set, and the request never reached the network. Probed against the + live gateway 2026-07-21: 19 models advertise more than 100000 and all + 19 accepted their advertised ceiling. + """ + for real_ceiling in (128_000, 262_144): + validate_max_tokens(real_ceiling) + + def test_reject_non_integer(self): + """Should reject non-integer.""" + with pytest.raises(ValueError, match="integer"): + validate_max_tokens(100.5) # type: ignore + + +class TestValidateTemperature: + def test_accept_valid_values(self): + """Should accept valid temperature.""" + validate_temperature(0.0) + validate_temperature(0.7) + validate_temperature(1.0) + validate_temperature(2.0) + + def test_accept_none(self): + """Should accept None.""" + validate_temperature(None) + + def test_reject_negative(self): + """Should reject negative values.""" + with pytest.raises(ValueError, match="between 0 and 2"): + validate_temperature(-0.1) + + def test_reject_too_large(self): + """Should reject values > 2.""" + with pytest.raises(ValueError, match="between 0 and 2"): + validate_temperature(2.1) + + def test_reject_non_number(self): + """Should reject non-numeric.""" + with pytest.raises(ValueError, match="number"): + validate_temperature("0.7") # type: ignore + + +class TestValidateTopP: + def test_accept_valid_values(self): + """Should accept valid top_p.""" + validate_top_p(0.0) + validate_top_p(0.5) + validate_top_p(0.9) + validate_top_p(1.0) + + def test_accept_none(self): + """Should accept None.""" + validate_top_p(None) + + def test_reject_negative(self): + """Should reject negative values.""" + with pytest.raises(ValueError, match="between 0 and 1"): + validate_top_p(-0.1) + + def test_reject_too_large(self): + """Should reject values > 1.""" + with pytest.raises(ValueError, match="between 0 and 1"): + validate_top_p(1.1) + + def test_reject_non_number(self): + """Should reject non-numeric.""" + with pytest.raises(ValueError, match="number"): + validate_top_p("0.9") # type: ignore + + +class TestSanitizeErrorResponse: + def test_extract_safe_fields(self): + """Should extract only safe error fields.""" + result = sanitize_error_response( + { + "error": "User-facing error", + "internal_stack": "/var/app/sensitive.py:123", + "api_key": "sk-secret", + "database_url": "postgres://user:pass@host/db", + } + ) + assert result == {"message": "User-facing error", "code": None} + + def test_include_code_if_present(self): + """Should include code if present.""" + result = sanitize_error_response( + {"error": "Invalid request", "code": "invalid_request_error"} + ) + assert result == {"message": "Invalid request", "code": "invalid_request_error"} + + def test_handle_non_dict(self): + """Should handle non-dict input.""" + assert sanitize_error_response("error") == { + "message": "API request failed", + "code": None, + } + assert sanitize_error_response(None) == { + "message": "API request failed", + "code": None, + } + assert sanitize_error_response(123) == { + "message": "API request failed", + "code": None, + } + + def test_handle_missing_error_field(self): + """Should handle missing error field.""" + result = sanitize_error_response({"something": "else"}) + assert result == {"message": "API request failed", "code": None} + + def test_nested_openai_error_shape(self): + """Should pass through the gateway's OpenAI-compatible nested error.""" + result = sanitize_error_response( + { + "error": { + "message": "Conversation too long — Message @bc1max on Telegram", + "type": "invalid_request_error", + "code": "CONTEXT_LENGTH_EXCEEDED", + "param": None, + }, + "message": "Message @bc1max on Telegram", + "code": "CONTEXT_LENGTH_EXCEEDED", + "debug": "/var/app/handler.py:123 SECRET_KEY=xyz", + } + ) + assert result["message"] == "Conversation too long — Message @bc1max on Telegram" + assert result["code"] == "CONTEXT_LENGTH_EXCEEDED" + assert result["type"] == "invalid_request_error" + # Raw upstream debug text must never be surfaced. + assert "debug" not in result + + def test_nested_error_falls_back_to_top_level_code(self): + """Should use top-level code when the nested object omits it.""" + result = sanitize_error_response( + { + "error": {"message": "Rate limited", "type": "rate_limit_error"}, + "code": "RATE_LIMITED", + } + ) + assert result == { + "message": "Rate limited", + "code": "RATE_LIMITED", + "type": "rate_limit_error", + } + + def test_nested_error_passes_param(self): + """Should pass through the OpenAI `param` field when present.""" + result = sanitize_error_response( + { + "error": { + "message": "Set stream: false", + "type": "invalid_request_error", + "code": "STREAM_UNSUPPORTED", + "param": "stream", + } + } + ) + assert result["param"] == "stream" + + def test_flat_string_error_still_supported(self): + """Should keep supporting the legacy flat string `error` shape.""" + result = sanitize_error_response({"error": "Unknown model: foo. Available models: gpt-5.2"}) + assert result == { + "message": "Unknown model: foo. Available models: gpt-5.2", + "code": None, + } + + +class TestValidateResourceUrl: + def test_allow_matching_domain(self): + """Should allow matching domain.""" + result = validate_resource_url("https://api.blockrun.ai/v1/chat", "https://api.blockrun.ai") + assert result == "https://api.blockrun.ai/v1/chat" + + def test_allow_different_path(self): + """Should allow different path on same domain.""" + result = validate_resource_url( + "https://api.blockrun.ai/v2/models", "https://api.blockrun.ai" + ) + assert result == "https://api.blockrun.ai/v2/models" + + def test_reject_different_domain(self): + """Should reject different domain.""" + result = validate_resource_url("https://malicious.com/steal", "https://api.blockrun.ai") + assert result == "https://api.blockrun.ai/v1/chat/completions" + + def test_reject_different_protocol(self): + """Should reject different protocol.""" + result = validate_resource_url("http://api.blockrun.ai/v1/chat", "https://api.blockrun.ai") + assert result == "https://api.blockrun.ai/v1/chat/completions" + + def test_handle_invalid_url(self): + """Should handle invalid URL format.""" + result = validate_resource_url("not-a-url", "https://api.blockrun.ai") + assert result == "https://api.blockrun.ai/v1/chat/completions" + + def test_reject_subdomain_difference(self): + """Should reject subdomain differences.""" + result = validate_resource_url( + "https://evil.api.blockrun.ai/v1/chat", "https://api.blockrun.ai" + ) + assert result == "https://api.blockrun.ai/v1/chat/completions" From 51ca63477193ded13419985e79ce393abdcfd312 Mon Sep 17 00:00:00 2001 From: 1bcMax Date: Tue, 21 Jul 2026 00:12:51 -0500 Subject: [PATCH 2/2] style: black-format the max_tokens validation tests CI runs black --check; the new test file was not formatted. --- tests/unit/test_validation.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_validation.py b/tests/unit/test_validation.py index 967a9ac..02933b9 100644 --- a/tests/unit/test_validation.py +++ b/tests/unit/test_validation.py @@ -132,8 +132,8 @@ def test_accept_valid_values(self): validate_max_tokens(100000) # Real ceilings the gateway serves — these were rejected before the # sanity bound was raised, despite every provider accepting them. - validate_max_tokens(128000) # opus-4.8 / sonnet-5 / gpt-5.6 / glm-5 - validate_max_tokens(262144) # zai/glm-5.2 + validate_max_tokens(128000) # opus-4.8 / sonnet-5 / gpt-5.6 / glm-5 + validate_max_tokens(262144) # zai/glm-5.2 def test_accept_none(self): """Should accept None."""