feat(ens): add local ensdns service and host-side test docs

scripts/requirements-ens.txt

@@ -0,0 +1 @@
+web3>=7,<8

scripts/resolve_ens.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import sys
+
+from web3 import HTTPProvider, Web3
+
+
+ENS_REGISTRY_ADDRESS = Web3.to_checksum_address("0x00000000000C2E074eC69A0dFb2997BA6C7d2e1e")
+
+ENS_REGISTRY_ABI = [
+    {
+        "constant": True,
+        "inputs": [{"name": "node", "type": "bytes32"}],
+        "name": "resolver",
+        "outputs": [{"name": "", "type": "address"}],
+        "type": "function",
+    }
+]
+
+PUBLIC_RESOLVER_ABI = [
+    {
+        "constant": True,
+        "inputs": [{"name": "node", "type": "bytes32"}],
+        "name": "addr",
+        "outputs": [{"name": "", "type": "address"}],
+        "type": "function",
+    },
+    {
+        "constant": True,
+        "inputs": [{"name": "node", "type": "bytes32"}],
+        "name": "contenthash",
+        "outputs": [{"name": "", "type": "bytes"}],
+        "type": "function",
+    },
+    {
+        "constant": True,
+        "inputs": [
+            {"name": "node", "type": "bytes32"},
+            {"name": "key", "type": "string"},
+        ],
+        "name": "text",
+        "outputs": [{"name": "", "type": "string"}],
+        "type": "function",
+    },
+]
+
+
+def namehash(name: str) -> bytes:
+    node = b"\x00" * 32
+    labels = [label for label in name.strip().lower().split(".") if label]
+    for label in reversed(labels):
+        label_hash = Web3.keccak(text=label)
+        node = Web3.keccak(node + label_hash)
+    return node
+
+
+def build_client(rpc_url: str) -> Web3:
+    web3 = Web3(HTTPProvider(rpc_url, request_kwargs={"timeout": 10}))
+    if not web3.is_connected():
+        raise RuntimeError(f"Could not connect to Ethereum RPC: {rpc_url}")
+    return web3
+
+
+def resolve_name(client: Web3, name: str, include_text: bool) -> dict:
+    node = namehash(name)
+
+    registry = client.eth.contract(address=ENS_REGISTRY_ADDRESS, abi=ENS_REGISTRY_ABI)
+    resolver_address = registry.functions.resolver(node).call()
+
+    empty_address = "0x0000000000000000000000000000000000000000"
+
+    if resolver_address == empty_address:
+        return {
+            "name": name,
+            "resolver": None,
+            "address": None,
+            "contenthash": None,
+            "text": {} if include_text else None,
+        }
+
+    resolver = client.eth.contract(address=resolver_address, abi=PUBLIC_RESOLVER_ABI)
+
+    resolved_address = None
+    try:
+        addr = resolver.functions.addr(node).call()
+        if addr and addr != empty_address:
+            resolved_address = Web3.to_checksum_address(addr)
+    except Exception:
+        resolved_address = None
+
+    resolved_contenthash = None
+    try:
+        ch = resolver.functions.contenthash(node).call()
+        if ch:
+            resolved_contenthash = "0x" + ch.hex()
+    except Exception:
+        resolved_contenthash = None
+
+    result = {
+        "name": name,
+        "resolver": resolver_address,
+        "address": resolved_address,
+        "contenthash": resolved_contenthash,
+    }
+
+    if include_text:
+        text_keys = ["url", "description", "com.twitter", "com.github"]
+        text_values = {}
+        for key in text_keys:
+            try:
+                value = resolver.functions.text(node, key).call()
+            except Exception:
+                value = None
+            if value:
+                text_values[key] = value
+        result["text"] = text_values
+
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Resolve ENS (.eth) domains locally")
+    parser.add_argument("name", help="ENS name, e.g. vitalik.eth")
+    parser.add_argument(
+        "--rpc",
+        default="https://ethereum-rpc.publicnode.com",
+        help="Ethereum JSON-RPC endpoint",
+    )
+    parser.add_argument(
+        "--include-text",
+        action="store_true",
+        help="Include common ENS text records",
+    )
+    parser.add_argument(
+        "--pretty",
+        action="store_true",
+        help="Pretty-print JSON output",
+    )
+    args = parser.parse_args()
+
+    try:
+        client = build_client(args.rpc)
+        output = resolve_name(client, args.name, args.include_text)
+    except Exception as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+
+    if args.pretty:
+        print(json.dumps(output, indent=2, sort_keys=True))
+    else:
+        print(json.dumps(output))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())