YOINK

pyproject.toml

@@ -1,21 +1,19 @@
 [tool.poetry]
 name = "snapcat"
-version = "0.1.3"
+version = "0.2.0"
 description = "snapcat is a Python command-line tool that helps monitor CATs (Chia Asset Tokens) on Chia Blockchain."
 authors = ["Dexie Contributors <pypi@dexie.space>"]
 readme = "README.md"
 packages = [{include = "snapcat", from = "src"}]
 
 [tool.poetry.dependencies]
 python = "^3.12"
-rich-click = "^1.8.0"
+rich-click = "^1.9.2"
 python-dotenv = "^1.0.1"
 asyncio = "^3.4.3"
 
-
 [tool.poetry.group.chia.dependencies]
-chia-blockchain = {version = "^2.3.1", allow-prereleases = true}
-
+chia-blockchain = {version = "^2.5.5", allow-prereleases = true}
 
 [tool.poetry.group.dev.dependencies]
 mypy = "^1.10.0"

src/snapcat/__init__.py

@@ -1,8 +1,8 @@
-from pkg_resources import DistributionNotFound, get_distribution
+from importlib.metadata import PackageNotFoundError, version as importlib_version
 
 try:
-    __version__ = get_distribution("snapcat").version
-except DistributionNotFound:
+    __version__ = importlib_version("snapcat")
+except PackageNotFoundError:
     # package is not installed
     __version__ = "unknown"
 

src/snapcat/cat_utils.py

@@ -1,18 +1,18 @@
 from typing import Dict, List, Tuple, Union
 from chia.types.blockchain_format.coin import Coin
-from chia.types.blockchain_format.sized_bytes import bytes32
+from chia_rs.sized_bytes import bytes32
 from chia.types.blockchain_format.program import Program
 from chia.types.coin_spend import CoinSpend
 from chia.types.condition_opcodes import ConditionOpcode
 from chia.types.condition_with_args import ConditionWithArgs
-from chia.util.condition_tools import (
-    conditions_dict_for_solution,
-)
-from chia.util.ints import uint64
+from chia.consensus.condition_tools import conditions_dict_for_solution
+from chia_rs.sized_ints import uint64
 from clvm.casts import int_from_bytes
 
 from chia.wallet.cat_wallet.cat_utils import match_cat_puzzle
+from chia.wallet.vc_wallet.vc_drivers import match_revocation_layer
 from chia.wallet.uncurried_puzzle import uncurry_puzzle
+from chia.wallet.cat_wallet.cat_utils import CAT_MOD_HASH
 
 
 def created_outputs_for_conditions_dict(
@@ -32,10 +32,11 @@ def created_outputs_for_conditions_dict(
 
 def extract_cat(
     expected_tail_hash: bytes32,
+    hidden_puzzle_hash: bytes32 | None,
     coin_spend: CoinSpend,
 ) -> Union[None, Tuple[Program, Program, Program, Program, Program]]:
-    outer_puzzle = coin_spend.puzzle_reveal.to_program()
-    outer_solution = coin_spend.solution.to_program()
+    outer_puzzle = coin_spend.puzzle_reveal
+    outer_solution = Program.from_bytes(coin_spend.solution.to_bytes())
     cat_curried_args = match_cat_puzzle(uncurry_puzzle(outer_puzzle))
     if cat_curried_args is None:
         return None
@@ -45,12 +46,36 @@ def extract_cat(
         return None
 
     # CAT2
-    _, tail_program_hash, inner_puzzle = cat_curried_args
+    cat_mod_hash, tail_program_hash, inner_puzzle = cat_curried_args
     tail_hash = bytes32(tail_program_hash.as_atom())
-    if tail_hash != expected_tail_hash:
+    cat_mod_hash = bytes32(cat_mod_hash.as_atom())
+    if tail_hash != expected_tail_hash or cat_mod_hash != CAT_MOD_HASH:
         return None
 
-    inner_solution = outer_solution.first()
+    inner_solution = None
+    if hidden_puzzle_hash is None:
+        inner_solution = outer_solution.first()
+    else:
+        revocation_layer_curried_args = match_revocation_layer(uncurry_puzzle(inner_puzzle))
+        if revocation_layer_curried_args is None:
+            return None
+
+        revocation_layer_curried_args = list(revocation_layer_curried_args)
+        if len(revocation_layer_curried_args) != 2:
+            return None
+
+        hidden_puzzle_hash_arg, inner_puzzle_hash = revocation_layer_curried_args
+        if hidden_puzzle_hash_arg != hidden_puzzle_hash:
+            return None
+
+        interim_solution = outer_solution.first()
+        hidden = bool(interim_solution.first().as_atom())
+        inner_puzzle = interim_solution.rest().first()
+        actual_inner_puzzle_hash = inner_puzzle.get_tree_hash()
+        inner_solution = interim_solution.rest().rest().first()
+
+        if (hidden and actual_inner_puzzle_hash != hidden_puzzle_hash) or (not hidden and actual_inner_puzzle_hash != inner_puzzle_hash):
+            return None
 
     return tail_hash, outer_puzzle, outer_solution, inner_puzzle, inner_solution
 

src/snapcat/shared.py

@@ -1,5 +1,10 @@
+from chia.full_node.full_node_rpc_client import FullNodeRpcClient
+from chia_rs.sized_bytes import bytes32
 import rich_click as click
-from chia.types.blockchain_format.sized_bytes import bytes32
+import aiohttp
+import random
+import time
+import json
 
 
 class Bytes32ParamType(click.ParamType):
@@ -11,3 +16,35 @@ def convert(self, value, param, ctx):  # type: ignore
             return bytes32_value
         except ValueError:
             self.fail(f"Invalid bytes32: {value}", param, ctx)
+
+class HttpFullNodeRpcClient(FullNodeRpcClient):
+    def __init__(self, rpc_url):
+        self.rpc_url = rpc_url
+        super().__init__(
+            url=rpc_url,
+            session=aiohttp.ClientSession(),
+            ssl_context=None,
+            hostname='localhost',
+            port=1337,
+        )
+        self.closing_task = None
+
+
+    async def fetch(self, path, request_json):
+        rpc_url = self.rpc_url
+        if "," in rpc_url:
+            rpc_url = rpc_url.split(",")[0]
+            if 'push_tx' in path or 'get_fee_estimate' in path:
+                rpc_url = random.choice(self.rpc_url.split(",")[1:])
+
+        async with self.session.post(rpc_url + path, json=request_json) as response:
+            if 'push_tx' in path or 'get_fee_estimate' in path:
+                print(f"Using {rpc_url} for {path}:", rpc_url)
+                print("Response:", await response.text())
+
+            response.raise_for_status()
+
+            res_json = json.loads(await response.text())
+            if not res_json["success"]:
+                raise ValueError(res_json)
+            return res_json

src/snapcat/show_cmd/__init__.py

@@ -6,7 +6,7 @@
 from rich.console import Console
 from typing import Optional
 
-from chia.types.blockchain_format.sized_bytes import bytes32
+from chia_rs.sized_bytes import bytes32
 
 from snapcat.shared import Bytes32ParamType
 

src/snapcat/sync_cmd/__init__.py

@@ -1,8 +1,8 @@
+from chia.full_node.full_node_rpc_client import FullNodeRpcClient
+from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint32
 import asyncio
 import aiosqlite
-from chia.rpc.full_node_rpc_client import FullNodeRpcClient
-from chia.types.blockchain_format.sized_bytes import bytes32
-from chia.util.ints import uint32
 
 import logging
 import rich_click as click
@@ -24,7 +24,7 @@
     target_height,
 )
 
-from snapcat.shared import Bytes32ParamType
+from snapcat.shared import Bytes32ParamType, HttpFullNodeRpcClient
 from snapcat.sync_cmd.sync import get_full_node_synced, process_block
 
 log = logging.getLogger("snapcat")
@@ -61,7 +61,13 @@ async def syncing_full_node(full_node_rpc, sync_progress):
         await asyncio.sleep(5)
 
 
-async def process_blocks(full_node_rpc, sync_progress, db, tail_hash: bytes32):
+async def process_blocks(
+    full_node_rpc,
+    sync_progress,
+    db,
+    tail_hash: bytes32,
+    hidden_puzzle_hash: bytes32 | None,
+):
     global abort_height
     async with db.execute(
         "SELECT value FROM config WHERE key = 'last_block_height'"
@@ -92,7 +98,7 @@ async def process_blocks(full_node_rpc, sync_progress, db, tail_hash: bytes32):
             print(message)
             break
 
-        await process_block(full_node_rpc, db, tail_hash, height)
+        await process_block(full_node_rpc, db, tail_hash, hidden_puzzle_hash, height)
 
         sync_progress.update(
             process_blocks_task_id,
@@ -111,8 +117,27 @@ async def process_blocks(full_node_rpc, sync_progress, db, tail_hash: bytes32):
     help="The TAIL hash of CAT",
     type=Bytes32ParamType(),
 )
+@click.option(
+    "-c",
+    "--coinset-url",
+    required=False,
+    help="Coinset.org API URL (if provided, local full node will not be used)",
+    type=str,
+)
+@click.option(
+    "--hidden-puzzle-hash",
+    required=False,
+    help="Optional hidden puzzle hash (Bytes32)",
+    type=Bytes32ParamType(),
+    default=None,
+)
 @click.pass_context
-def sync(ctx, tail_hash: bytes32):
+def sync(
+    ctx,
+    tail_hash: bytes32,
+    coinset_url: str = "",
+    hidden_puzzle_hash: bytes32 | None = None,
+):
     async def _sync(tail_hash: bytes32) -> None:
         db_file_name = (
             ctx.obj["db_file_name"]
@@ -155,6 +180,12 @@ async def _sync(tail_hash: bytes32) -> None:
                 """,
                 [tail_hash.hex()],
             )
+            await db.execute(
+                """
+                INSERT OR IGNORE INTO config(key, value) VALUES('hidden_puzzle_hash', ?);
+                """,
+                [hidden_puzzle_hash.hex() if hidden_puzzle_hash is not None else ''],
+            )
             await db.commit()
 
             block_progress = Progress(
@@ -166,14 +197,31 @@ async def _sync(tail_hash: bytes32) -> None:
                 refresh_per_second=5,
             )
             with block_progress:
-                async with FullNodeRpcClient.create_as_context(
-                    self_hostname,
-                    full_node_rpc_port,
-                    chia_root,
-                    chia_config,
-                ) as full_node_rpc:
+                if coinset_url:
+                    full_node_rpc = HttpFullNodeRpcClient(coinset_url)
                     await syncing_full_node(full_node_rpc, block_progress)
-                    await process_blocks(full_node_rpc, block_progress, db, tail_hash)
+                    await process_blocks(
+                        full_node_rpc,
+                        block_progress,
+                        db,
+                        tail_hash,
+                        hidden_puzzle_hash,
+                    )
+                else:
+                    async with FullNodeRpcClient.create_as_context(
+                        self_hostname,
+                        full_node_rpc_port,
+                        chia_root,
+                        chia_config,
+                    ) as full_node_rpc:
+                        await syncing_full_node(full_node_rpc, block_progress)
+                        await process_blocks(
+                            full_node_rpc,
+                            block_progress,
+                            db,
+                            tail_hash,
+                            hidden_puzzle_hash,
+                        )
 
     try:
         console.print("[bold red]press Ctrl+C to exit.")

src/snapcat/sync_cmd/sync.py

@@ -1,10 +1,11 @@
-from clvm.casts import int_to_bytes
-from chia.rpc.full_node_rpc_client import FullNodeRpcClient
-from chia.types.blockchain_format.sized_bytes import bytes32
+from chia.full_node.full_node_rpc_client import FullNodeRpcClient
 from chia.types.coin_spend import CoinSpend
+from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint32
+from clvm.casts import int_to_bytes
 from chia.util.hash import std_hash
-from chia.util.ints import uint32
-from chia.wallet.cat_wallet.cat_utils import CAT_MOD
+from chia.wallet.cat_wallet.cat_utils import CAT_MOD, CAT_MOD_HASH
+from chia.wallet.vc_wallet.vc_drivers import REVOCATION_LAYER, REVOCATION_LAYER_HASH
 import logging
 from typing import List, Optional, Tuple
 
@@ -28,6 +29,7 @@ async def get_full_node_synced(
 async def process_coin_spends(
     db,
     expected_tail_hash: bytes32,
+    hidden_puzzle_hash: bytes32 | None,
     height,
     header_hash: str,
     coin_spends: Optional[List[CoinSpend]],
@@ -43,15 +45,12 @@ async def process_coin_spends(
     )
 
     for coin_spend in coin_spends:
-        result = extract_cat(expected_tail_hash, coin_spend)
+        result = extract_cat(expected_tail_hash, hidden_puzzle_hash, coin_spend)
 
         if result is None:
             log.debug(f"{expected_tail_hash.hex()} CAT coin spend not found")
         else:
-            outer_puzzle = coin_spend.puzzle_reveal.to_program()
-            outer_solution = coin_spend.solution.to_program()
-            inner_solution = outer_solution.first()
-            (_, outer_puzzle, _, inner_puzzle, _) = result
+            (_, outer_puzzle, outer_solution, inner_puzzle, inner_solution) = result
 
             coin_spend_coin_name = coin_spend.coin.name().hex()
 
@@ -73,11 +72,19 @@ async def process_coin_spends(
                 ],
             )
             for coin in inner_puzzle_create_coin_conditions:
+                inner_puzzle_hash = coin.puzzle_hash
+                if hidden_puzzle_hash is not None:
+                    inner_puzzle_hash = REVOCATION_LAYER.curry(
+                        REVOCATION_LAYER_HASH,
+                        hidden_puzzle_hash,
+                        inner_puzzle_hash,
+                    ).get_tree_hash()
+
                 outer_puzzle_hash = CAT_MOD.curry(
-                    CAT_MOD.get_tree_hash(),
+                    CAT_MOD_HASH,
                     expected_tail_hash,
-                    coin.puzzle_hash,
-                ).get_tree_hash_precalc(coin.puzzle_hash)
+                    inner_puzzle_hash,
+                ).get_tree_hash_precalc(inner_puzzle_hash)
 
                 created_coin_name = std_hash(
                     bytes32.fromhex(coin_spend_coin_name)
@@ -99,7 +106,11 @@ async def process_coin_spends(
 
 
 async def process_block(
-    full_node_rpc: FullNodeRpcClient, db, tail_hash: bytes32, height: int
+    full_node_rpc: FullNodeRpcClient,
+    db,
+    tail_hash: bytes32,
+    hidden_puzzle_hash: bytes32 | None,
+    height: int,
 ):
     block_record = await full_node_rpc.get_block_record_by_height(height)
     if block_record is None:
@@ -116,7 +127,12 @@ async def process_block(
         if coin_spends is not None and len(coin_spends) > 0:
             log.debug("%i spends found in block %i", len(coin_spends), height)
             await process_coin_spends(
-                db, tail_hash, height, block_record.header_hash, coin_spends
+                db,
+                tail_hash,
+                hidden_puzzle_hash,
+                height,
+                block_record.header_hash,
+                coin_spends,
             )
 
         else: