From 92cbef90f9286cb451080845a90bee4c8d642d07 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Wed, 10 Jun 2026 22:01:53 -0700 Subject: [PATCH] Native login (increment 2/3): login/logout endpoints + require_login gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend auth for opt-in username/password mode (security.require_login, default off → zero change; the launch PIN + picker behave exactly as today). - core/security/login_gate.py: pure gate (mirrors launch_lock) — when login mode is on, an unauthenticated session reaches only the page shell, /api/auth/login, /api/auth/logout, /api/profiles/current, /api/setup/status, and the key-authed /api/v1 API. Deliberately does NOT expose the profile list pre-auth (you type your name, not pick from a roster). - _enforce_login before_request enforces it; _enforce_launch_pin no-ops when login mode is on (login replaces the shared PIN, per design). - POST /api/auth/login (username = profile name, case-insensitive; brute-force limited per IP; generic error so names don't leak) + POST /api/auth/logout. - Anti-lockout: the settings save refuses to turn ON login mode until the admin account has a password. Tests: gate blocks→login→access→logout→blocked; case-insensitive username; wrong password / passwordless profile / unknown user all 401 generically; login list not exposed pre-auth; can't enable login without an admin password. 12 tests pass. Next: the login screen + set-password UI + the toggle (increment 3). --- core/security/login_gate.py | 53 ++++++++++++++++++++ tests/test_login_endpoints.py | 82 ++++++++++++++++++++++++++++++ tests/test_login_gate.py | 36 ++++++++++++++ web_server.py | 93 +++++++++++++++++++++++++++++++++++ 4 files changed, 264 insertions(+) create mode 100644 core/security/login_gate.py create mode 100644 tests/test_login_endpoints.py create mode 100644 tests/test_login_gate.py diff --git a/core/security/login_gate.py b/core/security/login_gate.py new file mode 100644 index 00000000..bf521df4 --- /dev/null +++ b/core/security/login_gate.py @@ -0,0 +1,53 @@ +"""Pure gate decision for opt-in username/password login mode. + +When ``security.require_login`` is on, every request must come from an +authenticated session; unauthenticated requests are blocked except the page shell, +the login/logout flow, and the key-authed public API. This is the per-user +equivalent of (and replacement for) the shared launch-PIN gate. + +Deliberately does NOT allowlist the profile LIST or picker — in login mode you log +in by typing your name + password, you don't pick from an exposed roster. +""" + +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 +}) + +# POST endpoints that drive the login flow. +_ALLOWED_POST = frozenset({ + '/api/auth/login', + '/api/auth/logout', +}) + + +def login_request_is_blocked(path: str, method: str, *, + require_login: bool, authenticated: bool) -> bool: + """True when the login gate must reject this request (login mode on + the + session isn't authenticated and the path isn't part of the login flow).""" + if not require_login or authenticated: + return False + + path = path or '' + method = (method or 'GET').upper() + + # Page shell + assets needed to render the login screen. + if path == '/' or path.startswith('/static/') or path.startswith('/favicon'): + return False + + # Key-authed public API governs itself (its own key auth). + if path.startswith('/api/v1/') and not path.startswith('/api/v1/api-keys-internal'): + return False + + if method == 'GET' and path in _ALLOWED_GET: + return False + if method == 'POST' and path in _ALLOWED_POST: + return False + + return True + + +__all__ = ['login_request_is_blocked'] diff --git a/tests/test_login_endpoints.py b/tests/test_login_endpoints.py new file mode 100644 index 00000000..4b9854b0 --- /dev/null +++ b/tests/test_login_endpoints.py @@ -0,0 +1,82 @@ +"""Username/password login endpoints + gate (opt-in login mode).""" + +from __future__ import annotations + +import os +import tempfile + +import pytest + +_TMP = tempfile.mkdtemp(prefix='soulsync-testdb-login-') +os.environ['DATABASE_PATH'] = os.path.join(_TMP, 'l.db') +os.environ['SOULSYNC_TEST_DB_READY'] = '1' + +web_server = pytest.importorskip('web_server') + + +@pytest.fixture +def client(): + return web_server.app.test_client() + + +def _enable_login(monkeypatch): + real_get = web_server.config_manager.get + monkeypatch.setattr(web_server.config_manager, 'get', + lambda k, d=None: True if k == 'security.require_login' else real_get(k, d)) + web_server._login_limiter.record_success('127.0.0.1') # clean slate + + +_GATED = '/api/profiles/me/connections' # a normal, non-allowlisted endpoint + + +def test_login_gate_blocks_then_authenticated_access(client, monkeypatch): + db = web_server.get_database() + pid = db.create_profile(name='LoginUser') + db.set_profile_password(pid, 'secretpw') + _enable_login(monkeypatch) + + assert client.get(_GATED).status_code == 401 # not logged in → blocked + + r = client.post('/api/auth/login', json={'username': 'LoginUser', 'password': 'secretpw'}) + assert r.status_code == 200 and r.get_json()['success'] is True + + assert client.get(_GATED).status_code == 200 # authenticated → in + + assert client.post('/api/auth/logout').get_json()['success'] is True + assert client.get(_GATED).status_code == 401 # logged out → blocked again + + +def test_login_is_case_insensitive_on_username(client, monkeypatch): + db = web_server.get_database() + db.set_profile_password(db.create_profile(name='CaseUser'), 'pw') + _enable_login(monkeypatch) + assert client.post('/api/auth/login', json={'username': 'caseuser', 'password': 'pw'}).status_code == 200 + + +def test_wrong_password_401_generic(client, monkeypatch): + db = web_server.get_database() + db.set_profile_password(db.create_profile(name='WrongPwUser'), 'right') + _enable_login(monkeypatch) + r = client.post('/api/auth/login', json={'username': 'WrongPwUser', 'password': 'nope'}) + assert r.status_code == 401 + assert 'username or password' in r.get_json()['error'].lower() # generic — no name-leak + + +def test_passwordless_profile_cannot_login(client, monkeypatch): + db = web_server.get_database() + db.create_profile(name='NoPwUser') # no password set + _enable_login(monkeypatch) + assert client.post('/api/auth/login', json={'username': 'NoPwUser', 'password': 'x'}).status_code == 401 + + +def test_unknown_user_401(client, monkeypatch): + _enable_login(monkeypatch) + assert client.post('/api/auth/login', json={'username': 'ghost', 'password': 'x'}).status_code == 401 + + +def test_cannot_enable_login_without_admin_password(client): + # admin (1) has no password → enabling login mode is refused (anti-lockout) + web_server.get_database().set_profile_password(1, '') + r = client.post('/api/settings', json={'security': {'require_login': True}}) + assert r.status_code == 400 + assert 'password' in r.get_json().get('error', '').lower() diff --git a/tests/test_login_gate.py b/tests/test_login_gate.py new file mode 100644 index 00000000..51e9699e --- /dev/null +++ b/tests/test_login_gate.py @@ -0,0 +1,36 @@ +"""Pure login-gate decision (opt-in username/password mode).""" + +from __future__ import annotations + +from core.security.login_gate import login_request_is_blocked as blocked + + +def test_off_never_blocks(): + assert blocked('/api/anything', 'GET', require_login=False, authenticated=False) is False + + +def test_authenticated_never_blocked(): + assert blocked('/api/anything', 'GET', require_login=True, authenticated=True) is False + + +def test_unauthenticated_blocked_on_normal_api(): + assert blocked('/api/library', 'GET', require_login=True, authenticated=False) is True + + +def test_login_flow_and_shell_allowed_unauthenticated(): + for p in ('/', '/static/app.js', '/favicon.ico'): + assert blocked(p, 'GET', require_login=True, authenticated=False) is False + assert blocked('/api/auth/login', 'POST', require_login=True, authenticated=False) is False + assert blocked('/api/auth/logout', 'POST', require_login=True, authenticated=False) is False + assert blocked('/api/profiles/current', 'GET', require_login=True, authenticated=False) is False + assert blocked('/api/setup/status', 'GET', require_login=True, authenticated=False) is False + + +def test_profile_list_NOT_exposed_pre_auth(): + # login mode = type your name, don't pick from an exposed roster + assert blocked('/api/profiles', 'GET', require_login=True, authenticated=False) is True + + +def test_key_authed_public_api_allowed(): + assert blocked('/api/v1/search', 'GET', require_login=True, authenticated=False) is False + assert blocked('/api/v1/api-keys-internal', 'GET', require_login=True, authenticated=False) is True diff --git a/web_server.py b/web_server.py index c85c46f4..8345b3ee 100644 --- a/web_server.py +++ b/web_server.py @@ -402,6 +402,36 @@ def inject_webui_assets(): # PINs from one IP trips it — correct entry clears it instantly). from core.security.rate_limit import AttemptLimiter as _AttemptLimiter _launch_pin_limiter = _AttemptLimiter(max_attempts=10, window_seconds=300) +_login_limiter = _AttemptLimiter(max_attempts=10, window_seconds=300) + + +def _require_login_enabled(): + try: + return bool(config_manager.get('security.require_login', False)) if config_manager else False + except Exception: + return False + + +# --- Login gate (opt-in username/password mode; replaces the launch PIN) --- +@app.before_request +def _enforce_login(): + """Server-side enforcement of username/password login. No-op unless + security.require_login is on. When on, an unauthenticated session can only + reach the page shell + the login flow + the key-authed public API.""" + if not _require_login_enabled(): + return + from core.security.login_gate import login_request_is_blocked + from core.security.launch_lock import is_html_navigation + if login_request_is_blocked( + request.path, request.method, + require_login=True, + authenticated=bool(session.get('login_authenticated', False)), + ): + if is_html_navigation(request.method, request.headers.get('Accept', ''), + request.headers.get('Sec-Fetch-Mode', '')): + return redirect('/') + return jsonify({"error": "login_required", "login_required": True}), 401 + # --- Launch PIN gate (before_request hook) --- @app.before_request @@ -414,6 +444,10 @@ def _enforce_launch_pin(): on, except the page shell + the unlock flow + the key-authed public API. No-ops entirely when ``security.require_pin_on_launch`` is off (the default). """ + # Login mode replaces the launch PIN entirely — when it's on, _enforce_login + # owns the gate and this no-ops. + if _require_login_enabled(): + return try: require_pin = bool(config_manager.get('security.require_pin_on_launch', False)) if config_manager else False except Exception: @@ -3079,6 +3113,14 @@ def handle_settings(): if not new_settings: return jsonify({"success": False, "error": "No data received."}), 400 + # Anti-lockout: refuse to turn ON login mode until the admin account + # has a password — otherwise enabling it would lock everyone out. + _sec_in = new_settings.get('security') or {} + if _sec_in.get('require_login') and not config_manager.get('security.require_login', False): + if not get_database().profile_has_password(1): + return jsonify({"success": False, + "error": "Set an admin password before enabling login mode."}), 400 + if 'active_media_server' in new_settings: config_manager.set_active_media_server(new_settings['active_media_server']) @@ -25297,6 +25339,57 @@ def verify_launch_pin(): except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 + +@app.route('/api/auth/login', methods=['POST']) +def auth_login(): + """Username/password login (opt-in login mode). Username = profile name. + Brute-force limited per IP; a profile with no password set can't log in.""" + 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() + password = data.get('password') or '' + if not username or not password: + return jsonify({'success': False, 'error': 'Username and password required'}), 400 + + database = get_database() + profile = database.get_profile_by_name(username) + # Same generic error + a recorded failure whether the name or password is + # wrong — don't leak which names exist. + if not profile or not database.verify_profile_password(profile['id'], password): + _login_limiter.record_failure(_ip, _now) + return jsonify({'success': False, 'error': 'Invalid username or password'}), 401 + + _login_limiter.record_success(_ip) + session['login_authenticated'] = True + session['profile_id'] = profile['id'] + # A fresh login also clears any stale launch-PIN flag. + session.pop('launch_pin_verified', None) + return jsonify({'success': True, 'profile': { + 'id': profile['id'], 'name': profile['name'], 'is_admin': profile['is_admin'], + }}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + +@app.route('/api/auth/logout', methods=['POST']) +def auth_logout(): + """Log out — clears the authenticated session.""" + try: + session.pop('login_authenticated', None) + session.pop('profile_id', None) + 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/reset-pin-via-credential', methods=['POST']) def reset_pin_via_credential(): """Reset admin PIN by verifying a known API credential"""