davide
f27fe6243c
Session hardening: / and /admin now respond with Cache-Control: no-store, and
both pages re-derive their auth state on pageshow (event.persisted) as a
safety net against bfcache showing a stale logged-in/out view across
back/forward navigation. The user page also syncs logout across tabs via the
storage event, since localStorage is shared but in-memory JS state isn't.
Password recovery: admin gets a "Reset" button per user (POST
/admin/users/{id}/reset-password) that generates and sets a new password,
shown once — passwords are Argon2-hashed and can never be recovered, only
replaced. Users get self-service password change (POST
/users/me/change-password, requires the current password) under a new
Profilo tab, alongside read-only account info (username, address, balance,
join date).
Round display robustness: the user dashboard now refreshes immediately on
tab visibility change (background tabs get their timers throttled hard),
shows an explicit "connessione persa" state after repeated failed polls
instead of silently freezing on stale data, and polls faster both right when
the countdown hits zero and through the gap where the round is past its
deadline but still waiting for in-flight bets to confirm before the server
actually closes it.
214 lines
7.3 KiB
Python
214 lines
7.3 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_can_pause_and_resume_the_lottery(client):
|
|
headers = {"X-Admin-Token": "test-admin-token"}
|
|
|
|
resp = await client.get("/admin/config", headers=headers)
|
|
assert resp.json()["paused"] is False
|
|
|
|
resp = await client.post("/admin/pause", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["paused"] is True
|
|
|
|
resp = await client.get("/admin/config", headers=headers)
|
|
assert resp.json()["paused"] is True
|
|
|
|
resp = await client.post("/admin/resume", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["paused"] is False
|
|
|
|
|
|
async def test_admin_pause_requires_token(client):
|
|
resp = await client.post("/admin/pause")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
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
|
|
|
|
|
|
async def test_admin_resets_user_password(client):
|
|
from app.auth.security import hash_password, verify_password
|
|
from app.db import base as db_base
|
|
from app.db.models import User
|
|
from app.wallet.hd import derive_user_address
|
|
|
|
old_hash = hash_password("original-password")
|
|
async with db_base.AsyncSessionLocal() as session:
|
|
user = User(
|
|
username="carol",
|
|
password_hash=old_hash,
|
|
derivation_index=2,
|
|
address=derive_user_address(2),
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
user_id = user.id
|
|
|
|
headers = {"X-Admin-Token": "test-admin-token"}
|
|
resp = await client.post(f"/admin/users/{user_id}/reset-password", headers=headers)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["username"] == "carol"
|
|
new_password = body["new_password"]
|
|
assert new_password and new_password != "original-password"
|
|
|
|
async with db_base.AsyncSessionLocal() as session:
|
|
refreshed = await session.get(User, user_id)
|
|
assert refreshed.password_hash != old_hash
|
|
assert verify_password(new_password, refreshed.password_hash)
|
|
assert not verify_password("original-password", refreshed.password_hash)
|
|
|
|
|
|
async def test_admin_reset_password_requires_token(client):
|
|
resp = await client.post("/admin/users/1/reset-password")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_admin_reset_password_404_for_unknown_user(client):
|
|
headers = {"X-Admin-Token": "test-admin-token"}
|
|
resp = await client.post("/admin/users/999/reset-password", headers=headers)
|
|
assert resp.status_code == 404
|