Add single-user local authentication
This commit is contained in:
+57
-5
@@ -1,23 +1,27 @@
|
||||
"""Logins: none, CWA accounts, or a reverse proxy header. An API key works in every mode.
|
||||
"""Logins: none, one local account, CWA accounts, or a proxy. API keys work in every mode.
|
||||
|
||||
CWA accounts use an optional read-only app.db, with OPDS as the fallback when
|
||||
the database cannot answer. A rejected database login never falls back.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
from werkzeug.security import check_password_hash
|
||||
|
||||
from bindery import calibre_db, config
|
||||
from bindery.cwa import CwaClient, CwaError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
AUTH_NONE, AUTH_CWA, AUTH_PROXY = "none", "cwa", "proxy"
|
||||
AUTH_NONE, AUTH_LOCAL, AUTH_CWA, AUTH_PROXY = "none", "local", "cwa", "proxy"
|
||||
MAX_FAILURES = 5 # for one user name from one address
|
||||
MAX_ADDRESS_FAILURES = 20 # for all user names from one address
|
||||
LOCKOUT_SECONDS = 15 * 60
|
||||
@@ -33,14 +37,40 @@ class Identity:
|
||||
via: str # "none", "session", "proxy" or "api_key"
|
||||
|
||||
|
||||
@lru_cache(maxsize=8)
|
||||
def _valid_local_hash(value: str) -> bool:
|
||||
"""Validate once per configured hash, not with an expensive hash on each request."""
|
||||
try:
|
||||
method, salt, digest = value.split("$")
|
||||
if not salt or not method.startswith(("scrypt:", "pbkdf2:")):
|
||||
return False
|
||||
size = 64 if method.startswith("scrypt:") else hashlib.new(method.split(":")[1]).digest_size
|
||||
if len(digest) != size * 2 or bytes.fromhex(digest).hex() != digest:
|
||||
return False
|
||||
# This also rejects unsupported parameters and hashes of an empty password.
|
||||
return not check_password_hash(value, "")
|
||||
except (ValueError, OverflowError):
|
||||
return False
|
||||
|
||||
|
||||
def local_password_version() -> str:
|
||||
"""Invalidate local sessions when the configured password hash changes."""
|
||||
return hashlib.sha256(config.LOCAL_PASSWORD_HASH.encode()).hexdigest()
|
||||
|
||||
|
||||
def config_error() -> str | None:
|
||||
"""Why the login settings cannot work, or None.
|
||||
|
||||
Bindery then refuses every request, so a mistake in the settings never turns the logins off.
|
||||
"""
|
||||
method = config.AUTH_METHOD
|
||||
if method not in (AUTH_NONE, AUTH_CWA, AUTH_PROXY):
|
||||
return f"AUTH_METHOD must be none, cwa or proxy, not {method!r}"
|
||||
if method not in (AUTH_NONE, AUTH_LOCAL, AUTH_CWA, AUTH_PROXY):
|
||||
return f"AUTH_METHOD must be none, local, cwa or proxy, not {method!r}"
|
||||
if method == AUTH_LOCAL:
|
||||
if not config.LOCAL_USERNAME or not config.LOCAL_PASSWORD_HASH:
|
||||
return "AUTH_METHOD=local needs BINDERY_USERNAME and BINDERY_PASSWORD_HASH"
|
||||
if not _valid_local_hash(config.LOCAL_PASSWORD_HASH):
|
||||
return "BINDERY_PASSWORD_HASH must be a Werkzeug scrypt or PBKDF2 hash of a nonempty password"
|
||||
if method == AUTH_CWA and not (config.CWA_URL or config.CWA_DB_PATH):
|
||||
return "AUTH_METHOD=cwa needs CWA_URL or CWA_DB_PATH"
|
||||
if method == AUTH_PROXY and not config.PROXY_AUTH_USER_HEADER:
|
||||
@@ -83,7 +113,12 @@ def current_identity(headers: Mapping[str, str], session: Mapping[str, Any]) ->
|
||||
name = (headers.get(config.PROXY_AUTH_USER_HEADER) or "").strip()
|
||||
return Identity(name, "proxy") if name else None
|
||||
name = session.get("user")
|
||||
if name and session.get("mode") == AUTH_CWA and _allowed(name):
|
||||
if not name or session.get("mode") != mode:
|
||||
return None
|
||||
if mode == AUTH_LOCAL:
|
||||
if name == config.LOCAL_USERNAME and session.get("local_password_version") == local_password_version():
|
||||
return Identity(name, "session")
|
||||
elif mode == AUTH_CWA and _allowed(name):
|
||||
return Identity(name, "session")
|
||||
return None
|
||||
|
||||
@@ -142,6 +177,23 @@ def _login_ok(username: str, password: str, client: CwaClient) -> bool:
|
||||
return client.login_ok(username, password)
|
||||
|
||||
|
||||
def login_local(username: str, password: str, address: str) -> Identity:
|
||||
username = username.strip()
|
||||
if active_mode() != AUTH_LOCAL:
|
||||
raise AuthError("Bindery does not use local logins")
|
||||
if error := config_error():
|
||||
raise AuthError(error)
|
||||
if limiter.locked(username, address):
|
||||
raise AuthError("Too many wrong passwords. Try again in 15 minutes.")
|
||||
# Check the password even for a wrong user name, so both failures take the same work.
|
||||
password_ok = check_password_hash(config.LOCAL_PASSWORD_HASH, password)
|
||||
if not (password_ok and password and hmac.compare_digest(username.encode(), config.LOCAL_USERNAME.encode())):
|
||||
limiter.failed(username, address)
|
||||
raise AuthError("Wrong user name or password")
|
||||
limiter.clear(username, address)
|
||||
return Identity(config.LOCAL_USERNAME, "session")
|
||||
|
||||
|
||||
def login_cwa(username: str, password: str, client: CwaClient, address: str) -> Identity:
|
||||
"""Check a CWA login. Raises AuthError with a message for the user."""
|
||||
username = username.strip()
|
||||
|
||||
+3
-1
@@ -52,8 +52,10 @@ CALIBRE_DB_PATH = _optional_path("CALIBRE_DB_PATH") # calibre metadata.db, for
|
||||
UPDATE_INTERVAL_HOURS = float(os.getenv("UPDATE_INTERVAL_HOURS", "24"))
|
||||
DEFAULT_LOCKED_FIELDS = _list("DEFAULT_LOCKED_FIELDS", "title")
|
||||
|
||||
# Logins: none, cwa or proxy. An API key works in every mode.
|
||||
# Logins: none, local, cwa or proxy. An API key works in every mode.
|
||||
AUTH_METHOD = os.getenv("AUTH_METHOD", "none").strip().lower()
|
||||
LOCAL_USERNAME = os.getenv("BINDERY_USERNAME", "").strip()
|
||||
LOCAL_PASSWORD_HASH = os.getenv("BINDERY_PASSWORD_HASH", "").strip()
|
||||
CWA_ALLOWED_USERS = _list("CWA_ALLOWED_USERS")
|
||||
PROXY_AUTH_USER_HEADER = os.getenv("PROXY_AUTH_USER_HEADER", "X-Remote-User").strip()
|
||||
PROXY_AUTH_LOGOUT_URL = os.getenv("PROXY_AUTH_LOGOUT_URL", "").strip()
|
||||
|
||||
@@ -78,7 +78,7 @@ function shell() {
|
||||
auth.mode !== 'none' &&
|
||||
auth.user && [
|
||||
h('span', {}, auth.user),
|
||||
(auth.mode === 'cwa' || auth.logout_url) &&
|
||||
(auth.mode === 'cwa' || auth.mode === 'local' || auth.logout_url) &&
|
||||
h('button', { type: 'button', onclick: () => logout().catch(showError) }, 'Log out'),
|
||||
];
|
||||
nav = h('nav', { 'aria-label': 'Main navigation' }, h('a', { href: '#/' }, 'Stories'), h('a', { href: '#/settings' }, 'Settings'));
|
||||
@@ -134,7 +134,7 @@ function loginPage() {
|
||||
},
|
||||
},
|
||||
h('h1', { class: 'brand' }, h('img', { src: 'static/favicon.svg', alt: '', width: 28, height: 28 }), 'Bindery'),
|
||||
h('p', { class: 'hint' }, 'Log in with your CWA account.'),
|
||||
h('p', { class: 'hint' }, auth.mode === 'local' ? 'Log in with your Bindery account.' : 'Log in with your CWA account.'),
|
||||
h('label', {}, 'User name', username),
|
||||
h('label', {}, 'Password', password),
|
||||
h('label', {}, remember, ' Keep me logged in'),
|
||||
|
||||
+10
-7
@@ -179,17 +179,20 @@ def auth_check() -> Any:
|
||||
def auth_login() -> Any:
|
||||
data = _json()
|
||||
try:
|
||||
identity = auth.login_cwa(
|
||||
str(data.get("username", "")),
|
||||
str(data.get("password", "")),
|
||||
orchestrator.cwa(),
|
||||
auth.client_address(request.headers, request.remote_addr),
|
||||
)
|
||||
username = str(data.get("username", ""))
|
||||
password = str(data.get("password", ""))
|
||||
address = auth.client_address(request.headers, request.remote_addr)
|
||||
if auth.active_mode() == auth.AUTH_LOCAL:
|
||||
identity = auth.login_local(username, password, address)
|
||||
else:
|
||||
identity = auth.login_cwa(username, password, orchestrator.cwa(), address)
|
||||
except auth.AuthError as e:
|
||||
return {"error": str(e)}, 401
|
||||
session.clear()
|
||||
session["user"] = identity.name
|
||||
session["mode"] = auth.AUTH_CWA
|
||||
session["mode"] = auth.active_mode()
|
||||
if auth.active_mode() == auth.AUTH_LOCAL:
|
||||
session["local_password_version"] = auth.local_password_version()
|
||||
session.permanent = bool(data.get("remember"))
|
||||
logger.info(f"{identity.name} logged in")
|
||||
return {"authenticated": True, "user": identity.name}
|
||||
|
||||
Reference in New Issue
Block a user