From df72367f025c1c975b10cb18d2c8a0a088b4795e Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Mon, 20 Jul 2026 23:52:20 +0200 Subject: [PATCH] Implement full MVP: auth, HD wallet, Electrum client, deposits, bets, round/draw engine, payout, withdrawals, RBF, admin+audit All 10 build-order stages complete and unit-tested (49 tests). Verified live on mainnet: registration/address derivation, deposit crediting, a real 10 PLM bet (broadcast + confirmed + change credited). A full round close->draw->payout cycle was triggered live and was in progress at commit time. Withdrawal and RBF bump are unit-tested but not yet exercised against a live broadcast. Known gaps (scheduler doesn't resume mid-flight rounds after restart, payout has no retry, no deployment setup, etc.) are documented in CLAUDE.md. Co-Authored-By: Claude Sonnet 5 --- .env.example | 20 ++ .gitignore | 8 + CLAUDE.md | 38 +++- alembic.ini | 149 +++++++++++++ app/__init__.py | 0 app/api/__init__.py | 0 app/api/routes/__init__.py | 0 app/api/routes/admin.py | 44 ++++ app/api/routes/bets.py | 41 ++++ app/api/routes/users.py | 18 ++ app/api/routes/withdrawals.py | 49 +++++ app/audit/__init__.py | 0 app/audit/log.py | 22 ++ app/auth/__init__.py | 0 app/auth/dependencies.py | 25 +++ app/auth/routes.py | 70 ++++++ app/auth/security.py | 31 +++ app/bets/__init__.py | 0 app/bets/confirmation.py | 19 ++ app/bets/service.py | 104 +++++++++ app/config.py | 29 +++ app/db/__init__.py | 0 app/db/base.py | 11 + app/db/models.py | 131 +++++++++++ app/db/session.py | 10 + app/deposits/__init__.py | 0 app/deposits/service.py | 54 +++++ app/electrum/__init__.py | 0 app/electrum/client.py | 114 ++++++++++ app/electrum/listener.py | 111 ++++++++++ app/electrum/scripthash.py | 15 ++ app/main.py | 63 ++++++ app/rounds/__init__.py | 0 app/rounds/config.py | 17 ++ app/rounds/confirmation.py | 16 ++ app/rounds/draw.py | 21 ++ app/rounds/scheduler.py | 206 ++++++++++++++++++ app/rounds/service.py | 27 +++ app/tx/__init__.py | 0 app/tx/broadcast.py | 163 ++++++++++++++ app/tx/confirmation.py | 67 ++++++ app/tx/locks.py | 22 ++ app/wallet/__init__.py | 0 app/wallet/balance.py | 16 ++ app/wallet/hd.py | 52 +++++ app/wallet/keystore.py | 11 + app/wallet/plm_network.py | 28 +++ app/wallet/psbt_builder.py | 181 +++++++++++++++ app/withdrawals/__init__.py | 0 app/withdrawals/confirmation.py | 18 ++ app/withdrawals/service.py | 86 ++++++++ migrations/README | 1 + migrations/env.py | 85 ++++++++ migrations/script.py.mako | 28 +++ .../versions/274efdcbfbcc_initial_schema.py | 153 +++++++++++++ pyproject.toml | 27 +++ scripts/electrum_smoke_test.py | 27 +++ scripts/generate_master_key.py | 16 ++ tests/__init__.py | 0 tests/integration/__init__.py | 0 tests/unit/__init__.py | 0 tests/unit/test_admin.py | 64 ++++++ tests/unit/test_bets.py | 106 +++++++++ tests/unit/test_broadcast.py | 181 +++++++++++++++ tests/unit/test_confirmation.py | 82 +++++++ tests/unit/test_deposits.py | 54 +++++ tests/unit/test_draw.py | 32 +++ tests/unit/test_electrum_client.py | 77 +++++++ tests/unit/test_hd.py | 21 ++ tests/unit/test_payout_builder.py | 100 +++++++++ tests/unit/test_psbt_builder.py | 113 ++++++++++ tests/unit/test_rounds_service.py | 58 +++++ tests/unit/test_scheduler.py | 55 +++++ tests/unit/test_scripthash.py | 9 + tests/unit/test_security.py | 13 ++ tests/unit/test_withdrawals.py | 98 +++++++++ 76 files changed, 3506 insertions(+), 1 deletion(-) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 alembic.ini create mode 100644 app/__init__.py create mode 100644 app/api/__init__.py create mode 100644 app/api/routes/__init__.py create mode 100644 app/api/routes/admin.py create mode 100644 app/api/routes/bets.py create mode 100644 app/api/routes/users.py create mode 100644 app/api/routes/withdrawals.py create mode 100644 app/audit/__init__.py create mode 100644 app/audit/log.py create mode 100644 app/auth/__init__.py create mode 100644 app/auth/dependencies.py create mode 100644 app/auth/routes.py create mode 100644 app/auth/security.py create mode 100644 app/bets/__init__.py create mode 100644 app/bets/confirmation.py create mode 100644 app/bets/service.py create mode 100644 app/config.py create mode 100644 app/db/__init__.py create mode 100644 app/db/base.py create mode 100644 app/db/models.py create mode 100644 app/db/session.py create mode 100644 app/deposits/__init__.py create mode 100644 app/deposits/service.py create mode 100644 app/electrum/__init__.py create mode 100644 app/electrum/client.py create mode 100644 app/electrum/listener.py create mode 100644 app/electrum/scripthash.py create mode 100644 app/main.py create mode 100644 app/rounds/__init__.py create mode 100644 app/rounds/config.py create mode 100644 app/rounds/confirmation.py create mode 100644 app/rounds/draw.py create mode 100644 app/rounds/scheduler.py create mode 100644 app/rounds/service.py create mode 100644 app/tx/__init__.py create mode 100644 app/tx/broadcast.py create mode 100644 app/tx/confirmation.py create mode 100644 app/tx/locks.py create mode 100644 app/wallet/__init__.py create mode 100644 app/wallet/balance.py create mode 100644 app/wallet/hd.py create mode 100644 app/wallet/keystore.py create mode 100644 app/wallet/plm_network.py create mode 100644 app/wallet/psbt_builder.py create mode 100644 app/withdrawals/__init__.py create mode 100644 app/withdrawals/confirmation.py create mode 100644 app/withdrawals/service.py create mode 100644 migrations/README create mode 100644 migrations/env.py create mode 100644 migrations/script.py.mako create mode 100644 migrations/versions/274efdcbfbcc_initial_schema.py create mode 100644 pyproject.toml create mode 100644 scripts/electrum_smoke_test.py create mode 100644 scripts/generate_master_key.py create mode 100644 tests/__init__.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/test_admin.py create mode 100644 tests/unit/test_bets.py create mode 100644 tests/unit/test_broadcast.py create mode 100644 tests/unit/test_confirmation.py create mode 100644 tests/unit/test_deposits.py create mode 100644 tests/unit/test_draw.py create mode 100644 tests/unit/test_electrum_client.py create mode 100644 tests/unit/test_hd.py create mode 100644 tests/unit/test_payout_builder.py create mode 100644 tests/unit/test_psbt_builder.py create mode 100644 tests/unit/test_rounds_service.py create mode 100644 tests/unit/test_scheduler.py create mode 100644 tests/unit/test_scripthash.py create mode 100644 tests/unit/test_security.py create mode 100644 tests/unit/test_withdrawals.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..36e674b --- /dev/null +++ b/.env.example @@ -0,0 +1,20 @@ +DATABASE_URL=sqlite+aiosqlite:///./plm_lottery.db + +ELECTRUM_HOST=santantonio.sytes.net +ELECTRUM_PORT=50002 +ELECTRUM_USE_SSL=true + +# Fernet key protecting the master xprv at rest. Generate with: +# python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +XPRV_ENCRYPTION_KEY= +MASTER_KEY_PATH=./master.xprv.enc + +# Random secret for JWT session signing. Generate with: +# python -c "import secrets; print(secrets.token_urlsafe(32))" +JWT_SECRET= + +# Bearer token required on the admin endpoints (X-Admin-Token header). Generate with: +# python -c "import secrets; print(secrets.token_urlsafe(32))" +ADMIN_TOKEN= + +ROUND_DURATION_SECONDS=600 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ac30ea9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.venv/ +__pycache__/ +*.pyc +.env +*.db +master.xprv.enc +.pytest_cache/ +*.egg-info/ diff --git a/CLAUDE.md b/CLAUDE.md index 1e3346e..7c17104 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,10 +8,32 @@ The user communicates in Italian in chat — reply to them in Italian. Everythin ## Project status -This repository is at the **specification stage, not yet implemented**: it currently contains only [flowchart.mmd](flowchart.mmd), which is the source of truth for the project and describes the entire application flow. The tech stack is decided (see below) but no code, build system, or lint/test commands exist yet — once the project is scaffolded, this section must be updated with real commands (install, run, lint, test — including how to run a single test). +All 10 build-order stages from `/home/davide/.claude/plans/scalable-mixing-sloth.md` are code-complete and unit-tested (49 tests green): project skeleton, DB schema + Alembic migrations, auth, HD wallet derivation, Electrum client, deposit detection, bet flow, round/draw engine, payout, withdrawal, RBF fee-bump, admin config + audit log. + +Real-money verification on mainnet, done so far: registration + address derivation, deposit crediting (1-conf), a real 10 PLM bet (broadcast, confirmed, change credited back). A full round close → draw → payout cycle was triggered live and is in progress as of this commit (round closed, winner-draw was waiting on the next confirmed block) — **not yet confirmed complete**; withdrawal and the RBF bump path have only been verified with unit tests, never against a live broadcast. See "Known gaps" below before treating this as production-ready. Before writing code, always read [flowchart.mmd](flowchart.mmd) in full: every node in the diagram corresponds to a behavior that must be implemented exactly as described, including the labels on the edges (conditions, retries, loops). +## Commands + +```bash +source .venv/bin/activate # venv already created at .venv/ +pip install -e ".[dev]" # install/update deps + +alembic upgrade head # apply DB migrations +alembic revision --autogenerate -m "message" # generate a new migration after editing app/db/models.py + +PYTHONPATH=. python scripts/generate_master_key.py # one-time: create+encrypt the server's master xprv (requires XPRV_ENCRYPTION_KEY in .env) + +uvicorn app.main:app --reload --port 8123 # run the dev server + +python -m pytest # run all tests +python -m pytest tests/unit/test_hd.py # run one test file +python -m pytest tests/unit/test_hd.py::test_derivation_is_deterministic # run a single test +``` + +`.env` (gitignored) holds real secrets for local dev; `.env.example` documents the required keys and how to generate them. + ## Tech stack (MVP) - **Backend language**: Python. @@ -65,3 +87,17 @@ These choices were made explicitly during design (not derivable from reading a s - The user's personal deposit address always doubles as the winnings-receiving address: there is no separate "winner address". - 1 confirmation is the chosen threshold for all tx types (deposits, bets, payouts, withdrawals): don't introduce different thresholds (e.g. 3 or 6 confirmations) without an explicit decision. - The draw algorithm (node R) is deliberately simple and should be treated as a replaceable/pluggable component, not the final design — don't architect around its current implementation. + +## Known gaps / TODO + +Not blockers for reading the code, but must be addressed before this is production-ready: + +- **Scheduler doesn't resume mid-flight rounds after a restart.** `rounds/scheduler.py`'s `_tick()` only acts on rounds with `status == "open"`. If the process restarts while a round is `closing`/`drawing`/`paying_out`, it's permanently stuck — nothing re-enters `_wait_for_next_block` or retries `_trigger_payout`. Needs a startup routine that inspects in-progress rounds and resumes (or a periodic "unstick" check) before this can run unattended. +- **RBF bump only handles one case**: a single change output, paying back to the tx's own sender address, large enough to absorb the fee increase. No additional-input selection fallback — an exact-amount tx (no change) or a change output too small to absorb the bump raises `RbfError` and needs manual operator intervention. Documented in `tx/broadcast.py`. +- **Payout retry**: if `_trigger_payout` fails (e.g. insufficient pool UTXOs, Electrum disconnected), it just logs and returns — the round stays stuck in `paying_out` with no automatic retry. +- **Withdrawal and RBF bump have never been exercised against a live broadcast** — only deposit and bet flow are verified end-to-end with real PLM as of this commit. +- **No user-facing history endpoints** (list my bets / withdrawals / past rounds) — only `/users/me` (balance) exists. +- **No deployment setup**: no Dockerfile, process manager, or reconnect/supervision beyond the in-process asyncio tasks. Currently only run manually via `uvicorn` in a dev venv. +- **Admin auth is a single shared bearer token** (`ADMIN_TOKEN`, `X-Admin-Token` header) — no per-admin identity or audit trail of who changed config. +- **No rate limiting / abuse protection** on any endpoint (register, bet, withdrawal). +- No automated integration tests against a live Electrum connection — all live-network verification so far has been manual (ad hoc scripts + real mainnet transactions), not part of the `pytest` suite. diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..6a15082 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,149 @@ +# A generic, single database configuration. + +[alembic] +# path to migration scripts. +# this is typically a path given in POSIX (e.g. forward slashes) +# format, relative to the token %(here)s which refers to the location of this +# ini file +script_location = %(here)s/migrations + +# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s +# Uncomment the line below if you want the files to be prepended with date and time +# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file +# for all available tokens +# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s +# Or organize into date-based subdirectories (requires recursive_version_locations = true) +# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. for multiple paths, the path separator +# is defined by "path_separator" below. +prepend_sys_path = . + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the tzdata library which can be installed by adding +# `alembic[tz]` to the pip requirements. +# string value is passed to ZoneInfo() +# leave blank for localtime +# timezone = + +# max length of characters to apply to the "slug" field +# truncate_slug_length = 40 + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + +# set to 'true' to allow .pyc and .pyo files without +# a source .py file to be detected as revisions in the +# versions/ directory +# sourceless = false + +# version location specification; This defaults +# to /versions. When using multiple version +# directories, initial revisions must be specified with --version-path. +# The path separator used here should be the separator specified by "path_separator" +# below. +# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions + +# path_separator; This indicates what character is used to split lists of file +# paths, including version_locations and prepend_sys_path within configparser +# files such as alembic.ini. +# The default rendered in new alembic.ini files is "os", which uses os.pathsep +# to provide os-dependent path splitting. +# +# Note that in order to support legacy alembic.ini files, this default does NOT +# take place if path_separator is not present in alembic.ini. If this +# option is omitted entirely, fallback logic is as follows: +# +# 1. Parsing of the version_locations option falls back to using the legacy +# "version_path_separator" key, which if absent then falls back to the legacy +# behavior of splitting on spaces and/or commas. +# 2. Parsing of the prepend_sys_path option falls back to the legacy +# behavior of splitting on spaces, commas, or colons. +# +# Valid values for path_separator are: +# +# path_separator = : +# path_separator = ; +# path_separator = space +# path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +path_separator = os + + +# set to 'true' to search source files recursively +# in each "version_locations" directory +# new in Alembic version 1.10 +# recursive_version_locations = false + +# the output encoding used when revision files +# are written from script.py.mako +# output_encoding = utf-8 + +# database URL. This is consumed by the user-maintained env.py script only. +# other means of configuring database URLs may be customized within the env.py +# file. +sqlalchemy.url = driver://user:pass@localhost/dbname + + +[post_write_hooks] +# post_write_hooks defines scripts or Python functions that are run +# on newly generated revision scripts. See the documentation for further +# detail and examples + +# format using "black" - use the console_scripts runner, against the "black" entrypoint +# hooks = black +# black.type = console_scripts +# black.entrypoint = black +# black.options = -l 79 REVISION_SCRIPT_FILENAME + +# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module +# hooks = ruff +# ruff.type = module +# ruff.module = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Alternatively, use the exec runner to execute a binary found on your PATH +# hooks = ruff +# ruff.type = exec +# ruff.executable = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration. This is also consumed by the user-maintained +# env.py script only. +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routes/__init__.py b/app/api/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routes/admin.py b/app/api/routes/admin.py new file mode 100644 index 0000000..c045cac --- /dev/null +++ b/app/api/routes/admin.py @@ -0,0 +1,44 @@ +from fastapi import APIRouter, Depends, Header, HTTPException, status +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.db.session import get_session +from app.rounds.config import get_round_config + +router = APIRouter(prefix="/admin", tags=["admin"]) + + +async def require_admin(x_admin_token: str = Header(default="")) -> None: + if not settings.admin_token or x_admin_token != settings.admin_token: + raise HTTPException(status.HTTP_403_FORBIDDEN, "invalid admin token") + + +class RoundConfigResponse(BaseModel): + fee_address: str + bet_amount_sats: int + + +class RoundConfigUpdate(BaseModel): + fee_address: str | None = None + bet_amount_sats: int | None = None + + +@router.get("/config", response_model=RoundConfigResponse, dependencies=[Depends(require_admin)]) +async def read_config(session: AsyncSession = Depends(get_session)) -> RoundConfigResponse: + config = await get_round_config(session) + await session.commit() + return RoundConfigResponse(fee_address=config.fee_address, bet_amount_sats=config.bet_amount_sats) + + +@router.put("/config", response_model=RoundConfigResponse, dependencies=[Depends(require_admin)]) +async def update_config( + body: RoundConfigUpdate, session: AsyncSession = Depends(get_session) +) -> RoundConfigResponse: + config = await get_round_config(session) + if body.fee_address is not None: + config.fee_address = body.fee_address + if body.bet_amount_sats is not None: + config.bet_amount_sats = body.bet_amount_sats + await session.commit() + return RoundConfigResponse(fee_address=config.fee_address, bet_amount_sats=config.bet_amount_sats) diff --git a/app/api/routes/bets.py b/app/api/routes/bets.py new file mode 100644 index 0000000..4d34b8f --- /dev/null +++ b/app/api/routes/bets.py @@ -0,0 +1,41 @@ +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth.dependencies import get_current_user +from app.bets.service import BetError, place_bet +from app.db.models import User +from app.db.session import get_session + +router = APIRouter(prefix="/bets", tags=["bets"]) + + +class BetResponse(BaseModel): + round_id: int + bet_txid: str + bet_amount_sats: int + status: str + + +@router.post("", response_model=BetResponse, status_code=status.HTTP_201_CREATED) +async def create_bet( + request: Request, + user: User = Depends(get_current_user), + session: AsyncSession = Depends(get_session), +) -> BetResponse: + listener = request.app.state.electrum_listener + if listener.client is None: + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "not connected to the network, try again shortly") + + async with request.app.state.user_locks.acquire(user.id): + try: + participant = await place_bet(session, listener.client, user) + except BetError as exc: + raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc + + return BetResponse( + round_id=participant.round_id, + bet_txid=participant.bet_txid, + bet_amount_sats=participant.bet_amount_sats, + status=participant.status, + ) diff --git a/app/api/routes/users.py b/app/api/routes/users.py new file mode 100644 index 0000000..c8973a4 --- /dev/null +++ b/app/api/routes/users.py @@ -0,0 +1,18 @@ +from fastapi import APIRouter, Depends +from pydantic import BaseModel + +from app.auth.dependencies import get_current_user +from app.db.models import User + +router = APIRouter(prefix="/users", tags=["users"]) + + +class MeResponse(BaseModel): + username: str + address: str + balance_sats: int + + +@router.get("/me", response_model=MeResponse) +async def me(user: User = Depends(get_current_user)) -> MeResponse: + return MeResponse(username=user.username, address=user.address, balance_sats=user.cached_balance_sats) diff --git a/app/api/routes/withdrawals.py b/app/api/routes/withdrawals.py new file mode 100644 index 0000000..0cb3182 --- /dev/null +++ b/app/api/routes/withdrawals.py @@ -0,0 +1,49 @@ +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth.dependencies import get_current_user +from app.db.models import User +from app.db.session import get_session +from app.withdrawals.service import WithdrawalError, request_withdrawal + +router = APIRouter(prefix="/withdrawals", tags=["withdrawals"]) + + +class WithdrawalRequest(BaseModel): + external_address: str + amount_sats: int + + +class WithdrawalResponse(BaseModel): + txid: str + amount_requested_sats: int + amount_sent_sats: int + status: str + + +@router.post("", response_model=WithdrawalResponse, status_code=status.HTTP_201_CREATED) +async def create_withdrawal( + body: WithdrawalRequest, + request: Request, + user: User = Depends(get_current_user), + session: AsyncSession = Depends(get_session), +) -> WithdrawalResponse: + listener = request.app.state.electrum_listener + if listener.client is None: + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "not connected to the network, try again shortly") + + async with request.app.state.user_locks.acquire(user.id): + try: + withdrawal = await request_withdrawal( + session, listener.client, user, body.external_address, body.amount_sats + ) + except WithdrawalError as exc: + raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc + + return WithdrawalResponse( + txid=withdrawal.txid, + amount_requested_sats=withdrawal.amount_requested_sats, + amount_sent_sats=withdrawal.amount_sent_sats, + status=withdrawal.status, + ) diff --git a/app/audit/__init__.py b/app/audit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/audit/log.py b/app/audit/log.py new file mode 100644 index 0000000..4510ab6 --- /dev/null +++ b/app/audit/log.py @@ -0,0 +1,22 @@ +import json + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import AuditLog + + +async def write_audit_log( + session: AsyncSession, + event_type: str, + payload: dict, + user_id: int | None = None, + round_id: int | None = None, +) -> None: + session.add( + AuditLog( + event_type=event_type, + payload_json=json.dumps(payload), + user_id=user_id, + round_id=round_id, + ) + ) diff --git a/app/auth/__init__.py b/app/auth/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/auth/dependencies.py b/app/auth/dependencies.py new file mode 100644 index 0000000..18de63e --- /dev/null +++ b/app/auth/dependencies.py @@ -0,0 +1,25 @@ +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth.security import decode_access_token +from app.db.models import User +from app.db.session import get_session + +_bearer = HTTPBearer() + + +async def get_current_user( + credentials: HTTPAuthorizationCredentials = Depends(_bearer), + session: AsyncSession = Depends(get_session), +) -> User: + try: + user_id = decode_access_token(credentials.credentials) + except Exception as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") from exc + + user = await session.scalar(select(User).where(User.id == user_id)) + if user is None: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found") + return user diff --git a/app/auth/routes.py b/app/auth/routes.py new file mode 100644 index 0000000..6cfcc6a --- /dev/null +++ b/app/auth/routes.py @@ -0,0 +1,70 @@ +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth.security import create_access_token, hash_password, verify_password +from app.db.models import User +from app.db.session import get_session +from app.wallet.hd import derive_user_address + +router = APIRouter(prefix="/auth", tags=["auth"]) + +_MAX_REGISTER_RETRIES = 5 + + +class RegisterRequest(BaseModel): + username: str + password: str + + +class TokenResponse(BaseModel): + access_token: str + address: str + + +@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED) +async def register( + body: RegisterRequest, request: Request, session: AsyncSession = Depends(get_session) +) -> TokenResponse: + existing = await session.scalar(select(User).where(User.username == body.username)) + if existing is not None: + raise HTTPException(status.HTTP_409_CONFLICT, "username already taken") + + password_hash = hash_password(body.password) + + for _ in range(_MAX_REGISTER_RETRIES): + max_index = await session.scalar(select(func.max(User.derivation_index))) + next_index = 0 if max_index is None else max_index + 1 + address = derive_user_address(next_index) + user = User( + username=body.username, + password_hash=password_hash, + derivation_index=next_index, + address=address, + ) + session.add(user) + try: + await session.commit() + except IntegrityError: + await session.rollback() + continue + await session.refresh(user) + request.app.state.electrum_listener.address_for_new_user(user.id, user.address) + return TokenResponse(access_token=create_access_token(user.id), address=user.address) + + raise HTTPException(status.HTTP_409_CONFLICT, "could not allocate a derivation index, retry") + + +class LoginRequest(BaseModel): + username: str + password: str + + +@router.post("/login", response_model=TokenResponse) +async def login(body: LoginRequest, session: AsyncSession = Depends(get_session)) -> TokenResponse: + user = await session.scalar(select(User).where(User.username == body.username)) + if user is None or not verify_password(body.password, user.password_hash): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials") + return TokenResponse(access_token=create_access_token(user.id), address=user.address) diff --git a/app/auth/security.py b/app/auth/security.py new file mode 100644 index 0000000..947bcfc --- /dev/null +++ b/app/auth/security.py @@ -0,0 +1,31 @@ +from datetime import datetime, timedelta, timezone + +import jwt +from argon2 import PasswordHasher +from argon2.exceptions import VerifyMismatchError + +from app.config import settings + +_hasher = PasswordHasher() + + +def hash_password(password: str) -> str: + return _hasher.hash(password) + + +def verify_password(password: str, password_hash: str) -> bool: + try: + return _hasher.verify(password_hash, password) + except VerifyMismatchError: + return False + + +def create_access_token(user_id: int) -> str: + expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes) + payload = {"sub": str(user_id), "exp": expires_at} + return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) + + +def decode_access_token(token: str) -> int: + payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) + return int(payload["sub"]) diff --git a/app/bets/__init__.py b/app/bets/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/bets/confirmation.py b/app/bets/confirmation.py new file mode 100644 index 0000000..4162d9c --- /dev/null +++ b/app/bets/confirmation.py @@ -0,0 +1,19 @@ +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import PendingTransaction, RoundParticipant +from app.tx.confirmation import register_handler + + +async def _on_bet_confirmed(session: AsyncSession, pending: PendingTransaction) -> None: + participant = await session.scalar( + select(RoundParticipant).where(RoundParticipant.bet_txid == pending.current_txid) + ) + if participant is not None and participant.status == "broadcast": + participant.status = "confirmed" + participant.confirmed_at = datetime.now(timezone.utc) + + +register_handler("bet", _on_bet_confirmed) diff --git a/app/bets/service.py b/app/bets/service.py new file mode 100644 index 0000000..aeb22f2 --- /dev/null +++ b/app/bets/service.py @@ -0,0 +1,104 @@ +from datetime import datetime, timezone + +from embit import script +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.audit.log import write_audit_log +from app.config import settings +from app.db.models import PendingTransaction, RoundParticipant, User, UtxoEvent +from app.electrum.client import ElectrumClient +from app.rounds.config import get_round_config +from app.rounds.service import open_new_round_if_needed +from app.wallet.balance import recompute_balance +from app.wallet.hd import derive_pool_address, derive_user_key +from app.wallet.psbt_builder import BuiltTransaction, InsufficientFundsError, Utxo, build_signed_transaction + + +class BetError(Exception): + pass + + +async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) -> RoundParticipant: + round_ = await open_new_round_if_needed(session) + if round_.status != "open": + raise BetError("the current round is closing, please try again shortly") + + already_playing = await session.scalar( + select(RoundParticipant).where( + RoundParticipant.round_id == round_.id, RoundParticipant.user_id == user.id + ) + ) + if already_playing is not None: + raise BetError("you already have an active bet in the current round") + + config = await get_round_config(session) + bet_amount = config.bet_amount_sats + + unspent = ( + await session.scalars( + select(UtxoEvent).where(UtxoEvent.user_id == user.id, UtxoEvent.spent_txid.is_(None)) + ) + ).all() + if sum(u.amount_sats for u in unspent) < bet_amount: + raise BetError("insufficient balance") + + user_key = derive_user_key(user.derivation_index) + from_script = script.p2wpkh(user_key.to_public()) + utxos = [Utxo(u.txid, u.vout, u.amount_sats) for u in unspent] + + try: + built = build_signed_transaction( + signing_key=user_key, + from_script=from_script, + utxos=utxos, + to_address=derive_pool_address(), + amount_sats=bet_amount, + change_address=user.address, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + ) + except InsufficientFundsError as exc: + raise BetError(str(exc)) from exc + + await client.broadcast(built.raw_hex) + + spent_by_key = {(u.txid, u.vout): u for u in unspent} + for spent in built.spent_utxos: + row = spent_by_key[(spent.txid, spent.vout)] + row.spent_txid = built.txid + await recompute_balance(session, user.id) + + broadcast_at = datetime.now(timezone.utc) + participant = RoundParticipant( + round_id=round_.id, + user_id=user.id, + bet_amount_sats=built.recipient_sats, + bet_txid=built.txid, + broadcast_at=broadcast_at, + status="broadcast", + ) + session.add(participant) + session.add(_pending_transaction(round_.id, user.id, built)) + await write_audit_log( + session, + "bet_placed", + {"txid": built.txid, "amount_sats": built.recipient_sats}, + user_id=user.id, + round_id=round_.id, + ) + + await session.commit() + await session.refresh(participant) + return participant + + +def _pending_transaction(round_id: int, user_id: int, built: BuiltTransaction) -> PendingTransaction: + return PendingTransaction( + kind="bet", + round_id=round_id, + user_id=user_id, + current_txid=built.txid, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + raw_tx_hex=built.raw_hex, + status="pending", + ) diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..f1bbf52 --- /dev/null +++ b/app/config.py @@ -0,0 +1,29 @@ +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", extra="ignore") + + database_url: str = "sqlite+aiosqlite:///./plm_lottery.db" + + electrum_host: str = "santantonio.sytes.net" + electrum_port: int = 50002 + electrum_use_ssl: bool = True + + xprv_encryption_key: str = "" + master_key_path: str = "./master.xprv.enc" + jwt_secret: str = "" + jwt_algorithm: str = "HS256" + jwt_expire_minutes: int = 60 * 24 + admin_token: str = "" + + round_duration_seconds: int = 600 + + bet_amount_sats: int = 10 * 100_000_000 + min_amount_sats: int = 1 * 100_000_000 + confirmations_required: int = 1 + fee_rate_sat_vb: int = 1 + rbf_timeout_seconds: int = 900 + + +settings = Settings() diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/db/base.py b/app/db/base.py new file mode 100644 index 0000000..cac5738 --- /dev/null +++ b/app/db/base.py @@ -0,0 +1,11 @@ +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.orm import DeclarativeBase + +from app.config import settings + +engine = create_async_engine(settings.database_url) +AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + +class Base(DeclarativeBase): + pass diff --git a/app/db/models.py b/app/db/models.py new file mode 100644 index 0000000..12790ba --- /dev/null +++ b/app/db/models.py @@ -0,0 +1,131 @@ +from datetime import datetime, timezone + +from sqlalchemy import BigInteger, ForeignKey, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class User(Base): + __tablename__ = "users" + + id: Mapped[int] = mapped_column(primary_key=True) + username: Mapped[str] = mapped_column(String(64), unique=True, index=True) + password_hash: Mapped[str] = mapped_column(String(256)) + derivation_index: Mapped[int] = mapped_column(unique=True) + address: Mapped[str] = mapped_column(String(128), unique=True) + # Read cache only; must always be written in the same transaction as the + # utxo_events rows it summarizes. Source of truth is utxo_events. + cached_balance_sats: Mapped[int] = mapped_column(BigInteger, default=0) + created_at: Mapped[datetime] = mapped_column(default=utcnow) + + +class UtxoEvent(Base): + __tablename__ = "utxo_events" + __table_args__ = (UniqueConstraint("txid", "vout"),) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + txid: Mapped[str] = mapped_column(String(64)) + vout: Mapped[int] + amount_sats: Mapped[int] = mapped_column(BigInteger) + confirmed_height: Mapped[int] + confirmed_at: Mapped[datetime] = mapped_column(default=utcnow) + # Set once this UTXO is consumed by an outgoing bet/withdrawal build. + spent_txid: Mapped[str | None] = mapped_column(String(64), default=None) + + +class Round(Base): + __tablename__ = "rounds" + + id: Mapped[int] = mapped_column(primary_key=True) + status: Mapped[str] = mapped_column(String(16), default="open") + opened_at: Mapped[datetime] = mapped_column(default=utcnow) + closed_at: Mapped[datetime | None] = mapped_column(default=None) + draw_block_height: Mapped[int | None] = mapped_column(default=None) + draw_block_hash: Mapped[str | None] = mapped_column(String(64), default=None) + seed_int: Mapped[str | None] = mapped_column(String(128), default=None) + winner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None) + pool_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None) + winner_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None) + fee_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None) + payout_txid: Mapped[str | None] = mapped_column(String(64), default=None) + + +class RoundParticipant(Base): + __tablename__ = "round_participants" + __table_args__ = (UniqueConstraint("round_id", "user_id"),) + + id: Mapped[int] = mapped_column(primary_key=True) + round_id: Mapped[int] = mapped_column(ForeignKey("rounds.id"), index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + bet_amount_sats: Mapped[int] = mapped_column(BigInteger) + bet_txid: Mapped[str] = mapped_column(String(64)) + # Ordering / tie-break field per spec: broadcast time, not confirmation time. + broadcast_at: Mapped[datetime] = mapped_column(default=utcnow) + confirmed_at: Mapped[datetime | None] = mapped_column(default=None) + status: Mapped[str] = mapped_column(String(16), default="broadcast") + + +class RoundConfig(Base): + """Single-row operational config, DB-backed so it's editable without a redeploy. + + round_duration is intentionally NOT here: it stays env-var-driven per spec. + Don't move it here without an explicit decision to change that. + """ + + __tablename__ = "round_config" + + id: Mapped[int] = mapped_column(primary_key=True) + fee_address: Mapped[str] = mapped_column(String(128)) + bet_amount_sats: Mapped[int] = mapped_column(BigInteger) + updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow) + + +class PendingTransaction(Base): + """Single source of truth for the RBF timeout->bump->rebroadcast loop.""" + + __tablename__ = "pending_transactions" + + id: Mapped[int] = mapped_column(primary_key=True) + kind: Mapped[str] = mapped_column(String(16)) # bet | payout | withdrawal + round_id: Mapped[int | None] = mapped_column(ForeignKey("rounds.id"), default=None) + withdrawal_id: Mapped[int | None] = mapped_column(ForeignKey("withdrawals.id"), default=None) + user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None) + current_txid: Mapped[str] = mapped_column(String(64)) + fee_rate_sat_vb: Mapped[int] + raw_tx_hex: Mapped[str] = mapped_column(String) + broadcast_at: Mapped[datetime] = mapped_column(default=utcnow) + status: Mapped[str] = mapped_column(String(16), default="pending") + replaced_by_txid: Mapped[str | None] = mapped_column(String(64), default=None) + attempt_count: Mapped[int] = mapped_column(default=1) + updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow) + + +class Withdrawal(Base): + __tablename__ = "withdrawals" + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + external_address: Mapped[str] = mapped_column(String(128)) + amount_requested_sats: Mapped[int] = mapped_column(BigInteger) + amount_sent_sats: Mapped[int | None] = mapped_column(BigInteger, default=None) + txid: Mapped[str | None] = mapped_column(String(64), default=None) + status: Mapped[str] = mapped_column(String(16), default="pending") + created_at: Mapped[datetime] = mapped_column(default=utcnow) + confirmed_at: Mapped[datetime | None] = mapped_column(default=None) + + +class AuditLog(Base): + __tablename__ = "audit_log" + + id: Mapped[int] = mapped_column(primary_key=True) + event_type: Mapped[str] = mapped_column(String(32)) + payload_json: Mapped[str] = mapped_column(String) + user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None) + round_id: Mapped[int | None] = mapped_column(ForeignKey("rounds.id"), default=None) + created_at: Mapped[datetime] = mapped_column(default=utcnow) diff --git a/app/db/session.py b/app/db/session.py new file mode 100644 index 0000000..46a5245 --- /dev/null +++ b/app/db/session.py @@ -0,0 +1,10 @@ +from collections.abc import AsyncGenerator + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.base import AsyncSessionLocal + + +async def get_session() -> AsyncGenerator[AsyncSession, None]: + async with AsyncSessionLocal() as session: + yield session diff --git a/app/deposits/__init__.py b/app/deposits/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/deposits/service.py b/app/deposits/service.py new file mode 100644 index 0000000..142b74f --- /dev/null +++ b/app/deposits/service.py @@ -0,0 +1,54 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.audit.log import write_audit_log +from app.db.models import UtxoEvent +from app.wallet.balance import recompute_balance + + +async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: list[dict]) -> int: + """Insert utxo_events for newly-confirmed entries from an Electrum + `listunspent` response (idempotent on txid+vout), refresh the user's cached + balance. Returns the number of newly-credited UTXOs. + + entries: [{"tx_hash": ..., "tx_pos": ..., "height": ..., "value": ...}, ...] + height <= 0 means unconfirmed (mempool) per the Electrum protocol convention — + skipped, since the spec requires 1 confirmation before crediting. + """ + existing_keys = { + (txid, vout) + for txid, vout in ( + await session.execute(select(UtxoEvent.txid, UtxoEvent.vout).where(UtxoEvent.user_id == user_id)) + ).all() + } + + newly_credited = 0 + for entry in entries: + if entry["height"] <= 0: + continue + key = (entry["tx_hash"], entry["tx_pos"]) + if key in existing_keys: + continue + session.add( + UtxoEvent( + user_id=user_id, + txid=entry["tx_hash"], + vout=entry["tx_pos"], + amount_sats=entry["value"], + confirmed_height=entry["height"], + ) + ) + await write_audit_log( + session, + "deposit_credited", + {"txid": entry["tx_hash"], "vout": entry["tx_pos"], "amount_sats": entry["value"]}, + user_id=user_id, + ) + newly_credited += 1 + + if newly_credited: + await session.flush() + await recompute_balance(session, user_id) + await session.commit() + + return newly_credited diff --git a/app/electrum/__init__.py b/app/electrum/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/electrum/client.py b/app/electrum/client.py new file mode 100644 index 0000000..7e4b5e1 --- /dev/null +++ b/app/electrum/client.py @@ -0,0 +1,114 @@ +import asyncio +import itertools +import json +import ssl + + +class ElectrumError(Exception): + pass + + +class ElectrumClient: + """Minimal asyncio Electrum protocol client: line-delimited JSON-RPC over TLS. + + Push notifications (blockchain.headers.subscribe, blockchain.scripthash.subscribe) + arrive under the *same* method name as the subscribe call, multiplexed for every + scripthash subscribed — callers read `notifications(method)` and, for scripthash + pushes, dispatch on `params[0]` (the scripthash) themselves. + """ + + def __init__(self, host: str, port: int, use_ssl: bool = True): + self.host = host + self.port = port + self.use_ssl = use_ssl + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._id_counter = itertools.count(1) + self._pending: dict[int, asyncio.Future] = {} + self._subscriptions: dict[str, asyncio.Queue] = {} + self._read_task: asyncio.Task | None = None + + async def connect(self) -> None: + # Electrum servers commonly present self-signed certs; the protocol's trust + # model is server consensus, not TLS PKI, so we only use SSL for transport + # encryption and don't verify the certificate chain/hostname. + ssl_context = None + if self.use_ssl: + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + self._reader, self._writer = await asyncio.open_connection(self.host, self.port, ssl=ssl_context) + self._read_task = asyncio.create_task(self._read_loop()) + await self.request("server.version", ["plm-lottery", "1.4"]) + + async def close(self) -> None: + if self._read_task is not None: + self._read_task.cancel() + if self._writer is not None: + self._writer.close() + try: + await asyncio.wait_for(self._writer.wait_closed(), timeout=2) + except (ssl.SSLError, TimeoutError, asyncio.TimeoutError): + pass # some Electrum servers don't send a clean TLS close_notify + + async def request(self, method: str, params: list | None = None) -> object: + if self._writer is None: + raise ElectrumError("not connected") + request_id = next(self._id_counter) + future: asyncio.Future = asyncio.get_event_loop().create_future() + self._pending[request_id] = future + payload = json.dumps({"id": request_id, "method": method, "params": params or []}) + "\n" + self._writer.write(payload.encode()) + await self._writer.drain() + return await future + + def notifications(self, method: str) -> asyncio.Queue: + return self._subscriptions.setdefault(method, asyncio.Queue()) + + async def subscribe_headers(self) -> dict: + self.notifications("blockchain.headers.subscribe") + return await self.request("blockchain.headers.subscribe") + + async def subscribe_scripthash(self, scripthash: str) -> str | None: + self.notifications("blockchain.scripthash.subscribe") + return await self.request("blockchain.scripthash.subscribe", [scripthash]) + + async def listunspent(self, scripthash: str) -> list[dict]: + return await self.request("blockchain.scripthash.listunspent", [scripthash]) + + async def broadcast(self, raw_tx_hex: str) -> str: + return await self.request("blockchain.transaction.broadcast", [raw_tx_hex]) + + async def get_transaction(self, txid: str, verbose: bool = False) -> object: + return await self.request("blockchain.transaction.get", [txid, verbose]) + + async def _read_loop(self) -> None: + assert self._reader is not None + try: + while True: + line = await self._reader.readline() + if not line: + break + message = json.loads(line) + self._dispatch(message) + finally: + error = ElectrumError("connection closed") + for future in self._pending.values(): + if not future.done(): + future.set_exception(error) + self._pending.clear() + + def _dispatch(self, message: dict) -> None: + message_id = message.get("id") + if message_id is not None and message_id in self._pending: + future = self._pending.pop(message_id) + if future.done(): + return + if message.get("error"): + future.set_exception(ElectrumError(message["error"])) + else: + future.set_result(message.get("result")) + elif "method" in message: + queue = self._subscriptions.get(message["method"]) + if queue is not None: + queue.put_nowait(message.get("params")) diff --git a/app/electrum/listener.py b/app/electrum/listener.py new file mode 100644 index 0000000..48a7d7a --- /dev/null +++ b/app/electrum/listener.py @@ -0,0 +1,111 @@ +import asyncio +import logging +from collections.abc import Callable + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from app.db.models import User +from app.deposits.service import credit_confirmed_utxos +from app.electrum.client import ElectrumClient +from app.electrum.scripthash import address_to_scripthash + +logger = logging.getLogger(__name__) + + +class ElectrumListener: + """Long-lived background task: keeps one Electrum connection open, subscribes + every user's address (plus any address added later via add_address), and + credits confirmed deposits as scripthash-change notifications arrive. + + Reconnects with backoff on any failure; a fresh connection re-subscribes to + every user pulled straight from the DB, so no in-memory subscription state is + ever a stale source of truth. + """ + + def __init__(self, client_factory: Callable[[], ElectrumClient], session_factory: async_sessionmaker): + self._client_factory = client_factory + self._session_factory = session_factory + self._scripthash_to_user: dict[str, int] = {} + self.tip_height: int = 0 + self.tip_header_hex: str | None = None + self.client: ElectrumClient | None = None + + def address_for_new_user(self, user_id: int, address: str) -> None: + """Called right after a user registers so their deposit address starts + being watched immediately, without waiting for the next reconnect cycle.""" + scripthash = address_to_scripthash(address) + self._scripthash_to_user[scripthash] = user_id + if self.client is not None: + asyncio.create_task(self._subscribe_and_refresh(scripthash, user_id)) + + async def run(self) -> None: + backoff = 1 + while True: + try: + await self._run_once() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Electrum listener error, reconnecting in %ss", backoff) + self.client = None + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 30) + continue + backoff = 1 + + async def _run_once(self) -> None: + client = self._client_factory() + await client.connect() + self.client = client + + header = await client.subscribe_headers() + self.tip_height = header["height"] + self.tip_header_hex = header.get("hex") + + await self._subscribe_all_users() + + headers_queue = client.notifications("blockchain.headers.subscribe") + scripthash_queue = client.notifications("blockchain.scripthash.subscribe") + try: + await asyncio.gather( + self._consume_headers(headers_queue), + self._consume_scripthash(scripthash_queue), + ) + finally: + await client.close() + + async def _subscribe_all_users(self) -> None: + async with self._session_factory() as session: + users = (await session.scalars(select(User))).all() + for user in users: + scripthash = address_to_scripthash(user.address) + self._scripthash_to_user[scripthash] = user.id + await self._subscribe_and_refresh(scripthash, user.id) + + async def _subscribe_and_refresh(self, scripthash: str, user_id: int) -> None: + assert self.client is not None + await self.client.subscribe_scripthash(scripthash) + await self._refresh_user(user_id, scripthash) + + async def _consume_headers(self, queue: asyncio.Queue) -> None: + while True: + params = await queue.get() + for header in params: + self.tip_height = header["height"] + self.tip_header_hex = header.get("hex") + + async def _consume_scripthash(self, queue: asyncio.Queue) -> None: + while True: + scripthash, _status = await queue.get() + user_id = self._scripthash_to_user.get(scripthash) + if user_id is not None: + await self._refresh_user(user_id, scripthash) + + async def _refresh_user(self, user_id: int, scripthash: str) -> None: + assert self.client is not None + entries = await self.client.listunspent(scripthash) + async with self._session_factory() as session: + credited = await credit_confirmed_utxos(session, user_id, entries) + if credited: + logger.info("credited %s new UTXO(s) for user_id=%s", credited, user_id) diff --git a/app/electrum/scripthash.py b/app/electrum/scripthash.py new file mode 100644 index 0000000..dff74ac --- /dev/null +++ b/app/electrum/scripthash.py @@ -0,0 +1,15 @@ +import hashlib + +from embit.script import Script + + +def address_to_scripthash(address: str) -> str: + """Electrum protocol scripthash: sha256(scriptPubKey), byte-reversed, hex. + + Uses `.data` (the raw scriptPubKey bytes), not `.serialize()` — the latter + prefixes a compact-size length byte meant for embedding the script as pushdata + elsewhere (e.g. a P2SH redeemScript), which is not part of the actual on-chain + output script and produces a wrong (unmatchable) scripthash if used here. + """ + script_pubkey = Script.from_address(address).data + return hashlib.sha256(script_pubkey).digest()[::-1].hex() diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..c36cd28 --- /dev/null +++ b/app/main.py @@ -0,0 +1,63 @@ +import asyncio +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +import app.bets.confirmation # noqa: F401 (registers the "bet" confirmation handler) +import app.rounds.confirmation # noqa: F401 (registers the "payout" confirmation handler) +import app.withdrawals.confirmation # noqa: F401 (registers the "withdrawal" confirmation handler) +from app.api.routes.admin import router as admin_router +from app.api.routes.bets import router as bets_router +from app.api.routes.users import router as users_router +from app.api.routes.withdrawals import router as withdrawals_router +from app.auth.routes import router as auth_router +from app.config import settings +from app.db.base import AsyncSessionLocal +from app.electrum.client import ElectrumClient +from app.electrum.listener import ElectrumListener +from app.rounds.scheduler import RoundScheduler +from app.tx.broadcast import RbfBumper +from app.tx.confirmation import ConfirmationPoller +from app.tx.locks import UserLocks + + +def _make_electrum_client() -> ElectrumClient: + return ElectrumClient(settings.electrum_host, settings.electrum_port, settings.electrum_use_ssl) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + listener = ElectrumListener(_make_electrum_client, AsyncSessionLocal) + app.state.electrum_listener = listener + app.state.user_locks = UserLocks() + + scheduler = RoundScheduler(AsyncSessionLocal, listener) + poller = ConfirmationPoller(AsyncSessionLocal, lambda: listener.client) + bumper = RbfBumper(AsyncSessionLocal, lambda: listener.client) + + tasks = [ + asyncio.create_task(listener.run()), + asyncio.create_task(scheduler.run()), + asyncio.create_task(poller.run()), + asyncio.create_task(bumper.run()), + ] + try: + yield + finally: + for task in tasks: + task.cancel() + if listener.client is not None: + await listener.client.close() + + +app = FastAPI(title="PLM Lottery", lifespan=lifespan) +app.include_router(auth_router) +app.include_router(users_router) +app.include_router(bets_router) +app.include_router(withdrawals_router) +app.include_router(admin_router) + + +@app.get("/health") +async def health() -> dict[str, str]: + return {"status": "ok"} diff --git a/app/rounds/__init__.py b/app/rounds/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/rounds/config.py b/app/rounds/config.py new file mode 100644 index 0000000..55bee3f --- /dev/null +++ b/app/rounds/config.py @@ -0,0 +1,17 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.db.models import RoundConfig + + +async def get_round_config(session: AsyncSession) -> RoundConfig: + """Single-row operational config, lazily seeded from settings defaults on + first use. fee_address starts empty until an operator sets it (admin + endpoint, stage 10) — payouts must refuse to run until it's set.""" + config = await session.scalar(select(RoundConfig)) + if config is None: + config = RoundConfig(fee_address="", bet_amount_sats=settings.bet_amount_sats) + session.add(config) + await session.flush() + return config diff --git a/app/rounds/confirmation.py b/app/rounds/confirmation.py new file mode 100644 index 0000000..c11991c --- /dev/null +++ b/app/rounds/confirmation.py @@ -0,0 +1,16 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import PendingTransaction, Round +from app.tx.confirmation import register_handler + + +async def _on_payout_confirmed(session: AsyncSession, pending: PendingTransaction) -> None: + round_ = await session.scalar(select(Round).where(Round.payout_txid == pending.current_txid)) + if round_ is not None and round_.status == "paying_out": + round_.status = "closed" + # The winner's own address is already watched by the Electrum listener, so + # their balance is credited by the normal deposit path once this confirms. + + +register_handler("payout", _on_payout_confirmed) diff --git a/app/rounds/draw.py b/app/rounds/draw.py new file mode 100644 index 0000000..9e26ffc --- /dev/null +++ b/app/rounds/draw.py @@ -0,0 +1,21 @@ +import hashlib + + +def header_hex_to_block_hash(header_hex: str) -> str: + """Block hash from a raw Electrum header: sha256d, byte-reversed, hex. + Verified against a real mainnet block (blockchain.transaction.get's own + reported blockhash) during development.""" + header_bytes = bytes.fromhex(header_hex) + digest = hashlib.sha256(hashlib.sha256(header_bytes).digest()).digest() + return digest[::-1].hex() + + +def draw_winner(participants: list[str], block_hash_hex: str) -> str: + """v1 draw algorithm (flowchart.mmd, node R): seed = block hash as an integer, + index = seed mod participant_count, winner = participants[index]. Anyone can + recompute and verify it from public data. Deliberately simple/replaceable.""" + if not participants: + raise ValueError("no participants to draw from") + seed = int(block_hash_hex, 16) + index = seed % len(participants) + return participants[index] diff --git a/app/rounds/scheduler.py b/app/rounds/scheduler.py new file mode 100644 index 0000000..34db08c --- /dev/null +++ b/app/rounds/scheduler.py @@ -0,0 +1,206 @@ +import asyncio +import logging +from datetime import datetime, timedelta, timezone + +from embit import script +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from app.audit.log import write_audit_log +from app.config import settings +from app.db.models import PendingTransaction, Round, RoundParticipant, User +from app.electrum.listener import ElectrumListener +from app.electrum.scripthash import address_to_scripthash +from app.rounds.config import get_round_config +from app.rounds.draw import draw_winner, header_hex_to_block_hash +from app.rounds.service import open_new_round_if_needed +from app.wallet.hd import derive_pool_key +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_payout_transaction + +logger = logging.getLogger(__name__) + +_TICK_INTERVAL_SECONDS = 5 + + +class RoundScheduler: + """Background task implementing flowchart.mmd's DRAW subgraph: closes the + round on its timer (once any in-flight bets have confirmed), draws a winner + from the next confirmed block, and broadcasts the payout. The next round only + opens once this one is fully closed (rounds/service.get_active_round).""" + + def __init__(self, session_factory: async_sessionmaker, listener: ElectrumListener): + self._session_factory = session_factory + self._listener = listener + + async def run(self) -> None: + while True: + try: + await self._tick() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("round scheduler tick failed") + await asyncio.sleep(_TICK_INTERVAL_SECONDS) + + async def _tick(self) -> None: + if self._listener.client is None: + return + + async with self._session_factory() as session: + round_ = await open_new_round_if_needed(session) + await session.commit() + round_id, status, opened_at = round_.id, round_.status, round_.opened_at + + if status != "open": + return # already closing/drawing/paying_out; progress happens elsewhere + + opened_at = opened_at.replace(tzinfo=timezone.utc) + if datetime.now(timezone.utc) < opened_at + timedelta(seconds=settings.round_duration_seconds): + return + + async with self._session_factory() as session: + pending_count = await session.scalar( + select(func.count()) + .select_from(RoundParticipant) + .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "broadcast") + ) + if pending_count: + return # wait for in-flight bets to confirm before closing + + await self._close_and_draw(round_id) + + async def _close_and_draw(self, round_id: int) -> None: + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) + round_.status = "closing" + round_.closed_at = datetime.now(timezone.utc) + + participants = ( + await session.scalars( + select(RoundParticipant) + .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "confirmed") + .order_by(RoundParticipant.broadcast_at) + ) + ).all() + + if not participants: + round_.status = "closed" + await write_audit_log(session, "round_closed", {"participants": 0}, round_id=round_id) + await session.commit() + logger.info("round %s closed with no participants", round_id) + return + + pool_amount = sum(p.bet_amount_sats for p in participants) + addresses: list[str] = [] + user_by_address: dict[str, int] = {} + for p in participants: + user = await session.get(User, p.user_id) + addresses.append(user.address) + user_by_address[user.address] = user.id + + round_.status = "drawing" + await session.commit() + + tip_at_close = self._listener.tip_height + block_height, block_hash = await self._wait_for_next_block(tip_at_close) + winner_address = draw_winner(addresses, block_hash) + + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) + round_.draw_block_height = block_height + round_.draw_block_hash = block_hash + round_.seed_int = str(int(block_hash, 16)) + round_.winner_user_id = user_by_address[winner_address] + round_.pool_amount_sats = pool_amount + round_.status = "paying_out" + await write_audit_log( + session, + "winner_drawn", + { + "winner_address": winner_address, + "pool_amount_sats": pool_amount, + "block_height": block_height, + "block_hash": block_hash, + "participants": len(addresses), + }, + user_id=user_by_address[winner_address], + round_id=round_id, + ) + await session.commit() + + logger.info("round %s: winner=%s pool=%s", round_id, winner_address, pool_amount) + await self._trigger_payout(round_id) + + async def _wait_for_next_block(self, tip_at_close: int) -> tuple[int, str]: + while self._listener.tip_height <= tip_at_close or not self._listener.tip_header_hex: + await asyncio.sleep(_TICK_INTERVAL_SECONDS) + return self._listener.tip_height, header_hex_to_block_hash(self._listener.tip_header_hex) + + async def _trigger_payout(self, round_id: int) -> None: + client = self._listener.client + if client is None: + logger.error("round %s payout deferred: not connected", round_id) + return + + async with self._session_factory() as session: + round_ = await session.get(Round, round_id) + config = await get_round_config(session) + if not config.fee_address: + logger.error( + "round %s payout blocked: no fee_address configured (set it via the admin endpoint)", round_id + ) + return + + winner = await session.get(User, round_.winner_user_id) + winner_share = round_.pool_amount_sats * 70 // 100 + commission_share = round_.pool_amount_sats - winner_share # remainder from rounding goes to fees + + pool_key = derive_pool_key() + pool_script_obj = script.p2wpkh(pool_key.to_public()) + pool_address = pool_script_obj.address(network=PLM_MAINNET) + pool_scripthash = address_to_scripthash(pool_address) + entries = await client.listunspent(pool_scripthash) + utxos = [Utxo(e["tx_hash"], e["tx_pos"], e["value"]) for e in entries if e["height"] > 0] + + try: + built = build_payout_transaction( + signing_key=pool_key, + from_script=pool_script_obj, + utxos=utxos, + winner_address=winner.address, + winner_share_sats=winner_share, + fee_address=config.fee_address, + commission_sats=commission_share, + change_address=pool_address, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + ) + except InsufficientFundsError: + logger.exception("round %s payout failed: insufficient pool UTXOs", round_id) + return + + await client.broadcast(built.raw_hex) + + round_.winner_amount_sats = built.winner_sats + round_.fee_amount_sats = built.commission_sats + round_.payout_txid = built.txid + session.add( + PendingTransaction( + kind="payout", + round_id=round_id, + current_txid=built.txid, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + raw_tx_hex=built.raw_hex, + status="pending", + ) + ) + await write_audit_log( + session, + "payout_sent", + {"txid": built.txid, "winner_sats": built.winner_sats, "commission_sats": built.commission_sats}, + user_id=round_.winner_user_id, + round_id=round_id, + ) + await session.commit() + + logger.info("round %s payout broadcast: txid=%s", round_id, built.txid) diff --git a/app/rounds/service.py b/app/rounds/service.py new file mode 100644 index 0000000..e7c219c --- /dev/null +++ b/app/rounds/service.py @@ -0,0 +1,27 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Round + +_ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out") + + +async def get_active_round(session: AsyncSession) -> Round | None: + """The round currently in progress (in any non-closed state), if any. Rounds + never overlap: a new round only opens once the previous one is fully closed + (payout confirmed, or no participants to pay out).""" + return await session.scalar(select(Round).where(Round.status.in_(_ACTIVE_STATUSES)).order_by(Round.id.desc())) + + +async def open_new_round_if_needed(session: AsyncSession) -> Round: + """Returns the active round if one exists (whatever its status), otherwise + opens a fresh one. Callers that need to attach a bet must additionally check + the returned round's status == "open" — a round in closing/drawing/paying_out + isn't accepting new bets, but a new round can't open until it's done.""" + active = await get_active_round(session) + if active is not None: + return active + round_ = Round(status="open") + session.add(round_) + await session.flush() + return round_ diff --git a/app/tx/__init__.py b/app/tx/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/tx/broadcast.py b/app/tx/broadcast.py new file mode 100644 index 0000000..821fe6e --- /dev/null +++ b/app/tx/broadcast.py @@ -0,0 +1,163 @@ +import asyncio +import logging +from datetime import datetime, timedelta, timezone + +from embit import script +from embit.psbt import PSBT +from embit.transaction import Transaction, TransactionInput, TransactionOutput +from embit.finalizer import finalize_psbt +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.config import settings +from app.db.models import PendingTransaction, User +from app.electrum.client import ElectrumClient +from app.wallet.hd import derive_pool_key, derive_user_key +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import RBF_SEQUENCE, estimate_vsize + +logger = logging.getLogger(__name__) + +_POLL_INTERVAL_SECONDS = 30 +_FEE_RATE_INCREMENT = 1 # minimum relay-policy-friendly bump per BIP125 + + +class RbfError(Exception): + pass + + +def should_bump(pending: PendingTransaction, now: datetime, timeout_seconds: int | None = None) -> bool: + """Pure decision: has this pending tx been unconfirmed for longer than the + configured timeout? Kept separate from the I/O-heavy bump_fee() so it's + trivially unit-testable.""" + timeout = timeout_seconds if timeout_seconds is not None else settings.rbf_timeout_seconds + if pending.status != "pending": + return False + return now >= pending.broadcast_at.replace(tzinfo=timezone.utc) + timedelta(seconds=timeout) + + +async def _signing_context(session: AsyncSession, pending: PendingTransaction) -> tuple: + """Returns (signing_key, own_script, own_address) for the single sender that + controls every input of this tx — a user for bet/withdrawal, the pool for + payout. All our builders only ever spend one address's UTXOs per tx.""" + if pending.kind == "payout": + key = derive_pool_key() + else: + user = await session.get(User, pending.user_id) + key = derive_user_key(user.derivation_index) + own_script = script.p2wpkh(key.to_public()) + own_address = own_script.address(network=PLM_MAINNET) + return key, own_script, own_address + + +async def _prevout_amount(client: ElectrumClient, vin: TransactionInput) -> int: + txid_hex = vin.txid.hex() + tx = await client.get_transaction(txid_hex, verbose=True) + value_coins = tx["vout"][vin.vout]["value"] + return round(value_coins * 100_000_000) + + +def _find_change_output(tx: Transaction, change_address: str) -> int | None: + for i, out in enumerate(tx.vout): + if out.script_pubkey.address(network=PLM_MAINNET) == change_address: + return i + return None + + +async def bump_fee(session: AsyncSession, client: ElectrumClient, pending: PendingTransaction) -> str: + """Rebuild `pending`'s transaction with a higher fee (same inputs, same + recipient outputs, the extra fee taken from the change output) and + rebroadcast. Returns the new txid. + + Only handles the common case: exactly one change output paying back to the + tx's own sender address, large enough to absorb the increase. If there's no + such output (e.g. an exact-amount bet with no change), this raises RbfError — + bumping such a tx would require selecting additional inputs, which isn't + implemented for the MVP; it needs manual operator intervention. + """ + old_tx = Transaction.parse(bytes.fromhex(pending.raw_tx_hex)) + signing_key, own_script, own_address = await _signing_context(session, pending) + + input_amounts = [await _prevout_amount(client, vin) for vin in old_tx.vin] + total_in = sum(input_amounts) + old_fee = total_in - sum(o.value for o in old_tx.vout) + + new_fee_rate = pending.fee_rate_sat_vb + _FEE_RATE_INCREMENT + new_fee = estimate_vsize(len(old_tx.vin), len(old_tx.vout)) * new_fee_rate + fee_delta = new_fee - old_fee + if fee_delta <= 0: + fee_delta = _FEE_RATE_INCREMENT # already above the new target vsize*rate; bump by a token amount + + change_index = _find_change_output(old_tx, own_address) + if change_index is None or old_tx.vout[change_index].value <= fee_delta: + raise RbfError(f"pending_transaction {pending.id}: no change output large enough to absorb a fee bump") + + new_vout = list(old_tx.vout) + bumped_change = new_vout[change_index].value - fee_delta + new_vout[change_index] = TransactionOutput(bumped_change, new_vout[change_index].script_pubkey) + + new_vin = [TransactionInput(v.txid, v.vout, sequence=RBF_SEQUENCE) for v in old_tx.vin] + new_tx = Transaction(vin=new_vin, vout=new_vout) + psbt = PSBT(new_tx) + for i, amount in enumerate(input_amounts): + psbt.inputs[i].witness_utxo = TransactionOutput(amount, own_script) + + signed = psbt.sign_with(signing_key) + if signed != len(new_vin): + raise RuntimeError(f"expected {len(new_vin)} signatures, got {signed}") + + final_tx = finalize_psbt(psbt) + if final_tx is None: + raise RuntimeError("failed to finalize bumped PSBT") + + raw_hex = final_tx.serialize().hex() + new_txid = final_tx.txid().hex() + await client.broadcast(raw_hex) + + pending.current_txid = new_txid + pending.raw_tx_hex = raw_hex + pending.fee_rate_sat_vb = new_fee_rate + pending.attempt_count += 1 + pending.broadcast_at = datetime.now(timezone.utc) + await session.commit() + + logger.info("bumped %s pending_transaction %s: %s -> %s", pending.kind, pending.id, pending.current_txid, new_txid) + return new_txid + + +class RbfBumper: + def __init__(self, session_factory: async_sessionmaker, get_client): + self._session_factory = session_factory + self._get_client = get_client + + async def run(self) -> None: + while True: + client = self._get_client() + if client is not None: + try: + await self._tick(client) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("RBF bump tick failed") + await asyncio.sleep(_POLL_INTERVAL_SECONDS) + + async def _tick(self, client: ElectrumClient) -> None: + now = datetime.now(timezone.utc) + async with self._session_factory() as session: + candidates = ( + await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending")) + ).all() + due = [p for p in candidates if should_bump(p, now)] + + for pending in due: + async with self._session_factory() as session: + row = await session.get(PendingTransaction, pending.id) + if row is None or row.status != "pending": + continue + try: + await bump_fee(session, client, row) + except RbfError: + logger.exception("could not bump pending_transaction %s", row.id) + except Exception: + logger.exception("unexpected error bumping pending_transaction %s", row.id) diff --git a/app/tx/confirmation.py b/app/tx/confirmation.py new file mode 100644 index 0000000..edba732 --- /dev/null +++ b/app/tx/confirmation.py @@ -0,0 +1,67 @@ +import asyncio +import logging +from collections.abc import Awaitable, Callable + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.db.models import PendingTransaction +from app.electrum.client import ElectrumClient + +logger = logging.getLogger(__name__) + +_POLL_INTERVAL_SECONDS = 10 + +ConfirmationHandler = Callable[[AsyncSession, PendingTransaction], Awaitable[None]] +_handlers: dict[str, ConfirmationHandler] = {} + + +def register_handler(kind: str, handler: ConfirmationHandler) -> None: + """Domain modules (bets, rounds, withdrawals) register here so this generic + poller can notify them when one of their outgoing txs gets its 1st + confirmation, without this module importing them directly.""" + _handlers[kind] = handler + + +async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) -> int: + async with session_factory() as session: + pending = ( + await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending")) + ).all() + pending_ids = [p.id for p in pending] + + confirmed = 0 + for pending_id, txid, kind in [(p.id, p.current_txid, p.kind) for p in pending]: + tx = await client.get_transaction(txid, verbose=True) + if not tx or tx.get("confirmations", 0) < 1: + continue + async with session_factory() as session: + row = await session.get(PendingTransaction, pending_id) + if row is None or row.status != "pending": + continue + row.status = "confirmed" + handler = _handlers.get(kind) + if handler is not None: + await handler(session, row) + await session.commit() + confirmed += 1 + + return confirmed + + +class ConfirmationPoller: + def __init__(self, session_factory: async_sessionmaker, get_client: Callable[[], ElectrumClient | None]): + self._session_factory = session_factory + self._get_client = get_client + + async def run(self) -> None: + while True: + client = self._get_client() + if client is not None: + try: + await poll_once(self._session_factory, client) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("confirmation poll failed") + await asyncio.sleep(_POLL_INTERVAL_SECONDS) diff --git a/app/tx/locks.py b/app/tx/locks.py new file mode 100644 index 0000000..6439bb5 --- /dev/null +++ b/app/tx/locks.py @@ -0,0 +1,22 @@ +import asyncio +from contextlib import asynccontextmanager + + +class UserLocks: + """Per-user asyncio.Lock registry, shared by PLAY and WITHDRAW so a user can + never have a bet-build and a withdrawal-build in flight at once (both would + otherwise spend from the same UTXO set on the user's dedicated address). + + Single-process-only by design (an in-memory dict of asyncio.Lock) — this is an + accepted MVP constraint; a multi-process deployment would need a DB or Redis + lock instead (e.g. a Postgres advisory lock). + """ + + def __init__(self) -> None: + self._locks: dict[int, asyncio.Lock] = {} + + @asynccontextmanager + async def acquire(self, user_id: int): + lock = self._locks.setdefault(user_id, asyncio.Lock()) + async with lock: + yield diff --git a/app/wallet/__init__.py b/app/wallet/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/wallet/balance.py b/app/wallet/balance.py new file mode 100644 index 0000000..7b0fd4b --- /dev/null +++ b/app/wallet/balance.py @@ -0,0 +1,16 @@ +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import User, UtxoEvent + + +async def recompute_balance(session: AsyncSession, user_id: int) -> int: + """Source of truth: sum of this user's confirmed, unspent UTXOs. Updates and + returns the read-cache column (User.cached_balance_sats). Must be called + within the same transaction as whatever inserted/updated utxo_events rows.""" + balance = await session.scalar( + select(func.sum(UtxoEvent.amount_sats)).where(UtxoEvent.user_id == user_id, UtxoEvent.spent_txid.is_(None)) + ) + user = await session.get(User, user_id) + user.cached_balance_sats = balance or 0 + return user.cached_balance_sats diff --git a/app/wallet/hd.py b/app/wallet/hd.py new file mode 100644 index 0000000..b752f64 --- /dev/null +++ b/app/wallet/hd.py @@ -0,0 +1,52 @@ +import os + +from embit import script +from embit.bip32 import HDKey + +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_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) diff --git a/app/wallet/keystore.py b/app/wallet/keystore.py new file mode 100644 index 0000000..d013713 --- /dev/null +++ b/app/wallet/keystore.py @@ -0,0 +1,11 @@ +from cryptography.fernet import Fernet + +from app.config import settings + + +def encrypt_xprv(xprv_base58: str) -> bytes: + return Fernet(settings.xprv_encryption_key).encrypt(xprv_base58.encode()) + + +def decrypt_xprv(token: bytes) -> str: + return Fernet(settings.xprv_encryption_key).decrypt(token).decode() diff --git a/app/wallet/plm_network.py b/app/wallet/plm_network.py new file mode 100644 index 0000000..499feec --- /dev/null +++ b/app/wallet/plm_network.py @@ -0,0 +1,28 @@ +"""PLM mainnet params for embit, verified against PalladiumWallet/src/Core/Chain/ChainProfiles.cs. + +Threaded explicitly through every embit call via `network=PLM_MAINNET` rather than +registered globally, since this is a long-lived async server (embit has no +concept of "current network" beyond what you pass in). +""" + +PLM_MAINNET = { + "name": "PLM Mainnet", + "wif": bytes([0x80]), + "p2pkh": bytes([55]), + "p2sh": bytes([5]), + "bech32": "plm", + "xprv": bytes.fromhex("0488ade4"), + "xpub": bytes.fromhex("0488b21e"), + "yprv": bytes.fromhex("049d7878"), + "ypub": bytes.fromhex("049d7cb2"), + "zprv": bytes.fromhex("04b2430c"), + "zpub": bytes.fromhex("04b24746"), + "Yprv": bytes.fromhex("0295b005"), + "Ypub": bytes.fromhex("0295b43f"), + "Zprv": bytes.fromhex("02aa7a99"), + "Zpub": bytes.fromhex("02aa7ed3"), + "bip32": 0, +} + +BIP44_COIN_TYPE = 746 +ACCOUNT_PATH = f"m/84h/{BIP44_COIN_TYPE}h/0h" diff --git a/app/wallet/psbt_builder.py b/app/wallet/psbt_builder.py new file mode 100644 index 0000000..251a8fe --- /dev/null +++ b/app/wallet/psbt_builder.py @@ -0,0 +1,181 @@ +from dataclasses import dataclass + +from embit import script +from embit.bip32 import HDKey +from embit.finalizer import finalize_psbt +from embit.psbt import PSBT +from embit.transaction import Transaction, TransactionInput, TransactionOutput + +# Standard P2WPKH size estimates (vbytes): 10.5-byte overhead (version+counts+locktime+ +# segwit marker/flag), ~68 vbytes per input, ~31 vbytes per output. Used to size the fee +# before signing (fee only needs to be "minimized ~1 sat/vB", not maximally precise). +_TX_OVERHEAD_VBYTES = 11 +_P2WPKH_INPUT_VBYTES = 68 +_P2WPKH_OUTPUT_VBYTES = 31 + +# BIP125 opt-in RBF: any sequence < 0xfffffffe signals replaceability. Set on every +# input we create so a stuck tx can later be fee-bumped (tx/broadcast.py, stage 9). +RBF_SEQUENCE = 0xFFFFFFFD + + +class InsufficientFundsError(Exception): + pass + + +@dataclass +class Utxo: + txid: str + vout: int + amount_sats: int + + +@dataclass +class BuiltTransaction: + raw_hex: str + txid: str + fee_sats: int + recipient_sats: int + change_sats: int + spent_utxos: list[Utxo] + + +def estimate_vsize(n_inputs: int, n_outputs: int) -> int: + return _TX_OVERHEAD_VBYTES + n_inputs * _P2WPKH_INPUT_VBYTES + n_outputs * _P2WPKH_OUTPUT_VBYTES + + +def select_utxos(utxos: list[Utxo], target_sats: int) -> tuple[list[Utxo], int]: + """Greedily select UTXOs (largest first, to minimize input count) covering + target_sats — the amount deducted from the sender's balance. The fee is paid + out of target_sats (see build_signed_transaction), not added on top of it.""" + ordered = sorted(utxos, key=lambda u: u.amount_sats, reverse=True) + selected: list[Utxo] = [] + total = 0 + for utxo in ordered: + selected.append(utxo) + total += utxo.amount_sats + if total >= target_sats: + return selected, total + raise InsufficientFundsError("not enough confirmed balance to cover amount") + + +def build_signed_transaction( + *, + signing_key: HDKey, + from_script: script.Script, + utxos: list[Utxo], + to_address: str, + amount_sats: int, + change_address: str, + fee_rate_sat_vb: int, +) -> BuiltTransaction: + """Build, sign and finalize a single-recipient P2WPKH transaction with change + back to change_address. + + `amount_sats` is deducted from the sender's balance in full: the recipient + receives `amount_sats - fee`, change = total_in - amount_sats. This matches the + spec everywhere a single-recipient tx is used (bet, withdrawal): "fee deducted + from the amount being moved", not paid on top by the sender. + """ + selected, total_in = select_utxos(utxos, amount_sats) + fee = estimate_vsize(len(selected), 2) * fee_rate_sat_vb + recipient_amount = amount_sats - fee + if recipient_amount <= 0: + raise InsufficientFundsError("amount too small to cover the network fee") + change = total_in - amount_sats + + # TransactionInput.txid is natural/display byte order (as in tx_hash from Electrum); + # embit reverses it internally when serializing to wire format. + vin = [TransactionInput(bytes.fromhex(u.txid), u.vout, sequence=RBF_SEQUENCE) for u in selected] + vout = [TransactionOutput(recipient_amount, script.Script.from_address(to_address))] + if change > 0: + vout.append(TransactionOutput(change, script.Script.from_address(change_address))) + + tx = Transaction(vin=vin, vout=vout) + psbt = PSBT(tx) + for i, utxo in enumerate(selected): + psbt.inputs[i].witness_utxo = TransactionOutput(utxo.amount_sats, from_script) + + signed_count = psbt.sign_with(signing_key) + if signed_count != len(selected): + raise RuntimeError(f"expected {len(selected)} signatures, got {signed_count}") + + final_tx = finalize_psbt(psbt) + if final_tx is None: + raise RuntimeError("failed to finalize PSBT") + + raw = final_tx.serialize() + return BuiltTransaction( + raw_hex=raw.hex(), + txid=final_tx.txid().hex(), + fee_sats=fee, + recipient_sats=recipient_amount, + change_sats=change, + spent_utxos=selected, + ) + + +@dataclass +class PayoutTransaction: + raw_hex: str + txid: str + fee_sats: int + winner_sats: int + commission_sats: int + change_sats: int + spent_utxos: list[Utxo] + + +def build_payout_transaction( + *, + signing_key: HDKey, + from_script: script.Script, + utxos: list[Utxo], + winner_address: str, + winner_share_sats: int, + fee_address: str, + commission_sats: int, + change_address: str, + fee_rate_sat_vb: int, +) -> PayoutTransaction: + """Build, sign and finalize the round payout: pool -> winner + fee address, + with change back to the pool itself. Per spec, only the winner's share + absorbs the tx fee — the commission (fee_address) output is untouched.""" + target = winner_share_sats + commission_sats + selected, total_in = select_utxos(utxos, target) + fee = estimate_vsize(len(selected), 3) * fee_rate_sat_vb # winner + commission + pool change + winner_amount = winner_share_sats - fee + if winner_amount <= 0: + raise InsufficientFundsError("winner share too small to cover the network fee") + change = total_in - target + + vin = [TransactionInput(bytes.fromhex(u.txid), u.vout, sequence=RBF_SEQUENCE) for u in selected] + vout = [ + TransactionOutput(winner_amount, script.Script.from_address(winner_address)), + TransactionOutput(commission_sats, script.Script.from_address(fee_address)), + ] + if change > 0: + vout.append(TransactionOutput(change, script.Script.from_address(change_address))) + + tx = Transaction(vin=vin, vout=vout) + psbt = PSBT(tx) + for i, utxo in enumerate(selected): + psbt.inputs[i].witness_utxo = TransactionOutput(utxo.amount_sats, from_script) + + signed_count = psbt.sign_with(signing_key) + if signed_count != len(selected): + raise RuntimeError(f"expected {len(selected)} signatures, got {signed_count}") + + final_tx = finalize_psbt(psbt) + if final_tx is None: + raise RuntimeError("failed to finalize PSBT") + + raw = final_tx.serialize() + return PayoutTransaction( + raw_hex=raw.hex(), + txid=final_tx.txid().hex(), + fee_sats=fee, + winner_sats=winner_amount, + commission_sats=commission_sats, + change_sats=change, + spent_utxos=selected, + ) diff --git a/app/withdrawals/__init__.py b/app/withdrawals/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/withdrawals/confirmation.py b/app/withdrawals/confirmation.py new file mode 100644 index 0000000..04d098b --- /dev/null +++ b/app/withdrawals/confirmation.py @@ -0,0 +1,18 @@ +from datetime import datetime, timezone + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import PendingTransaction, Withdrawal +from app.tx.confirmation import register_handler + + +async def _on_withdrawal_confirmed(session: AsyncSession, pending: PendingTransaction) -> None: + if pending.withdrawal_id is None: + return + withdrawal = await session.get(Withdrawal, pending.withdrawal_id) + if withdrawal is not None and withdrawal.status == "broadcast": + withdrawal.status = "confirmed" + withdrawal.confirmed_at = datetime.now(timezone.utc) + + +register_handler("withdrawal", _on_withdrawal_confirmed) diff --git a/app/withdrawals/service.py b/app/withdrawals/service.py new file mode 100644 index 0000000..8e1a76c --- /dev/null +++ b/app/withdrawals/service.py @@ -0,0 +1,86 @@ +from embit import script +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.audit.log import write_audit_log +from app.config import settings +from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal +from app.electrum.client import ElectrumClient +from app.wallet.balance import recompute_balance +from app.wallet.hd import derive_user_key +from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_signed_transaction + + +class WithdrawalError(Exception): + pass + + +async def request_withdrawal( + session: AsyncSession, client: ElectrumClient, user: User, external_address: str, amount_sats: int +) -> Withdrawal: + if amount_sats < settings.min_amount_sats: + raise WithdrawalError(f"amount below the minimum of {settings.min_amount_sats} sats") + + unspent = ( + await session.scalars( + select(UtxoEvent).where(UtxoEvent.user_id == user.id, UtxoEvent.spent_txid.is_(None)) + ) + ).all() + if sum(u.amount_sats for u in unspent) < amount_sats: + raise WithdrawalError("insufficient balance") + + user_key = derive_user_key(user.derivation_index) + from_script = script.p2wpkh(user_key.to_public()) + utxos = [Utxo(u.txid, u.vout, u.amount_sats) for u in unspent] + + try: + built = build_signed_transaction( + signing_key=user_key, + from_script=from_script, + utxos=utxos, + to_address=external_address, + amount_sats=amount_sats, + change_address=user.address, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + ) + except InsufficientFundsError as exc: + raise WithdrawalError(str(exc)) from exc + + await client.broadcast(built.raw_hex) + + spent_by_key = {(u.txid, u.vout): u for u in unspent} + for spent in built.spent_utxos: + spent_by_key[(spent.txid, spent.vout)].spent_txid = built.txid + await recompute_balance(session, user.id) + + withdrawal = Withdrawal( + user_id=user.id, + external_address=external_address, + amount_requested_sats=amount_sats, + amount_sent_sats=built.recipient_sats, + txid=built.txid, + status="broadcast", + ) + session.add(withdrawal) + await session.flush() + session.add( + PendingTransaction( + kind="withdrawal", + withdrawal_id=withdrawal.id, + user_id=user.id, + current_txid=built.txid, + fee_rate_sat_vb=settings.fee_rate_sat_vb, + raw_tx_hex=built.raw_hex, + status="pending", + ) + ) + await write_audit_log( + session, + "withdrawal_sent", + {"txid": built.txid, "amount_sent_sats": built.recipient_sats, "external_address": external_address}, + user_id=user.id, + ) + + await session.commit() + await session.refresh(withdrawal) + return withdrawal diff --git a/migrations/README b/migrations/README new file mode 100644 index 0000000..e0d0858 --- /dev/null +++ b/migrations/README @@ -0,0 +1 @@ +Generic single-database configuration with an async dbapi. \ No newline at end of file diff --git a/migrations/env.py b/migrations/env.py new file mode 100644 index 0000000..0df7a1f --- /dev/null +++ b/migrations/env.py @@ -0,0 +1,85 @@ +import asyncio +from logging.config import fileConfig + +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config + +from alembic import context + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +# Interpret the config file for Python logging. +# This line sets up loggers basically. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +from app.config import settings +from app.db.base import Base +from app.db import models # noqa: F401 (registers models on Base.metadata) + +target_metadata = Base.metadata +config.set_main_option("sqlalchemy.url", settings.database_url) + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode. + + This configures the context with just a URL + and not an Engine, though an Engine is acceptable + here as well. By skipping the Engine creation + we don't even need a DBAPI to be available. + + Calls to context.execute() here emit the given string to the + script output. + + """ + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + """In this scenario we need to create an Engine + and associate a connection with the context. + + """ + + connectable = async_engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + + await connectable.dispose() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode.""" + + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/migrations/script.py.mako b/migrations/script.py.mako new file mode 100644 index 0000000..1101630 --- /dev/null +++ b/migrations/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + """Upgrade schema.""" + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """Downgrade schema.""" + ${downgrades if downgrades else "pass"} diff --git a/migrations/versions/274efdcbfbcc_initial_schema.py b/migrations/versions/274efdcbfbcc_initial_schema.py new file mode 100644 index 0000000..0a48d2e --- /dev/null +++ b/migrations/versions/274efdcbfbcc_initial_schema.py @@ -0,0 +1,153 @@ +"""initial schema + +Revision ID: 274efdcbfbcc +Revises: +Create Date: 2026-07-20 22:11:47.766165 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '274efdcbfbcc' +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('round_config', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('fee_address', sa.String(length=128), nullable=False), + sa.Column('bet_amount_sats', sa.BigInteger(), nullable=False), + sa.Column('updated_at', sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('users', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('username', sa.String(length=64), nullable=False), + sa.Column('password_hash', sa.String(length=256), nullable=False), + sa.Column('derivation_index', sa.Integer(), nullable=False), + sa.Column('address', sa.String(length=128), nullable=False), + sa.Column('cached_balance_sats', sa.BigInteger(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('address'), + sa.UniqueConstraint('derivation_index') + ) + op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True) + op.create_table('rounds', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('opened_at', sa.DateTime(), nullable=False), + sa.Column('closed_at', sa.DateTime(), nullable=True), + sa.Column('draw_block_height', sa.Integer(), nullable=True), + sa.Column('draw_block_hash', sa.String(length=64), nullable=True), + sa.Column('seed_int', sa.String(length=128), nullable=True), + sa.Column('winner_user_id', sa.Integer(), nullable=True), + sa.Column('pool_amount_sats', sa.BigInteger(), nullable=True), + sa.Column('winner_amount_sats', sa.BigInteger(), nullable=True), + sa.Column('fee_amount_sats', sa.BigInteger(), nullable=True), + sa.Column('payout_txid', sa.String(length=64), nullable=True), + sa.ForeignKeyConstraint(['winner_user_id'], ['users.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('utxo_events', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('user_id', sa.Integer(), nullable=False), + sa.Column('txid', sa.String(length=64), nullable=False), + sa.Column('vout', sa.Integer(), nullable=False), + sa.Column('amount_sats', sa.BigInteger(), nullable=False), + sa.Column('confirmed_height', sa.Integer(), nullable=False), + sa.Column('confirmed_at', sa.DateTime(), nullable=False), + sa.Column('spent_txid', sa.String(length=64), nullable=True), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('txid', 'vout') + ) + op.create_index(op.f('ix_utxo_events_user_id'), 'utxo_events', ['user_id'], unique=False) + op.create_table('withdrawals', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('user_id', sa.Integer(), nullable=False), + sa.Column('external_address', sa.String(length=128), nullable=False), + sa.Column('amount_requested_sats', sa.BigInteger(), nullable=False), + sa.Column('amount_sent_sats', sa.BigInteger(), nullable=True), + sa.Column('txid', sa.String(length=64), nullable=True), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('confirmed_at', sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_withdrawals_user_id'), 'withdrawals', ['user_id'], unique=False) + op.create_table('audit_log', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('event_type', sa.String(length=32), nullable=False), + sa.Column('payload_json', sa.String(), nullable=False), + sa.Column('user_id', sa.Integer(), nullable=True), + sa.Column('round_id', sa.Integer(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['round_id'], ['rounds.id'], ), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('pending_transactions', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('kind', sa.String(length=16), nullable=False), + sa.Column('round_id', sa.Integer(), nullable=True), + sa.Column('withdrawal_id', sa.Integer(), nullable=True), + sa.Column('user_id', sa.Integer(), nullable=True), + sa.Column('current_txid', sa.String(length=64), nullable=False), + sa.Column('fee_rate_sat_vb', sa.Integer(), nullable=False), + sa.Column('raw_tx_hex', sa.String(), nullable=False), + sa.Column('broadcast_at', sa.DateTime(), nullable=False), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('replaced_by_txid', sa.String(length=64), nullable=True), + sa.Column('attempt_count', sa.Integer(), nullable=False), + sa.Column('updated_at', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['round_id'], ['rounds.id'], ), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), + sa.ForeignKeyConstraint(['withdrawal_id'], ['withdrawals.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('round_participants', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('round_id', sa.Integer(), nullable=False), + sa.Column('user_id', sa.Integer(), nullable=False), + sa.Column('bet_amount_sats', sa.BigInteger(), nullable=False), + sa.Column('bet_txid', sa.String(length=64), nullable=False), + sa.Column('broadcast_at', sa.DateTime(), nullable=False), + sa.Column('confirmed_at', sa.DateTime(), nullable=True), + sa.Column('status', sa.String(length=16), nullable=False), + sa.ForeignKeyConstraint(['round_id'], ['rounds.id'], ), + sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('round_id', 'user_id') + ) + op.create_index(op.f('ix_round_participants_round_id'), 'round_participants', ['round_id'], unique=False) + op.create_index(op.f('ix_round_participants_user_id'), 'round_participants', ['user_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_round_participants_user_id'), table_name='round_participants') + op.drop_index(op.f('ix_round_participants_round_id'), table_name='round_participants') + op.drop_table('round_participants') + op.drop_table('pending_transactions') + op.drop_table('audit_log') + op.drop_index(op.f('ix_withdrawals_user_id'), table_name='withdrawals') + op.drop_table('withdrawals') + op.drop_index(op.f('ix_utxo_events_user_id'), table_name='utxo_events') + op.drop_table('utxo_events') + op.drop_table('rounds') + op.drop_index(op.f('ix_users_username'), table_name='users') + op.drop_table('users') + op.drop_table('round_config') + # ### end Alembic commands ### diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..60bc12d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[project] +name = "plm-lottery" +version = "0.1.0" +requires-python = ">=3.12" +dependencies = [ + "fastapi>=0.115", + "uvicorn[standard]>=0.32", + "sqlalchemy>=2.0", + "aiosqlite>=0.20", + "alembic>=1.13", + "pydantic-settings>=2.6", + "argon2-cffi>=23.1", + "pyjwt>=2.7", + "cryptography>=43.0", + "embit>=0.7", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.3", + "pytest-asyncio>=0.24", + "httpx>=0.27", +] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/scripts/electrum_smoke_test.py b/scripts/electrum_smoke_test.py new file mode 100644 index 0000000..585c961 --- /dev/null +++ b/scripts/electrum_smoke_test.py @@ -0,0 +1,27 @@ +"""Manual smoke test: connect to the configured Electrum server, do the +server.version handshake, subscribe to headers and confirm a sane tip height. + +Usage: PYTHONPATH=. python scripts/electrum_smoke_test.py +""" + +import asyncio + +from app.config import settings +from app.electrum.client import ElectrumClient + + +async def main() -> None: + client = ElectrumClient(settings.electrum_host, settings.electrum_port, settings.electrum_use_ssl) + await client.connect() + print("connected + handshake OK") + + header = await client.subscribe_headers() + print("tip:", header) + assert header["height"] > 0 + + await client.close() + print("OK") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/generate_master_key.py b/scripts/generate_master_key.py new file mode 100644 index 0000000..4b9e3f1 --- /dev/null +++ b/scripts/generate_master_key.py @@ -0,0 +1,16 @@ +"""One-time ops bootstrap: generate and encrypt the server's master xprv. + +Usage: + python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" + # put that value in XPRV_ENCRYPTION_KEY (.env), then: + python scripts/generate_master_key.py +""" + +from app.config import settings +from app.wallet.hd import generate_master_key + +if __name__ == "__main__": + if not settings.xprv_encryption_key: + raise SystemExit("XPRV_ENCRYPTION_KEY is not set") + generate_master_key() + print(f"Master key written (encrypted) to {settings.master_key_path}") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_admin.py b/tests/unit/test_admin.py new file mode 100644 index 0000000..df9c633 --- /dev/null +++ b/tests/unit/test_admin.py @@ -0,0 +1,64 @@ +import pytest +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") + + 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) + + 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" diff --git a/tests/unit/test_bets.py b/tests/unit/test_bets.py new file mode 100644 index 0000000..8e1d2bb --- /dev/null +++ b/tests/unit/test_bets.py @@ -0,0 +1,106 @@ +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.bets.service import BetError, place_bet +from app.config import settings +from app.db.base import Base +from app.db.models import AuditLog, PendingTransaction, RoundParticipant, User, UtxoEvent +from app.wallet.hd import derive_user_address + + +class FakeElectrumClient: + def __init__(self): + self.broadcasted: list[str] = [] + + async def broadcast(self, raw_tx_hex: str) -> str: + self.broadcasted.append(raw_tx_hex) + return "fake-network-txid" + + +@pytest.fixture +async def session_factory(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc")) + monkeypatch.setattr(settings, "xprv_encryption_key", __import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode()) + from app.wallet import hd + + hd._account_key = None + hd.generate_master_key() + + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + hd._account_key = None + + +async def _make_funded_user(session_factory, index: int, funded_sats: int) -> int: + async with session_factory() as session: + address = derive_user_address(index) + user = User(username=f"user{index}", password_hash="x", derivation_index=index, address=address) + session.add(user) + await session.commit() + session.add( + UtxoEvent( + user_id=user.id, + txid=f"{index:02x}" * 32, + vout=0, + amount_sats=funded_sats, + confirmed_height=100, + ) + ) + await session.commit() + return user.id + + +async def test_place_bet_broadcasts_and_records_participant(session_factory): + user_id = await _make_funded_user(session_factory, 0, 1_500_000_000) + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + participant = await place_bet(session, client, user) + + assert client.broadcasted # a raw tx was broadcast + assert participant.status == "broadcast" + assert participant.bet_txid + + async with session_factory() as session: + utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one() + assert utxo.spent_txid == participant.bet_txid + + pending = (await session.scalars(select(PendingTransaction))).one() + assert pending.kind == "bet" + + audit_events = (await session.scalars(select(AuditLog))).all() + assert any(e.event_type == "bet_placed" for e in audit_events) + assert pending.current_txid == participant.bet_txid + + +async def test_place_bet_rejects_insufficient_balance(session_factory): + user_id = await _make_funded_user(session_factory, 1, 1_000_000) # below bet_amount_sats + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(BetError, match="insufficient balance"): + await place_bet(session, client, user) + + +async def test_place_bet_rejects_second_bet_same_round(session_factory): + user_id = await _make_funded_user(session_factory, 2, 3_000_000_000) + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + await place_bet(session, client, user) + + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(BetError, match="already"): + await place_bet(session, client, user) + + async with session_factory() as session: + participants = (await session.scalars(select(RoundParticipant))).all() + assert len(participants) == 1 diff --git a/tests/unit/test_broadcast.py b/tests/unit/test_broadcast.py new file mode 100644 index 0000000..072b483 --- /dev/null +++ b/tests/unit/test_broadcast.py @@ -0,0 +1,181 @@ +from datetime import datetime, timedelta, timezone + +import pytest +from embit import script +from embit.bip32 import HDKey +from embit.transaction import Transaction +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.config import settings +from app.db.base import Base +from app.db.models import PendingTransaction, User +from app.tx.broadcast import RbfError, bump_fee, should_bump +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import Utxo, build_signed_transaction + + +def _key(seed_byte: int) -> HDKey: + root = HDKey.from_seed(bytes([seed_byte]) * 32, version=PLM_MAINNET["xprv"]) + return root.derive("m/84h/746h/0h/0/0") + + +def test_should_bump_false_before_timeout(): + pending = PendingTransaction( + kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending", + broadcast_at=datetime.now(timezone.utc), + ) + assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is False + + +def test_should_bump_true_after_timeout(): + pending = PendingTransaction( + kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending", + broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000), + ) + assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is True + + +def test_should_bump_false_when_not_pending(): + pending = PendingTransaction( + kind="bet", current_txid="x", fee_rate_sat_vb=1, raw_tx_hex="00", status="confirmed", + broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000), + ) + assert should_bump(pending, datetime.now(timezone.utc), timeout_seconds=900) is False + + +class FakeClient: + def __init__(self, prevout_values: dict[str, int]): + self._prevout_values = prevout_values + self.broadcasted: list[str] = [] + + async def get_transaction(self, txid: str, verbose: bool = False) -> dict: + return {"vout": {0: {"value": self._prevout_values[txid] / 100_000_000}}} + + async def broadcast(self, raw_tx_hex: str) -> str: + self.broadcasted.append(raw_tx_hex) + return "network-txid" + + +@pytest.fixture +async def session_factory(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc")) + monkeypatch.setattr( + settings, + "xprv_encryption_key", + __import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode(), + ) + from app.wallet import hd + + hd._account_key = None + hd.generate_master_key() + + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + hd._account_key = None + + +async def test_bump_fee_shrinks_change_and_rebroadcasts(session_factory): + from app.wallet.hd import derive_user_address, derive_user_key + + signer = derive_user_key(0) + my_address = derive_user_address(0) + from_script = script.p2wpkh(signer.to_public()) + to_address = script.p2wpkh(_key(99).to_public()).address(network=PLM_MAINNET) + + utxo_amount = 150_000_000 + utxo_txid = "11" * 32 + built = build_signed_transaction( + signing_key=signer, + from_script=from_script, + utxos=[Utxo(utxo_txid, 0, utxo_amount)], + to_address=to_address, + amount_sats=10_000_000, + change_address=my_address, + fee_rate_sat_vb=1, + ) + + async with session_factory() as session: + user = User(username="alice", password_hash="x", derivation_index=0, address=my_address) + session.add(user) + await session.commit() + pending = PendingTransaction( + kind="bet", + user_id=user.id, + current_txid=built.txid, + fee_rate_sat_vb=1, + raw_tx_hex=built.raw_hex, + status="pending", + broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000), + ) + session.add(pending) + await session.commit() + pending_id = pending.id + + client = FakeClient({utxo_txid: utxo_amount}) + + async with session_factory() as session: + row = await session.get(PendingTransaction, pending_id) + new_txid = await bump_fee(session, client, row) + + assert client.broadcasted + assert new_txid != built.txid + + new_tx = Transaction.parse(bytes.fromhex(client.broadcasted[0])) + old_tx = Transaction.parse(bytes.fromhex(built.raw_hex)) + old_change = next(o.value for o in old_tx.vout if o.script_pubkey.address(network=PLM_MAINNET) == my_address) + new_change = next(o.value for o in new_tx.vout if o.script_pubkey.address(network=PLM_MAINNET) == my_address) + assert new_change < old_change # fee bump came out of the change output + + async with session_factory() as session: + row = await session.get(PendingTransaction, pending_id) + assert row.current_txid == new_txid + assert row.fee_rate_sat_vb == 2 + assert row.attempt_count == 2 + + +async def test_bump_fee_raises_when_no_change_output(session_factory): + from app.wallet.hd import derive_user_address, derive_user_key + + signer = derive_user_key(0) + my_address = derive_user_address(0) + from_script = script.p2wpkh(signer.to_public()) + to_address = script.p2wpkh(_key(98).to_public()).address(network=PLM_MAINNET) + + utxo_amount = 10_000_000 # exact amount, no change output + utxo_txid = "22" * 32 + built = build_signed_transaction( + signing_key=signer, + from_script=from_script, + utxos=[Utxo(utxo_txid, 0, utxo_amount)], + to_address=to_address, + amount_sats=10_000_000, + change_address=my_address, + fee_rate_sat_vb=1, + ) + + async with session_factory() as session: + user = User(username="bob", password_hash="x", derivation_index=0, address=my_address) + session.add(user) + await session.commit() + pending = PendingTransaction( + kind="bet", + user_id=user.id, + current_txid=built.txid, + fee_rate_sat_vb=1, + raw_tx_hex=built.raw_hex, + status="pending", + broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000), + ) + session.add(pending) + await session.commit() + pending_id = pending.id + + client = FakeClient({utxo_txid: utxo_amount}) + + async with session_factory() as session: + row = await session.get(PendingTransaction, pending_id) + with pytest.raises(RbfError): + await bump_fee(session, client, row) diff --git a/tests/unit/test_confirmation.py b/tests/unit/test_confirmation.py new file mode 100644 index 0000000..24bf060 --- /dev/null +++ b/tests/unit/test_confirmation.py @@ -0,0 +1,82 @@ +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +import app.bets.confirmation # noqa: F401 (registers the "bet" handler) +import app.rounds.confirmation # noqa: F401 (registers the "payout" handler) +from app.db.base import Base +from app.db.models import PendingTransaction, Round, RoundParticipant +from app.tx.confirmation import poll_once + + +class FakeClient: + def __init__(self, confirmations_by_txid: dict[str, int]): + self._confirmations = confirmations_by_txid + + async def get_transaction(self, txid: str, verbose: bool = False) -> dict: + return {"confirmations": self._confirmations.get(txid, 0)} + + +@pytest.fixture +async def session_factory(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + + +async def test_bet_confirmation_marks_participant_confirmed(session_factory): + async with session_factory() as session: + session.add(Round(id=1, status="open")) + session.add( + RoundParticipant( + round_id=1, user_id=1, bet_amount_sats=1_000, bet_txid="tx1", status="broadcast" + ) + ) + session.add( + PendingTransaction(kind="bet", round_id=1, user_id=1, current_txid="tx1", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending") + ) + await session.commit() + + client = FakeClient({"tx1": 1}) + confirmed = await poll_once(session_factory, client) + assert confirmed == 1 + + async with session_factory() as session: + participant = (await session.scalars(select(RoundParticipant))).one() + assert participant.status == "confirmed" + assert participant.confirmed_at is not None + pending = (await session.scalars(select(PendingTransaction))).one() + assert pending.status == "confirmed" + + +async def test_unconfirmed_tx_is_left_pending(session_factory): + async with session_factory() as session: + session.add(Round(id=2, status="open")) + session.add(RoundParticipant(round_id=2, user_id=1, bet_amount_sats=1_000, bet_txid="tx2", status="broadcast")) + session.add(PendingTransaction(kind="bet", round_id=2, user_id=1, current_txid="tx2", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending")) + await session.commit() + + client = FakeClient({"tx2": 0}) + confirmed = await poll_once(session_factory, client) + assert confirmed == 0 + + async with session_factory() as session: + participant = (await session.scalars(select(RoundParticipant))).one() + assert participant.status == "broadcast" + + +async def test_payout_confirmation_closes_round(session_factory): + async with session_factory() as session: + session.add(Round(id=3, status="paying_out", payout_txid="tx3")) + session.add(PendingTransaction(kind="payout", round_id=3, current_txid="tx3", fee_rate_sat_vb=1, raw_tx_hex="00", status="pending")) + await session.commit() + + client = FakeClient({"tx3": 2}) + confirmed = await poll_once(session_factory, client) + assert confirmed == 1 + + async with session_factory() as session: + round_ = await session.get(Round, 3) + assert round_.status == "closed" diff --git a/tests/unit/test_deposits.py b/tests/unit/test_deposits.py new file mode 100644 index 0000000..8571dd3 --- /dev/null +++ b/tests/unit/test_deposits.py @@ -0,0 +1,54 @@ +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import User +from app.deposits.service import credit_confirmed_utxos + + +@pytest.fixture +async def session_factory(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + + +@pytest.fixture +async def user_id(session_factory): + async with session_factory() as session: + user = User(username="alice", password_hash="x", derivation_index=0, address="plm1qxxx") + session.add(user) + await session.commit() + return user.id + + +async def test_credits_confirmed_utxo_and_updates_balance(session_factory, user_id): + entries = [{"tx_hash": "aa" * 32, "tx_pos": 0, "height": 100, "value": 10_000_000}] + async with session_factory() as session: + credited = await credit_confirmed_utxos(session, user_id, entries) + assert credited == 1 + user = await session.get(User, user_id) + assert user.cached_balance_sats == 10_000_000 + + +async def test_unconfirmed_entry_is_ignored(session_factory, user_id): + entries = [{"tx_hash": "bb" * 32, "tx_pos": 0, "height": 0, "value": 5_000_000}] + async with session_factory() as session: + credited = await credit_confirmed_utxos(session, user_id, entries) + assert credited == 0 + user = await session.get(User, user_id) + assert user.cached_balance_sats == 0 + + +async def test_idempotent_on_repeated_notification(session_factory, user_id): + entries = [{"tx_hash": "cc" * 32, "tx_pos": 0, "height": 100, "value": 7_000_000}] + async with session_factory() as session: + first = await credit_confirmed_utxos(session, user_id, entries) + async with session_factory() as session: + second = await credit_confirmed_utxos(session, user_id, entries) + user = await session.get(User, user_id) + assert first == 1 + assert second == 0 + assert user.cached_balance_sats == 7_000_000 diff --git a/tests/unit/test_draw.py b/tests/unit/test_draw.py new file mode 100644 index 0000000..4f1bea6 --- /dev/null +++ b/tests/unit/test_draw.py @@ -0,0 +1,32 @@ +import pytest + +from app.rounds.draw import draw_winner, header_hex_to_block_hash + + +def test_header_hex_to_block_hash_matches_known_mainnet_block(): + # Real PLM mainnet block 477486: header from blockchain.block.header, hash + # cross-checked against the blockhash reported by blockchain.transaction.get + # for a tx confirmed in that block. + header_hex = ( + "0020ed30fbc39ee45200d10214f5107c5f36e7bf753032fd1d3279810c170000000000" + "009a4ea723f732be4c538f3a2490837dd6560103d73ece606aa7d578a66700447b28875" + "e6a47a61b1ad8012582" + ) + known_block_hash = "00000000000008788b55ade13b74d54ceffda9e54315b802411be1ca65064e86" + assert header_hex_to_block_hash(header_hex) == known_block_hash + + +def test_draw_winner_is_deterministic_and_within_range(): + participants = ["addrA", "addrB", "addrC"] + block_hash = "00" * 31 + "05" # seed = 5, index = 5 % 3 = 2 + assert draw_winner(participants, block_hash) == "addrC" + + +def test_draw_winner_single_participant_always_wins(): + block_hash = "ff" * 32 + assert draw_winner(["only"], block_hash) == "only" + + +def test_draw_winner_raises_on_empty_participants(): + with pytest.raises(ValueError): + draw_winner([], "00" * 32) diff --git a/tests/unit/test_electrum_client.py b/tests/unit/test_electrum_client.py new file mode 100644 index 0000000..83a9653 --- /dev/null +++ b/tests/unit/test_electrum_client.py @@ -0,0 +1,77 @@ +import asyncio +import json + +from app.electrum.client import ElectrumClient, ElectrumError + + +class FakeWriter: + def __init__(self): + self.written = b"" + + def write(self, data: bytes) -> None: + self.written += data + + async def drain(self) -> None: + pass + + def close(self) -> None: + pass + + async def wait_closed(self) -> None: + pass + + +async def _client_with_fake_transport() -> tuple[ElectrumClient, asyncio.StreamReader, FakeWriter]: + client = ElectrumClient("localhost", 1234) + reader = asyncio.StreamReader() + writer = FakeWriter() + client._reader = reader + client._writer = writer + client._read_task = asyncio.create_task(client._read_loop()) + return client, reader, writer + + +async def test_request_resolves_on_matching_response(): + client, reader, writer = await _client_with_fake_transport() + + task = asyncio.create_task(client.request("blockchain.headers.subscribe")) + await asyncio.sleep(0) # let request() write the payload + sent = json.loads(writer.written.decode()) + assert sent["method"] == "blockchain.headers.subscribe" + + reader.feed_data((json.dumps({"id": sent["id"], "result": {"height": 100}}) + "\n").encode()) + result = await task + assert result == {"height": 100} + + client._read_task.cancel() + + +async def test_error_response_raises_electrum_error(): + client, reader, writer = await _client_with_fake_transport() + + task = asyncio.create_task(client.request("blockchain.transaction.broadcast", ["deadbeef"])) + await asyncio.sleep(0) + sent = json.loads(writer.written.decode()) + + reader.feed_data((json.dumps({"id": sent["id"], "error": "bad tx"}) + "\n").encode()) + try: + await task + assert False, "expected ElectrumError" + except ElectrumError: + pass + + client._read_task.cancel() + + +async def test_notification_delivered_to_subscription_queue(): + client, reader, writer = await _client_with_fake_transport() + queue = client.notifications("blockchain.scripthash.subscribe") + + push = {"method": "blockchain.scripthash.subscribe", "params": ["abcd", "newstatus"]} + reader.feed_data((json.dumps(push) + "\n").encode()) + await asyncio.sleep(0) + + params = await asyncio.wait_for(queue.get(), timeout=1) + assert params == ["abcd", "newstatus"] + + client._read_task.cancel() diff --git a/tests/unit/test_hd.py b/tests/unit/test_hd.py new file mode 100644 index 0000000..18a1143 --- /dev/null +++ b/tests/unit/test_hd.py @@ -0,0 +1,21 @@ +from embit.bip32 import HDKey + +from app.wallet.plm_network import PLM_MAINNET + + +def test_p2wpkh_address_uses_plm_hrp(): + from embit import script + + root = HDKey.from_seed(b"\x01" * 32, version=PLM_MAINNET["xprv"]) + child = root.derive("m/84h/746h/0h/0/0") + address = script.p2wpkh(child.to_public()).address(network=PLM_MAINNET) + assert address.startswith("plm1q") + + +def test_derivation_is_deterministic(): + root = HDKey.from_seed(b"\x02" * 32, version=PLM_MAINNET["xprv"]) + a = root.derive("m/84h/746h/0h/0/5").sec() + b = root.derive("m/84h/746h/0h/0/5").sec() + c = root.derive("m/84h/746h/0h/0/6").sec() + assert a == b + assert a != c diff --git a/tests/unit/test_payout_builder.py b/tests/unit/test_payout_builder.py new file mode 100644 index 0000000..6d2dcae --- /dev/null +++ b/tests/unit/test_payout_builder.py @@ -0,0 +1,100 @@ +import pytest +from embit import script +from embit.bip32 import HDKey +from embit.transaction import Transaction + +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_payout_transaction, estimate_vsize + + +def _key(seed_byte: int) -> HDKey: + root = HDKey.from_seed(bytes([seed_byte]) * 32, version=PLM_MAINNET["xprv"]) + return root.derive("m/84h/746h/0h/0/0") + + +def test_payout_deducts_fee_only_from_winner_share(): + pool_key = _key(10) + pool_script = script.p2wpkh(pool_key.to_public()) + pool_address = pool_script.address(network=PLM_MAINNET) + winner_address = script.p2wpkh(_key(11).to_public()).address(network=PLM_MAINNET) + fee_address = script.p2wpkh(_key(12).to_public()).address(network=PLM_MAINNET) + + pool_amount = 10_000_000_000 # 100 PLM pot + winner_share = pool_amount * 70 // 100 + commission = pool_amount - winner_share + + utxos = [Utxo("aa" * 32, 0, pool_amount)] + built = build_payout_transaction( + signing_key=pool_key, + from_script=pool_script, + utxos=utxos, + winner_address=winner_address, + winner_share_sats=winner_share, + fee_address=fee_address, + commission_sats=commission, + change_address=pool_address, + fee_rate_sat_vb=1, + ) + + fee = estimate_vsize(1, 3) + assert built.fee_sats == fee + assert built.winner_sats == winner_share - fee + assert built.commission_sats == commission # untouched by the fee + assert built.change_sats == pool_amount - (winner_share + commission) + + parsed = Transaction.parse(bytes.fromhex(built.raw_hex)) + assert len(parsed.vout) == 2 # no change needed: winner_share + commission == pool_amount exactly + amounts = sorted(o.value for o in parsed.vout) + assert amounts == sorted([built.winner_sats, built.commission_sats]) + + +def test_payout_adds_change_output_when_pool_utxos_exceed_target(): + pool_key = _key(20) + pool_script = script.p2wpkh(pool_key.to_public()) + pool_address = pool_script.address(network=PLM_MAINNET) + winner_address = script.p2wpkh(_key(21).to_public()).address(network=PLM_MAINNET) + fee_address = script.p2wpkh(_key(22).to_public()).address(network=PLM_MAINNET) + + winner_share = 700_000_000 + commission = 300_000_000 + utxos = [Utxo("bb" * 32, 0, 2_000_000_000)] # more than winner_share+commission + built = build_payout_transaction( + signing_key=pool_key, + from_script=pool_script, + utxos=utxos, + winner_address=winner_address, + winner_share_sats=winner_share, + fee_address=fee_address, + commission_sats=commission, + change_address=pool_address, + fee_rate_sat_vb=1, + ) + assert built.change_sats == 2_000_000_000 - (winner_share + commission) + + parsed = Transaction.parse(bytes.fromhex(built.raw_hex)) + assert len(parsed.vout) == 3 + + +def test_payout_raises_when_winner_share_too_small(): + pool_key = _key(30) + pool_script = script.p2wpkh(pool_key.to_public()) + pool_address = pool_script.address(network=PLM_MAINNET) + winner_address = script.p2wpkh(_key(31).to_public()).address(network=PLM_MAINNET) + fee_address = script.p2wpkh(_key(32).to_public()).address(network=PLM_MAINNET) + + winner_share = 100 # smaller than the ~172 sat fee at 1 sat/vB for 1-in-3-out + commission = 50 + assert winner_share < estimate_vsize(1, 3) + utxos = [Utxo("cc" * 32, 0, winner_share + commission)] + with pytest.raises(InsufficientFundsError): + build_payout_transaction( + signing_key=pool_key, + from_script=pool_script, + utxos=utxos, + winner_address=winner_address, + winner_share_sats=winner_share, + fee_address=fee_address, + commission_sats=commission, + change_address=pool_address, + fee_rate_sat_vb=1, + ) diff --git a/tests/unit/test_psbt_builder.py b/tests/unit/test_psbt_builder.py new file mode 100644 index 0000000..7f8d34b --- /dev/null +++ b/tests/unit/test_psbt_builder.py @@ -0,0 +1,113 @@ +import pytest +from embit import script +from embit.bip32 import HDKey + +from app.wallet.plm_network import PLM_MAINNET +from app.wallet.psbt_builder import ( + InsufficientFundsError, + Utxo, + build_signed_transaction, + estimate_vsize, + select_utxos, +) + + +def _key(seed_byte: int) -> HDKey: + root = HDKey.from_seed(bytes([seed_byte]) * 32, version=PLM_MAINNET["xprv"]) + return root.derive("m/84h/746h/0h/0/0") + + +def test_estimate_vsize_grows_with_inputs_and_outputs(): + assert estimate_vsize(1, 2) < estimate_vsize(2, 2) + assert estimate_vsize(1, 1) < estimate_vsize(1, 2) + + +def test_select_utxos_picks_largest_first(): + utxos = [Utxo("a" * 64, 0, 5_000_000), Utxo("b" * 64, 0, 20_000_000), Utxo("c" * 64, 0, 1_000_000)] + selected, total = select_utxos(utxos, target_sats=10_000_000) + assert selected == [utxos[1]] # the 20M UTXO alone covers 10M + assert total == 20_000_000 + + +def test_select_utxos_raises_when_insufficient(): + utxos = [Utxo("a" * 64, 0, 1_000_000)] + with pytest.raises(InsufficientFundsError): + select_utxos(utxos, target_sats=10_000_000) + + +def test_build_signed_transaction_deducts_fee_from_amount_not_change(): + signer = _key(1) + from_script = script.p2wpkh(signer.to_public()) + my_address = from_script.address(network=PLM_MAINNET) + to_address = script.p2wpkh(_key(2).to_public()).address(network=PLM_MAINNET) + + utxos = [Utxo("11" * 32, 0, 150_000_000)] + built = build_signed_transaction( + signing_key=signer, + from_script=from_script, + utxos=utxos, + to_address=to_address, + amount_sats=10_000_000, + change_address=my_address, + fee_rate_sat_vb=1, + ) + + fee = estimate_vsize(1, 2) + assert built.fee_sats == fee + assert built.recipient_sats == 10_000_000 - fee + # change reflects the full amount_sats deducted from the sender, fee comes out + # of what the recipient gets, not out of the sender's remaining balance + assert built.change_sats == 150_000_000 - 10_000_000 + assert built.spent_utxos == utxos + assert len(built.txid) == 64 + + from embit.transaction import Transaction + + parsed = Transaction.parse(bytes.fromhex(built.raw_hex)) + assert len(parsed.vin[0].witness.items) == 2 + assert len(parsed.vout) == 2 + + +def test_build_signed_transaction_omits_change_output_when_exact_amount(): + signer = _key(3) + from_script = script.p2wpkh(signer.to_public()) + my_address = from_script.address(network=PLM_MAINNET) + to_address = script.p2wpkh(_key(4).to_public()).address(network=PLM_MAINNET) + + utxos = [Utxo("22" * 32, 0, 10_000_000)] # exactly amount_sats, zero change + built = build_signed_transaction( + signing_key=signer, + from_script=from_script, + utxos=utxos, + to_address=to_address, + amount_sats=10_000_000, + change_address=my_address, + fee_rate_sat_vb=1, + ) + assert built.change_sats == 0 + + from embit.transaction import Transaction + + parsed = Transaction.parse(bytes.fromhex(built.raw_hex)) + assert len(parsed.vout) == 1 + + +def test_build_signed_transaction_raises_when_amount_smaller_than_fee(): + signer = _key(5) + from_script = script.p2wpkh(signer.to_public()) + my_address = from_script.address(network=PLM_MAINNET) + to_address = script.p2wpkh(_key(6).to_public()).address(network=PLM_MAINNET) + + small_amount = 100 # smaller than the ~141 sat fee at 1 sat/vB for 1-in-2-out + assert small_amount < estimate_vsize(1, 2) + utxos = [Utxo("33" * 32, 0, small_amount)] + with pytest.raises(InsufficientFundsError): + build_signed_transaction( + signing_key=signer, + from_script=from_script, + utxos=utxos, + to_address=to_address, + amount_sats=small_amount, + change_address=my_address, + fee_rate_sat_vb=1, + ) diff --git a/tests/unit/test_rounds_service.py b/tests/unit/test_rounds_service.py new file mode 100644 index 0000000..0ef8937 --- /dev/null +++ b/tests/unit/test_rounds_service.py @@ -0,0 +1,58 @@ +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import Round +from app.rounds.service import get_active_round, open_new_round_if_needed + + +@pytest.fixture +async def session_factory(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + + +async def test_opens_a_round_when_none_exists(session_factory): + async with session_factory() as session: + round_ = await open_new_round_if_needed(session) + await session.commit() + assert round_.status == "open" + + +async def test_reuses_existing_open_round(session_factory): + async with session_factory() as session: + first = await open_new_round_if_needed(session) + await session.commit() + first_id = first.id + + async with session_factory() as session: + second = await open_new_round_if_needed(session) + assert second.id == first_id + + +@pytest.mark.parametrize("status", ["closing", "drawing", "paying_out"]) +async def test_does_not_open_new_round_while_previous_is_in_progress(session_factory, status): + async with session_factory() as session: + session.add(Round(status=status)) + await session.commit() + + async with session_factory() as session: + active = await get_active_round(session) + assert active is not None + assert active.status == status + # open_new_round_if_needed must return the in-progress round, not open a new one + returned = await open_new_round_if_needed(session) + assert returned.status == status + + +async def test_opens_new_round_after_previous_is_closed(session_factory): + async with session_factory() as session: + session.add(Round(status="closed")) + await session.commit() + + async with session_factory() as session: + round_ = await open_new_round_if_needed(session) + assert round_.status == "open" diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py new file mode 100644 index 0000000..05b2791 --- /dev/null +++ b/tests/unit/test_scheduler.py @@ -0,0 +1,55 @@ +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.config import settings +from app.db.base import Base +from app.db.models import Round +from app.rounds.scheduler import RoundScheduler + + +class FakeListener: + client = object() # truthy sentinel; _tick only checks "is not None" + tip_height = 100 + tip_header_hex = "00" + + +@pytest.fixture +async def session_factory(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + + +async def test_tick_survives_sqlite_naive_datetime_roundtrip(session_factory, monkeypatch): + """Regression test: SQLite drops tzinfo on round-trip, so opened_at comes back + naive even though it was written as an aware UTC datetime. A prior bug compared + it directly against datetime.now(timezone.utc) and crashed with + "can't compare offset-naive and offset-aware datetimes" on every tick once a + round existed — this must not happen.""" + monkeypatch.setattr(settings, "round_duration_seconds", 3600) # not due yet + async with session_factory() as session: + session.add(Round(status="open", opened_at=datetime.now(timezone.utc))) + await session.commit() + + scheduler = RoundScheduler(session_factory, FakeListener()) + await scheduler._tick() # must not raise + + +async def test_tick_closes_round_with_no_participants_once_due(session_factory, monkeypatch): + monkeypatch.setattr(settings, "round_duration_seconds", 1) + past = datetime.now(timezone.utc) - timedelta(seconds=10) + async with session_factory() as session: + session.add(Round(status="open", opened_at=past)) + await session.commit() + + scheduler = RoundScheduler(session_factory, FakeListener()) + await scheduler._tick() + + async with session_factory() as session: + round_ = (await session.scalars(select(Round))).one() + assert round_.status == "closed" diff --git a/tests/unit/test_scripthash.py b/tests/unit/test_scripthash.py new file mode 100644 index 0000000..d09d3bd --- /dev/null +++ b/tests/unit/test_scripthash.py @@ -0,0 +1,9 @@ +from app.electrum.scripthash import address_to_scripthash + + +def test_matches_known_mainnet_vector(): + # Ground truth: scriptPubKey "0014a195473740aea3b4df1690fbdcb51243fe4e7a20" of a real + # confirmed mainnet output to this address (txid 1ebd0219...b0d48a, vout 1), cross-checked + # against the Electrum server's own listunspent for this scripthash. + address = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n" + assert address_to_scripthash(address) == "5b7744e34b1d6ee3ae6eef7e1ce82aa4c28ff13f10d4aa905640c722c9249111" diff --git a/tests/unit/test_security.py b/tests/unit/test_security.py new file mode 100644 index 0000000..b0ded21 --- /dev/null +++ b/tests/unit/test_security.py @@ -0,0 +1,13 @@ +from app.auth import security + + +def test_password_hash_roundtrip(): + hashed = security.hash_password("s3cret!") + assert security.verify_password("s3cret!", hashed) + assert not security.verify_password("wrong", hashed) + + +def test_jwt_roundtrip(monkeypatch): + monkeypatch.setattr(security.settings, "jwt_secret", "test-secret") + token = security.create_access_token(user_id=42) + assert security.decode_access_token(token) == 42 diff --git a/tests/unit/test_withdrawals.py b/tests/unit/test_withdrawals.py new file mode 100644 index 0000000..57a6a3a --- /dev/null +++ b/tests/unit/test_withdrawals.py @@ -0,0 +1,98 @@ +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.config import settings +from app.db.base import Base +from app.db.models import PendingTransaction, User, UtxoEvent +from app.wallet.hd import derive_user_address +from app.withdrawals.service import WithdrawalError, request_withdrawal + + +class FakeElectrumClient: + def __init__(self): + self.broadcasted: list[str] = [] + + async def broadcast(self, raw_tx_hex: str) -> str: + self.broadcasted.append(raw_tx_hex) + return "fake-network-txid" + + +EXTERNAL_ADDRESS = "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx" + + +@pytest.fixture +async def session_factory(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc")) + monkeypatch.setattr( + settings, + "xprv_encryption_key", + __import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode(), + ) + from app.wallet import hd + + hd._account_key = None + hd.generate_master_key() + + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + yield async_sessionmaker(engine, expire_on_commit=False) + await engine.dispose() + hd._account_key = None + + +async def _make_funded_user(session_factory, index: int, funded_sats: int) -> int: + async with session_factory() as session: + address = derive_user_address(index) + user = User(username=f"user{index}", password_hash="x", derivation_index=index, address=address) + session.add(user) + await session.commit() + session.add( + UtxoEvent(user_id=user.id, txid=f"{index:02x}" * 32, vout=0, amount_sats=funded_sats, confirmed_height=100) + ) + await session.commit() + return user.id + + +async def test_withdrawal_broadcasts_and_updates_balance(session_factory): + user_id = await _make_funded_user(session_factory, 0, 500_000_000) + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + withdrawal = await request_withdrawal(session, client, user, EXTERNAL_ADDRESS, 100_000_000) + + assert client.broadcasted + assert withdrawal.status == "broadcast" + assert withdrawal.amount_sent_sats < 100_000_000 # fee deducted from the amount + + async with session_factory() as session: + user = await session.get(User, user_id) + # The spent UTXO is gone immediately; the change output isn't credited + # until it's independently observed as confirmed on-chain (same as bets) — + # so the cached balance is transiently 0 until then, not the pre-fee delta. + assert user.cached_balance_sats == 0 + pending = (await session.scalars(select(PendingTransaction))).one() + assert pending.kind == "withdrawal" + assert pending.withdrawal_id == withdrawal.id + + +async def test_withdrawal_rejects_amount_below_minimum(session_factory): + user_id = await _make_funded_user(session_factory, 1, 500_000_000) + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(WithdrawalError, match="minimum"): + await request_withdrawal(session, client, user, EXTERNAL_ADDRESS, settings.min_amount_sats - 1) + + +async def test_withdrawal_rejects_insufficient_balance(session_factory): + user_id = await _make_funded_user(session_factory, 2, 1_000_000) + client = FakeElectrumClient() + + async with session_factory() as session: + user = await session.get(User, user_id) + with pytest.raises(WithdrawalError, match="insufficient balance"): + await request_withdrawal(session, client, user, EXTERNAL_ADDRESS, settings.min_amount_sats)