Make usernames one case-insensitive namespace (B-57)

The login throttle keyed on body.username.lower() while the lookup matched
User.username exactly, so "Bob" and "bob" were two accounts sharing one
rate-limit bucket — each able to lock the other out — and registration happily
accepted near-duplicate names, which on a custodial system is an impersonation
vector.

Uniqueness is now the database's job: a unique index on lower(username), with
register and login both matching through func.lower(). The name is still stored
exactly as typed, since that's what /admin and the audit log display, and the
username pattern is ASCII-only so lower() is the whole of the normalization.

The migration refuses to run if two existing accounts differ only by case. It
can't merge or rename one automatically: both are custodial accounts that may
hold funds, so that would be the migration silently deciding who owns what. It
names the collisions and leaves them to the operator — the container runs
`alembic upgrade head` at startup, so it surfaces as a refusal to start rather
than a half-applied schema. Verified both directions against a scratch DB, plus
`alembic check` (clean) and the collision guard actually firing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-03 22:27:15 +02:00
co-authored by Claude Opus 5
parent ab65728bdc
commit 57721355f0
6 changed files with 109 additions and 16 deletions
+6 -2
View File
@@ -75,7 +75,11 @@ async def register(
raise _rate_limited_error(retry_after)
limiters.register_ip.record_failure(ip_key)
existing = await session.scalar(select(User).where(User.username == body.username))
# B-57: case-insensitive, matching the unique index on lower(username) — and
# matching the throttle key below, which has always been lowercased.
existing = await session.scalar(
select(User).where(func.lower(User.username) == body.username.lower())
)
if existing is not None:
raise http_error(status.HTTP_409_CONFLICT, "username_taken", "username already taken")
@@ -134,7 +138,7 @@ async def login(
if retry_after > 0:
raise _rate_limited_error(retry_after)
user = await session.scalar(select(User).where(User.username == body.username))
user = await session.scalar(select(User).where(func.lower(User.username) == body.username.lower()))
if user is None or not await verify_password_async(body.password, user.password_hash):
# Same code path (and therefore the same response) whether the username
# doesn't exist or the password is wrong — no enumeration oracle here.