Wallet: connect and read
Every Owner API v3 call travels inside an encrypted envelope. Once the client holds the shared secret, every method is an ordinary JSON-RPC call.
Start the API:
epic-wallet owner_api
The handshake
The client generates a secp256k1 keypair, exchanges public keys through init_secure_api, and
wraps every later request in encrypted_request_v3. The first wrapped call is open_wallet, which
returns the token every subsequent call carries. Message by message, it is documented in
the Owner API reference.
- Generate a keypair. A secp256k1 keypair in your client, used once per session.
- init_secure_api. Send your compressed public key. The wallet replies with its own. This is the only unencrypted call.
- Derive the shared secret. Multiply the wallet key by your scalar and keep the raw x coordinate, 32 bytes, unhashed. Hashing it is the most common implementation mistake.
- open_wallet, encrypted. The first call inside encrypted_request_v3. It needs the wallet password and returns a token.
- Every later call. AES-256-GCM with a fresh 12-byte nonce, the 16-byte tag appended to the ciphertext, and the token in the params.
A complete client
Needs requests, coincurve and pycryptodome. Run it from the examples directory.
python/epic_wallet.py, the client
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))
Using it:
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
Reading a balance
retrieve_summary_info returns a two-element array: whether the figures were validated against the
node, and the summary itself.
| Field | Meaning |
|---|---|
last_confirmed_height | Height the figures are current as of |
total | Everything the wallet knows about |
amount_currently_spendable | What you can send right now |
amount_awaiting_confirmation | Received but not yet deep enough |
amount_awaiting_finalization | Part of a transfer still in progress |
amount_locked | Reserved as inputs to a transfer you started |
amount_immature | Coinbase outputs still inside the maturity window |
A gap between total and amount_currently_spendable comes from
outputs and locking.
Amounts are strings in JSON
u64 values are serialised as strings to survive JavaScript's number precision. Parse them as
integers, and divide by 100,000,000 for EPIC.
Listing outputs and transactions
retrieve_outputs and retrieve_txs each return an object carrying a pager alongside the
records, not a bare array. Supply limit, offset and sort_order: without a limit a large wallet
returns every record in one response.
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 in the same file walks the whole set a page at a time.
result.Ok
{
"refresh_from_node": true,
"pager": {
"records_read": 20,
"total_records": 143,
"limit": 20,
"offset": 0,
"sort_order": "desc"
},
"txs": [{ "...": "TxLogEntry" }]
}
retrieve_outputs is identical with outputs in place of txs. Types are
RetrieveTxsResult,
RetrieveOutputsResult
and Pager.