Add admin endpoints to list users and export a user's private key

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>
This commit is contained in:
2026-07-21 14:21:55 +02:00
co-authored by Claude Sonnet 5
parent f21ecbd4ee
commit 39c6ea1950
3 changed files with 141 additions and 0 deletions
+81
View File
@@ -1,4 +1,5 @@
import pytest
from cryptography.fernet import Fernet
from httpx import ASGITransport, AsyncClient
from app.config import settings
@@ -9,6 +10,13 @@ async def client(monkeypatch, tmp_path):
monkeypatch.setattr(settings, "database_url", f"sqlite+aiosqlite:///{tmp_path}/test.db")
monkeypatch.setattr(settings, "admin_token", "test-admin-token")
monkeypatch.setattr(settings, "jwt_secret", "test-jwt-secret")
monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode())
monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
import app.wallet.hd as hd
hd._account_key = None
hd.generate_master_key()
from sqlalchemy.ext.asyncio import create_async_engine
@@ -19,6 +27,15 @@ async def client(monkeypatch, tmp_path):
db_base.AsyncSessionLocal = async_sessionmaker(db_base.engine, expire_on_commit=False)
# app.db.session did `from app.db.base import AsyncSessionLocal` at its own
# first import, which only copies the reference as it was at that moment —
# reassigning db_base.AsyncSessionLocal above doesn't reach it. get_session()
# looks up its module global at call time, so rebinding it here (every test)
# keeps it pointed at *this* test's engine instead of whichever ran first.
from app.db import session as db_session
db_session.AsyncSessionLocal = db_base.AsyncSessionLocal
async with db_base.engine.begin() as conn:
await conn.run_sync(db_base.Base.metadata.create_all)
@@ -62,3 +79,67 @@ async def test_admin_reads_and_updates_config(client):
resp = await client.get("/admin/config", headers=headers)
assert resp.json()["fee_address"] == "plm1qfeeaddress"
async def test_admin_lists_users(client):
from app.db import base as db_base
from app.db.models import User
from app.wallet.hd import derive_user_address
async with db_base.AsyncSessionLocal() as session:
session.add(
User(
username="alice",
password_hash="unused",
derivation_index=0,
address=derive_user_address(0),
cached_balance_sats=1_000_000_000,
)
)
await session.commit()
headers = {"X-Admin-Token": "test-admin-token"}
resp = await client.get("/admin/users", headers=headers)
assert resp.status_code == 200
body = resp.json()
assert len(body) == 1
assert body[0]["username"] == "alice"
assert body[0]["balance_sats"] == 1_000_000_000
async def test_admin_exports_user_privkey(client):
from embit.ec import PrivateKey
from app.db import base as db_base
from app.db.models import User
from app.wallet.hd import derive_user_address, derive_user_key
from app.wallet.plm_network import PLM_MAINNET
async with db_base.AsyncSessionLocal() as session:
user = User(
username="bob",
password_hash="unused",
derivation_index=1,
address=derive_user_address(1),
)
session.add(user)
await session.commit()
await session.refresh(user)
user_id = user.id
headers = {"X-Admin-Token": "test-admin-token"}
resp = await client.get(f"/admin/users/{user_id}/privkey", headers=headers)
assert resp.status_code == 200
body = resp.json()
expected_wif = PrivateKey(derive_user_key(1).secret, compressed=True, network=PLM_MAINNET).wif(
network=PLM_MAINNET
)
assert body["wif"] == expected_wif
assert body["address"] == derive_user_address(1)
async def test_admin_privkey_404_for_unknown_user(client):
headers = {"X-Admin-Token": "test-admin-token"}
resp = await client.get("/admin/users/999/privkey", headers=headers)
assert resp.status_code == 404