Split the auth module into hashing, password policy, passwords and login
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c72bc8c6af
commit
4e71c809da
8 files changed
+133
-120
No files matched your search
@@ -1,10 +1,7 @@
|
|||||||
from .auth import ( # noqa
|
from .hashing import check_password, hash_password # noqa
|
||||||
change_password,
|
|
||||||
check_password,
|
|
||||||
hash_password,
|
|
||||||
validate_password,
|
|
||||||
)
|
|
||||||
from .middlewares import AuthMiddleware, no_authentication_required # noqa
|
from .middlewares import AuthMiddleware, no_authentication_required # noqa
|
||||||
|
from .password_policy import validate_password # noqa
|
||||||
|
from .passwords import change_password # noqa
|
||||||
from .routes import init_routes
|
from .routes import init_routes
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,112 +0,0 @@
|
|||||||
import bcrypt
|
|
||||||
from flask import current_app, session
|
|
||||||
|
|
||||||
from max.crypto import force_bytes
|
|
||||||
from max.db import get_user_by_email, update_user
|
|
||||||
from max.exceptions import InvalidPassword
|
|
||||||
from max.sessions import session_storage
|
|
||||||
from max.translations import t
|
|
||||||
|
|
||||||
BLF_CRYPT_PREFIX = "{BLF-CRYPT}"
|
|
||||||
BCRYPT_ROUNDS = 12
|
|
||||||
# bcrypt only uses the first 72 bytes
|
|
||||||
MAX_PASSWORD_BYTES = 72
|
|
||||||
# Hash of a random, unknown password. It is checked for unknown users, so the
|
|
||||||
# response time does not tell whether an email exists.
|
|
||||||
DUMMY_HASH = (
|
|
||||||
BLF_CRYPT_PREFIX + "$2b$12$J27D9dEpQ5ff/I48HG05p.97oJGTO7UiaKGIbVuzV6nw0lo6gZ5wq"
|
|
||||||
)
|
|
||||||
|
|
||||||
LOGIN_FAILED_MESSAGE = "Email or password is wrong, or the account is deactivated"
|
|
||||||
PASSWORD_TOO_LONG_MESSAGE = (
|
|
||||||
"The password must be at most 72 bytes long "
|
|
||||||
"(72 characters, fewer with umlauts or special characters)"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def login(email, password):
|
|
||||||
if not isinstance(email, str) or not isinstance(password, str):
|
|
||||||
return t(LOGIN_FAILED_MESSAGE)
|
|
||||||
if not email or not password:
|
|
||||||
return t(LOGIN_FAILED_MESSAGE)
|
|
||||||
|
|
||||||
user = get_user_by_email(email)
|
|
||||||
if user is None:
|
|
||||||
check_password(password, DUMMY_HASH)
|
|
||||||
return t(LOGIN_FAILED_MESSAGE)
|
|
||||||
|
|
||||||
if not check_password(password, user.passwordhash):
|
|
||||||
return t(LOGIN_FAILED_MESSAGE)
|
|
||||||
|
|
||||||
if not user.enabled:
|
|
||||||
return t(LOGIN_FAILED_MESSAGE)
|
|
||||||
|
|
||||||
rehash_password_if_needed(user, password)
|
|
||||||
|
|
||||||
# Start a new session (new id, new CSRF token) against session fixation.
|
|
||||||
session.clear()
|
|
||||||
session["user_id"] = user.id
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def logout():
|
|
||||||
session.clear()
|
|
||||||
|
|
||||||
|
|
||||||
def change_password(user, password):
|
|
||||||
hash = hash_password(password)
|
|
||||||
update_user(user.id, passwordhash=hash)
|
|
||||||
if session.get("user_id") == user.id:
|
|
||||||
# Own password: keep this session (with a new id), end all others
|
|
||||||
session.regenerate()
|
|
||||||
session_storage.remove_user_sessions(user.id, except_id=session.id)
|
|
||||||
else:
|
|
||||||
# An admin changes the password of another user: end all of their sessions
|
|
||||||
session_storage.remove_user_sessions(user.id)
|
|
||||||
|
|
||||||
|
|
||||||
def rehash_password_if_needed(user, password):
|
|
||||||
"""Replaces an outdated hash, the plain password is only known at the login"""
|
|
||||||
if not needs_rehash(user.passwordhash):
|
|
||||||
return
|
|
||||||
error = validate_password(password)
|
|
||||||
if error:
|
|
||||||
current_app.logger.warning(
|
|
||||||
'Cannot rehash the password of user %s: "%s"', user.id, error
|
|
||||||
)
|
|
||||||
return
|
|
||||||
update_user(user.id, passwordhash=hash_password(password))
|
|
||||||
current_app.logger.info("Rehashed the password of user %s", user.id)
|
|
||||||
|
|
||||||
|
|
||||||
def validate_password(password: str):
|
|
||||||
"""Returns the (untranslated) error message, or None if the password is valid"""
|
|
||||||
if len(force_bytes(password)) > MAX_PASSWORD_BYTES:
|
|
||||||
return PASSWORD_TOO_LONG_MESSAGE
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def hash_password(password: str):
|
|
||||||
error = validate_password(password)
|
|
||||||
if error:
|
|
||||||
raise InvalidPassword(error)
|
|
||||||
|
|
||||||
hash = bcrypt.hashpw(force_bytes(password), bcrypt.gensalt(rounds=BCRYPT_ROUNDS))
|
|
||||||
return BLF_CRYPT_PREFIX + hash.decode()
|
|
||||||
|
|
||||||
|
|
||||||
def needs_rehash(hash: str):
|
|
||||||
# "$2b$12$…": the cost factor follows the version
|
|
||||||
rounds = int(hash[len(BLF_CRYPT_PREFIX) :].split("$")[2]) # noqa
|
|
||||||
return rounds < BCRYPT_ROUNDS
|
|
||||||
|
|
||||||
|
|
||||||
def check_password(password: str, hash: str):
|
|
||||||
if hash.startswith(BLF_CRYPT_PREFIX):
|
|
||||||
# Such a password cannot have been set, and bcrypt would raise
|
|
||||||
if validate_password(password):
|
|
||||||
return False
|
|
||||||
bcrypt_hash = hash[len(BLF_CRYPT_PREFIX) :] # noqa
|
|
||||||
return bcrypt.checkpw(force_bytes(password), force_bytes(bcrypt_hash))
|
|
||||||
|
|
||||||
raise RuntimeError("The hash has an unknown scheme")
|
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
import bcrypt
|
||||||
|
|
||||||
|
from max.crypto import force_bytes
|
||||||
|
from max.exceptions import InvalidPassword
|
||||||
|
|
||||||
|
BLF_CRYPT_PREFIX = "{BLF-CRYPT}"
|
||||||
|
BCRYPT_ROUNDS = 12
|
||||||
|
# bcrypt only uses the first 72 bytes
|
||||||
|
MAX_PASSWORD_BYTES = 72
|
||||||
|
|
||||||
|
|
||||||
|
def is_password_too_long(password: str):
|
||||||
|
return len(force_bytes(password)) > MAX_PASSWORD_BYTES
|
||||||
|
|
||||||
|
|
||||||
|
def hash_password(password: str):
|
||||||
|
if is_password_too_long(password):
|
||||||
|
raise InvalidPassword(f"The password is longer than {MAX_PASSWORD_BYTES} bytes")
|
||||||
|
|
||||||
|
hash = bcrypt.hashpw(force_bytes(password), bcrypt.gensalt(rounds=BCRYPT_ROUNDS))
|
||||||
|
return BLF_CRYPT_PREFIX + hash.decode()
|
||||||
|
|
||||||
|
|
||||||
|
def needs_rehash(hash: str):
|
||||||
|
# "$2b$12$…": the cost factor follows the version
|
||||||
|
rounds = int(hash[len(BLF_CRYPT_PREFIX) :].split("$")[2]) # noqa
|
||||||
|
return rounds < BCRYPT_ROUNDS
|
||||||
|
|
||||||
|
|
||||||
|
def check_password(password: str, hash: str):
|
||||||
|
if hash.startswith(BLF_CRYPT_PREFIX):
|
||||||
|
# Such a password cannot have been set, and bcrypt would raise
|
||||||
|
if is_password_too_long(password):
|
||||||
|
return False
|
||||||
|
bcrypt_hash = hash[len(BLF_CRYPT_PREFIX) :] # noqa
|
||||||
|
return bcrypt.checkpw(force_bytes(password), force_bytes(bcrypt_hash))
|
||||||
|
|
||||||
|
raise RuntimeError("The hash has an unknown scheme")
|
||||||
@@ -0,0 +1,44 @@
|
|||||||
|
from flask import session
|
||||||
|
|
||||||
|
from max.db import get_user_by_email
|
||||||
|
from max.translations import t
|
||||||
|
|
||||||
|
from .hashing import BLF_CRYPT_PREFIX, check_password
|
||||||
|
from .passwords import rehash_password_if_needed
|
||||||
|
|
||||||
|
# Hash of a random, unknown password. It is checked for unknown users, so the
|
||||||
|
# response time does not tell whether an email exists.
|
||||||
|
DUMMY_HASH = (
|
||||||
|
BLF_CRYPT_PREFIX + "$2b$12$J27D9dEpQ5ff/I48HG05p.97oJGTO7UiaKGIbVuzV6nw0lo6gZ5wq"
|
||||||
|
)
|
||||||
|
|
||||||
|
LOGIN_FAILED_MESSAGE = "Email or password is wrong, or the account is deactivated"
|
||||||
|
|
||||||
|
|
||||||
|
def login(email, password):
|
||||||
|
if not isinstance(email, str) or not isinstance(password, str):
|
||||||
|
return t(LOGIN_FAILED_MESSAGE)
|
||||||
|
if not email or not password:
|
||||||
|
return t(LOGIN_FAILED_MESSAGE)
|
||||||
|
|
||||||
|
user = get_user_by_email(email)
|
||||||
|
if user is None:
|
||||||
|
check_password(password, DUMMY_HASH)
|
||||||
|
return t(LOGIN_FAILED_MESSAGE)
|
||||||
|
|
||||||
|
if not check_password(password, user.passwordhash):
|
||||||
|
return t(LOGIN_FAILED_MESSAGE)
|
||||||
|
|
||||||
|
if not user.enabled:
|
||||||
|
return t(LOGIN_FAILED_MESSAGE)
|
||||||
|
|
||||||
|
rehash_password_if_needed(user, password)
|
||||||
|
|
||||||
|
# Start a new session (new id, new CSRF token) against session fixation.
|
||||||
|
session.clear()
|
||||||
|
session["user_id"] = user.id
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def logout():
|
||||||
|
session.clear()
|
||||||
@@ -7,7 +7,7 @@ from max.db import get_user_by_user_id
|
|||||||
from max.middlewares import BaseMiddleware
|
from max.middlewares import BaseMiddleware
|
||||||
from max.views import redirect
|
from max.views import redirect
|
||||||
|
|
||||||
from .auth import logout
|
from .login import logout
|
||||||
|
|
||||||
DEBUG_DEFAULT_LOGIN = False
|
DEBUG_DEFAULT_LOGIN = False
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,13 @@
|
|||||||
|
from .hashing import is_password_too_long
|
||||||
|
|
||||||
|
PASSWORD_TOO_LONG_MESSAGE = (
|
||||||
|
"The password must be at most 72 bytes long "
|
||||||
|
"(72 characters, fewer with umlauts or special characters)"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def validate_password(password: str):
|
||||||
|
"""Returns the (untranslated) error message, or None if the password is valid"""
|
||||||
|
if is_password_too_long(password):
|
||||||
|
return PASSWORD_TOO_LONG_MESSAGE
|
||||||
|
return None
|
||||||
@@ -0,0 +1,33 @@
|
|||||||
|
from flask import current_app, session
|
||||||
|
|
||||||
|
from max.db import update_user
|
||||||
|
from max.sessions import session_storage
|
||||||
|
|
||||||
|
from .hashing import hash_password, needs_rehash
|
||||||
|
from .password_policy import validate_password
|
||||||
|
|
||||||
|
|
||||||
|
def change_password(user, password):
|
||||||
|
hash = hash_password(password)
|
||||||
|
update_user(user.id, passwordhash=hash)
|
||||||
|
if session.get("user_id") == user.id:
|
||||||
|
# Own password: keep this session (with a new id), end all others
|
||||||
|
session.regenerate()
|
||||||
|
session_storage.remove_user_sessions(user.id, except_id=session.id)
|
||||||
|
else:
|
||||||
|
# An admin changes the password of another user: end all of their sessions
|
||||||
|
session_storage.remove_user_sessions(user.id)
|
||||||
|
|
||||||
|
|
||||||
|
def rehash_password_if_needed(user, password):
|
||||||
|
"""Replaces an outdated hash, the plain password is only known at the login"""
|
||||||
|
if not needs_rehash(user.passwordhash):
|
||||||
|
return
|
||||||
|
error = validate_password(password)
|
||||||
|
if error:
|
||||||
|
current_app.logger.warning(
|
||||||
|
'Cannot rehash the password of user %s: "%s"', user.id, error
|
||||||
|
)
|
||||||
|
return
|
||||||
|
update_user(user.id, passwordhash=hash_password(password))
|
||||||
|
current_app.logger.info("Rehashed the password of user %s", user.id)
|
||||||
+1
-1
@@ -5,7 +5,7 @@ from max.permissions import AllowAnyAccess
|
|||||||
from max.translations import t
|
from max.translations import t
|
||||||
from max.views import BaseTemplateGetView, BaseView
|
from max.views import BaseTemplateGetView, BaseView
|
||||||
|
|
||||||
from .auth import login, logout
|
from .login import login, logout
|
||||||
from .middlewares import no_authentication_required
|
from .middlewares import no_authentication_required
|
||||||
|
|
||||||
# Messages for the login page (text, category), selected by ?msg=<key>.
|
# Messages for the login page (text, category), selected by ?msg=<key>.
|
||||||
|
|||||||
Reference in new issue
Block a user