Profiles Phase 3: gate shared-destructive endpoints to admin server-side

The audit found these were UI-hidden but API-open — any profile (or the
anonymous default-admin) could call them directly. Added @admin_only to the 15
that mutate SHARED/global state:
- DB: update, backup, backups DELETE, restore, vacuum
- library: track DELETE, album DELETE, tracks delete-batch, clear-match
- plex clear-library; metadata-cache clear + clear-musicbrainz
- internal API keys: list, generate, revoke

Deliberately NOT gated: profile-scoped own-data ops like /api/wishlist/clear
(clears the caller's OWN wishlist via profile_id) — gating that would wrongly
block a non-admin from managing their own data. Verified by test.

Zero change for single-profile installs (the default profile IS admin), so
existing users are unaffected; only genuine non-admin profiles get 403.

Tests: non-admin → 403 on all 15 (the 403 fires before the view body, so no
destructive op runs); admin not blocked on the read-only one; wishlist/clear
stays open to non-admins (over-gating guard). 17 tests.
This commit is contained in:
BoulderBadgeDad 2026-06-10 00:37:56 -07:00
parent d1dd6ed714
commit af24a08cc7
2 changed files with 95 additions and 0 deletions

View file

@ -0,0 +1,80 @@
"""Phase 3: server-side admin gating of shared/global-destructive endpoints.
The audit found these were callable by any profile (UI hid them, the API didn't).
For a real multi-user setup that's unsafe — a non-admin could restore/vacuum the
DB, wipe the shared library, clear the Plex library, or mint API keys. These
assert the @admin_only gate now blocks non-admins, that admin is NOT blocked
(zero change for single-profile installs, where everyone is the default admin),
and crucially that a PROFILE-SCOPED op (clearing your OWN wishlist) was NOT
over-gated.
"""
from __future__ import annotations
import os
import tempfile
import pytest
_TMP = tempfile.mkdtemp(prefix='ss-admin-gate-')
os.environ['DATABASE_PATH'] = os.path.join(_TMP, 'gate.db')
os.environ['SOULSYNC_TEST_DB_READY'] = '1'
web_server = pytest.importorskip('web_server')
# (method, path) for every endpoint that must be admin-only.
GATED = [
('GET', '/api/v1/api-keys-internal'),
('POST', '/api/v1/api-keys-internal/generate'),
('DELETE', '/api/v1/api-keys-internal/revoke/abc'),
('POST', '/api/plex/clear-library'),
('PUT', '/api/library/clear-match'),
('DELETE', '/api/library/track/123'),
('DELETE', '/api/library/album/123'),
('POST', '/api/library/tracks/delete-batch'),
('POST', '/api/database/update'),
('POST', '/api/database/backup'),
('DELETE', '/api/database/backups/x.db'),
('POST', '/api/database/backups/x.db/restore'),
('POST', '/api/database/maintenance/vacuum'),
('DELETE', '/api/metadata-cache/clear'),
('DELETE', '/api/metadata-cache/clear-musicbrainz'),
]
@pytest.fixture
def client():
return web_server.app.test_client()
@pytest.fixture
def nonadmin(client):
pid = web_server.get_database().create_profile(name=f'u_{os.urandom(3).hex()}')
with client.session_transaction() as sess:
sess['profile_id'] = pid
return pid
def _call(client, method, path):
return client.open(path, method=method, json={})
@pytest.mark.parametrize('method,path', GATED)
def test_nonadmin_blocked(client, nonadmin, method, path):
# @admin_only returns 403 BEFORE the view body runs, so this never triggers
# the underlying destructive operation — safe to assert across all of them.
assert _call(client, method, path).status_code == 403, f"{method} {path} should be 403 for non-admin"
def test_admin_not_blocked_by_the_gate(client):
# Default session = profile 1 (admin). Prove the gate lets admin through on a
# SAFE, read-only gated endpoint (listing API keys) — confirming the no-change
# guarantee for single-profile installs without triggering a destructive op.
assert client.get('/api/v1/api-keys-internal').status_code != 403
def test_profile_scoped_wishlist_clear_not_overgated(client, nonadmin):
# Clearing your OWN wishlist is profile-scoped data — a non-admin MUST still
# be allowed. This is the guard against a blanket sweep.
assert _call(client, 'POST', '/api/wishlist/clear').status_code != 403

View file

@ -2927,6 +2927,7 @@ def get_activity_logs():
# --- Internal API Key Management (browser-only, no auth) --- # --- Internal API Key Management (browser-only, no auth) ---
@app.route('/api/v1/api-keys-internal', methods=['GET']) @app.route('/api/v1/api-keys-internal', methods=['GET'])
@admin_only
def list_api_keys_internal(): def list_api_keys_internal():
"""List API keys for the settings page (no auth required — same as all UI routes).""" """List API keys for the settings page (no auth required — same as all UI routes)."""
keys = config_manager.get('api_keys', []) keys = config_manager.get('api_keys', [])
@ -2943,6 +2944,7 @@ def list_api_keys_internal():
return jsonify({"success": True, "data": {"keys": safe_keys}}) return jsonify({"success": True, "data": {"keys": safe_keys}})
@app.route('/api/v1/api-keys-internal/generate', methods=['POST']) @app.route('/api/v1/api-keys-internal/generate', methods=['POST'])
@admin_only
def generate_api_key_internal(): def generate_api_key_internal():
"""Generate API key from settings page (no auth required).""" """Generate API key from settings page (no auth required)."""
from api.auth import generate_api_key from api.auth import generate_api_key
@ -2961,6 +2963,7 @@ def generate_api_key_internal():
}}), 201 }}), 201
@app.route('/api/v1/api-keys-internal/revoke/<key_id>', methods=['DELETE']) @app.route('/api/v1/api-keys-internal/revoke/<key_id>', methods=['DELETE'])
@admin_only
def revoke_api_key_internal(key_id): def revoke_api_key_internal(key_id):
"""Revoke API key from settings page (no auth required).""" """Revoke API key from settings page (no auth required)."""
keys = config_manager.get('api_keys', []) keys = config_manager.get('api_keys', [])
@ -4040,6 +4043,7 @@ def get_plex_pin_status():
@app.route('/api/plex/clear-library', methods=['POST']) @app.route('/api/plex/clear-library', methods=['POST'])
@admin_only
def clear_plex_library_preference(): def clear_plex_library_preference():
try: try:
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
@ -11636,6 +11640,7 @@ def library_manual_match():
logger.error(f"Error manual matching: {e}") logger.error(f"Error manual matching: {e}")
@app.route('/api/library/clear-match', methods=['PUT']) @app.route('/api/library/clear-match', methods=['PUT'])
@admin_only
def library_clear_match(): def library_clear_match():
"""Clear a service ID match for an entity, reverting it to not_found. """Clear a service ID match for an entity, reverting it to not_found.
Body: { entity_type: str, entity_id: str, service: str } Body: { entity_type: str, entity_id: str, service: str }
@ -11751,6 +11756,7 @@ def library_import_existing_track_for_missing_slot(album_id):
@app.route('/api/library/track/<track_id>', methods=['DELETE']) @app.route('/api/library/track/<track_id>', methods=['DELETE'])
@admin_only
def library_delete_track(track_id): def library_delete_track(track_id):
"""Delete a track from the database, optionally deleting the file and blacklisting the source.""" """Delete a track from the database, optionally deleting the file and blacklisting the source."""
try: try:
@ -12322,6 +12328,7 @@ def sync_artist_library(artist_id):
return jsonify({"success": False, "error": str(e)}), 500 return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/library/album/<album_id>', methods=['DELETE']) @app.route('/api/library/album/<album_id>', methods=['DELETE'])
@admin_only
def library_delete_album(album_id): def library_delete_album(album_id):
"""Delete an album and all its tracks from the database, optionally deleting files on disk.""" """Delete an album and all its tracks from the database, optionally deleting files on disk."""
try: try:
@ -12399,6 +12406,7 @@ def library_delete_album(album_id):
@app.route('/api/library/tracks/delete-batch', methods=['POST']) @app.route('/api/library/tracks/delete-batch', methods=['POST'])
@admin_only
def library_delete_tracks_batch(): def library_delete_tracks_batch():
"""Delete multiple track records from the database (does NOT delete files on disk). """Delete multiple track records from the database (does NOT delete files on disk).
Body: { track_ids: [int] } Body: { track_ids: [int] }
@ -16055,6 +16063,7 @@ def add_album_track_to_wishlist():
return jsonify({"success": False, "error": str(e)}), 500 return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/database/update', methods=['POST']) @app.route('/api/database/update', methods=['POST'])
@admin_only
def start_database_update(): def start_database_update():
"""Endpoint to start the database update process.""" """Endpoint to start the database update process."""
global db_update_worker global db_update_worker
@ -16111,6 +16120,7 @@ def stop_database_update():
_BACKUP_FILENAME_RE = re.compile(r'^music_library\.db\.backup_\d{8}_\d{6}$') _BACKUP_FILENAME_RE = re.compile(r'^music_library\.db\.backup_\d{8}_\d{6}$')
@app.route('/api/database/backup', methods=['POST']) @app.route('/api/database/backup', methods=['POST'])
@admin_only
def backup_database_endpoint(): def backup_database_endpoint():
"""Create a rolling backup of the database (max 5).""" """Create a rolling backup of the database (max 5)."""
try: try:
@ -16202,6 +16212,7 @@ def list_backups_endpoint():
return jsonify({"success": False, "error": str(e)}), 500 return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/database/backups/<filename>', methods=['DELETE']) @app.route('/api/database/backups/<filename>', methods=['DELETE'])
@admin_only
def delete_backup_endpoint(filename): def delete_backup_endpoint(filename):
"""Delete a specific database backup.""" """Delete a specific database backup."""
try: try:
@ -16224,6 +16235,7 @@ def delete_backup_endpoint(filename):
return jsonify({"success": False, "error": str(e)}), 500 return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/database/backups/<filename>/restore', methods=['POST']) @app.route('/api/database/backups/<filename>/restore', methods=['POST'])
@admin_only
def restore_backup_endpoint(filename): def restore_backup_endpoint(filename):
"""Restore the database from a specific backup.""" """Restore the database from a specific backup."""
try: try:
@ -16367,6 +16379,7 @@ def database_maintenance_info():
@app.route('/api/database/maintenance/vacuum', methods=['POST']) @app.route('/api/database/maintenance/vacuum', methods=['POST'])
@admin_only
def database_vacuum(): def database_vacuum():
"""Run VACUUM to compact the database. Locks DB during operation.""" """Run VACUUM to compact the database. Locks DB during operation."""
try: try:
@ -16556,6 +16569,7 @@ def metadata_cache_browse_musicbrainz():
return jsonify({"error": str(e)}), 500 return jsonify({"error": str(e)}), 500
@app.route('/api/metadata-cache/clear', methods=['DELETE']) @app.route('/api/metadata-cache/clear', methods=['DELETE'])
@admin_only
def metadata_cache_clear(): def metadata_cache_clear():
"""Clear cached metadata. Optional query params: source, type.""" """Clear cached metadata. Optional query params: source, type."""
try: try:
@ -16583,6 +16597,7 @@ def metadata_cache_evict():
return jsonify({"success": False, "error": str(e)}), 500 return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/metadata-cache/clear-musicbrainz', methods=['DELETE']) @app.route('/api/metadata-cache/clear-musicbrainz', methods=['DELETE'])
@admin_only
def metadata_cache_clear_musicbrainz(): def metadata_cache_clear_musicbrainz():
"""Clear MusicBrainz cache entries. Optional query param: failed_only=true.""" """Clear MusicBrainz cache entries. Optional query param: failed_only=true."""
try: try: