Support signed snapshot manifests in CLI

Author: Endogen

README.md

@@ -78,6 +78,9 @@ uv run xian doctor devnet-node --skip-live-checks
 uv run xian snapshot restore devnet-node
 ```
 
+For remote snapshot bootstrap, prefer a signed snapshot manifest plus trusted
+snapshot signing keys in the network manifest or node profile.
+
 ## Principles
 
 - `xian-cli` owns operator UX. Deterministic node logic stays in `xian-abci`,

src/xian_cli/cli.py

@@ -639,6 +639,7 @@ def _handle_network_create(args: argparse.Namespace) -> int:
         ),
         genesis_source=genesis_source,
         snapshot_url=args.snapshot_url,
+        snapshot_signing_keys=args.snapshot_signing_key or [],
         seed_nodes=args.seed or [],
         block_policy_mode=_pick_template_value(
             args.block_policy_mode,
@@ -705,6 +706,11 @@ def _handle_network_create(args: argparse.Namespace) -> int:
             snapshot_url=(
                 args.snapshot_url if validator["is_bootstrap"] else None
             ),
+            snapshot_signing_keys=(
+                list(args.snapshot_signing_key or [])
+                if validator["is_bootstrap"]
+                else []
+            ),
             service_node=(
                 _pick_template_value(
                     args.service_node,
@@ -1133,6 +1139,7 @@ def _handle_network_join(args: argparse.Namespace) -> int:
         seeds=args.seed or [],
         genesis_url=args.genesis_url,
         snapshot_url=args.snapshot_url,
+        snapshot_signing_keys=args.snapshot_signing_key or [],
         service_node=_pick_template_value(
             args.service_node,
             None if template is None else template.get("service_node"),
@@ -1623,6 +1630,20 @@ def _resolve_effective_snapshot_url(
     )
 
 
+def _resolve_effective_snapshot_signing_keys(
+    *,
+    profile: dict,
+    network: dict,
+) -> list[str]:
+    profile_keys = profile.get("snapshot_signing_keys")
+    if isinstance(profile_keys, list) and profile_keys:
+        return list(profile_keys)
+    network_keys = network.get("snapshot_signing_keys")
+    if isinstance(network_keys, list):
+        return list(network_keys)
+    return []
+
+
 def _restore_snapshot(
     *,
     base_dir: Path,
@@ -1654,6 +1675,10 @@ def _restore_snapshot(
         network=network,
         explicit_snapshot_url=explicit_snapshot_url,
     )
+    snapshot_signing_keys = _resolve_effective_snapshot_signing_keys(
+        profile=profile,
+        network=network,
+    )
     if not snapshot_url:
         raise ValueError(
             "no snapshot source configured; "
@@ -1662,7 +1687,10 @@ def _restore_snapshot(
 
     node_admin = get_node_admin_module()
     snapshot_archive_name = node_admin.apply_snapshot_archive(
-        snapshot_url, home
+        snapshot_url,
+        home,
+        trusted_manifest_public_keys=snapshot_signing_keys,
+        expected_chain_id=network.get("chain_id"),
     )
     return {
         "home": str(home),

src/xian_cli/models.py

@@ -409,6 +409,9 @@ def normalize_network_manifest(payload: dict) -> dict:
         "genesis_preset": genesis_preset,
         "genesis_time": genesis_time,
         "snapshot_url": _require_optional_str(payload, "snapshot_url"),
+        "snapshot_signing_keys": _require_str_list(
+            payload, "snapshot_signing_keys"
+        ),
         "seed_nodes": _require_str_list(payload, "seed_nodes"),
         "block_policy_mode": _require_block_policy_mode(
             payload, "block_policy_mode"
@@ -464,6 +467,9 @@ def normalize_node_profile(payload: dict) -> dict:
         "seeds": _require_str_list(payload, "seeds"),
         "genesis_url": _require_optional_str(payload, "genesis_url"),
         "snapshot_url": _require_optional_str(payload, "snapshot_url"),
+        "snapshot_signing_keys": _require_str_list(
+            payload, "snapshot_signing_keys"
+        ),
         "service_node": _require_bool(payload, "service_node", default=False),
         "home": _require_optional_str(payload, "home"),
         "pruning_enabled": _require_bool(
@@ -844,6 +850,7 @@ class NetworkManifest:
     genesis_preset: str | None = None
     genesis_time: str | None = None
     snapshot_url: str | None = None
+    snapshot_signing_keys: list[str] = field(default_factory=list)
     seed_nodes: list[str] = field(default_factory=list)
     block_policy_mode: str = "on_demand"
     block_policy_interval: str = "0s"
@@ -869,6 +876,7 @@ class NodeProfile:
     seeds: list[str] = field(default_factory=list)
     genesis_url: str | None = None
     snapshot_url: str | None = None
+    snapshot_signing_keys: list[str] = field(default_factory=list)
     service_node: bool = False
     home: str | None = None
     pruning_enabled: bool = False

src/xian_cli/parser.py

@@ -306,6 +306,14 @@ def build_parser() -> argparse.ArgumentParser:
         help="voting power for the generated initial validator entry",
     )
     create_parser.add_argument("--snapshot-url", help="optional snapshot URL")
+    create_parser.add_argument(
+        "--snapshot-signing-key",
+        action="append",
+        help=(
+            "trusted Ed25519 public key for signed snapshot manifests; "
+            "may be repeated"
+        ),
+    )
     create_parser.add_argument(
         "--seed",
         action="append",
@@ -673,6 +681,14 @@ def build_parser() -> argparse.ArgumentParser:
         "--snapshot-url",
         help="node-local snapshot URL override",
     )
+    join_parser.add_argument(
+        "--snapshot-signing-key",
+        action="append",
+        help=(
+            "trusted Ed25519 public key for signed snapshot manifests; "
+            "may be repeated"
+        ),
+    )
     join_parser.add_argument(
         "--init-node",
         action="store_true",

tests/test_cli.py

@@ -1990,7 +1990,9 @@ def test_node_init_can_restore_snapshot_from_manifest(self) -> None:
                         "--genesis-source",
                         str(genesis_source),
                         "--snapshot-url",
-                        "https://example.invalid/snapshot.tar.gz",
+                        "https://example.invalid/snapshot-manifest.json",
+                        "--snapshot-signing-key",
+                        "a" * 64,
                         "--output",
                         str(
                             base_dir
@@ -2052,14 +2054,16 @@ def test_node_init_can_restore_snapshot_from_manifest(self) -> None:
 
             self.assertEqual(exit_code, 0)
             snapshot_mock.assert_called_once_with(
-                "https://example.invalid/snapshot.tar.gz",
+                "https://example.invalid/snapshot-manifest.json",
                 home,
+                trusted_manifest_public_keys=["a" * 64],
+                expected_chain_id="xian-testnet-12",
             )
             result = json.loads(stdout.getvalue())
             self.assertTrue(result["snapshot_restored"])
             self.assertEqual(
                 result["effective_snapshot_url"],
-                "https://example.invalid/snapshot.tar.gz",
+                "https://example.invalid/snapshot-manifest.json",
             )
             self.assertEqual(
                 result["snapshot"]["snapshot_archive_name"],
@@ -3169,7 +3173,9 @@ def test_snapshot_restore_uses_effective_snapshot_url(self) -> None:
                         "--chain-id",
                         "xian-local-1",
                         "--snapshot-url",
-                        "https://example.invalid/network-snapshot.tar.gz",
+                        "https://example.invalid/network-snapshot-manifest.json",
+                        "--snapshot-signing-key",
+                        "b" * 64,
                         "--output",
                         str(
                             base_dir
@@ -3219,14 +3225,16 @@ def test_snapshot_restore_uses_effective_snapshot_url(self) -> None:
 
             self.assertEqual(exit_code, 0)
             snapshot_mock.assert_called_once_with(
-                "https://example.invalid/network-snapshot.tar.gz",
+                "https://example.invalid/network-snapshot-manifest.json",
                 home,
+                trusted_manifest_public_keys=["b" * 64],
+                expected_chain_id="xian-local-1",
             )
             result = json.loads(stdout.getvalue())
             self.assertEqual(result["home"], str(home))
             self.assertEqual(
                 result["snapshot_url"],
-                "https://example.invalid/network-snapshot.tar.gz",
+                "https://example.invalid/network-snapshot-manifest.json",
             )
             self.assertEqual(
                 result["snapshot_archive_name"],