From 17440329c189dd20b8801473753539b0d0b7268a Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Thu, 11 Jun 2026 10:43:40 -0700 Subject: [PATCH] #845 follow-up: admin-gate the mutating verification-review endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The merged PR left the review-queue's mutating endpoints ungated. Both now require admin, matching the Phase 3 destructive-endpoint convention: - /api/verification//delete (os.remove + drops the history row) — @admin_only, so a non-admin on a login/multi-profile instance can't delete library files. - /api/verification//approve (flips verification_status + writes the tag) — @admin_only; also wrapped its DB writes in `with db._get_connection()` for rollback-on-error + codebase consistency (was a bare conn). Read/playback endpoints (stream/play/compare/entry/config) stay open — the app's LAN-read model. Tests: non-admin gets 403 on delete + approve; admin isn't blocked. --- tests/test_verification_admin_gate.py | 45 +++++++++++++++++++++++++++ web_server.py | 28 ++++++++++------- 2 files changed, 61 insertions(+), 12 deletions(-) create mode 100644 tests/test_verification_admin_gate.py diff --git a/tests/test_verification_admin_gate.py b/tests/test_verification_admin_gate.py new file mode 100644 index 00000000..b927c458 --- /dev/null +++ b/tests/test_verification_admin_gate.py @@ -0,0 +1,45 @@ +"""#845 follow-up: the mutating verification-review endpoints (delete removes a +file from disk; approve flips verification state) must be admin-only, matching the +Phase 3 destructive-endpoint gating. The read/playback ones stay open.""" + +from __future__ import annotations + +import os +import tempfile + +import pytest + +_TMP = tempfile.mkdtemp(prefix='soulsync-testdb-vgate-') +os.environ['DATABASE_PATH'] = os.path.join(_TMP, 'v.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 _as_nonadmin(client): + pid = web_server.get_database().create_profile(name=f'u_{os.urandom(4).hex()}') + with client.session_transaction() as s: + s['profile_id'] = pid + return pid + + +def test_delete_is_admin_only(client): + _as_nonadmin(client) + assert client.post('/api/verification/1/delete').status_code == 403 + + +def test_approve_is_admin_only(client): + _as_nonadmin(client) + assert client.post('/api/verification/1/approve').status_code == 403 + + +def test_admin_not_blocked(client): + # default session resolves to admin (profile 1) — must pass the gate (a + # missing history id yields 404, NOT 403). + assert client.post('/api/verification/999999/delete').status_code != 403 + assert client.post('/api/verification/999999/approve').status_code != 403 diff --git a/web_server.py b/web_server.py index 5480f63e..8d4e147c 100644 --- a/web_server.py +++ b/web_server.py @@ -7914,10 +7914,12 @@ def get_quarantine_file_tags(entry_id): @app.route('/api/verification//approve', methods=['POST']) +@admin_only def approve_verification_item(history_id): """User confirmed the file IS the right track: set human_verified on the history row, the file tag, and (best-effort) the tracks row. The AcoustID - scanner skips human-verified files entirely.""" + scanner skips human-verified files entirely. Admin-only: mutates shared + library/verification state.""" try: from core.matching.verification_status import HUMAN_VERIFIED from core.tag_writer import write_verification_status @@ -7927,16 +7929,16 @@ def approve_verification_item(history_id): return jsonify({"success": False, "error": "History entry not found"}), 404 file_path = row.get('file_path') or '' on_disk = _resolve_history_audio_path(row) - conn = db._get_connection() - conn.cursor().execute( - "UPDATE library_history SET verification_status = ? WHERE id = ?", - (HUMAN_VERIFIED, history_id)) - # The tracks row may carry either the recorded or the resolved path. - for p in {p for p in (file_path, on_disk) if p}: - conn.cursor().execute( - "UPDATE tracks SET verification_status = ? WHERE file_path = ?", - (HUMAN_VERIFIED, p)) - conn.commit() + with db._get_connection() as conn: + conn.execute( + "UPDATE library_history SET verification_status = ? WHERE id = ?", + (HUMAN_VERIFIED, history_id)) + # The tracks row may carry either the recorded or the resolved path. + for p in {p for p in (file_path, on_disk) if p}: + conn.execute( + "UPDATE tracks SET verification_status = ? WHERE file_path = ?", + (HUMAN_VERIFIED, p)) + conn.commit() tag_written = bool(on_disk) and write_verification_status(on_disk, HUMAN_VERIFIED) return jsonify({"success": True, "tag_written": tag_written}) except Exception as e: @@ -7945,9 +7947,11 @@ def approve_verification_item(history_id): @app.route('/api/verification//delete', methods=['POST']) +@admin_only def delete_verification_item(history_id): """User decided the file is wrong: delete it from disk and drop the - history row (the media-server mirror cleans the tracks row on next scan).""" + history row (the media-server mirror cleans the tracks row on next scan). + Admin-only: it removes a file from disk + the library.""" try: db = get_database() row = _get_library_history_row(history_id)