Кошелёк: подключение и чтение
Каждый вызов 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-байтовым нонсом, 16-байтовый тег добавляется к шифртексту, токен передаётся в 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 обусловлен выходами (outputs) и блокировкой.
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 вместо txs. Типы: RetrieveTxsResult,
RetrieveOutputsResult
и Pager.