Native login (increment 2/3): login/logout endpoints + require_login gate

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).
This commit is contained in:
BoulderBadgeDad 2026-06-10 22:01:53 -07:00
parent 8e1b678d6f
commit 92cbef90f9
4 changed files with 264 additions and 0 deletions

View file

@ -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']

View file

@ -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()

36
tests/test_login_gate.py Normal file
View file

@ -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

View file

@ -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"""