davideandClaude Sonnet 5
39c6ea1950
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>
146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import settings
|
|
|
|
|
|
@pytest.fixture
|
|
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
|
|
|
|
from app.db import base as db_base
|
|
|
|
db_base.engine = create_async_engine(settings.database_url)
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
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)
|
|
|
|
from app.api.routes.admin import router as admin_router
|
|
from fastapi import FastAPI
|
|
|
|
app = FastAPI()
|
|
app.include_router(admin_router)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
await db_base.engine.dispose()
|
|
|
|
|
|
async def test_admin_requires_token(client):
|
|
resp = await client.get("/admin/config")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_admin_rejects_wrong_token(client):
|
|
resp = await client.get("/admin/config", headers={"X-Admin-Token": "wrong"})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_admin_reads_and_updates_config(client):
|
|
headers = {"X-Admin-Token": "test-admin-token"}
|
|
|
|
resp = await client.get("/admin/config", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["fee_address"] == ""
|
|
|
|
resp = await client.put(
|
|
"/admin/config", headers=headers, json={"fee_address": "plm1qfeeaddress", "bet_amount_sats": 500_000_000}
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["fee_address"] == "plm1qfeeaddress"
|
|
assert body["bet_amount_sats"] == 500_000_000
|
|
|
|
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
|