fix: verify TLS for deployment RPCs

deploy_gcp/seismic_deploy/network/bootnode.py

@@ -2,15 +2,17 @@
 
 from __future__ import annotations
 
+import ipaddress
 import json
 import ssl
 import urllib.request
+from urllib.parse import urlparse
 
 import click
 
 
 def _rpc_call(url: str, method: str, params: list | None = None) -> dict:
-    """Make a JSON-RPC call. Skips SSL verification for IP-based URLs."""
+    """Make a JSON-RPC call with certificate verification for HTTPS URLs."""
     payload = json.dumps(
         {
             "jsonrpc": "2.0",
@@ -26,12 +28,24 @@ def _rpc_call(url: str, method: str, params: list | None = None) -> dict:
         headers={"Content-Type": "application/json"},
     )
 
-    # Skip SSL verification for direct IP access (no matching cert)
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
+    parsed = urlparse(url)
+    if parsed.scheme not in {"http", "https"}:
+        raise click.ClickException(
+            f"Unsupported RPC URL scheme {parsed.scheme!r}; use http:// or https://"
+        )
 
-    with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
+    if parsed.scheme == "https":
+        # Keep hostname and certificate verification enabled. The previous code
+        # used CERT_NONE for every URL, including the HTTPS hostnames used by
+        # production deployments, allowing an on-path attacker to spoof RPC
+        # responses during bootstrap and staking.
+        response = urllib.request.urlopen(
+            req, timeout=30, context=ssl.create_default_context()
+        )
+    else:
+        response = urllib.request.urlopen(req, timeout=30)
+
+    with response as resp:
         data = json.loads(resp.read())
 
     if "error" in data:
@@ -41,32 +55,31 @@ def _rpc_call(url: str, method: str, params: list | None = None) -> dict:
 
 
 def _normalize_rpc_base(host: str) -> str:
-    """Normalize a host/URL to a base URL, probing HTTPS then HTTP for bare IPs.
+    """Normalize a host/URL to a base URL without silently downgrading HTTPS.
 
     Accepts:
-      - bare IP: 35.193.138.168 -> tries https first, falls back to http
+      - bare IP: 35.193.138.168 -> http://35.193.138.168 (explicitly insecure)
+      - bare hostname: node.example -> https://node.example
       - full URL: https://35.193.138.168 -> https://35.193.138.168
       - URL with path: https://node-0.seismictest.net/rpc -> https://node-0.seismictest.net
     """
     from urllib.parse import urlparse
 
     s = host.strip().rstrip("/")
 
-    # If user gave a full URL, respect the scheme
-    if s.startswith("http"):
+    # If user gave a full URL, respect the scheme and strip its path.
+    if "://" in s:
         parsed = urlparse(s)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            raise click.ClickException(
+                f"Invalid RPC URL {host!r}; use http:// or https://"
+            )
         return f"{parsed.scheme}://{parsed.netloc}"
 
-    # Bare IP/hostname — try HTTPS first, fall back to HTTP
-    for scheme in ("https", "http"):
-        base = f"{scheme}://{s}"
-        try:
-            _rpc_call(f"{base}/summit/", "health")
-            return base
-        except Exception:
-            continue
-
-    # If neither worked, default to http (let the actual call produce the error)
+    try:
+        ipaddress.ip_address(s)
+    except ValueError:
+        return f"https://{s}"
     return f"http://{s}"
 
 

deploy_gcp/seismic_deploy/network/stake.py

@@ -2,9 +2,6 @@
 
 from __future__ import annotations
 
-import json
-import ssl
-import urllib.request
 from typing import cast
 
 import click
@@ -31,33 +28,12 @@ def _summit_url(node_host: str) -> str:
 
 def get_deposit_signature(summit_url: str, amount: int, address: str) -> dict:
     """Call summit's getDepositSignature RPC."""
-    payload = json.dumps(
-        {
-            "jsonrpc": "2.0",
-            "method": "getDepositSignature",
-            "params": [amount, address],
-            "id": 1,
-        }
-    ).encode()
-
-    req = urllib.request.Request(
-        summit_url,
-        data=payload,
-        headers={"Content-Type": "application/json"},
-    )
-
-    # Skip SSL verification for direct IP access
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
-
-    with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
-        data = json.loads(resp.read())
-
-    if "error" in data:
-        raise click.ClickException(f"Error from summit: {data['error']}")
+    from deploy_gcp.seismic_deploy.network.bootnode import _rpc_call
 
-    return data["result"]
+    result = _rpc_call(summit_url, "getDepositSignature", [amount, address])
+    if not result:
+        raise click.ClickException("No deposit signature returned from summit")
+    return result
 
 
 def stake_node(

deploy_gcp/seismic_deploy/network/sync.py

@@ -2,10 +2,7 @@
 
 from __future__ import annotations
 
-import json
-import ssl
 import time
-import urllib.request
 from pathlib import Path
 
 import click
@@ -15,33 +12,10 @@
 
 def fetch_checkpoint(rpc_url: str) -> dict:
     """Call getLatestCheckpoint on a summit RPC endpoint."""
-    payload = json.dumps(
-        {
-            "jsonrpc": "2.0",
-            "method": "getLatestCheckpoint",
-            "params": [],
-            "id": 1,
-        }
-    ).encode()
-
-    req = urllib.request.Request(
-        rpc_url,
-        data=payload,
-        headers={"Content-Type": "application/json"},
-    )
-
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
+    from deploy_gcp.seismic_deploy.network.bootnode import _rpc_call
 
     click.echo(f"  Fetching checkpoint from {rpc_url}...")
-    with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
-        data = json.loads(resp.read())
-
-    if "error" in data:
-        raise click.ClickException(f"Error from summit RPC: {data['error']}")
-
-    result = data.get("result")
+    result = _rpc_call(rpc_url, "getLatestCheckpoint")
     if not result:
         raise click.ClickException("No checkpoint data returned from RPC")
 

deploy_gcp/seismic_deploy/network/test_bootnode.py

@@ -0,0 +1,50 @@
+"""Tests for safe RPC URL handling."""
+
+import json
+import ssl
+import unittest
+from unittest import mock
+
+from deploy_gcp.seismic_deploy.network.bootnode import _normalize_rpc_base, _rpc_call
+
+
+class _Response:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, traceback):
+        return False
+
+    def read(self):
+        return json.dumps({"result": {"ok": True}}).encode()
+
+
+class RpcSecurityTests(unittest.TestCase):
+    @mock.patch("deploy_gcp.seismic_deploy.network.bootnode.urllib.request.urlopen")
+    def test_https_keeps_certificate_and_hostname_verification(self, urlopen):
+        urlopen.return_value = _Response()
+
+        _rpc_call("https://internal-0.seismictest.net/rpc", "health")
+
+        context = urlopen.call_args.kwargs["context"]
+        self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
+        self.assertTrue(context.check_hostname)
+
+    @mock.patch("deploy_gcp.seismic_deploy.network.bootnode.urllib.request.urlopen")
+    def test_http_does_not_create_an_insecure_tls_context(self, urlopen):
+        urlopen.return_value = _Response()
+
+        _rpc_call("http://127.0.0.1/rpc", "health")
+
+        self.assertNotIn("context", urlopen.call_args.kwargs)
+
+    def test_normalization_does_not_probe_or_downgrade(self):
+        self.assertEqual(_normalize_rpc_base("node.example"), "https://node.example")
+        self.assertEqual(_normalize_rpc_base("203.0.113.10"), "http://203.0.113.10")
+        self.assertEqual(
+            _normalize_rpc_base("https://node.example/rpc"), "https://node.example"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()