#845 follow-up: admin-gate the mutating verification-review endpoints

The merged PR left the review-queue's mutating endpoints ungated. Both now require
admin, matching the Phase 3 destructive-endpoint convention:

- /api/verification/<id>/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/<id>/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.
This commit is contained in:
BoulderBadgeDad 2026-06-11 10:43:40 -07:00
parent eb35ba86fb
commit 17440329c1
2 changed files with 61 additions and 12 deletions

View file

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

View file

@ -7914,10 +7914,12 @@ def get_quarantine_file_tags(entry_id):
@app.route('/api/verification/<int:history_id>/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,13 +7929,13 @@ 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(
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.cursor().execute(
conn.execute(
"UPDATE tracks SET verification_status = ? WHERE file_path = ?",
(HUMAN_VERIFIED, p))
conn.commit()
@ -7945,9 +7947,11 @@ def approve_verification_item(history_id):
@app.route('/api/verification/<int:history_id>/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)