diff --git a/blockrun_llm/solana_client.py b/blockrun_llm/solana_client.py index 48f11c3..c6543c3 100644 --- a/blockrun_llm/solana_client.py +++ b/blockrun_llm/solana_client.py @@ -20,6 +20,7 @@ import asyncio import json as _json import os +import re import sys import threading from typing import Any, Dict, Iterator, List, Optional, Tuple, Union @@ -53,6 +54,8 @@ ) from .solana_wallet import get_solana_public_key from .tx_log import TransactionLogger, decode_settlement_header, _resolve_log_dir +from .price import Category, Market, Resolution, Session +from .realface import _GROUP_ID_RE from .validation import ( build_payment_rejected_error, sanitize_error_response, @@ -320,6 +323,46 @@ def _should_fallback_solana(exc: Exception) -> bool: return False +# Characters safe to interpolate into a single URL path segment. network / +# symbol / market / wallet address all get f-string'd into a paid endpoint +# path; a '/', '..', '?' or '#' would silently re-target the payment-signing +# request. These values often come from LLM output in agent use, so validate +# before building the URL. +_SAFE_PATH_SEGMENT_RE = re.compile(r"^[A-Za-z0-9._-]+$") + + +def _safe_path_segment(value: str, field: str) -> str: + """Return ``value`` if it is a single safe URL path segment, else raise.""" + if not value or not _SAFE_PATH_SEGMENT_RE.match(value): + raise ValueError( + f"{field} must contain only letters, digits, '.', '_' or '-' " f"(got {value!r})" + ) + return value + + +def _receipt_from_headers(headers: Any) -> Optional[str]: + """Pull the x402 settlement tx hash from a paid response's headers.""" + if headers is None: + return None + return headers.get("x-payment-receipt") or headers.get("X-Payment-Receipt") + + +def _assert_same_payment_terms(signed_payload: Any, orig_amount: Any, orig_pay_to: Any) -> None: + """Guard a mid-poll re-sign: the fresh 402 challenge must charge the same + amount to the same recipient as the payment originally authorized for this + job. A gateway (buggy or hostile) that reprices or redirects the re-challenge + would otherwise extract an unbounded, unrelated payment from the wallet. + Raises :class:`PaymentError` on any mismatch so no signature is submitted.""" + accepted = signed_payload.accepted + if str(accepted.amount) != str(orig_amount) or accepted.pay_to != orig_pay_to: + raise PaymentError( + "Mid-poll re-sign challenge changed the payment terms " + f"(amount {orig_amount!r} -> {accepted.amount!r}, " + f"pay_to {orig_pay_to!r} -> {accepted.pay_to!r}); refusing to " + "authorize a different payment for the same job." + ) + + class SolanaLLMClient: """ BlockRun LLM Client for Solana — pays via Solana USDC x402. @@ -346,7 +389,10 @@ class SolanaLLMClient: VIDEO_DEFAULT_MODEL = "xai/grok-imagine-video" VIDEO_POLL_INTERVAL_SECONDS = 5.0 VIDEO_POLL_BUDGET_SECONDS = 900.0 - MEDIA_POLL_MAX_RESIGNS = 3 + # Matches Base VideoClient.MAX_POLL_RESIGNS (2) — each re-sign is only used + # to refresh an expired blockhash, and every fresh signature is validated + # against the original payment terms before use. + MEDIA_POLL_MAX_RESIGNS = 2 # Media generation defaults (mirror the Base MusicClient/SpeechClient). MUSIC_DEFAULT_MODEL = "minimax/music-2.5+" @@ -433,6 +479,11 @@ def __init__( TransactionLogger(log_dir) if log_dir is not None else None ) self._last_settlement: Optional[Dict[str, Any]] = None + # Response headers from the most recent raw paid POST — consumed by + # rpc()/music()/speech() to surface the settlement receipt + gateway + # metadata the shared JSON-only helper would otherwise drop. Read it + # immediately after the helper returns (no intervening await). + self._last_raw_headers: Optional[httpx.Headers] = None # Initialize x402 SDK client for Solana payment signing. self._x402_client = x402ClientSync() @@ -480,6 +531,14 @@ def _capture_settlement(self, response: httpx.Response) -> Optional[Dict[str, An self._last_settlement = settlement return settlement + def _attach_receipt(self, data: Any) -> None: + """Inject the settlement tx hash from the most recent paid POST into a + raw response dict under ``txHash`` (mirrors the Base Music/Speech + clients). No-op on free responses (no receipt header).""" + tx_hash = _receipt_from_headers(self._last_raw_headers) + if tx_hash and isinstance(data, dict) and not data.get("txHash"): + data["txHash"] = tx_hash + def get_wallet_address(self) -> str: if not self._address: self._address = get_solana_public_key(self._private_key) @@ -1117,6 +1176,10 @@ def _request_with_payment_raw( if cached is not None: return cached + # Reset per-call receipt headers; only a paid retry repopulates them, so + # a free/cached model can't inherit a prior call's settlement receipt. + self._last_raw_headers = None + url = f"{self._api_url}{endpoint}" headers = {"Content-Type": "application/json", "User-Agent": _get_user_agent()} eff_timeout = timeout if timeout is not None else self._timeout @@ -1210,6 +1273,7 @@ def _handle_payment_and_retry_raw( self._session_total_usd += cost_usd self._last_call_cost = cost_usd self._capture_settlement(retry_response) + self._last_raw_headers = retry_response.headers return retry_response.json() @@ -1316,6 +1380,7 @@ def _handle_get_payment_and_retry( self._session_total_usd += cost_usd self._last_call_cost = cost_usd self._capture_settlement(retry_response) + self._last_raw_headers = retry_response.headers return retry_response.json() @@ -1413,6 +1478,9 @@ def _request_image_with_payment( payment_payload_obj = self._sign_payment(payment_required) encoded_payment = encode_payment_signature_header(payment_payload_obj) cost_usd = float(payment_payload_obj.accepted.amount) / 1e6 + # Terms this job is authorized to pay — any mid-poll re-sign must match. + orig_amount = payment_payload_obj.accepted.amount + orig_pay_to = payment_payload_obj.accepted.pay_to paid_headers = { "Content-Type": "application/json", @@ -1510,17 +1578,33 @@ def _request_image_with_payment( # fresh signature that 402s again is a genuine payment problem. if resigns_left > 0: resigns_left -= 1 - challenge = self._client.get( - poll_url, - headers={"User-Agent": _get_user_agent()}, - timeout=eff_timeout, - ) - resign_header = self._extract_payment_header(challenge) - if challenge.status_code == 402 and resign_header: - resign_required = decode_payment_required_header(resign_header) - resign_payload = self._sign_payment(resign_required) - encoded_payment = encode_payment_signature_header(resign_payload) - poll_headers["PAYMENT-SIGNATURE"] = encoded_payment + resign_payload = None + try: + challenge = self._client.get( + poll_url, + headers={"User-Agent": _get_user_agent()}, + timeout=eff_timeout, + ) + resign_header = self._extract_payment_header(challenge) + if challenge.status_code == 402 and resign_header: + resign_required = decode_payment_required_header(resign_header) + resign_payload = self._sign_payment(resign_required) + except (PaymentError, httpx.HTTPError): + # Challenge GET failed, or signing was rejected — fall + # through to surface the gateway's real 402 reason rather + # than masking it with a network/signing error. Nothing + # settled here. + resign_payload = None + if resign_payload is not None: + # Refuse a re-challenge that reprices or redirects the + # payment vs. what this job originally authorized. This + # PaymentError must propagate (NOT fall through to the + # generic 402). The guard also pins the amount, so the + # submit-time cost_usd stays correct for the ledger. + _assert_same_payment_terms(resign_payload, orig_amount, orig_pay_to) + poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( + resign_payload + ) continue raise build_payment_rejected_error(poll_resp) @@ -1671,60 +1755,21 @@ def video( **no payment** and leaves the job claimable ~48h. Default model is ``xai/grok-imagine-video``. """ - if image_url and real_face_asset_id: - raise ValueError( - "image_url and real_face_asset_id are mutually exclusive; pass at most one." - ) - if last_frame_url and not image_url: - raise ValueError( - "last_frame_url requires image_url: image_url seeds the FIRST frame and " - "last_frame_url the FINAL frame — send both." - ) - if last_frame_url and real_face_asset_id: - raise ValueError( - "last_frame_url and real_face_asset_id are mutually exclusive; " - "first-and-last-frame uses image_url + last_frame_url." - ) - if reference_image_urls: - if image_url or last_frame_url or real_face_asset_id: - raise ValueError( - "reference_image_urls is mutually exclusive with image_url, " - "last_frame_url, and real_face_asset_id." - ) - if len(reference_image_urls) > 9: - raise ValueError("reference_image_urls accepts at most 9 images.") - if real_face_asset_id is not None and not real_face_asset_id.startswith("ta_"): - raise ValueError( - "real_face_asset_id must start with 'ta_' " - "(a Virtual Portrait or RealFace asset id, e.g. 'ta_abc123xyz')" - ) - - body: Dict[str, Any] = { - "model": model or self.VIDEO_DEFAULT_MODEL, - "prompt": prompt, - } - if image_url: - body["image_url"] = image_url - if last_frame_url: - body["last_frame_url"] = last_frame_url - if reference_image_urls: - body["reference_image_urls"] = reference_image_urls - if real_face_asset_id: - body["real_face_asset_id"] = real_face_asset_id - if duration_seconds is not None: - body["duration_seconds"] = duration_seconds - if aspect_ratio is not None: - body["aspect_ratio"] = aspect_ratio - if resolution is not None: - body["resolution"] = resolution - if generate_audio is not None: - body["generate_audio"] = generate_audio - if seed is not None: - body["seed"] = seed - if watermark is not None: - body["watermark"] = watermark - if return_last_frame is not None: - body["return_last_frame"] = return_last_frame + body = self._build_video_body( + prompt, + model=model, + image_url=image_url, + last_frame_url=last_frame_url, + reference_image_urls=reference_image_urls, + real_face_asset_id=real_face_asset_id, + duration_seconds=duration_seconds, + aspect_ratio=aspect_ratio, + resolution=resolution, + generate_audio=generate_audio, + seed=seed, + watermark=watermark, + return_last_frame=return_last_frame, + ) data = self._request_image_with_payment( "/v1/videos/generations", @@ -1800,6 +1845,7 @@ def music( if lyrics and lyrics.strip(): body["lyrics"] = lyrics.strip() data = self._request_with_payment_raw("/v1/audio/generations", body, timeout=timeout) + self._attach_receipt(data) return MusicResponse(**data) # ------------------------------------------------------------------ @@ -1833,6 +1879,7 @@ def speech( if speed is not None: body["speed"] = speed data = self._request_with_payment_raw("/v1/audio/speech", body, timeout=timeout) + self._attach_receipt(data) return SpeechResponse(**data) def sound_effect( @@ -1858,12 +1905,15 @@ def sound_effect( if response_format: body["response_format"] = response_format data = self._request_with_payment_raw("/v1/audio/sound-effects", body, timeout=timeout) + self._attach_receipt(data) return SpeechResponse(**data) def list_voices(self) -> List[Dict[str, Any]]: """List available speech voices (free).""" url = f"{self._api_url}/v1/audio/voices" - resp = self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -1875,7 +1925,8 @@ def list_voices(self) -> List[Dict[str, Any]]: sanitize_error_response(error_body), ) data = resp.json() - return data.get("voices", data) if isinstance(data, dict) else data + # Gateway wraps the voice list under "data" (mirrors SpeechClient.list_voices). + return data.get("data", []) if isinstance(data, dict) else data # ------------------------------------------------------------------ # Virtual Portrait enrollment (Solana payment) @@ -1897,9 +1948,11 @@ def portrait_enroll(self, name: str, image_url: str) -> PortraitEnrollment: def list_portraits(self, wallet_address: Optional[str] = None) -> PortraitList: """List Virtual Portraits enrolled by a wallet (free, rate-limited).""" - addr = wallet_address or self.get_wallet_address() + addr = _safe_path_segment(wallet_address or self.get_wallet_address(), "wallet_address") url = f"{self._api_url}/v1/wallet/{addr}/portraits" - resp = self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -1924,6 +1977,8 @@ def realface_init(self, name: str, group_id: Optional[str] = None) -> RealFaceIn raise ValueError("name is required (1-64 chars)") if len(name) > 64: raise ValueError(f"name must be 64 chars or fewer (got {len(name)})") + if group_id is not None and not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") body: Dict[str, Any] = {"name": name} if group_id: body["groupId"] = group_id @@ -1932,6 +1987,7 @@ def realface_init(self, name: str, group_id: Optional[str] = None) -> RealFaceIn url, json=body, headers={"Content-Type": "application/json", "User-Agent": _get_user_agent()}, + timeout=DEFAULT_FAST_TIMEOUT, ) if resp.status_code != 200: try: @@ -1945,11 +2001,14 @@ def realface_init(self, name: str, group_id: Optional[str] = None) -> RealFaceIn def realface_status(self, group_id: str) -> RealFaceStatus: """Poll a RealFace group's state (free, rate-limited).""" - if not group_id: - raise ValueError("group_id is required") + if not group_id or not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") url = f"{self._api_url}/v1/realface/status" resp = self._client.get( - url, params={"groupId": group_id}, headers={"User-Agent": _get_user_agent()} + url, + params={"groupId": group_id}, + headers={"User-Agent": _get_user_agent()}, + timeout=DEFAULT_FAST_TIMEOUT, ) if resp.status_code != 200: try: @@ -1996,17 +2055,19 @@ def realface_enroll(self, name: str, image_url: str, group_id: str) -> RealFaceE raise ValueError(f"name must be 64 chars or fewer (got {len(name)})") if not image_url or not image_url.lower().startswith(("https://", "http://")): raise ValueError("image_url must be an http(s) URL") - if not group_id: - raise ValueError("group_id is required") + if not group_id or not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") body: Dict[str, Any] = {"name": name, "image_url": image_url, "group_id": group_id} data = self._request_with_payment_raw("/v1/realface/enroll", body) return RealFaceEnrollment(**data) def list_realfaces(self, wallet_address: Optional[str] = None) -> RealFaceList: """List RealFace assets enrolled by a wallet (free, rate-limited).""" - addr = wallet_address or self.get_wallet_address() + addr = _safe_path_segment(wallet_address or self.get_wallet_address(), "wallet_address") url = f"{self._api_url}/v1/wallet/{addr}/realfaces" - resp = self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -2023,6 +2084,101 @@ def list_realfaces(self, wallet_address: Optional[str] = None) -> RealFaceList: # Pyth market data (Solana payment for paid categories) # ------------------------------------------------------------------ + @staticmethod + def _build_video_body( + prompt: str, + *, + model: Optional[str], + image_url: Optional[str], + last_frame_url: Optional[str], + reference_image_urls: Optional[List[str]], + real_face_asset_id: Optional[str], + duration_seconds: Optional[int], + aspect_ratio: Optional[str], + resolution: Optional[str], + generate_audio: Optional[bool], + seed: Optional[int], + watermark: Optional[bool], + return_last_frame: Optional[bool], + ) -> Dict[str, Any]: + """Validate video kwargs and build the request body. Shared by the sync + and async ``video()`` so their validation and payload never drift.""" + if image_url and real_face_asset_id: + raise ValueError( + "image_url and real_face_asset_id are mutually exclusive; pass at most one." + ) + if last_frame_url and not image_url: + raise ValueError( + "last_frame_url requires image_url: image_url seeds the FIRST frame and " + "last_frame_url the FINAL frame — send both." + ) + if last_frame_url and real_face_asset_id: + raise ValueError( + "last_frame_url and real_face_asset_id are mutually exclusive; " + "first-and-last-frame uses image_url + last_frame_url." + ) + if reference_image_urls: + if image_url or last_frame_url or real_face_asset_id: + raise ValueError( + "reference_image_urls is mutually exclusive with image_url, " + "last_frame_url, and real_face_asset_id." + ) + if len(reference_image_urls) > 9: + raise ValueError("reference_image_urls accepts at most 9 images.") + if real_face_asset_id is not None and not real_face_asset_id.startswith("ta_"): + raise ValueError( + "real_face_asset_id must start with 'ta_' " + "(a Virtual Portrait or RealFace asset id, e.g. 'ta_abc123xyz')" + ) + + body: Dict[str, Any] = { + "model": model or SolanaLLMClient.VIDEO_DEFAULT_MODEL, + "prompt": prompt, + } + if image_url: + body["image_url"] = image_url + if last_frame_url: + body["last_frame_url"] = last_frame_url + if reference_image_urls: + body["reference_image_urls"] = reference_image_urls + if real_face_asset_id: + body["real_face_asset_id"] = real_face_asset_id + if duration_seconds is not None: + body["duration_seconds"] = duration_seconds + if aspect_ratio is not None: + body["aspect_ratio"] = aspect_ratio + if resolution is not None: + body["resolution"] = resolution + if generate_audio is not None: + body["generate_audio"] = generate_audio + if seed is not None: + body["seed"] = seed + if watermark is not None: + body["watermark"] = watermark + if return_last_frame is not None: + body["return_last_frame"] = return_last_frame + return body + + @staticmethod + def _rpc_response( + data: Any, headers: Optional[httpx.Headers], fallback_network: str + ) -> RpcResponse: + """Build an RpcResponse, surfacing gateway metadata from the paid + response headers (canonical network, cache hit, settlement tx) exactly + like the Base RPCClient. Strips body keys that would collide with those + metadata kwargs.""" + if not isinstance(data, dict): + data = {"result": data} + else: + data = {k: v for k, v in data.items() if k not in ("network", "cache_hit", "tx_hash")} + hdrs = headers if headers is not None else httpx.Headers() + return RpcResponse( + **data, + network=hdrs.get("x-network") or fallback_network, + cache_hit=(hdrs.get("x-cache", "") or "").upper() == "HIT", + tx_hash=_receipt_from_headers(hdrs), + ) + @staticmethod def _price_category_path( category: str, market: Optional[str], kind: str, symbol: Optional[str] @@ -2030,22 +2186,22 @@ def _price_category_path( if category == "stocks": if not market: raise ValueError("market is required for category='stocks' (e.g. market='us')") - base = f"/v1/stocks/{market}" + base = f"/v1/stocks/{_safe_path_segment(market, 'market')}" elif category in ("crypto", "fx", "commodity", "usstock"): base = f"/v1/{category}" else: raise ValueError(f"Unknown category: {category}") if symbol is None: return f"{base}/{kind}" - return f"{base}/{kind}/{symbol.upper()}" + return f"{base}/{kind}/{_safe_path_segment(symbol.upper(), 'symbol')}" def price( self, - category: str, + category: Category, symbol: str, *, - market: Optional[str] = None, - session: Optional[str] = None, + market: Optional[Market] = None, + session: Optional[Session] = None, ) -> PricePoint: """Fetch a realtime Pyth price quote (Solana payment for paid categories). ``market`` is required for ``category='stocks'``.""" @@ -2053,10 +2209,12 @@ def price( params: Dict[str, Any] = {} if session is not None: params["session"] = session - data = self._get_with_payment_raw(endpoint, params=params or None) + data = self._get_with_payment_raw( + endpoint, params=params or None, timeout=DEFAULT_FAST_TIMEOUT + ) return PricePoint( symbol=data.get("symbol", symbol.upper()), - price=data["price"], + price=data.get("price"), publish_time=data.get("publishTime"), confidence=data.get("confidence"), feed_id=data.get("feedId"), @@ -2069,21 +2227,21 @@ def price( def price_history( self, - category: str, + category: Category, symbol: str, *, - resolution: str = "D", + resolution: Resolution = "D", from_ts: int, to_ts: int, - market: Optional[str] = None, - session: Optional[str] = None, + market: Optional[Market] = None, + session: Optional[Session] = None, ) -> PriceHistoryResponse: """Fetch OHLC bars between two Unix timestamps (seconds).""" endpoint = self._price_category_path(category, market, "history", symbol) params: Dict[str, Any] = {"resolution": resolution, "from": from_ts, "to": to_ts} if session is not None: params["session"] = session - data = self._get_with_payment_raw(endpoint, params=params) + data = self._get_with_payment_raw(endpoint, params=params, timeout=DEFAULT_FAST_TIMEOUT) return PriceHistoryResponse( symbol=data.get("symbol", symbol.upper()), resolution=data.get("resolution", resolution), @@ -2093,18 +2251,18 @@ def price_history( def list_symbols( self, - category: str, + category: Category, *, q: Optional[str] = None, limit: int = 100, - market: Optional[str] = None, + market: Optional[Market] = None, ) -> SymbolListResponse: """List available symbols in a Pyth category (free discovery).""" endpoint = self._price_category_path(category, market, "list", None) params: Dict[str, Any] = {"limit": limit} if q: params["q"] = q - data = self._get_with_payment_raw(endpoint, params=params) + data = self._get_with_payment_raw(endpoint, params=params, timeout=DEFAULT_FAST_TIMEOUT) if isinstance(data, list): return SymbolListResponse(symbols=data, count=len(data)) return SymbolListResponse( @@ -2130,32 +2288,28 @@ def rpc( Mirrors ``RPCClient.call``. ``network`` may be a chain name or alias (``eth``, ``sol``, ``base`` …); the gateway resolves it. """ + _safe_path_segment(network, "network") body: Dict[str, Any] = {"jsonrpc": "2.0", "id": id, "method": method} if params is not None: body["params"] = params data = self._request_with_payment_raw(f"/v1/rpc/{network}", body) - if not isinstance(data, dict): - data = {"result": data} - return RpcResponse(**data, network=network) + return self._rpc_response(data, self._last_raw_headers, network) def rpc_batch(self, network: str, requests: List[Dict[str, Any]]) -> List[RpcResponse]: """Make a JSON-RPC 2.0 batch call (Solana payment, $0.002 x N).""" if not requests: raise ValueError("batch requires at least one request") + _safe_path_segment(network, "network") body: List[Dict[str, Any]] = [] for i, req in enumerate(requests): if "method" not in req: raise ValueError(f"batch request {i} is missing 'method'") body.append({"jsonrpc": "2.0", "id": i + 1, **req}) data = self._request_with_payment_raw(f"/v1/rpc/{network}", body) # type: ignore[arg-type] + headers = self._last_raw_headers if not isinstance(data, list): data = [data] - out: List[RpcResponse] = [] - for item in data: - if not isinstance(item, dict): - item = {"result": item} - out.append(RpcResponse(**item, network=network)) - return out + return [self._rpc_response(item, headers, network) for item in data] def search( self, @@ -2531,6 +2685,11 @@ def __init__( TransactionLogger(log_dir) if log_dir is not None else None ) self._last_settlement: Optional[Dict[str, Any]] = None + # Response headers from the most recent raw paid POST — consumed by + # rpc()/music()/speech() to surface the settlement receipt + gateway + # metadata the shared JSON-only helper would otherwise drop. Read it + # immediately after the helper returns (no intervening await). + self._last_raw_headers: Optional[httpx.Headers] = None # Async x402 client + same SVM signer the sync class uses. from x402 import x402Client # local import to keep optional dep clean @@ -2572,6 +2731,14 @@ def _capture_settlement(self, response: httpx.Response) -> Optional[Dict[str, An self._last_settlement = settlement return settlement + def _attach_receipt(self, data: Any) -> None: + """Inject the settlement tx hash from the most recent paid POST into a + raw response dict under ``txHash`` (mirrors the Base Music/Speech + clients). No-op on free responses (no receipt header).""" + tx_hash = _receipt_from_headers(self._last_raw_headers) + if tx_hash and isinstance(data, dict) and not data.get("txHash"): + data["txHash"] = tx_hash + def _log_transaction( self, endpoint: str, @@ -3100,6 +3267,10 @@ async def _request_with_payment_raw( if cached is not None: return cached + # Reset per-call receipt headers; only a paid retry repopulates them, so + # a free/cached model can't inherit a prior call's settlement receipt. + self._last_raw_headers = None + url = f"{self._api_url}{endpoint}" headers = {"Content-Type": "application/json", "User-Agent": _get_user_agent()} eff_timeout = timeout if timeout is not None else self._timeout @@ -3135,6 +3306,7 @@ async def _request_with_payment_raw( self._session_total_usd += cost_usd self._last_call_cost = cost_usd self._capture_settlement(retry_response) + self._last_raw_headers = retry_response.headers result = retry_response.json() save_to_cache(endpoint, body, result, cost_usd=cost_usd, **self._billing_meta()) self._log_transaction(endpoint, body, result, cost_usd) @@ -3360,44 +3532,21 @@ async def video( ) -> VideoResponse: """Generate a video clip (Solana payment). Async mirror of :meth:`SolanaLLMClient.video`.""" - if image_url and real_face_asset_id: - raise ValueError( - "image_url and real_face_asset_id are mutually exclusive; pass at most one." - ) - if last_frame_url and not image_url: - raise ValueError("last_frame_url requires image_url (seeds the FIRST frame).") - if last_frame_url and real_face_asset_id: - raise ValueError("last_frame_url and real_face_asset_id are mutually exclusive.") - if reference_image_urls: - if image_url or last_frame_url or real_face_asset_id: - raise ValueError( - "reference_image_urls is mutually exclusive with image_url, " - "last_frame_url, and real_face_asset_id." - ) - if len(reference_image_urls) > 9: - raise ValueError("reference_image_urls accepts at most 9 images.") - if real_face_asset_id is not None and not real_face_asset_id.startswith("ta_"): - raise ValueError("real_face_asset_id must start with 'ta_'.") - - body: Dict[str, Any] = { - "model": model or SolanaLLMClient.VIDEO_DEFAULT_MODEL, - "prompt": prompt, - } - for k, v in ( - ("image_url", image_url), - ("last_frame_url", last_frame_url), - ("reference_image_urls", reference_image_urls), - ("real_face_asset_id", real_face_asset_id), - ("duration_seconds", duration_seconds), - ("aspect_ratio", aspect_ratio), - ("resolution", resolution), - ("generate_audio", generate_audio), - ("seed", seed), - ("watermark", watermark), - ("return_last_frame", return_last_frame), - ): - if v is not None: - body[k] = v + body = SolanaLLMClient._build_video_body( + prompt, + model=model, + image_url=image_url, + last_frame_url=last_frame_url, + reference_image_urls=reference_image_urls, + real_face_asset_id=real_face_asset_id, + duration_seconds=duration_seconds, + aspect_ratio=aspect_ratio, + resolution=resolution, + generate_audio=generate_audio, + seed=seed, + watermark=watermark, + return_last_frame=return_last_frame, + ) data = await self._request_image_with_payment( "/v1/videos/generations", @@ -3464,6 +3613,7 @@ async def music( if lyrics and lyrics.strip(): body["lyrics"] = lyrics.strip() data = await self._request_with_payment_raw("/v1/audio/generations", body, timeout=timeout) + self._attach_receipt(data) return MusicResponse(**data) async def speech( @@ -3488,6 +3638,7 @@ async def speech( if speed is not None: body["speed"] = speed data = await self._request_with_payment_raw("/v1/audio/speech", body, timeout=timeout) + self._attach_receipt(data) return SpeechResponse(**data) async def sound_effect( @@ -3514,12 +3665,15 @@ async def sound_effect( data = await self._request_with_payment_raw( "/v1/audio/sound-effects", body, timeout=timeout ) + self._attach_receipt(data) return SpeechResponse(**data) async def list_voices(self) -> List[Dict[str, Any]]: """List available speech voices (free).""" url = f"{self._api_url}/v1/audio/voices" - resp = await self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = await self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -3531,7 +3685,8 @@ async def list_voices(self) -> List[Dict[str, Any]]: sanitize_error_response(error_body), ) data = resp.json() - return data.get("voices", data) if isinstance(data, dict) else data + # Gateway wraps the voice list under "data" (mirrors SpeechClient.list_voices). + return data.get("data", []) if isinstance(data, dict) else data async def portrait_enroll(self, name: str, image_url: str) -> PortraitEnrollment: """Enroll a Virtual Portrait ($0.01 USDC). Returns a ``ta_`` asset id.""" @@ -3548,9 +3703,11 @@ async def portrait_enroll(self, name: str, image_url: str) -> PortraitEnrollment async def list_portraits(self, wallet_address: Optional[str] = None) -> PortraitList: """List Virtual Portraits enrolled by a wallet (free, rate-limited).""" - addr = wallet_address or self.get_wallet_address() + addr = _safe_path_segment(wallet_address or self.get_wallet_address(), "wallet_address") url = f"{self._api_url}/v1/wallet/{addr}/portraits" - resp = await self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = await self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -3567,6 +3724,8 @@ async def realface_init(self, name: str, group_id: Optional[str] = None) -> Real raise ValueError("name is required (1-64 chars)") if len(name) > 64: raise ValueError(f"name must be 64 chars or fewer (got {len(name)})") + if group_id is not None and not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") body: Dict[str, Any] = {"name": name} if group_id: body["groupId"] = group_id @@ -3575,6 +3734,7 @@ async def realface_init(self, name: str, group_id: Optional[str] = None) -> Real url, json=body, headers={"Content-Type": "application/json", "User-Agent": _get_user_agent()}, + timeout=DEFAULT_FAST_TIMEOUT, ) if resp.status_code != 200: try: @@ -3588,11 +3748,14 @@ async def realface_init(self, name: str, group_id: Optional[str] = None) -> Real async def realface_status(self, group_id: str) -> RealFaceStatus: """Poll a RealFace group's state (free, rate-limited).""" - if not group_id: - raise ValueError("group_id is required") + if not group_id or not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") url = f"{self._api_url}/v1/realface/status" resp = await self._client.get( - url, params={"groupId": group_id}, headers={"User-Agent": _get_user_agent()} + url, + params={"groupId": group_id}, + headers={"User-Agent": _get_user_agent()}, + timeout=DEFAULT_FAST_TIMEOUT, ) if resp.status_code != 200: try: @@ -3638,8 +3801,8 @@ async def realface_enroll(self, name: str, image_url: str, group_id: str) -> Rea raise ValueError(f"name must be 64 chars or fewer (got {len(name)})") if not image_url or not image_url.lower().startswith(("https://", "http://")): raise ValueError("image_url must be an http(s) URL") - if not group_id: - raise ValueError("group_id is required") + if not group_id or not _GROUP_ID_RE.match(group_id): + raise ValueError("group_id must look like 'legacy_rf_'") data = await self._request_with_payment_raw( "/v1/realface/enroll", {"name": name, "image_url": image_url, "group_id": group_id} ) @@ -3647,9 +3810,11 @@ async def realface_enroll(self, name: str, image_url: str, group_id: str) -> Rea async def list_realfaces(self, wallet_address: Optional[str] = None) -> RealFaceList: """List RealFace assets enrolled by a wallet (free, rate-limited).""" - addr = wallet_address or self.get_wallet_address() + addr = _safe_path_segment(wallet_address or self.get_wallet_address(), "wallet_address") url = f"{self._api_url}/v1/wallet/{addr}/realfaces" - resp = await self._client.get(url, headers={"User-Agent": _get_user_agent()}) + resp = await self._client.get( + url, headers={"User-Agent": _get_user_agent()}, timeout=DEFAULT_FAST_TIMEOUT + ) if resp.status_code != 200: try: error_body = resp.json() @@ -3662,21 +3827,23 @@ async def list_realfaces(self, wallet_address: Optional[str] = None) -> RealFace async def price( self, - category: str, + category: Category, symbol: str, *, - market: Optional[str] = None, - session: Optional[str] = None, + market: Optional[Market] = None, + session: Optional[Session] = None, ) -> PricePoint: """Fetch a realtime Pyth price quote (Solana payment for paid categories).""" endpoint = SolanaLLMClient._price_category_path(category, market, "price", symbol) params: Dict[str, Any] = {} if session is not None: params["session"] = session - data = await self._get_with_payment_raw(endpoint, params=params or None) + data = await self._get_with_payment_raw( + endpoint, params=params or None, timeout=DEFAULT_FAST_TIMEOUT + ) return PricePoint( symbol=data.get("symbol", symbol.upper()), - price=data["price"], + price=data.get("price"), publish_time=data.get("publishTime"), confidence=data.get("confidence"), feed_id=data.get("feedId"), @@ -3689,21 +3856,23 @@ async def price( async def price_history( self, - category: str, + category: Category, symbol: str, *, - resolution: str = "D", + resolution: Resolution = "D", from_ts: int, to_ts: int, - market: Optional[str] = None, - session: Optional[str] = None, + market: Optional[Market] = None, + session: Optional[Session] = None, ) -> PriceHistoryResponse: """Fetch OHLC bars between two Unix timestamps (seconds).""" endpoint = SolanaLLMClient._price_category_path(category, market, "history", symbol) params: Dict[str, Any] = {"resolution": resolution, "from": from_ts, "to": to_ts} if session is not None: params["session"] = session - data = await self._get_with_payment_raw(endpoint, params=params) + data = await self._get_with_payment_raw( + endpoint, params=params, timeout=DEFAULT_FAST_TIMEOUT + ) return PriceHistoryResponse( symbol=data.get("symbol", symbol.upper()), resolution=data.get("resolution", resolution), @@ -3713,18 +3882,20 @@ async def price_history( async def list_symbols( self, - category: str, + category: Category, *, q: Optional[str] = None, limit: int = 100, - market: Optional[str] = None, + market: Optional[Market] = None, ) -> SymbolListResponse: """List available symbols in a Pyth category (free discovery).""" endpoint = SolanaLLMClient._price_category_path(category, market, "list", None) params: Dict[str, Any] = {"limit": limit} if q: params["q"] = q - data = await self._get_with_payment_raw(endpoint, params=params) + data = await self._get_with_payment_raw( + endpoint, params=params, timeout=DEFAULT_FAST_TIMEOUT + ) if isinstance(data, list): return SymbolListResponse(symbols=data, count=len(data)) return SymbolListResponse( @@ -3742,32 +3913,28 @@ async def rpc( id: Union[str, int] = 1, ) -> RpcResponse: """Make a single JSON-RPC 2.0 call (Solana payment, flat $0.002).""" + _safe_path_segment(network, "network") body: Dict[str, Any] = {"jsonrpc": "2.0", "id": id, "method": method} if params is not None: body["params"] = params data = await self._request_with_payment_raw(f"/v1/rpc/{network}", body) - if not isinstance(data, dict): - data = {"result": data} - return RpcResponse(**data, network=network) + return SolanaLLMClient._rpc_response(data, self._last_raw_headers, network) async def rpc_batch(self, network: str, requests: List[Dict[str, Any]]) -> List[RpcResponse]: """Make a JSON-RPC 2.0 batch call (Solana payment, $0.002 x N).""" if not requests: raise ValueError("batch requires at least one request") + _safe_path_segment(network, "network") body: List[Dict[str, Any]] = [] for i, req in enumerate(requests): if "method" not in req: raise ValueError(f"batch request {i} is missing 'method'") body.append({"jsonrpc": "2.0", "id": i + 1, **req}) data = await self._request_with_payment_raw(f"/v1/rpc/{network}", body) # type: ignore[arg-type] + headers = self._last_raw_headers if not isinstance(data, list): data = [data] - out: List[RpcResponse] = [] - for item in data: - if not isinstance(item, dict): - item = {"result": item} - out.append(RpcResponse(**item, network=network)) - return out + return [SolanaLLMClient._rpc_response(item, headers, network) for item in data] async def _request_image_with_payment( self, @@ -3908,18 +4075,20 @@ async def _request_image_with_payment( # Base VideoClient. if resigns_left > 0: resigns_left -= 1 - challenge = await self._client.get( - poll_url, - headers={"User-Agent": _get_user_agent()}, - timeout=eff_timeout, - ) - if challenge.status_code == 402: - try: + try: + challenge = await self._client.get( + poll_url, + headers={"User-Agent": _get_user_agent()}, + timeout=eff_timeout, + ) + if challenge.status_code == 402: resign_headers, _ = await self._sign_payment_from_response(challenge) poll_headers["PAYMENT-SIGNATURE"] = resign_headers["PAYMENT-SIGNATURE"] continue - except PaymentError: - pass + except (PaymentError, httpx.HTTPError): + # Challenge GET or re-sign failed — surface the gateway's + # real 402 reason, not a network/signing error. + pass raise build_payment_rejected_error(poll_resp) if last_status == "failed": diff --git a/tests/unit/test_solana_media.py b/tests/unit/test_solana_media.py new file mode 100644 index 0000000..f7a70b5 --- /dev/null +++ b/tests/unit/test_solana_media.py @@ -0,0 +1,236 @@ +"""Unit tests for the Solana media surface added in #16 (video/music/speech/ +sound-effects/price/list_voices) plus the mid-poll re-sign payment-terms guard. + +Payment flow is mocked at the httpx transport level (402 on the unsigned probe, +success once a PAYMENT-SIGNATURE is present); the x402 codec + signer are +stubbed so no wallet or network is needed — same approach as +test_solana_timeout_routing.py. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any, Dict, List +from unittest import mock + +import httpx +import pytest + +from blockrun_llm.solana_client import SolanaLLMClient, _assert_same_payment_terms +from blockrun_llm.types import ( + MusicResponse, + PaymentError, + SpeechResponse, +) + + +# --------------------------------------------------------------------------- +# _assert_same_payment_terms — the mid-poll re-sign guard +# --------------------------------------------------------------------------- + + +def _payload(amount: str, pay_to: str) -> SimpleNamespace: + return SimpleNamespace(accepted=SimpleNamespace(amount=amount, pay_to=pay_to)) + + +class TestPaymentTermsGuard: + def test_same_terms_pass(self) -> None: + # Identical amount + recipient (the normal stale-blockhash re-sign) is + # allowed through with no exception. + _assert_same_payment_terms(_payload("1000000", "WALLET_A"), "1000000", "WALLET_A") + + def test_amount_change_rejected(self) -> None: + with pytest.raises(PaymentError, match="changed the payment terms"): + _assert_same_payment_terms(_payload("9999999", "WALLET_A"), "1000000", "WALLET_A") + + def test_recipient_change_rejected(self) -> None: + with pytest.raises(PaymentError, match="changed the payment terms"): + _assert_same_payment_terms(_payload("1000000", "ATTACKER"), "1000000", "WALLET_A") + + def test_amount_type_coerced_before_compare(self) -> None: + # int vs str for the same value must not trip the guard. + _assert_same_payment_terms(_payload(1000000, "WALLET_A"), "1000000", "WALLET_A") + + +# --------------------------------------------------------------------------- +# Media dispatch — body construction + response parsing over the mocked flow +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _stub_x402_codec(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "blockrun_llm.solana_client.decode_payment_required_header", + lambda header: {"stub": True}, + ) + monkeypatch.setattr( + "blockrun_llm.solana_client.encode_payment_signature_header", + lambda payload: "stub-signature", + ) + + +@pytest.fixture(autouse=True) +def _no_disk_cache(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("blockrun_llm.cache.get_cached", lambda *a, **k: None) + monkeypatch.setattr("blockrun_llm.cache.save_to_cache", lambda *a, **k: None) + + +def _make_client(handler: Any) -> SolanaLLMClient: + with ( + mock.patch("blockrun_llm.solana_client.register_exact_svm_client"), + mock.patch("blockrun_llm.solana_client._create_signer"), + ): + client = SolanaLLMClient( + private_key="bogus_signer_is_patched", + api_url="https://sol.blockrun.ai/api", + rpc_url="http://test", + ) + + class _FakePayload: + class accepted: + amount = "1000000" + pay_to = "GsbwXfJraMomNxBcpR3DBNxnKwZbyq7YCoDdSLDwzxdV" + + client._x402_client = mock.MagicMock() + client._x402_client.create_payment_payload.return_value = _FakePayload() + client._client = httpx.Client(transport=httpx.MockTransport(handler)) + client._address = "11111111111111111111111111111111" + return client + + +def _paid_flow(calls: List[httpx.Request], ok_body: Dict[str, Any]): + """402 on the unsigned probe, then ``ok_body`` once signed. Captures the + signed request so tests can assert the forwarded JSON body + path.""" + + def handler(request: httpx.Request) -> httpx.Response: + if "PAYMENT-SIGNATURE" not in request.headers: + return httpx.Response( + 402, + headers={"content-type": "application/json", "payment-required": "stub"}, + json={"error": "Payment Required"}, + ) + calls.append(request) + return httpx.Response(200, json=ok_body, headers={"content-type": "application/json"}) + + return handler + + +_MUSIC_OK = {"created": 1, "model": "minimax/music-2.5+", "data": [{"url": "https://cdn/x.mp3"}]} +_SPEECH_OK = { + "created": 1, + "model": "elevenlabs/flash-v2.5", + "data": [{"url": "https://cdn/x.wav"}], +} + + +class TestMediaDispatch: + def test_music_body_and_response(self) -> None: + import json + + calls: List[httpx.Request] = [] + client = _make_client(_paid_flow(calls, _MUSIC_OK)) + resp = client.music("lo-fi beats") + assert isinstance(resp, MusicResponse) + assert resp.data[0].url == "https://cdn/x.mp3" + assert calls[-1].url.path == "/api/v1/audio/generations" + sent = json.loads(calls[-1].content) + assert sent["model"] == "minimax/music-2.5+" + assert sent["instrumental"] is True + + def test_speech_body_and_response(self) -> None: + import json + + calls: List[httpx.Request] = [] + client = _make_client(_paid_flow(calls, _SPEECH_OK)) + resp = client.speech("hello world", voice="sarah") + assert isinstance(resp, SpeechResponse) + assert resp.data[0].url == "https://cdn/x.wav" + assert calls[-1].url.path == "/api/v1/audio/speech" + sent = json.loads(calls[-1].content) + assert sent["input"] == "hello world" + assert sent["voice"] == "sarah" + + def test_sound_effect_endpoint(self) -> None: + calls: List[httpx.Request] = [] + client = _make_client(_paid_flow(calls, _SPEECH_OK)) + client.sound_effect("thunder clap") + assert calls[-1].url.path == "/api/v1/audio/sound-effects" + + def test_list_voices_returns_list_not_envelope(self) -> None: + # Regression: the gateway returns {"data": [...]}, and list_voices must + # return the list, not the whole dict. + voices = [{"id": "sarah"}, {"id": "adam"}] + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"data": voices}) + + client = _make_client(handler) + assert client.list_voices() == voices + + +# --------------------------------------------------------------------------- +# Local validation — must reject before any HTTP / payment +# --------------------------------------------------------------------------- + + +class TestLocalValidation: + def test_music_lyrics_with_instrumental_rejected(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) # never reached + with pytest.raises(ValueError, match="lyrics"): + client.music("pop", instrumental=True, lyrics="la la la") + + def test_video_mutually_exclusive_image_and_face(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) + with pytest.raises(ValueError, match="mutually exclusive"): + client.video("a cat", image_url="https://x/y.png", real_face_asset_id="ta_abc") + + def test_video_bad_face_id_prefix(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) + with pytest.raises(ValueError, match="ta_"): + client.video("a cat", real_face_asset_id="not_a_valid_id") + + def test_portrait_enroll_requires_http_url(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) + with pytest.raises(ValueError, match="image_url"): + client.portrait_enroll("Alice", "ftp://bad/url") + + +# --------------------------------------------------------------------------- +# price() — missing "price" in a paid body must not raise a raw KeyError +# --------------------------------------------------------------------------- + + +class TestPriceRobustness: + def test_missing_price_field_is_clean_error_not_keyerror(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if "PAYMENT-SIGNATURE" not in request.headers: + return httpx.Response( + 402, + headers={"content-type": "application/json", "payment-required": "stub"}, + json={"error": "Payment Required"}, + ) + # Paid 200 but the body is missing "price" — must surface as a + # pydantic validation error, not a bare KeyError. + return httpx.Response(200, json={"symbol": "BTCUSD"}) + + client = _make_client(handler) + with pytest.raises(Exception) as exc_info: + client.price("crypto", "BTCUSD") + assert not isinstance(exc_info.value, KeyError) + + +# --------------------------------------------------------------------------- +# Path-segment guard — LLM-controlled values can't escape the URL path +# --------------------------------------------------------------------------- + + +class TestPathSegmentGuard: + def test_symbol_with_slash_rejected(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) + with pytest.raises(ValueError, match="symbol"): + client.price("crypto", "../../secret") + + def test_network_with_traversal_rejected(self) -> None: + client = _make_client(lambda r: httpx.Response(500)) + with pytest.raises(ValueError, match="network"): + client.rpc("../evil", "eth_blockNumber") diff --git a/tests/unit/test_solana_timeout_routing.py b/tests/unit/test_solana_timeout_routing.py index f07c0fa..1909181 100644 --- a/tests/unit/test_solana_timeout_routing.py +++ b/tests/unit/test_solana_timeout_routing.py @@ -74,6 +74,7 @@ def _make_client(transport: httpx.MockTransport, **kwargs: float) -> SolanaLLMCl class _FakePayload: class accepted: amount = "1000000" + pay_to = "GsbwXfJraMomNxBcpR3DBNxnKwZbyq7YCoDdSLDwzxdV" client._x402_client = mock.MagicMock() client._x402_client.create_payment_payload.return_value = _FakePayload() @@ -228,6 +229,7 @@ def _make_async_client(transport: httpx.MockTransport, **kwargs: float): class _FakePayload: class accepted: amount = "1000000" + pay_to = "GsbwXfJraMomNxBcpR3DBNxnKwZbyq7YCoDdSLDwzxdV" client._x402_client = mock.MagicMock() client._x402_client.create_payment_payload = mock.AsyncMock(return_value=_FakePayload())