From d1dd6ed7149aef8085d70decf00877eb0f536f33 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Wed, 10 Jun 2026 00:33:13 -0700 Subject: [PATCH] Profiles Phase 1: admin endpoints to manage service-credential sets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Admin-only CRUD over the named credential sets from Phase 0: - GET /api/credentials list all sets grouped by service (NO secrets) - POST /api/credentials create {service,label,payload}, validated - PUT /api/credentials/ update label and/or payload (partial) - DELETE /api/credentials/ delete (clears any profile selections → fallback) All four are @admin_only (non-admin → 403), payloads validated via core.credentials.store, secrets never returned to the browser. Additive — no existing endpoint or behaviour changes. Tests: real web_server app + Flask test client (8) — create/list/update/delete roundtrip, payload never leaks in list, missing-field/unsupported-service/blank- label/duplicate(409)/404 validation, and the non-admin 403 gate on every write. Verified the web_server import coexists with the rest of the suite (175 mixed tests pass). --- tests/test_credentials_endpoints.py | 108 ++++++++++++++++++++++++++++ web_server.py | 100 ++++++++++++++++++++++++++ 2 files changed, 208 insertions(+) create mode 100644 tests/test_credentials_endpoints.py diff --git a/tests/test_credentials_endpoints.py b/tests/test_credentials_endpoints.py new file mode 100644 index 00000000..5542736a --- /dev/null +++ b/tests/test_credentials_endpoints.py @@ -0,0 +1,108 @@ +"""Phase 1: service-credential-set admin endpoints (real app, real HTTP). + +These import the actual web_server app and drive the endpoints through a Flask +test client — the only way to verify the @admin_only gating and the request +validation wrappers for real. Secrets must never come back in any response. + +Heavy (imports web_server once), so isolated in its own module. The default +session is profile 1 (admin); a non-admin session is simulated to prove the +gate blocks writes. +""" + +from __future__ import annotations + +import os +import tempfile + +import pytest + +# Redirect the DB before importing web_server so it never touches a real library. +_TMP = tempfile.mkdtemp(prefix='ss-cred-ep-') +os.environ['DATABASE_PATH'] = os.path.join(_TMP, 'creds_ep.db') +os.environ['SOULSYNC_TEST_DB_READY'] = '1' + +web_server = pytest.importorskip('web_server') + + +@pytest.fixture +def client(): + return web_server.app.test_client() + + +@pytest.fixture +def nonadmin_profile(): + """Create a real non-admin profile and yield its id.""" + db = web_server.get_database() + pid = db.create_profile(name=f'tester_{os.urandom(3).hex()}', avatar_color='#fff') + yield pid + + +# ── admin happy paths ──────────────────────────────────────────────────────── + +def test_admin_create_list_update_delete_roundtrip(client): + r = client.post('/api/credentials', json={ + 'service': 'plex', 'label': 'Living Room', + 'payload': {'base_url': 'http://plex:32400', 'token': 'sekret'}}) + assert r.status_code == 200 and r.get_json()['success'] + cid = r.get_json()['id'] + + # list shows it, and NEVER leaks the payload/secret + body = client.get('/api/credentials').get_json() + assert any(c['label'] == 'Living Room' for c in body['services']['plex']) + assert 'sekret' not in str(body) and 'payload' not in str(body) + + # update label + assert client.put(f'/api/credentials/{cid}', json={'label': 'Den'}).get_json()['success'] + body = client.get('/api/credentials').get_json() + assert any(c['label'] == 'Den' for c in body['services']['plex']) + + # delete + assert client.delete(f'/api/credentials/{cid}').get_json()['success'] + body = client.get('/api/credentials').get_json() + assert not any(c['id'] == cid for c in body['services']['plex']) + + +# ── validation ─────────────────────────────────────────────────────────────── + +def test_create_rejects_missing_fields(client): + r = client.post('/api/credentials', json={ + 'service': 'plex', 'label': 'X', 'payload': {'base_url': 'http://p'}}) + assert r.status_code == 400 and 'token' in r.get_json()['error'] + + +def test_create_rejects_unsupported_service(client): + r = client.post('/api/credentials', json={'service': 'itunes', 'label': 'X', 'payload': {}}) + assert r.status_code == 400 + + +def test_create_rejects_blank_label(client): + r = client.post('/api/credentials', json={ + 'service': 'deezer', 'label': ' ', 'payload': {'arl': 'x'}}) + assert r.status_code == 400 + + +def test_duplicate_label_conflict(client): + p = {'service': 'qobuz', 'label': 'Dup', 'payload': {'user_auth_token': 't'}} + assert client.post('/api/credentials', json=p).status_code == 200 + assert client.post('/api/credentials', json=p).status_code == 409 + + +def test_update_missing_set_404(client): + assert client.put('/api/credentials/999999', json={'label': 'x'}).status_code == 404 + + +def test_delete_missing_set_404(client): + assert client.delete('/api/credentials/999999').status_code == 404 + + +# ── the security gate: non-admin cannot manage credential sets ─────────────── + +def test_nonadmin_blocked_from_all_credential_writes(client, nonadmin_profile): + with client.session_transaction() as sess: + sess['profile_id'] = nonadmin_profile + assert client.get('/api/credentials').status_code == 403 + assert client.post('/api/credentials', json={ + 'service': 'plex', 'label': 'Sneaky', + 'payload': {'base_url': 'http://p', 'token': 't'}}).status_code == 403 + assert client.put('/api/credentials/1', json={'label': 'x'}).status_code == 403 + assert client.delete('/api/credentials/1').status_code == 403 diff --git a/web_server.py b/web_server.py index 59bf8074..36fb9aeb 100644 --- a/web_server.py +++ b/web_server.py @@ -25421,6 +25421,106 @@ def save_profile_server_library(): except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 + +# ================================================================================== +# SERVICE CREDENTIAL SETS (admin-created named "pills" per auth service; #profiles) +# ---------------------------------------------------------------------------------- +# Admin manages the named credential sets; any profile only SELECTS among them +# (see /api/credentials/active + /select below). Payloads (the actual secrets) +# are NEVER returned to the browser — only id/service/label. +# ================================================================================== + +@app.route('/api/credentials', methods=['GET']) +@admin_only +def list_service_credentials_endpoint(): + """List all credential sets grouped by service (metadata only, no secrets).""" + try: + from core.credentials.store import SERVICE_CREDENTIAL_SCHEMA + rows = get_database().list_service_credentials() + grouped = {svc: [] for svc in SERVICE_CREDENTIAL_SCHEMA} + for r in rows: + grouped.setdefault(r['service'], []).append( + {'id': r['id'], 'label': r['label'], 'updated_at': r['updated_at']} + ) + return jsonify({'success': True, 'services': grouped}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + +@app.route('/api/credentials', methods=['POST']) +@admin_only +def create_service_credential_endpoint(): + """Create a named credential set for a service. Body: {service, label, payload}.""" + try: + from core.credentials.store import is_supported_service, validate_credential_payload + data = request.json or {} + service = (data.get('service') or '').strip() + label = (data.get('label') or '').strip() + payload = data.get('payload') or {} + + if not is_supported_service(service): + return jsonify({'success': False, 'error': f'Unsupported service: {service}'}), 400 + if not label: + return jsonify({'success': False, 'error': 'A name is required'}), 400 + ok, missing = validate_credential_payload(service, payload) + if not ok: + return jsonify({'success': False, 'error': f'Missing required fields: {", ".join(missing)}'}), 400 + + cred_id = get_database().create_service_credential(service, label, payload) + if cred_id is None: + return jsonify({'success': False, 'error': f'A "{label}" set already exists for {service}'}), 409 + return jsonify({'success': True, 'id': cred_id}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + +@app.route('/api/credentials/', methods=['PUT']) +@admin_only +def update_service_credential_endpoint(credential_id): + """Update a credential set's label and/or payload. Only provided fields change.""" + try: + from core.credentials.store import validate_credential_payload + data = request.json or {} + label = data.get('label') + payload = data.get('payload') # None = leave secrets untouched + + existing = get_database().get_service_credential(credential_id) + if not existing: + return jsonify({'success': False, 'error': 'Credential set not found'}), 404 + + if label is not None and not str(label).strip(): + return jsonify({'success': False, 'error': 'Name cannot be empty'}), 400 + if payload is not None: + ok, missing = validate_credential_payload(existing['service'], payload) + if not ok: + return jsonify({'success': False, 'error': f'Missing required fields: {", ".join(missing)}'}), 400 + + updated = get_database().update_service_credential( + credential_id, + label=str(label).strip() if label is not None else None, + payload=payload, + ) + if not updated: + return jsonify({'success': False, 'error': 'Nothing to update or name already in use'}), 400 + return jsonify({'success': True}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + +@app.route('/api/credentials/', methods=['DELETE']) +@admin_only +def delete_service_credential_endpoint(credential_id): + """Delete a credential set. Any profile that had it selected falls back to + the global/admin default automatically (selection is cleared in the DB).""" + try: + ok = get_database().delete_service_credential(credential_id) + if not ok: + return jsonify({'success': False, 'error': 'Credential set not found'}), 404 + return jsonify({'success': True}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + # --- Watchlist API Endpoints --- @app.route('/api/watchlist/count', methods=['GET'])