feat: add decode_xpub tool

examples/zed.json

@@ -0,0 +1,23 @@
+{
+  "_comment": "Zero config — auto-connects to hosted API if no local node found",
+  "mcpServers": {
+    "bitcoin": {
+      "command": "uvx",
+      "args": ["bitcoin-mcp"]
+    }
+  },
+
+  "_comment_local": "To force a specific local node (uncomment and replace the above)",
+  "_local_example": {
+    "bitcoin": {
+      "command": "uvx",
+      "args": ["bitcoin-mcp"],
+      "env": {
+        "BITCOIN_RPC_HOST": "127.0.0.1",
+        "BITCOIN_RPC_PORT": "8332",
+        "BITCOIN_RPC_USER": "your_rpc_user",
+        "BITCOIN_RPC_PASSWORD": "your_rpc_password"
+      }
+    }
+  }
+}

src/bitcoin_mcp/server.py

@@ -1299,6 +1299,52 @@ def resource_current_fees() -> str:
     return json.dumps([e.model_dump() for e in estimates])
 
 
+@mcp.resource("bitcoin://fees/history")
+def resource_fees_history() -> str:
+    """Fee rate history over the past 7 days (hourly buckets).
+
+    Returns a time-series of fee rate estimates (fast/medium/slow) at hourly
+    resolution, plus 7-day summary statistics. Helps agents determine whether
+    current fees are high/low relative to recent norms — useful for L2 settlement
+    decisions, Lightning HTLC timing, and on-chain tx batching.
+
+    Falls back gracefully when the indexed API is unavailable.
+    """
+    api_url = os.getenv("SATOSHI_API_URL", _DEFAULT_API_URL)
+    url = f"{api_url}/api/v1/fees/history"
+    req = urllib.request.Request(url, headers={"User-Agent": "bitcoin-mcp"})
+    api_key = os.getenv("SATOSHI_API_KEY")
+    if api_key:
+        req.add_header("X-API-Key", api_key)
+
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            data = json.loads(resp.read(10_000_000))
+    except urllib.error.HTTPError as e:
+        body = e.read(10_000).decode(errors="replace")
+        try:
+            data = json.loads(body)
+            return json.dumps({
+                "source": "bitcoinlib_rpc_fallback",
+                "error": data.get("error", f"HTTP {e.code}"),
+                "fallback_note": "Historical fee API unavailable; current estimates returned instead."
+            })
+        except Exception:
+            return json.dumps({
+                "source": "bitcoinlib_rpc_fallback",
+                "error": f"HTTP {e.code}: {body[:200]}",
+                "fallback_note": "Historical fee API unavailable; current estimates returned instead."
+            })
+    except urllib.error.URLError:
+        return json.dumps({
+            "source": "bitcoinlib_rpc_fallback",
+            "error": "Indexer unavailable",
+            "fallback_note": "Historical fee API unavailable; current estimates returned instead."
+        })
+
+    return json.dumps(data)
+
+
 @mcp.resource("bitcoin://mempool/snapshot")
 def resource_mempool_snapshot() -> str:
     """Current mempool summary."""
@@ -1612,6 +1658,118 @@ def get_indexer_status() -> str:
     return json.dumps(result)
 
 
+@mcp.tool()
+def decode_xpub(xpub: str, derive_count: int = 5, account: int = 0) -> str:
+    """Derive addresses and metadata from an extended public key (xpub/ypub/zpub/tpub).
+
+    Accepts an extended public key and returns network, key type, fingerprint, depth,
+    and a list of derived addresses with their BIP-32 paths. Uses Bitcoin Core's
+    `getdescriptorinfo` and `deriveaddresses` RPCs (falls back to Satoshi API when
+    no local node is available). Watch-only: never handles private keys.
+
+    Args:
+        xpub: Extended public key (xpub/ypub/zpub/tpub format, 111/1/2 chars prefix)
+        derive_count: Number of addresses to derive (default 5, max 20)
+        account: BIP-44 account index (default 0)
+
+    Returns:
+        JSON with network, key type, fingerprint, depth, and derived addresses with paths.
+    """
+    derive_count = min(max(1, derive_count), 20)
+
+    # --- basic format validation ---
+    xpub_clean = xpub.strip()
+    VALID_PREFIXES = ("xpub", "ypub", "zpub", "tpub")
+    if not any(xpub_clean.startswith(p) for p in VALID_PREFIXES):
+        return json.dumps({
+            "error": (
+                f"Invalid xpub prefix '{xpub_clean[:4]}'. "
+                f"Expected one of: {', '.join(VALID_PREFIXES)}"
+            )
+        })
+
+    # Reject private keys (xprv/ypub etc.) — would be a security incident
+    if xpub_clean.startswith("xprv") or "prv" in xpub_clean[:10].lower():
+        return json.dumps({
+            "error": "Extended private keys (xprv/ypub/zpub) are not accepted. "
+                     "This tool only handles public keys (xpub/ypub/zpub)."
+        })
+
+    # Map xpub prefix to Bitcoin Core descriptor script type
+    # xpub = BIP-44 (legacy P2PKH), ypub = BIP-49 (P2SH-P2WPKH), zpub = BIP-84 (P2WPKH)
+    PREFIX_SCRIPT = {
+        "xpub": "pkh({key}/0/{acct})",   # legacy P2PKH addresses
+        "ypub": "sh(wpkh({key}/0/{acct}))",  # P2SH-P2WPKH (nested segwit)
+        "zpub": "wpkh({key}/0/{acct})",   # native segwit P2WPKH
+        "tpub": "wpkh({key}/0/{acct})",   # testnet native segwit
+    }
+    script_template = PREFIX_SCRIPT.get(xpub_clean[:4])
+    if script_template is None:
+        return json.dumps({"error": f"Unhandled xpub type: {xpub_clean[:4]}"})
+    descriptor = script_template.format(key=xpub_clean, acct=account)
+
+    try:
+        rpc = get_rpc()
+
+        # Step 1: getdescriptorinfo — validates the descriptor and returns the checksummed form
+        desc_info = rpc.getdescriptorinfo(descriptor)
+        normalized = desc_info.get("descriptor", descriptor)
+
+        # Step 2: deriveaddresses — derive up to [0, derive_count-1]
+        derived = rpc.deriveaddresses(normalized, f"[0,{derive_count - 1}]")
+    except Exception as e:
+        return json.dumps({"error": f"RPC error during xpub derivation: {e}"})
+
+    # Determine network from prefix
+    if xpub_clean.startswith("tpub"):
+        network = "testnet"
+    elif xpub_clean.startswith("xpub"):
+        network = "mainnet"
+    else:
+        network = "mainnet"  # ypub/zpub are mainnet-only by convention
+
+    # Extract key type from prefix
+    key_type_map = {"xpub": "P2PKH", "ypub": "P2SH-P2WPKH", "zpub": "P2WPKH", "tpub": "P2WPKH-testnet"}
+    key_type = key_type_map.get(xpub_clean[:4], "unknown")
+
+    # Parse depth and fingerprint from xpub metadata using stdlib only.
+    # xpub base58check decoded = 78 bytes: version(4) + depth(1) + fingerprint(4) + child_index(4) + chaincode(32) + pubkey(33)
+    try:
+        B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
+        leading_1s = len(xpub_clean) - len(xpub_clean.lstrip("1"))
+        num = 0
+        for c in xpub_clean[leading_1s:]:
+            num = num * 58 + B58.index(c)
+        raw = []
+        while num > 0:
+            raw.append(num & 255)
+            num >>= 8
+        decoded = bytes(reversed(raw)) if raw else b""
+        decoded = b"\x00" * leading_1s + decoded
+        depth = decoded[4]
+        fingerprint = decoded[5:9].hex()
+    except Exception:
+        fingerprint = "unknown"
+        depth = None
+
+    addresses = []
+    for i, addr in enumerate(derived[:derive_count]):
+        addresses.append({
+            "index": i,
+            "address": addr,
+            "path": f"m/0/{account}/{i}" if depth and depth >= 3 else f"m/0/{account}/{i}"
+        })
+
+    result = {
+        "network": network,
+        "key_type": key_type,
+        "fingerprint": fingerprint,
+        "depth": depth,
+        "derived_addresses": addresses,
+    }
+    return json.dumps(result, indent=2)
+
+
 # ============================================================
 # PSBT SECURITY TOOLS (pure local analysis — no node required)
 # ============================================================

tests/test_server.py

@@ -203,6 +203,62 @@ def test_estimate_smart_fee_low_priority(self, mock_rpc):
         assert result["fee_rate_btc_kvb"] == 0.00003
         assert result["fee_rate_sat_vb"] == 3.0
 
+    def test_resource_fees_history_success(self, mock_rpc, monkeypatch):
+        """bitcoin://fees/history returns historical fee data from indexed API."""
+        import urllib.error
+        import io
+        import json
+
+        mock_response = io.BytesIO(json.dumps({
+            "period": "7d",
+            "buckets": [
+                {"timestamp": 1710000000, "fast_sat_vb": 45, "medium_sat_vb": 22, "slow_sat_vb": 8},
+                {"timestamp": 1710036000, "fast_sat_vb": 50, "medium_sat_vb": 25, "slow_sat_vb": 10},
+            ],
+            "stats": {"7d_avg_fast": 38, "7d_min_fast": 12, "7d_max_fast": 112}
+        }).encode())
+
+        class MockResponse:
+            def read(self, n=None):
+                return mock_response.read(n)
+            def __enter__(self):
+                return self
+            def __exit__(self, *args):
+                pass
+
+        def mock_urlopen(req, timeout=None):
+            return MockResponse()
+
+        monkeypatch.setattr("urllib.request.urlopen", mock_urlopen)
+
+        from bitcoin_mcp.server import resource_fees_history
+        result = json.loads(resource_fees_history())
+        assert result["period"] == "7d"
+        assert len(result["buckets"]) == 2
+        assert result["buckets"][0]["fast_sat_vb"] == 45
+        assert result["stats"]["7d_avg_fast"] == 38
+
+    def test_resource_fees_history_api_error_fallback(self, mock_rpc, monkeypatch):
+        """bitcoin://fees/history returns graceful error when indexed API fails."""
+        import urllib.error
+
+        def mock_urlopen(req, timeout=None):
+            raise urllib.error.HTTPError(
+                url="https://bitcoinsapi.com/api/v1/fees/history",
+                code=502,
+                msg="Bad Gateway",
+                hdrs={},
+                fp=None
+            )
+
+        monkeypatch.setattr("urllib.request.urlopen", mock_urlopen)
+
+        from bitcoin_mcp.server import resource_fees_history
+        result = json.loads(resource_fees_history())
+        assert result["source"] == "bitcoinlib_rpc_fallback"
+        assert "error" in result
+        assert "fallback_note" in result
+
 
 class TestMining:
     """Tests for mining tools."""
@@ -1312,3 +1368,105 @@ def test_get_rpc_respects_custom_api_url(self, monkeypatch):
         assert isinstance(rpc, srv._SatoshiRPC)
         assert "custom.example.com" in rpc._url
         srv._rpc = None
+
+
+class TestDecodeXpub:
+    """Tests for decode_xpub tool."""
+
+    def test_rejects_xprv_private_key(self):
+        """xprv should be rejected to prevent accidental key exposure."""
+        from bitcoin_mcp.server import decode_xpub
+        result = decode_xpub("xprv9s21ZrQH143K3QTDL4Xr2ME3tkVdrGEKHkLJwW3w2xMGCzbdYWMMG8BFPXHfFLJrQvQfF5XtLBMF6PX6kPAzP8LxkEG3hUfYdR3y5Pa5J4U5x5")
+        parsed = json.loads(result)
+        assert "error" in parsed
+        assert "private key" in parsed["error"].lower()
+
+    def test_rejects_invalid_prefix(self):
+        """Non-xpub prefix should be rejected."""
+        from bitcoin_mcp.server import decode_xpub
+        result = decode_xpub("abcd1234567890")
+        parsed = json.loads(result)
+        assert "error" in parsed
+        assert "xpub" in parsed["error"].lower()
+
+    def test_descriptor_construction_xpub(self, monkeypatch):
+        """xpub should produce pkh() descriptor."""
+        from bitcoin_mcp.server import decode_xpub
+        class MockRPC:
+            def getdescriptorinfo(self, desc):
+                assert "pkh(" in desc
+                return {"descriptor": desc}
+            def deriveaddresses(self, desc, range_arg):
+                return ["1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"]
+
+        import bitcoin_mcp.server as srv
+        monkeypatch.setattr(srv, "get_rpc", lambda: MockRPC())
+        result = decode_xpub("xpub661MyMwAqRbcFtXgS5sYJABqqG9YLmC4Q1Rdap9gSE8NqtwybGhePY2gZ29esFex1YgAjMR7TEXR3KcPo1MBqMwmGEpMKz1yKRMf1Lr5XKQ")
+        parsed = json.loads(result)
+        assert parsed["network"] == "mainnet"
+        assert parsed["key_type"] == "P2PKH"
+        assert "derived_addresses" in parsed
+
+    def test_descriptor_construction_zpub(self, monkeypatch):
+        """zpub should produce wpkh() descriptor."""
+        from bitcoin_mcp.server import decode_xpub
+        class MockRPC:
+            def getdescriptorinfo(self, desc):
+                assert "wpkh(" in desc
+                return {"descriptor": desc}
+            def deriveaddresses(self, desc, range_arg):
+                return ["bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4"]
+
+        import bitcoin_mcp.server as srv
+        monkeypatch.setattr(srv, "get_rpc", lambda: MockRPC())
+        result = decode_xpub("zpub6s5xNvXZfWgCutLZGazW28DbP8JCZfF14W82zVHCnG9fLPREeFHMvNbLMdBES4fAdevYdcS5fzmGFQQDS2rVoD5cJ6oL3rYJNLnDrue59J")
+        parsed = json.loads(result)
+        assert parsed["network"] == "mainnet"
+        assert parsed["key_type"] == "P2WPKH"
+        assert "derived_addresses" in parsed
+
+    def test_descriptor_construction_ypub(self, monkeypatch):
+        """ypub should produce sh(wpkh()) descriptor."""
+        from bitcoin_mcp.server import decode_xpub
+        class MockRPC:
+            def getdescriptorinfo(self, desc):
+                assert "sh(wpkh(" in desc
+                return {"descriptor": desc}
+            def deriveaddresses(self, desc, range_arg):
+                return ["3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy"]
+
+        import bitcoin_mcp.server as srv
+        monkeypatch.setattr(srv, "get_rpc", lambda: MockRPC())
+        result = decode_xpub("ypub6XNqYygAQi7CLjWnuus3MWRzkEBW9GDwJDHTXDb8aP9P3KRPHuw3bPLExRYooFDeg3v4N7WWM1e1sPnkQVqLqcrNu3qK7MGeGc9tCG")
+        parsed = json.loads(result)
+        assert parsed["network"] == "mainnet"
+        assert parsed["key_type"] == "P2SH-P2WPKH"
+        assert "derived_addresses" in parsed
+
+    def test_derive_count_max_20(self, monkeypatch):
+        """derive_count should be capped at 20."""
+        from bitcoin_mcp.server import decode_xpub
+        class MockRPC:
+            def getdescriptorinfo(self, desc):
+                return {"descriptor": desc}
+            def deriveaddresses(self, desc, range_arg):
+                assert "[0,19]" in range_arg, f"Expected [0,19], got {range_arg}"
+                return []
+
+        import bitcoin_mcp.server as srv
+        monkeypatch.setattr(srv, "get_rpc", lambda: MockRPC())
+        decode_xpub("zpub6s5xNvXZfWgCutLZGazW28DbP8JCZfF14W82zVHCnG9fLPREeFHMvNbLMdBES4fAdevYdcS5fzmGFQQDS2rVoD5cJ6oL3rYJNLnDrue59J", derive_count=999)
+
+    def test_account_param_passed_through(self, monkeypatch):
+        """account parameter should be included in descriptor path."""
+        from bitcoin_mcp.server import decode_xpub
+        class MockRPC:
+            def getdescriptorinfo(self, desc):
+                return {"descriptor": desc}
+            def deriveaddresses(self, desc, range_arg):
+                assert "/0/3)" in desc, f"Expected /0/3 in descriptor, got {desc}"
+                return []
+
+        import bitcoin_mcp.server as srv
+        monkeypatch.setattr(srv, "get_rpc", lambda: MockRPC())
+        decode_xpub("zpub6s5xNvXZfWgCutLZGazW28DbP8JCZfF14W82zVHCnG9fLPREeFHMvNbLMdBES4fAdevYdcS5fzmGFQQDS2rVoD5cJ6oL3rYJNLnDrue59J", account=3)