DRY pass #2, batch 4, F11: one email normaliser, one 401, one password-length refusal (#5372)

common.normalize_email is the trim-and-lowercase that sign-up, sign-in,
reset and invite each wrote inline. auth._unauthenticated is the 401 the
six signed-out paths returned, and auth._password_refusal is the
minimum-length check that register, reset and change repeated.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/inkwell/auth.py

@@ -10,7 +10,7 @@ from quart import Blueprint, current_app, g, jsonify, request, session
 from sqlalchemy import delete, func, select
 
 from . import audit
-from .common import iso
+from .common import iso, normalize_email
 from .db import session_scope
 from .invites import INVALID as INVALID_INVITE, record_redeemer, redeem
 from .mailer import send_later
@@ -78,6 +78,19 @@ def _sign_in(user: User) -> None:
     session.permanent = True
 
 
+def _unauthenticated():
+    """401 for a caller who isn't signed in, or no longer is."""
+    return json_error("authentication required", 401)
+
+
+def _password_refusal(password: str):
+    """The 400 for a password too short to keep, or None. Sign-up, reset and change
+    all ask this one question."""
+    if len(password) < MIN_PASSWORD_LEN:
+        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
+    return None
+
+
 def _sign_out() -> None:
     session.pop(SESSION_KEY, None)
     session.pop(EPOCH_KEY, None)
@@ -146,7 +159,7 @@ def login_required(fn):
         if uid is None:
             uid = await _user_id_from_bearer()
         if uid is None:
-            return json_error("authentication required", 401)
+            return _unauthenticated()
         g.user_id = uid
         return await fn(*args, **kwargs)
 
@@ -161,11 +174,11 @@ def require_admin(fn):
     async def wrapper(*args, **kwargs):
         uid = _session_user_id()
         if uid is None:
-            return json_error("authentication required", 401)
+            return _unauthenticated()
         async with session_scope() as db:
             user = await db.get(User, uid)
             if not _session_is_current(user.session_epoch if user else None):
-                return json_error("authentication required", 401)
+                return _unauthenticated()
             if not user.is_admin:
                 return json_error("admin access required", 403)
         g.user_id = uid
@@ -220,15 +233,15 @@ def _sign_in_succeeded(email: str) -> None:
 @bp.post("/register")
 async def register():
     data = await request.get_json(silent=True) or {}
-    email = (data.get("email") or "").strip().lower()
+    email = normalize_email(data.get("email"))
     password = data.get("password") or ""
     display_name = (data.get("display_name") or "").strip()
     invite = (data.get("invite") or "").strip()
 
     if not email or "@" not in email:
         return json_error("a valid email is required", 400)
-    if len(password) < MIN_PASSWORD_LEN:
+    if (refusal := _password_refusal(password)) is not None:
-        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
+        return refusal
     if not display_name:
         display_name = email.split("@", 1)[0]
 
@@ -335,7 +348,7 @@ def _bad_credentials():
 @bp.post("/login")
 async def login():
     data = await request.get_json(silent=True) or {}
-    email = (data.get("email") or "").strip().lower()
+    email = normalize_email(data.get("email"))
     password = data.get("password") or ""
 
     wait = _sign_in_block(email)
@@ -365,7 +378,7 @@ async def me():
         user = await db.get(User, g.user_id)
         if user is None:
             _sign_out()
-            return json_error("authentication required", 401)
+            return _unauthenticated()
         return jsonify(_serialize_user(user))
 
 
@@ -397,7 +410,7 @@ async def forgot_password():
     the answer is unchanged and nothing is sent, so the cap is no oracle either.
     """
     data = await request.get_json(silent=True) or {}
-    email = (data.get("email") or "").strip().lower()
+    email = normalize_email(data.get("email"))
     if not email or "@" not in email:
         return json_error("a valid email is required", 400)
 
@@ -432,8 +445,8 @@ async def reset_password():
     data = await request.get_json(silent=True) or {}
     token = (data.get("token") or "").strip()
     password = data.get("password") or ""
-    if len(password) < MIN_PASSWORD_LEN:
+    if (refusal := _password_refusal(password)) is not None:
-        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
+        return refusal
 
     # The sign-in budget for this address. A token can't be guessed, but each attempt
     # is still a credential attempt, and the log should show a run of them.
@@ -502,13 +515,13 @@ async def change_password():
     data = await request.get_json(silent=True) or {}
     current = data.get("current_password") or ""
     password = data.get("new_password") or ""
-    if len(password) < MIN_PASSWORD_LEN:
+    if (refusal := _password_refusal(password)) is not None:
-        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
+        return refusal
 
     async with session_scope() as db:
         user = await db.get(User, g.user_id)
         if user is None:
-            return json_error("authentication required", 401)
+            return _unauthenticated()
         wait = _sign_in_block(user.email)
         if wait is not None:
             return _throttled(wait)
@@ -536,7 +549,7 @@ async def sign_out_elsewhere():
     async with session_scope() as db:
         user = await db.get(User, g.user_id)
         if user is None:
-            return json_error("authentication required", 401)
+            return _unauthenticated()
         unlinked = await _sign_out_elsewhere(db, user, keep_token=_bearer_token())
         _stay_signed_in(user)
         logger.info("signed out elsewhere email=%s devices_unlinked=%s from=%s", user.email, unlinked, client_address())
@@ -582,7 +595,7 @@ async def device_login():
     """Native first-link: exchange email+password for a device bearer token. Public
     (no existing session) — this is how a fresh native install authenticates."""
     data = await request.get_json(silent=True) or {}
-    email = (data.get("email") or "").strip().lower()
+    email = normalize_email(data.get("email"))
     password = data.get("password") or ""
     if not email or not password:
         return json_error("email and password are required", 400)

src/inkwell/common.py

@@ -40,3 +40,10 @@ def coerce_bool(raw: object) -> bool:
     if isinstance(raw, str):
         return raw.strip().lower() in ("true", "1", "yes", "on")
     return bool(raw)
+
+
+def normalize_email(raw: str | None) -> str:
+    """An address as accounts hold it: trimmed and lowercased. Every route that reads
+    one from a request goes through here, so "Ana@Example.com " and "ana@example.com"
+    are one account at sign-up, sign-in, reset and invite alike."""
+    return (raw or "").strip().lower()

src/inkwell/invites_api.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import aliased
 
 from . import audit
 from .auth import require_admin
+from .common import normalize_email
 from .db import session_scope
 from .invites import MAX_DAYS, lifetime_days, serialize
 from .models.invite import Invite
@@ -31,7 +32,7 @@ logger = logging.getLogger(__name__)
 @require_admin
 async def create_invite():
     data = await request.get_json(silent=True) or {}
-    email = (data.get("email") or "").strip().lower() or None
+    email = normalize_email(data.get("email")) or None
     if email is not None and "@" not in email:
         return json_error("that doesn't look like an email address", 400)
     days = lifetime_days(data.get("days"))