Validate and bound the QR endpoint (B-67)

/qr/{address} is reachable without auth (the dashboard renders it with a
plain <img> tag, which cannot carry a bearer token), and it did two things
it should not: it accepted any plm1-prefixed string matching a shape regex,
without checking the bech32 checksum, and it ran qrcode.make on the event
loop — a free CPU amplifier that also stalled the scheduler and the
listener for the duration of every request.

Validation now goes through is_valid_plm_address, the same check
withdrawals and the admin fee_address validator use, behind a length guard
so an oversized path never reaches embit. The render is memoized per
address in a bounded LRU (valid addresses are cheap to generate, so an
unbounded cache would just move the amplification to memory) and pushed
off the loop with run_in_threadpool, and the response carries a
Cache-Control so a browser stops re-asking for an image that never changes.
The rejection now uses the structured error contract (invalid_address),
which already has its i18n key.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-04 14:18:22 +02:00
co-authored by Claude Opus 5
parent 23d58796b6
commit 5f6abe5b32
3 changed files with 147 additions and 27 deletions
-17
View File
@@ -38,23 +38,6 @@ remains the last prerequisite for running unattended.
---
## Medium — correctness and robustness
### B-67 — `/qr/{address}` is unauthenticated, synchronous and only shape-validated
`app/api/routes/qr.py:11-21`.
`qrcode.make` runs on the event loop, so the endpoint is a cheap CPU amplifier for
an unauthenticated caller, and the regex accepts any `plm1[a-z0-9]{10,90}` string
without validating the bech32 checksum — so it happily renders a QR for a
non-address.
Fix: validate with `is_valid_plm_address` (already used by withdrawals and by the
admin `fee_address` validator), and either offload the render or cache it per
address.
---
## Low — documentation and consistency drift
### B-68 — CLAUDE.md and README describe a state the code has moved past
+46 -10
View File
@@ -1,22 +1,58 @@
"""PNG QR codes for PLM addresses.
Deliberately unauthenticated: the dashboard renders it with a plain `<img>`
tag, which cannot carry the bearer token, and the payload is an address the
caller already has. What the endpoint must not be is a free CPU amplifier
(B-67), so two things bound the work an anonymous caller can ask for:
- the address is validated for real (bech32 checksum + PLM HRP) via
`is_valid_plm_address`, the same check withdrawals and the admin
`fee_address` validator use, instead of a shape-only regex that happily
rendered a QR for any `plm1`-prefixed junk string;
- the render itself is memoized per address and pushed off the event loop, so
a repeat request costs a dict lookup and a first one never blocks the
scheduler, the listener or any other request.
"""
import io
import re
from functools import lru_cache
import qrcode
from fastapi import APIRouter, HTTPException, status
from fastapi import APIRouter
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import Response
from app.api.errors import http_error
from app.wallet.address import is_valid_plm_address
router = APIRouter(tags=["qr"])
# PLM P2WPKH addresses: bech32 HRP "plm" + separator + witness program.
_ADDRESS_RE = re.compile(r"^plm1[a-z0-9]{10,90}$")
# Long enough for any bech32 address, short enough that a multi-kilobyte path
# is rejected before embit ever looks at it.
_MAX_ADDRESS_LENGTH = 100
# Bounded on purpose: valid addresses are cheap to generate, so an unbounded
# cache would just move the amplification from CPU to memory.
_CACHE_SIZE = 512
@lru_cache(maxsize=_CACHE_SIZE)
def _render_png(address: str) -> bytes:
image = qrcode.make(address)
buf = io.BytesIO()
image.save(buf, format="PNG")
return buf.getvalue()
@router.get("/qr/{address}")
async def address_qr(address: str) -> Response:
if not _ADDRESS_RE.match(address):
raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid address")
if len(address) > _MAX_ADDRESS_LENGTH or not is_valid_plm_address(address):
raise http_error(400, "invalid_address", "not a valid PLM bech32 address")
image = qrcode.make(address)
buf = io.BytesIO()
image.save(buf, format="PNG")
return Response(content=buf.getvalue(), media_type="image/png")
png = await run_in_threadpool(_render_png, address)
# An address' QR never changes; let the browser stop asking for it.
return Response(
content=png,
media_type="image/png",
headers={"Cache-Control": "private, max-age=86400, immutable"},
)
+101
View File
@@ -0,0 +1,101 @@
"""B-67: /qr/{address} must validate the address for real and must not render
QR codes on the event loop for every anonymous request."""
import asyncio
import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from app.api.routes import qr
VALID_ADDRESS = "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd"
@pytest.fixture
async def client():
qr._render_png.cache_clear()
app = FastAPI()
app.include_router(qr.router)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
async def test_valid_address_renders_a_png(client):
resp = await client.get(f"/qr/{VALID_ADDRESS}")
assert resp.status_code == 200
assert resp.headers["content-type"] == "image/png"
assert resp.content.startswith(b"\x89PNG")
assert "max-age" in resp.headers["cache-control"]
@pytest.mark.parametrize(
"address",
[
"plm1qbogus0000000000000000000000000000000000", # right shape, broken checksum
"bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4", # valid bech32, wrong chain
"plm1q", # too short to be anything
"P" + "a" * 40, # not bech32 at all
"plm1" + "q" * 200, # over the length guard
],
)
async def test_non_addresses_are_rejected_without_rendering(client, address, monkeypatch):
def explode(*args, **kwargs): # pragma: no cover - must never run
raise AssertionError("rendered a QR for a non-address")
monkeypatch.setattr(qr.qrcode, "make", explode)
resp = await client.get(f"/qr/{address}")
assert resp.status_code == 400
assert resp.json()["detail"]["code"] == "invalid_address"
async def test_render_is_memoized_per_address(client):
calls = 0
original = qr.qrcode.make
def counting_make(data, *args, **kwargs):
nonlocal calls
calls += 1
return original(data, *args, **kwargs)
qr.qrcode.make = counting_make
try:
for _ in range(3):
assert (await client.get(f"/qr/{VALID_ADDRESS}")).status_code == 200
finally:
qr.qrcode.make = original
assert calls == 1
async def test_render_does_not_block_the_event_loop(client):
"""The render runs in a threadpool, so the loop stays responsive while it does."""
ticks = 0
async def ticker():
nonlocal ticks
while True:
ticks += 1
await asyncio.sleep(0)
original = qr.qrcode.make
def slow_make(data, *args, **kwargs):
# Blocking sleep: on the event loop this would freeze the ticker.
import time
time.sleep(0.05)
return original(data, *args, **kwargs)
qr.qrcode.make = slow_make
task = asyncio.create_task(ticker())
try:
assert (await client.get(f"/qr/{VALID_ADDRESS}")).status_code == 200
finally:
qr.qrcode.make = original
task.cancel()
assert ticks > 1