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