GET /admin/users lists id/username/address/balance_sats/created_at.
GET /admin/users/{id}/privkey derives and returns that user's raw WIF
private key, for manual intervention (e.g. sweeping funds back if
something's stuck) — this is already a custodial system, the server
holds the master key everything is derived from, so this doesn't grant
a new capability, just exposes an existing one through the API. Every
access is audit-logged (admin_privkey_accessed).
Also fixes a pre-existing test-isolation bug in test_admin.py's client
fixture: app.db.session.get_session had `from app.db.base import
AsyncSessionLocal`, a one-time reference copy at first import — later
tests reassigning db_base.AsyncSessionLocal never reached it, so any
test mixing direct DB writes with router calls silently read/wrote
against a stale, possibly-disposed engine from whichever test ran
first. Fixed by also rebinding app.db.session.AsyncSessionLocal in the
fixture on every run.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
import os
|
|
|
|
from embit import script
|
|
from embit.bip32 import HDKey
|
|
from embit.ec import PrivateKey
|
|
|
|
from app.config import settings
|
|
from app.wallet.keystore import decrypt_xprv, encrypt_xprv
|
|
from app.wallet.plm_network import ACCOUNT_PATH, PLM_MAINNET
|
|
|
|
_account_key: HDKey | None = None
|
|
|
|
|
|
def generate_master_key(overwrite: bool = False) -> None:
|
|
"""One-time ops bootstrap: create a random master seed, encrypt it, write it to
|
|
disk. Not exposed via any API endpoint — run manually before first launch."""
|
|
if os.path.exists(settings.master_key_path) and not overwrite:
|
|
raise FileExistsError(f"{settings.master_key_path} already exists")
|
|
root = HDKey.from_seed(os.urandom(32), version=PLM_MAINNET["xprv"])
|
|
with open(settings.master_key_path, "wb") as f:
|
|
f.write(encrypt_xprv(root.to_base58(version=PLM_MAINNET["xprv"])))
|
|
|
|
|
|
def _load_account_key() -> HDKey:
|
|
global _account_key
|
|
if _account_key is None:
|
|
with open(settings.master_key_path, "rb") as f:
|
|
token = f.read()
|
|
root = HDKey.from_base58(decrypt_xprv(token))
|
|
_account_key = root.derive(ACCOUNT_PATH)
|
|
return _account_key
|
|
|
|
|
|
def derive_user_key(derivation_index: int) -> HDKey:
|
|
return _load_account_key().derive(f"0/{derivation_index}")
|
|
|
|
|
|
def derive_user_address(derivation_index: int) -> str:
|
|
pub = derive_user_key(derivation_index).to_public()
|
|
return script.p2wpkh(pub).address(network=PLM_MAINNET)
|
|
|
|
|
|
def derive_user_wif(derivation_index: int) -> str:
|
|
"""Exports a user's raw private key (WIF) for manual server-side intervention
|
|
(e.g. sweeping funds back to a user, or out, if something gets stuck). This is
|
|
a custodial system — the server already holds the master key this is derived
|
|
from — but callers must still treat the result as a live secret: log access,
|
|
never persist it, never return it over an unauthenticated channel."""
|
|
key = derive_user_key(derivation_index)
|
|
return PrivateKey(key.secret, compressed=True, network=PLM_MAINNET).wif(network=PLM_MAINNET)
|
|
|
|
|
|
def derive_pool_key() -> HDKey:
|
|
"""The "indirizzo padre" from the flowchart: all bets are sent here, and
|
|
payouts are signed with this key. Reserved on branch 1 of the account (branch 0
|
|
is user addresses), index 0 — not a spec requirement, an implementation choice
|
|
to keep it in the same encrypted master key rather than a separate secret."""
|
|
return _load_account_key().derive("1/0")
|
|
|
|
|
|
def derive_pool_address() -> str:
|
|
pub = derive_pool_key().to_public()
|
|
return script.p2wpkh(pub).address(network=PLM_MAINNET)
|