git / moonbite-network / tests/test_rpc_proxy_dupkey.py · raw

"""Regression: the read-only RPC proxy must never forward raw client bytes.Python's json keeps the LAST duplicate key; the node's UniValue parser returnsthe FIRST. A body such as    {"method":"sendtoaddress",...,"method":"getblockcount"}passed the allowlist (last = getblockcount) while the node ran sendtoaddress.The proxy now rejects duplicate keys and re-serializes only validated fields."""import base64import jsonimport osimport subprocessimport sysimport threadingimport timeimport urllib.errorimport urllib.requestfrom http.server import BaseHTTPRequestHandler, HTTPServerfrom pathlib import Pathimport pytestPROXY = Path(__file__).resolve().parent.parent / "explorer" / "rpc_readonly_proxy.py"NODE_PORT, PROXY_PORT = 19332, 19350class _EchoNode(BaseHTTPRequestHandler):    seen = []    def do_POST(self):        body = self.rfile.read(int(self.headers["Content-Length"]))        _EchoNode.seen.append(body)        self.send_response(200)        self.send_header("Content-Type", "application/json")        self.end_headers()        self.wfile.write(b'{"result":"echo","error":null,"id":1}')    def log_message(self, *a):  # silence        pass@pytest.fixture(scope="module")def proxy():    srv = HTTPServer(("127.0.0.1", NODE_PORT), _EchoNode)    threading.Thread(target=srv.serve_forever, daemon=True).start()    env = dict(        os.environ,        PROXY_BIND="127.0.0.1", PROXY_PORT=str(PROXY_PORT),        NODE_RPC_URL=f"http://127.0.0.1:{NODE_PORT}/",        NODE_RPC_USER="n", NODE_RPC_PASSWORD="np",        PROXY_USER="u", PROXY_PASSWORD="p",    )    proc = subprocess.Popen([sys.executable, str(PROXY)], env=env,                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)    for _ in range(50):        try:            urllib.request.urlopen(f"http://127.0.0.1:{PROXY_PORT}/health", timeout=1)            break        except Exception:            time.sleep(0.1)    yield    proc.kill()    srv.shutdown()def _call(body: str):    auth = "Basic " + base64.b64encode(b"u:p").decode()    req = urllib.request.Request(        f"http://127.0.0.1:{PROXY_PORT}/", data=body.encode(),        headers={"Authorization": auth, "Content-Type": "application/json"})    try:        with urllib.request.urlopen(req, timeout=5) as r:            return r.status    except urllib.error.HTTPError as e:        return e.codedef test_duplicate_method_key_is_rejected(proxy):    _EchoNode.seen.clear()    body = '{"id":1,"method":"sendtoaddress","params":["M",1000],"method":"getblockcount"}'    assert _call(body) == 400    assert _EchoNode.seen == [], "node must receive nothing"def test_forwarded_body_is_reserialized_not_raw(proxy):    _EchoNode.seen.clear()    raw = '{"id":1,"method":"getblockcount","params":[], "junk":{"method":"stop"}}'    assert _call(raw) == 200    assert len(_EchoNode.seen) == 1    got = json.loads(_EchoNode.seen[0])    assert got == {"jsonrpc": "1.0", "id": 1, "method": "getblockcount", "params": []}    assert b"junk" not in _EchoNode.seen[0]def test_batch_with_one_bad_method_is_fully_refused(proxy):    _EchoNode.seen.clear()    assert _call('[{"id":1,"method":"getblockcount"},{"id":2,"method":"stop"}]') == 403    assert _EchoNode.seen == []def test_non_string_method_refused(proxy):    assert _call('{"id":1,"method":["getblockcount"]}') == 403def test_dockerignore_keeps_secrets_out_of_images():    src = (Path(__file__).resolve().parent.parent / ".dockerignore").read_text()    for pat in (".claude", ".git", "*.db", "*.jsonl", ".env"):        assert pat in src.split(), f"{pat} missing from .dockerignore"