feat: add Python keyless public chain read

README.md

@@ -67,6 +67,19 @@ The command reads the public chain directory, confirms the selected chain’s ex
 
 If a request fails with an HTTP status or times out, the command identifies the directory or RPC method involved and exits with code `1`. For HTTP `429`, wait before trying again manually. The command does not retry automatically and does not print response bodies or raw exceptions. Other failures use a generic error message. Successful runs print the same JSON result.
 
+### Python standard-library entry
+
+Prefer Python? Python 3.10 or newer can run the same keyless identity check with no Node.js, npm, third-party packages, wallet, or API key:
+
+```bash
+git clone https://github.com/blockvectra/agent-quickstart.git
+cd agent-quickstart
+python3 examples/public_chain.py
+python3 examples/public_chain.py 56  # Select a chain from the live directory
+```
+
+The script makes at most two serial HTTP requests: it selects the chain from the public directory, then verifies `eth_chainId` against that selection. It checks chain identity, not the latest block or method coverage. It never retries automatically. Each socket operation has a 15-second timeout; this is a socket timeout, not a hard deadline for DNS or the full request.
+
 Beyond the keyless read, the full example covers:
 1. **Wallet identity**: Reads an Ethereum EOA private key from `AGENT_PRIVATE_KEY` or from a local `.agent-wallet.json` file. If neither is configured, generates a new wallet locally and saves it to `.agent-wallet.json` (permission `0600`) so subsequent runs reuse the same identity instead of creating new accounts.
 2. **Programmatic account sign-up**: Authenticates via Sign-In with Ethereum (SIWE / EIP-191) against `https://console-api.blockvectra.com/v1` in programmatic mode (omitting the `Origin` header). First-time wallet sign-in automatically provisions an account.

README.zh.md

@@ -67,6 +67,19 @@ npm run read:public
 
 请求返回 HTTP 错误状态或超时时，命令会指出出错的目录请求或 RPC 方法，并以退出码 `1` 结束。遇到 HTTP `429` 时，请等待后再手动重试。命令不会自动重试，也不会输出响应正文或原始异常；其他失败仍使用通用错误提示。成功时继续输出相同的 JSON 结果。
 
+### Python 标准库入口
+
+也可以使用 Python 3.10 或更高版本运行无 Key 链身份校验，无需 Node.js、npm、第三方包、钱包或 API Key：
+
+```bash
+git clone https://github.com/blockvectra/agent-quickstart.git
+cd agent-quickstart
+python3 examples/public_chain.py
+python3 examples/public_chain.py 56  # 从实时目录选择链
+```
+
+脚本最多串行发送两次 HTTP 请求：先从公开目录选择链，再核对 `eth_chainId` 是否与所选链一致。它验证的是链身份，不读取最新区块，也不验证方法覆盖情况。失败时不会自动重试。每次 socket 操作的超时为 15 秒；这不是 DNS 或整个请求流程的硬截止时间。
+
 除无 Key 读取外，完整示例还涵盖：
 1. **钱包身份**：从环境变量 `AGENT_PRIVATE_KEY` 或本地 `.agent-wallet.json` 文件读取以太坊 EOA 私钥；若均未配置，则在本地生成新钱包并保存至 `.agent-wallet.json`（权限 `0600`），以便后续运行复用相同身份，避免重复开户。
 2. **程序化开户与登录**：通过以太坊签名（SIWE / EIP-191）调用 `https://console-api.blockvectra.com/v1` 进行鉴权（程序化模式，不带 `Origin` 请求头）。新钱包首次登录自动完成开户。

examples/public_chain.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Read and verify the public JSON-RPC chain identity without third-party packages."""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+import time
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+
+DIRECTORY_URL = "https://api.blockvectra.com/v1/chains"
+RPC_BASE_URL = "https://api.blockvectra.com/v1/"
+DEFAULT_CHAIN_ID = 4663
+MAX_SAFE_INTEGER = 2**53 - 1
+MAX_BODY_BYTES = 128 * 1024
+SOCKET_TIMEOUT_SECONDS = 15
+JSONRPC_REQUEST = {"jsonrpc": "2.0", "method": "eth_chainId", "params": [], "id": 1}
+DECIMAL_ID_RE = re.compile(r"^[1-9][0-9]*$", re.ASCII)
+SLUG_RE = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$", re.ASCII)
+QUANTITY_RE = re.compile(r"^0x(?:0|[1-9a-f][0-9a-f]*)$", re.ASCII)
+
+
+class _NoRedirectHandler(HTTPRedirectHandler):
+    def redirect_request(self, request, response, code, message, headers, new_url):
+        return None
+
+
+class PublicReadError(Exception):
+    """A safe, user-facing failure without response or transport details."""
+
+
+def parse_chain_id_args(args: list[str]) -> int:
+    if not args:
+        return DEFAULT_CHAIN_ID
+    if len(args) != 1 or not DECIMAL_ID_RE.fullmatch(args[0]):
+        raise PublicReadError("Usage: python3 examples/public_chain.py [positive-decimal-chain-id]")
+    value = int(args[0])
+    if value > MAX_SAFE_INTEGER:
+        raise PublicReadError("Invalid chain ID")
+    return value
+
+
+def _is_object(value: Any) -> bool:
+    return isinstance(value, dict)
+
+
+def select_public_chain(directory: Any, expected_chain_id: int) -> dict[str, Any]:
+    if isinstance(expected_chain_id, bool) or not isinstance(expected_chain_id, int) or not 1 <= expected_chain_id <= MAX_SAFE_INTEGER:
+        raise PublicReadError("Invalid chain ID")
+    if not _is_object(directory) or not isinstance(directory.get("chains"), list):
+        raise PublicReadError("Invalid chain directory")
+
+    entries = [entry for entry in directory["chains"] if _is_object(entry)]
+    matches = [entry for entry in entries if type(entry.get("chain_id")) is int and entry["chain_id"] == expected_chain_id]
+    if len(matches) != 1:
+        raise PublicReadError("Chain ID is missing or ambiguous in the directory")
+    entry = matches[0]
+    slug = entry.get("chain")
+    if not isinstance(slug, str) or not SLUG_RE.fullmatch(slug) or sum(candidate.get("chain") == slug for candidate in entries) != 1:
+        raise PublicReadError("Chain slug is invalid or ambiguous in the directory")
+    if expected_chain_id == DEFAULT_CHAIN_ID and slug != "robinhood_mainnet":
+        raise PublicReadError("Robinhood mainnet directory entry is invalid")
+    rpc_url = RPC_BASE_URL + slug + "/public"
+    if entry.get("jsonrpc") is not True:
+        raise PublicReadError("Selected chain does not advertise JSON-RPC")
+    public = entry.get("public")
+    if not _is_object(public) or public.get("url") != rpc_url:
+        raise PublicReadError("Selected chain public endpoint is invalid")
+    methods = public.get("methods")
+    if not isinstance(methods, list) or any(not isinstance(method, str) for method in methods) or "eth_chainId" not in methods:
+        raise PublicReadError("Selected chain does not advertise eth_chainId")
+    return {"chain": slug, "chainId": expected_chain_id, "rpcUrl": rpc_url}
+
+
+def _user_agent() -> str:
+    import os
+
+    suffix = " BlockVectraQA/1" if os.environ.get("BLOCKVECTRA_INTERNAL_QA") == "1" else ""
+    return "BlockVectraAgentQuickstartPython/1" + suffix
+
+
+def _read_limited(response: Any, deadline: float) -> bytes:
+    chunks: list[bytes] = []
+    total = 0
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise PublicReadError("Request timed out")
+        if hasattr(response, "settimeout"):
+            response.settimeout(min(SOCKET_TIMEOUT_SECONDS, remaining))
+        chunk = response.read(min(8192, MAX_BODY_BYTES + 1 - total))
+        if not chunk:
+            return b"".join(chunks)
+        total += len(chunk)
+        if total > MAX_BODY_BYTES:
+            raise PublicReadError("Response body is too large")
+        chunks.append(chunk)
+
+
+def _request_json(url: str, method: str, body: bytes | None) -> Any:
+    headers = {"Accept": "application/json", "User-Agent": _user_agent()}
+    if body is not None:
+        headers["Content-Type"] = "application/json"
+    request = Request(url, data=body, headers=headers, method=method)
+    started = time.monotonic()
+    response = None
+    try:
+        response = _open_request(request, timeout=SOCKET_TIMEOUT_SECONDS)
+        if getattr(response, "status", 200) < 200 or getattr(response, "status", 200) >= 300:
+            raise PublicReadError("HTTP request failed")
+        raw = _read_limited(response, started + SOCKET_TIMEOUT_SECONDS)
+        return json.loads(raw)
+    except PublicReadError:
+        raise
+    except HTTPError as exc:
+        exc.close()
+        raise PublicReadError("HTTP request failed") from None
+    except (URLError, TimeoutError, OSError, ValueError, json.JSONDecodeError):
+        raise PublicReadError("Request failed") from None
+    finally:
+        if response is not None:
+            response.close()
+
+
+def _open_request(request: Request, timeout: float):
+    return build_opener(_NoRedirectHandler).open(request, timeout=timeout)
+
+
+def _validate_rpc_response(payload: Any, expected_chain_id: int) -> None:
+    if not _is_object(payload) or payload.get("jsonrpc") != "2.0":
+        raise PublicReadError("Invalid JSON-RPC response")
+    response_id = payload.get("id")
+    if type(response_id) is not int or response_id != 1 or "error" in payload or "result" not in payload:
+        raise PublicReadError("Invalid JSON-RPC response")
+    quantity = payload["result"]
+    if not isinstance(quantity, str) or not QUANTITY_RE.fullmatch(quantity):
+        raise PublicReadError("Invalid chain ID response")
+    if int(quantity, 16) != expected_chain_id:
+        raise PublicReadError("RPC chain ID does not match the selected chain")
+
+
+def run(expected_chain_id: int) -> dict[str, Any]:
+    directory = _request_json(DIRECTORY_URL, "GET", None)
+    selected = select_public_chain(directory, expected_chain_id)
+    rpc_body = json.dumps(JSONRPC_REQUEST, separators=(",", ":")).encode("utf-8")
+    result = _request_json(selected["rpcUrl"], "POST", rpc_body)
+    _validate_rpc_response(result, expected_chain_id)
+    return {"chain": selected["chain"], "chainId": expected_chain_id}
+
+
+def main(args: list[str] | None = None) -> int:
+    try:
+        chain_id = parse_chain_id_args(sys.argv[1:] if args is None else args)
+        print(json.dumps(run(chain_id), separators=(",", ":")))
+        return 0
+    except PublicReadError as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+    except Exception:
+        print("Public chain read failed", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_public_chain.py

@@ -0,0 +1,179 @@
+import contextlib
+import io
+import json
+import unittest
+from unittest.mock import patch
+
+from examples import public_chain
+
+
+def chain(chain_id=4663, slug="robinhood_mainnet", **overrides):
+    entry = {
+        "chain_id": chain_id,
+        "chain": slug,
+        "jsonrpc": True,
+        "public": {
+            "url": f"https://api.blockvectra.com/v1/{slug}/public",
+            "methods": ["eth_chainId"],
+        },
+    }
+    entry.update(overrides)
+    return entry
+
+
+class FakeResponse:
+    def __init__(self, payload, status=200, read_error=None):
+        self.data = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
+        self.status = status
+        self.read_error = read_error
+        self.closed = False
+        self.timeouts = []
+
+    def read(self, size=-1):
+        if self.read_error:
+            raise self.read_error
+        part, self.data = self.data[:size], self.data[size:]
+        return part
+
+    def settimeout(self, value):
+        self.timeouts.append(value)
+
+    def close(self):
+        self.closed = True
+
+
+class PublicChainTests(unittest.TestCase):
+    def setup_network(self, rpc_result="0x1237", directory=None, rpc_payload=None):
+        response_list = [FakeResponse(directory if directory is not None else {"chains": [chain()]})]
+        if rpc_payload is not None:
+            response_list.append(FakeResponse(rpc_payload))
+        elif isinstance(rpc_result, FakeResponse):
+            response_list.append(rpc_result)
+        else:
+            response_list.append(FakeResponse({"jsonrpc": "2.0", "id": 1, "result": rpc_result}))
+        calls = []
+
+        def fake_urlopen(request, timeout):
+            calls.append((request, timeout))
+            if not response_list:
+                raise AssertionError("unexpected extra request")
+            return response_list.pop(0)
+
+        return calls, response_list, patch.object(public_chain, "_open_request", side_effect=fake_urlopen)
+
+    def invoke(self, args=None):
+        stdout, stderr = io.StringIO(), io.StringIO()
+        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
+            rc = public_chain.main(args)
+        return rc, stdout.getvalue(), stderr.getvalue()
+
+    def test_default_success_exact_requests_and_output(self):
+        calls, remaining, mocked = self.setup_network()
+        with mocked:
+            rc, out, err = self.invoke([])
+        self.assertEqual((rc, out, err), (0, '{"chain":"robinhood_mainnet","chainId":4663}\n', ""))
+        self.assertFalse(remaining)
+        self.assertEqual(len(calls), 2)
+        get, post = (call[0] for call in calls)
+        self.assertEqual((get.method, get.full_url, get.data), ("GET", public_chain.DIRECTORY_URL, None))
+        self.assertEqual((post.method, post.full_url), ("POST", "https://api.blockvectra.com/v1/robinhood_mainnet/public"))
+        self.assertEqual(post.data, b'{"jsonrpc":"2.0","method":"eth_chainId","params":[],"id":1}')
+        self.assertEqual(get.get_header("User-agent"), "BlockVectraAgentQuickstartPython/1")
+        self.assertEqual(post.get_header("User-agent"), "BlockVectraAgentQuickstartPython/1")
+        self.assertEqual([call[1] for call in calls], [15, 15])
+
+    def test_other_dynamic_chain_success_and_qa_user_agent(self):
+        calls, _, mocked = self.setup_network(
+            rpc_result="0x38", directory={"chains": [chain(56, "bsc_mainnet")]}
+        )
+        with patch.dict("os.environ", {"BLOCKVECTRA_INTERNAL_QA": "1"}), mocked:
+            rc, out, err = self.invoke(["56"])
+        self.assertEqual((rc, out, err), (0, '{"chain":"bsc_mainnet","chainId":56}\n', ""))
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(calls[1][0].full_url, "https://api.blockvectra.com/v1/bsc_mainnet/public")
+        self.assertEqual(calls[0][0].get_header("User-agent"), "BlockVectraAgentQuickstartPython/1 BlockVectraQA/1")
+
+    def test_default_slug_must_be_exact(self):
+        calls, _, mocked = self.setup_network(directory={"chains": [chain(slug="robinhood")]})
+        with mocked:
+            rc, out, _ = self.invoke([])
+        self.assertEqual((rc, out), (1, ""))
+        self.assertEqual(len(calls), 1)
+
+    def test_duplicate_id_slug_bool_and_malicious_metadata_rejected(self):
+        cases = [
+            [chain(), chain(4663, "other_mainnet")],
+            [chain(), chain(1, "robinhood_mainnet")],
+            [{**chain(), "chain_id": True}],
+            [chain(slug="evil.example/path")],
+            [{**chain(), "public": {"url": "https://attacker.invalid/x", "methods": ["eth_chainId"]}}],
+        ]
+        for entries in cases:
+            with self.subTest(entries=entries):
+                calls, _, mocked = self.setup_network(directory={"chains": entries})
+                with mocked:
+                    rc, out, _ = self.invoke([])
+                self.assertEqual((rc, out), (1, ""))
+                self.assertEqual(len(calls), 1)
+
+    def test_rpc_method_must_be_advertised(self):
+        bad = chain()
+        bad["public"]["methods"] = ["eth_blockNumber"]
+        calls, _, mocked = self.setup_network(directory={"chains": [bad]})
+        with mocked:
+            rc, out, _ = self.invoke([])
+        self.assertEqual((rc, out), (1, ""))
+        self.assertEqual(len(calls), 1)
+
+    def test_rpc_response_validation(self):
+        payloads = [
+            {"jsonrpc": "1.0", "id": 1, "result": "0x1237"},
+            {"jsonrpc": "2.0", "id": True, "result": "0x1237"},
+            {"jsonrpc": "2.0", "id": "1", "result": "0x1237"},
+            {"jsonrpc": "2.0", "id": 1, "error": {"code": -1, "message": "secret"}},
+            {"jsonrpc": "2.0", "id": 1, "result": "0X1237"},
+            {"jsonrpc": "2.0", "id": 1, "result": "0x01237"},
+            {"jsonrpc": "2.0", "id": 1, "result": "0x"},
+            {"jsonrpc": "2.0", "id": 1, "result": "0x1238"},
+        ]
+        for payload in payloads:
+            with self.subTest(payload=payload):
+                calls, _, mocked = self.setup_network(rpc_payload=payload)
+                with mocked:
+                    rc, out, err = self.invoke([])
+                self.assertEqual((rc, out), (1, ""))
+                self.assertNotIn("secret", err)
+                self.assertEqual(len(calls), 2)
+
+    def test_http_429_redirect_and_errors_do_not_retry_and_close(self):
+        for status in (429, 302):
+            with self.subTest(status=status):
+                fake = FakeResponse(b"sensitive response body", status=status)
+                calls, _, mocked = self.setup_network(rpc_result=fake)
+                with mocked:
+                    rc, out, err = self.invoke([])
+                self.assertEqual((rc, out), (1, ""))
+                self.assertNotIn("sensitive", err)
+                self.assertEqual(len(calls), 2)
+                self.assertTrue(fake.closed)
+
+    def test_oversized_body_and_timeout_close_response(self):
+        for response in (FakeResponse(b"x" * (public_chain.MAX_BODY_BYTES + 1)), FakeResponse(b"", read_error=TimeoutError())):
+            with self.subTest(response=response):
+                calls, _, mocked = self.setup_network(rpc_result=response)
+                with mocked:
+                    rc, out, _ = self.invoke([])
+                self.assertEqual((rc, out), (1, ""))
+                self.assertTrue(response.closed)
+                self.assertEqual(len(calls), 2)
+
+    def test_invalid_cli_fails_before_any_request(self):
+        for args in (["0"], ["-1"], ["+1"], ["01"], ["1.0"], ["9007199254740992"], ["56", "1"]):
+            with self.subTest(args=args), patch.object(public_chain, "_open_request") as network:
+                rc, out, _ = self.invoke(list(args))
+            self.assertEqual((rc, out), (1, ""))
+            network.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()