auto-ci
158 lines
4.2 KiB
Python
158 lines
4.2 KiB
Python
#!/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())
|