Login recovery (DB + backend): security question to reset a forgotten password
Closes the forgot-login-password gap. A per-profile recovery question + answer lets
a locked-out user reset their own password.
- DB: additive recovery_question + recovery_answer_hash columns (idempotent
migration). set/get-question/verify/has methods; answer is hashed (pbkdf2) and
matched forgivingly (trim + lowercase + collapse whitespace). No recovery set →
never verifies.
- Endpoints (allowlisted in the login gate so they work pre-auth):
GET /api/auth/recovery-question?username= (generic 404 when absent),
POST /api/auth/recovery-reset {username, answer, new_password} — brute-force
limited; a correct answer sets the new password + authenticates the session.
POST /api/profiles/<id>/set-recovery (admin or self) to configure it.
Tests: set/get/verify, forgiving match, hashed-not-plaintext, no-recovery-never-
verifies, full reset flow (wrong answer rejected + password intact; correct answer
resets), unknown-user 404. 25 tests pass. Next: the Settings + login-screen UI.
This commit is contained in:
BoulderBadgeDad
parent
9e40f5c12d
commit
613688a9ad
5 changed files with 259 additions and 2 deletions
|
|
@ -13,14 +13,16 @@ from __future__ import annotations
|
|||
|
||||
# GET endpoints the login screen itself needs before auth.
|
||||
_ALLOWED_GET = frozenset({
|
||||
'/api/profiles/current', # how the frontend detects login state
|
||||
'/api/setup/status', # first-run check runs before the login screen
|
||||
'/api/profiles/current', # how the frontend detects login state
|
||||
'/api/setup/status', # first-run check runs before the login screen
|
||||
'/api/auth/recovery-question', # forgot-password: fetch the security question
|
||||
})
|
||||
|
||||
# POST endpoints that drive the login flow.
|
||||
_ALLOWED_POST = frozenset({
|
||||
'/api/auth/login',
|
||||
'/api/auth/logout',
|
||||
'/api/auth/recovery-reset', # forgot-password: answer + set a new password
|
||||
})
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -461,6 +461,7 @@ class MusicDatabase:
|
|||
self._add_profile_settings(cursor)
|
||||
self._add_profile_listenbrainz_support(cursor)
|
||||
self._add_profile_password_support(cursor)
|
||||
self._add_profile_recovery_support(cursor)
|
||||
self._add_profile_service_credentials(cursor)
|
||||
self._add_service_credential_sets(cursor)
|
||||
self._add_soul_id_columns(cursor)
|
||||
|
|
@ -5455,6 +5456,93 @@ class MusicDatabase:
|
|||
except Exception as e:
|
||||
logger.error(f"Error in login-password migration: {e}")
|
||||
|
||||
# ── Login-password recovery (security question + answer) ──────────────────
|
||||
|
||||
@staticmethod
|
||||
def _normalize_recovery_answer(answer: str) -> str:
|
||||
"""Forgiving match: trim + lowercase + collapse internal whitespace."""
|
||||
return ' '.join((answer or '').strip().lower().split())
|
||||
|
||||
def set_profile_recovery(self, profile_id: int, question: str, answer: str) -> bool:
|
||||
"""Set (or clear, when either is empty) a profile's recovery Q + answer."""
|
||||
try:
|
||||
from werkzeug.security import generate_password_hash
|
||||
q = (question or '').strip()
|
||||
norm = self._normalize_recovery_answer(answer)
|
||||
if not q or not norm:
|
||||
question_val, answer_hash = None, None # clear
|
||||
else:
|
||||
question_val = q
|
||||
answer_hash = generate_password_hash(norm, method='pbkdf2:sha256')
|
||||
with self._get_connection() as conn:
|
||||
conn.execute(
|
||||
"UPDATE profiles SET recovery_question = ?, recovery_answer_hash = ? WHERE id = ?",
|
||||
(question_val, answer_hash, profile_id))
|
||||
conn.commit()
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error setting recovery for profile {profile_id}: {e}")
|
||||
return False
|
||||
|
||||
def get_profile_recovery_question(self, profile_id: int) -> Optional[str]:
|
||||
"""The recovery question text, or None if none set."""
|
||||
try:
|
||||
with self._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT recovery_question FROM profiles WHERE id = ?", (profile_id,))
|
||||
row = cursor.fetchone()
|
||||
return row['recovery_question'] if row and row['recovery_question'] else None
|
||||
except Exception as e:
|
||||
logger.error(f"Error reading recovery question for profile {profile_id}: {e}")
|
||||
return None
|
||||
|
||||
def verify_profile_recovery_answer(self, profile_id: int, answer: str) -> bool:
|
||||
"""Verify the recovery answer. No recovery set → never verifies."""
|
||||
try:
|
||||
from werkzeug.security import check_password_hash
|
||||
with self._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT recovery_answer_hash FROM profiles WHERE id = ?", (profile_id,))
|
||||
row = cursor.fetchone()
|
||||
if not row or not row['recovery_answer_hash']:
|
||||
return False
|
||||
return check_password_hash(row['recovery_answer_hash'], self._normalize_recovery_answer(answer))
|
||||
except Exception as e:
|
||||
logger.error(f"Error verifying recovery answer for profile {profile_id}: {e}")
|
||||
return False
|
||||
|
||||
def profile_has_recovery(self, profile_id: int) -> bool:
|
||||
try:
|
||||
with self._get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT recovery_answer_hash FROM profiles WHERE id = ?", (profile_id,))
|
||||
row = cursor.fetchone()
|
||||
return bool(row and row['recovery_answer_hash'])
|
||||
except Exception as e:
|
||||
logger.error(f"Error checking recovery for profile {profile_id}: {e}")
|
||||
return False
|
||||
|
||||
def _add_profile_recovery_support(self, cursor):
|
||||
"""Add recovery question + answer-hash columns (login-password recovery)."""
|
||||
try:
|
||||
cursor.execute("SELECT value FROM metadata WHERE key = 'profiles_recovery_v1' LIMIT 1")
|
||||
if cursor.fetchone():
|
||||
return # Already migrated
|
||||
logger.info("Applying per-profile recovery-question migration...")
|
||||
for col_sql in (
|
||||
"ALTER TABLE profiles ADD COLUMN recovery_question TEXT DEFAULT NULL",
|
||||
"ALTER TABLE profiles ADD COLUMN recovery_answer_hash TEXT DEFAULT NULL",
|
||||
):
|
||||
try:
|
||||
cursor.execute(col_sql)
|
||||
except sqlite3.OperationalError:
|
||||
pass # Column already exists
|
||||
cursor.execute(
|
||||
"INSERT OR REPLACE INTO metadata (key, value) VALUES ('profiles_recovery_v1', '1')")
|
||||
logger.info("Per-profile recovery-question migration completed")
|
||||
except Exception as e:
|
||||
logger.error(f"Error in recovery-question migration: {e}")
|
||||
|
||||
def close(self):
|
||||
"""Close database connection (no-op since we create connections per operation)"""
|
||||
# Each operation creates and closes its own connection, so nothing to do here
|
||||
|
|
|
|||
|
|
@ -125,3 +125,41 @@ def test_everything_normal_when_both_off(client, monkeypatch):
|
|||
monkeypatch.setattr(web_server.config_manager, 'get',
|
||||
lambda k, d=None: False if k in ('security.require_login', 'security.require_pin_on_launch') else real_get(k, d))
|
||||
assert client.get('/api/profiles/me/connections').status_code == 200 # reachable, unguarded
|
||||
|
||||
|
||||
def test_recovery_flow_resets_password(client, monkeypatch):
|
||||
db = web_server.get_database()
|
||||
pid = db.create_profile(name='RecoverMe')
|
||||
db.set_profile_password(pid, 'oldpassword')
|
||||
db.set_profile_recovery(pid, 'First pet?', 'Rex')
|
||||
_enable_login(monkeypatch)
|
||||
|
||||
# forgot-password flow is reachable pre-auth
|
||||
q = client.get('/api/auth/recovery-question?username=RecoverMe').get_json()
|
||||
assert q['success'] and q['question'] == 'First pet?'
|
||||
|
||||
# wrong answer → 401, password unchanged
|
||||
bad = client.post('/api/auth/recovery-reset',
|
||||
json={'username': 'RecoverMe', 'answer': 'Fido', 'new_password': 'newpass1'})
|
||||
assert bad.status_code == 401
|
||||
assert db.verify_profile_password(pid, 'oldpassword') is True
|
||||
|
||||
# correct answer → password reset + authenticated
|
||||
ok = client.post('/api/auth/recovery-reset',
|
||||
json={'username': 'RecoverMe', 'answer': 'rex', 'new_password': 'brandnew1'})
|
||||
assert ok.status_code == 200 and ok.get_json()['success'] is True
|
||||
assert db.verify_profile_password(pid, 'brandnew1') is True
|
||||
assert db.verify_profile_password(pid, 'oldpassword') is False
|
||||
|
||||
|
||||
def test_recovery_question_404_for_unknown(client, monkeypatch):
|
||||
_enable_login(monkeypatch)
|
||||
assert client.get('/api/auth/recovery-question?username=ghost').status_code == 404
|
||||
|
||||
|
||||
def test_set_recovery_endpoint(client):
|
||||
db = web_server.get_database()
|
||||
pid = db.create_profile(name='SetRec')
|
||||
r = client.post(f'/api/profiles/{pid}/set-recovery', json={'question': 'Q?', 'answer': 'A'})
|
||||
assert r.get_json()['has_recovery'] is True
|
||||
assert db.verify_profile_recovery_answer(pid, 'a') is True
|
||||
|
|
|
|||
60
tests/test_profile_recovery.py
Normal file
60
tests/test_profile_recovery.py
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
"""Login-password recovery via security question + answer — DB layer."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from database.music_database import MusicDatabase
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db(tmp_path):
|
||||
return MusicDatabase(str(tmp_path / "music.db"))
|
||||
|
||||
|
||||
def test_migration_adds_recovery_columns(db):
|
||||
with db._get_connection() as conn:
|
||||
cols = [r[1] for r in conn.execute("PRAGMA table_info(profiles)").fetchall()]
|
||||
assert 'recovery_question' in cols and 'recovery_answer_hash' in cols
|
||||
|
||||
|
||||
def test_set_get_verify(db):
|
||||
pid = db.create_profile(name='RecUser')
|
||||
assert db.profile_has_recovery(pid) is False
|
||||
assert db.get_profile_recovery_question(pid) is None
|
||||
|
||||
db.set_profile_recovery(pid, 'First pet?', 'Rex')
|
||||
assert db.profile_has_recovery(pid) is True
|
||||
assert db.get_profile_recovery_question(pid) == 'First pet?'
|
||||
assert db.verify_profile_recovery_answer(pid, 'Rex') is True
|
||||
assert db.verify_profile_recovery_answer(pid, 'Fido') is False
|
||||
|
||||
|
||||
def test_answer_match_is_forgiving(db):
|
||||
pid = db.create_profile(name='Forgiving')
|
||||
db.set_profile_recovery(pid, 'City?', ' New York ')
|
||||
assert db.verify_profile_recovery_answer(pid, 'new york') is True # case + spacing
|
||||
assert db.verify_profile_recovery_answer(pid, 'NEW YORK') is True
|
||||
|
||||
|
||||
def test_no_recovery_never_verifies(db):
|
||||
pid = db.create_profile(name='NoRec')
|
||||
assert db.verify_profile_recovery_answer(pid, '') is False
|
||||
assert db.verify_profile_recovery_answer(pid, 'anything') is False
|
||||
|
||||
|
||||
def test_clearing_recovery(db):
|
||||
pid = db.create_profile(name='ClearRec')
|
||||
db.set_profile_recovery(pid, 'Q?', 'A')
|
||||
assert db.profile_has_recovery(pid) is True
|
||||
db.set_profile_recovery(pid, '', '')
|
||||
assert db.profile_has_recovery(pid) is False
|
||||
assert db.get_profile_recovery_question(pid) is None
|
||||
|
||||
|
||||
def test_answer_is_hashed_not_plaintext(db):
|
||||
pid = db.create_profile(name='Hashed')
|
||||
db.set_profile_recovery(pid, 'Q?', 'secretanswer')
|
||||
with db._get_connection() as conn:
|
||||
stored = conn.execute("SELECT recovery_answer_hash FROM profiles WHERE id = ?", (pid,)).fetchone()[0]
|
||||
assert 'secretanswer' not in stored and stored.startswith('pbkdf2:')
|
||||
|
|
@ -25397,6 +25397,75 @@ def auth_logout():
|
|||
return jsonify({'success': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/auth/recovery-question', methods=['GET'])
|
||||
def auth_recovery_question():
|
||||
"""Return the recovery security-question for a username (forgot-password flow).
|
||||
Generic when the user/question is absent — don't confirm which names exist."""
|
||||
try:
|
||||
username = (request.args.get('username') or '').strip()
|
||||
database = get_database()
|
||||
profile = database.get_profile_by_name(username) if username else None
|
||||
question = database.get_profile_recovery_question(profile['id']) if profile else None
|
||||
if not question:
|
||||
return jsonify({'success': False, 'error': 'No recovery question available'}), 404
|
||||
return jsonify({'success': True, 'question': question})
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/auth/recovery-reset', methods=['POST'])
|
||||
def auth_recovery_reset():
|
||||
"""Reset a login password by answering the recovery question. Brute-force
|
||||
limited; a correct answer sets the new password and authenticates the session."""
|
||||
try:
|
||||
_ip = request.remote_addr or 'unknown'
|
||||
_now = time.time()
|
||||
_locked, _retry_after = _login_limiter.is_locked(_ip, _now)
|
||||
if _locked:
|
||||
return (jsonify({'success': False, 'error': 'Too many attempts — please wait and try again'}),
|
||||
429, {'Retry-After': str(_retry_after)})
|
||||
|
||||
data = request.json or {}
|
||||
username = (data.get('username') or '').strip()
|
||||
answer = data.get('answer') or ''
|
||||
new_password = data.get('new_password') or ''
|
||||
if not username or not answer or not new_password:
|
||||
return jsonify({'success': False, 'error': 'Username, answer and new password are required'}), 400
|
||||
if len(new_password) < 6:
|
||||
return jsonify({'success': False, 'error': 'New password must be at least 6 characters'}), 400
|
||||
|
||||
database = get_database()
|
||||
profile = database.get_profile_by_name(username)
|
||||
if not profile or not database.verify_profile_recovery_answer(profile['id'], answer):
|
||||
_login_limiter.record_failure(_ip, _now)
|
||||
return jsonify({'success': False, 'error': 'Incorrect answer'}), 401
|
||||
|
||||
_login_limiter.record_success(_ip)
|
||||
database.set_profile_password(profile['id'], new_password)
|
||||
session['login_authenticated'] = True
|
||||
session['profile_id'] = profile['id']
|
||||
session.pop('launch_pin_verified', None)
|
||||
return jsonify({'success': True})
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/profiles/<int:profile_id>/set-recovery', methods=['POST'])
|
||||
def set_profile_recovery_endpoint(profile_id):
|
||||
"""Set or clear a profile's recovery question + answer (admin, or self)."""
|
||||
try:
|
||||
database = get_database()
|
||||
current_pid = get_current_profile_id()
|
||||
current = database.get_profile(current_pid)
|
||||
if not current or (not current['is_admin'] and current_pid != profile_id):
|
||||
return jsonify({'success': False, 'error': 'Unauthorized'}), 403
|
||||
data = request.json or {}
|
||||
ok = database.set_profile_recovery(profile_id, data.get('question', ''), data.get('answer', ''))
|
||||
return jsonify({'success': bool(ok), 'has_recovery': database.profile_has_recovery(profile_id)})
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/profiles/reset-pin-via-credential', methods=['POST'])
|
||||
def reset_pin_via_credential():
|
||||
"""Reset admin PIN by verifying a known API credential"""
|
||||
|
|
|
|||
Loading…
Reference in a new issue