Add a password recovery page and way to generate links from the admin console.

bemani/common/__init__.py

@@ -1,6 +1,6 @@
 from bemani.common.model import Model
 from bemani.common.validateddict import ValidatedDict, Profile, PlayStatistics, intish
-from bemani.common.http import HTTP
+from bemani.common.http import HTTP, format_recovery_link
 from bemani.common.constants import (
     APIConstants,
     GameConstants,
@@ -41,4 +41,5 @@ __all__ = [
     "InvalidOffsetException",
     "cache",
     "debugonly",
+    "format_recovery_link",
 ]

bemani/common/http.py

@@ -1,4 +1,7 @@
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from bemani.data import Config
 
 
 class HTTP:
@@ -160,3 +163,12 @@ class HTTP:
 
         # Concatenate it with the binary data
         return "\r\n".join(out).encode("ascii") + b"\r\n\r\n" + data
+
+
+def format_recovery_link(config: "Config", token: str) -> str:
+    url = f"{config.server.uri}/recover/{token}"
+    while "//" in url:
+        url = url.replace("//", "/")
+    url = url.replace("http:/", "http://")
+    url = url.replace("https:/", "https://")
+    return url

bemani/data/mysql/base.py

@@ -1,5 +1,6 @@
 import json
 import random
+import string
 from typing import Final, Dict, Any, Optional, cast
 
 from bemani.common import Time
@@ -37,7 +38,8 @@ class _BytesEncoder(json.JSONEncoder):
 
 
 class BaseData:
-    SESSION_LENGTH: Final[int] = 32
+    # Chosen to have 512 unique bits of information.
+    SESSION_LENGTH: Final[int] = 24
 
     def __init__(self, config: Config, conn: scoped_session) -> None:
         """
@@ -158,7 +160,7 @@ class BaseData:
         """
         # Create a new session that is unique
         while True:
-            session = "".join(random.choice("0123456789ABCDEF") for _ in range(BaseData.SESSION_LENGTH))
+            session = "".join(random.choice(string.digits + string.ascii_lowercase) for _ in range(BaseData.SESSION_LENGTH))
             sql = "SELECT session FROM session WHERE session = :session"
             cursor = self.execute(sql, {"session": session})
             if cursor.rowcount == 0:

bemani/data/mysql/user.py

@@ -284,6 +284,26 @@ class UserData(BaseData):
 
         return UserID(userid)
 
+    def from_recovery(self, recovery: str) -> Optional[UserID]:
+        """
+        Given a previously-generated recovery token, look up a user ID.
+
+        Parameters:
+            recovery - String identifying a password recovery token that was returned by create_recovery.
+
+        Returns:
+            User ID as an integer if found, or None if the recovery is expired or doesn't exist.
+        """
+        userid = self._from_session(recovery, "recovery")
+        if userid is None:
+            return None
+        sql = "SELECT id FROM user WHERE id = :userid LIMIT 1"
+        cursor = self.execute(sql, {"userid": userid})
+        if cursor.rowcount != 1:
+            return None
+
+        return UserID(userid)
+
     def get_user(self, userid: UserID) -> Optional[User]:
         """
         Given a userid, look up details about the account.
@@ -1344,6 +1364,28 @@ class UserData(BaseData):
         """
         self._destroy_session(session, "userid")
 
+    def create_recovery(self, userid: UserID, expiration: int = (1 * 86400)) -> str:
+        """
+        Given a user ID, create a recovery string for password recovery.
+
+        Parameters:
+            userid - User ID we wish to generate a recovery token for.
+            expiration - Number of seconds before this recovery token is invalid.
+
+        Returns:
+            A string that can be used as a recovery token.
+        """
+        return self._create_session(userid, "recovery", expiration)
+
+    def destroy_recovery(self, recovery: str) -> None:
+        """
+        Destroy a previously-created recovery token.
+
+        Parameters:
+            recovery - A recovery token as returned from create_recovery.
+        """
+        self._destroy_session(recovery, "recovery")
+
     def create_refid(self, game: GameConstants, version: int, userid: UserID) -> str:
         """
         Given a game/version/userid, create a RefID and an ExtID if necessary.

bemani/frontend/account/account.py

@@ -1,4 +1,4 @@
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from flask import (
     Blueprint,
     request,
@@ -75,6 +75,66 @@ def viewlogin() -> Response:
     return Response(render_template("account/login.html", **{"title": "Log In", "show_navigation": False}))
 
 
+def recover_display(username: str, token: Optional[str]) -> Response:
+    return Response(render_template(
+        "account/recover.html",
+        **{"title": "Recover Password", "show_navigation": False, "token": token, "username": username},
+    ))
+
+
+@account_pages.route("/recover", methods=["POST"])
+@loginprohibited
+def recover() -> Response:
+    username = request.form["username"]
+    token = request.form.get("token", "")
+    password1 = request.form["password1"]
+    password2 = request.form["password2"]
+
+    # Now, make sure this account recovery token is valid.
+    if token:
+        userid = g.data.local.user.from_recovery(token)
+    else:
+        userid = None
+    if userid is None:
+        error("Recovery token is invalid or expired!")
+        return recover_display(username, token)
+
+    # And make sure the user is valid and matches this recovery token.
+    user = g.data.local.user.get_user(userid)
+    if user is None:
+        error("Recovery token is invalid or expired!")
+        return recover_display(username, token)
+
+    # Be a little lenient with username spelling here.
+    if user.username.lower() != username.lower():
+        error("Recovery token is not for this account!")
+        return recover_display(username, token)
+
+    # Now, make sure that the passwords match
+    if password1 != password2:
+        error("Passwords do not match each other!")
+        return recover_display(username, token)
+
+    # Now, make sure passwords are long enough
+    if len(password1) < 6:
+        error("Password is not long enough!")
+        return recover_display(username, token)
+
+    # Now, update the account.
+    g.data.local.user.update_password(userid, password1)
+    g.data.local.user.destroy_recovery(token)
+    success("Successfully updated password!")
+    response = make_response(redirect(url_for("account_pages.login")))
+    return response
+
+
+@account_pages.route("/recover/", defaults={"recovery": None})
+@account_pages.route("/recover/<recovery>")
+@loginprohibited
+def viewrecover(recovery: Optional[str]) -> Response:
+    return recover_display("", recovery)
+
+
 def register_display(card_number: str, username: str, email: str) -> Response:
     return Response(
         render_template(

bemani/frontend/templates/account/recover.html

@@ -0,0 +1,22 @@
+{% extends "base.html" %}
+{% block content %}
+    <div class="section">
+        <h2>Recover Account Password</h2>
+
+        <form action="{{ url_for('account_pages.recover') }}" method=post>
+            {% if token %}
+                <input type="hidden" name="token" value="{{ token }}" />
+            {% endif %}
+            <dl>
+                <dt>Username</dt>
+                <dd><input type="text" name="username" value="{{ username }}" /></dd>
+                <dt>Desired Password</dt>
+                <dd><input type="password" name="password1" /></dd>
+                <dt>Desired Password (again)</dt>
+                <dd><input type="password" name="password2" /></dd>
+                <dt></dt>
+                <dd><input type="submit" value="update password" /></dd>
+            </dl>
+        </form>
+    </div>
+{% endblock %}

bemani/utils/dbutils.py

@@ -3,6 +3,7 @@ import getpass
 import sys
 from typing import Optional
 
+from bemani.common import format_recovery_link
 from bemani.data import Config, Data, DBCreateException
 from bemani.utils.config import load_config
 
@@ -30,18 +31,30 @@ def upgrade(config: Config) -> None:
 def change_password(config: Config, username: Optional[str]) -> None:
     if username is None:
         raise Exception("Please provide a username!")
-    password1 = getpass.getpass("Password: ")
-    password2 = getpass.getpass("Re-enter password: ")
-    if password1 != password2:
-        raise Exception("Passwords don't match!")
     data = Data(config)
     userid = data.local.user.from_username(username)
     if userid is None:
         raise Exception("User not found!")
+    password1 = getpass.getpass("Password: ")
+    password2 = getpass.getpass("Re-enter password: ")
+    if password1 != password2:
+        raise Exception("Passwords don't match!")
     data.local.user.update_password(userid, password1)
     print(f"User {username} changed password.")
 
 
+def generate_recovery(config: Config, username: Optional[str]) -> None:
+    if username is None:
+        raise Exception("Please provide a username!")
+    data = Data(config)
+    userid = data.local.user.from_username(username)
+    if userid is None:
+        raise Exception("User not found!")
+    token = data.local.user.create_recovery(userid)
+    url = format_recovery_link(config, token)
+    print(f"User {username} can use the following URL for password recovery: {url}")
+
+
 def add_admin(config: Config, username: Optional[str]) -> None:
     if username is None:
         raise Exception("Please provide a username!")
@@ -72,7 +85,7 @@ def main() -> None:
     parser = argparse.ArgumentParser(description="A utility for working with databases created with this codebase.")
     parser.add_argument(
         "operation",
-        help="Operation to perform, options include 'create', 'generate', 'upgrade', 'change-password', 'add-admin' and 'remove-admin'.",
+        help="Operation to perform, options include 'create', 'generate', 'upgrade', 'change-password', 'generate-recovery', 'add-admin' and 'remove-admin'.",
         type=str,
     )
     parser.add_argument(
@@ -118,6 +131,8 @@ def main() -> None:
             remove_admin(config, args.username)
         elif args.operation == "change-password":
             change_password(config, args.username)
+        elif args.operation == "generate-recovery":
+            generate_recovery(config, args.username)
         else:
             raise Exception(f"Unknown operation '{args.operation}'")
     except DBCreateException as e: