钱包：连接与读取
每个Owner API v3调用都在加密信封内传输。客户端获得共享机密后，每个方法都是普通的JSON-RPC调用。
启动API：
epic-wallet owner_api
握手流程
客户端生成一个secp256k1密钥对，通过init_secure_api交换公钥，并将后续每个请求包装在encrypted_request_v3中。第一个包装调用是open_wallet，它返回后续所有调用所携带的令牌。逐条消息的说明见Owner API参考。
- 生成密钥对. 在客户端生成一个secp256k1密钥对，每次会话使用一次。
- init_secure_api. 发送你的压缩公钥，钱包返回其自身的公钥。这是唯一的未加密调用。
- 推导共享机密. 用你的标量乘以钱包公钥，保留原始x坐标，32字节，不做哈希处理。对其进行哈希是最常见的实现错误。
- open_wallet，已加密. encrypted_request_v3内的第一个调用。需要钱包密码，返回一个令牌。
- 后续所有调用. 使用AES-256-GCM，每次生成新的12字节nonce，16字节tag附加在密文之后，令牌置于params中。
完整客户端
需要requests、coincurve和pycryptodome。从examples目录运行。
python/epic_wallet.py，客户端
examples/python/epic_wallet.py
"""Minimal Epic wallet Owner API v3 client.
Requires: requests, coincurve, pycryptodome.
Run `epic-wallet owner_api` first. See https://devdocs.epiccash.com/examples/wallet-connect
Override the defaults for a wallet outside the usual location:
EPIC_OWNER_URL=http://127.0.0.1:3420/v3/owner
EPIC_OWNER_SECRET=/path/to/.owner_api_secret
"""
from __future__ import annotations
import base64
import json
import os
from pathlib import Path
from typing import Any
import requests
from coincurve import PrivateKey, PublicKey
from Crypto.Cipher import AES
from tx_args import EPIC, init_tx_args, invoice_tx_args
__all__ = ["EPIC", "EpicWallet", "init_tx_args", "invoice_tx_args"]
class EpicWallet:
"""Talks to `epic-wallet owner_api` over the encrypted v3 surface."""
def __init__(
self,
url: str | None = None,
secret_path: Path | None = None,
) -> None:
self.url = url or os.environ.get("EPIC_OWNER_URL", "http://127.0.0.1:3420/v3/owner")
if secret_path is None:
env_secret = os.environ.get("EPIC_OWNER_SECRET")
secret_path = (
Path(env_secret)
if env_secret
else Path.home() / ".epic" / "main" / ".owner_api_secret"
)
# The token returned by open_wallet authorises every v3 call. An HTTP credential is
# sent only as a pass-through, for a listener published behind a proxy that wants one.
self.auth = (
("epic", secret_path.read_text().strip()) if secret_path.is_file() else None
)
self.shared_secret: str | None = None
self.token: str | None = None
# --- transport -----------------------------------------------------------
def _post(self, payload: dict[str, Any]) -> Any:
response = requests.post(self.url, json=payload, auth=self.auth, timeout=120)
if response.status_code == 401:
raise RuntimeError("Unauthorized. A proxy in front of the listener rejected the request")
response.raise_for_status()
return self._unwrap(response.json())
@staticmethod
def _unwrap(body: dict[str, Any]) -> Any:
"""Unwrap the JSON-RPC error field and Epic's inner Ok/Err envelope."""
if "error" in body:
raise RuntimeError(f"JSON-RPC error: {body['error']}")
result = body.get("result", body)
if isinstance(result, dict):
if "Err" in result:
raise RuntimeError(f"Wallet error: {result['Err']}")
if "Ok" in result:
return result["Ok"]
return result
# --- handshake -----------------------------------------------------------
def connect(self) -> None:
"""Perform the ECDH handshake and store the shared secret."""
ephemeral = PrivateKey(os.urandom(32))
wallet_pubkey_hex = self._post(
{
"jsonrpc": "2.0",
"id": 1,
"method": "init_secure_api",
"params": {"ecdh_pubkey": ephemeral.public_key.format().hex()},
}
)
# Multiply the wallet's point by our scalar, then keep only the x coordinate.
# format() returns 33 bytes: a 1-byte parity prefix followed by x.
point = PublicKey(bytes.fromhex(wallet_pubkey_hex)).multiply(ephemeral.secret)
self.shared_secret = point.format().hex()[2:]
def open_wallet(self, password: str, name: str | None = None) -> str:
"""Unlock the wallet and store the session token."""
if self.shared_secret is None:
self.connect()
self.token = self.call("open_wallet", {"name": name, "password": password})
return self.token
# --- encryption ----------------------------------------------------------
def call(self, method: str, params: dict[str, Any] | list[Any]) -> Any:
"""Call an Owner API method inside the encrypted envelope."""
if self.shared_secret is None:
raise RuntimeError("call connect() first")
inner = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}
key = bytes.fromhex(self.shared_secret)
nonce = os.urandom(12)
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
ciphertext, tag = cipher.encrypt_and_digest(json.dumps(inner).encode())
envelope = self._post(
{
"jsonrpc": "2.0",
"id": 1,
"method": "encrypted_request_v3",
"params": {
"nonce": nonce.hex(),
# The auth tag is appended to the ciphertext.
"body_enc": base64.b64encode(ciphertext + tag).decode(),
},
}
)
blob = base64.b64decode(envelope["body_enc"])
decipher = AES.new(key, AES.MODE_GCM, nonce=bytes.fromhex(envelope["nonce"]))
plaintext = decipher.decrypt_and_verify(blob[:-16], blob[-16:])
# The decrypted body is itself a JSON-RPC response, so unwrap again.
return self._unwrap(json.loads(plaintext))
使用方法：
examples/python/balance.py
"""Read an Epic wallet balance through the Owner API v3.
Usage: uv run balance.py, or python balance.py, with `epic-wallet owner_api` running.
"""
from __future__ import annotations
import os
from epic_wallet import EpicWallet
def main() -> None:
wallet = EpicWallet()
wallet.open_wallet(password=os.environ["EPIC_WALLET_PASSWORD"])
# retrieve_summary_info returns (validated_against_node, WalletInfo)
validated, summary = wallet.call(
"retrieve_summary_info",
{"token": wallet.token, "refresh_from_node": True, "minimum_confirmations": 3},
)
def epic(freemen: str | int) -> str:
return f"{int(freemen) / 1e8:.8f}"
print(f"validated against node: {validated}")
print(f"height: {summary['last_confirmed_height']}")
print(f"total: {epic(summary['total'])} EPIC")
print(f"spendable: {epic(summary['amount_currently_spendable'])} EPIC")
print(f"awaiting confirmation: {epic(summary['amount_awaiting_confirmation'])} EPIC")
print(f"locked: {epic(summary['amount_locked'])} EPIC")
if __name__ == "__main__":
main()
EPIC_WALLET_PASSWORD=... python python/balance.py
读取余额
retrieve_summary_info返回一个二元数组：数据是否已通过节点验证，以及汇总信息本身。
| 字段 | 含义 |
|---|---|
last_confirmed_height | 数据对应的区块高度 |
total | 钱包已知的全部信息 |
amount_currently_spendable | 当前可发送的金额 |
amount_awaiting_confirmation | 已接收但确认数不足 |
amount_awaiting_finalization | 转账仍在进行中的部分 |
amount_locked | 已预留为您发起的转账的输入 |
amount_immature | 仍在成熟期内的coinbase输出 |
total与amount_currently_spendable之间的差额来自输出（output）与锁定。
Amounts are strings in JSON
u64的值序列化为字符串，以避免JavaScript数字精度问题。将其解析为整数，除以100,000,000得到EPIC。
列出输出与交易
retrieve_outputs和retrieve_txs各自返回一个对象，其中包含pager以及记录列表，而非裸数组。需提供limit、offset和sort_order：不设限制时，大型钱包会在单次响应中返回所有记录。
examples/python/list_paged.py
"""Page through a wallet's outputs and transactions.
Both calls return an object carrying a `pager` alongside the records.
Usage: python list_paged.py, with `epic-wallet owner_api` running.
See https://devdocs.epiccash.com/examples/wallet-connect
"""
from __future__ import annotations
import os
from epic_wallet import EpicWallet
PAGE = 100
def outputs(wallet: EpicWallet, offset: int = 0, limit: int = PAGE) -> dict:
"""One page of outputs, newest first."""
return wallet.call(
"retrieve_outputs",
{
"token": wallet.token,
"include_spent": False,
"refresh_from_node": True,
"tx_id": None,
"limit": limit,
"offset": offset,
"sort_order": "desc",
},
)
def transactions(wallet: EpicWallet, offset: int = 0, limit: int = PAGE) -> dict:
"""One page of transaction log entries, newest first."""
return wallet.call(
"retrieve_txs",
{
"token": wallet.token,
"refresh_from_node": True,
"tx_id": None,
"tx_slate_id": None,
"limit": limit,
"offset": offset,
"sort_order": "desc",
},
)
def every_output(wallet: EpicWallet):
"""Walk the whole output set a page at a time."""
offset = 0
while True:
page = outputs(wallet, offset=offset)
yield from page["outputs"]
offset += page["pager"]["records_read"]
if offset >= page["pager"]["total_records"] or page["pager"]["records_read"] == 0:
return
def main() -> None:
wallet = EpicWallet()
wallet.open_wallet(password=os.environ["EPIC_WALLET_PASSWORD"])
page = outputs(wallet)
pager = page["pager"]
print(f"{pager['records_read']} of {pager['total_records']} outputs")
for entry in page["outputs"]:
output = entry["output"]
value = int(output["value"]) / 1e8
print(f"{output['status']:<12} {value:>16.8f} EPIC height {output['height']}")
if __name__ == "__main__":
main()
同一文件中的every_output每次翻页遍历整个集合。
result.Ok
{
"refresh_from_node": true,
"pager": {
"records_read": 20,
"total_records": 143,
"limit": 20,
"offset": 0,
"sort_order": "desc"
},
"txs": [{ "...": "TxLogEntry" }]
}
retrieve_outputs与outputs相同，以outputs替代txs。类型为RetrieveTxsResult，
RetrieveOutputsResult
以及Pager。