235 lines
9.6 KiB
Python
235 lines
9.6 KiB
Python
"""Regression tests for security hardening.
|
|
|
|
Covers the vulnerabilities found during the 2026-08 audit:
|
|
- SSRF / open proxy on /api/v1/ai/proxy
|
|
- path traversal on render output and audio file ids
|
|
- unauthenticated filesystem access via /api/v1/media/*
|
|
- hardcoded SECRET_KEY
|
|
- quota bypass on project update
|
|
- audio resampling correctness in the render engine
|
|
"""
|
|
import os
|
|
import json
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import soundfile as sf
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.config import settings
|
|
from app.core import auth as core_auth
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def get_admin_token():
|
|
# test_auth_and_quota.py may have rotated the admin password; try both.
|
|
for pwd in ("admin123", "admin_new_password_2026"):
|
|
resp = client.post("/api/v1/auth/login", json={"username": "admin", "password": pwd})
|
|
if resp.status_code != 200:
|
|
continue
|
|
token = resp.json()["access_token"]
|
|
user = resp.json()["user"]
|
|
if user.get("must_change_password"):
|
|
r = client.post("/api/v1/auth/change-password", headers={"Authorization": f"Bearer {token}"},
|
|
json={"old_password": pwd, "new_password": pwd})
|
|
if r.status_code == 200:
|
|
token = r.json()["access_token"]
|
|
return token
|
|
return None
|
|
|
|
|
|
def auth_headers():
|
|
return {"Authorization": f"Bearer {get_admin_token()}"}
|
|
|
|
|
|
# ── 1. SSRF / open proxy ──
|
|
|
|
class TestAIProxySSRF:
|
|
def test_proxy_requires_auth(self):
|
|
resp = client.post("/api/v1/ai/proxy", json={"url": "https://api.openai.com/v1", "body": {}})
|
|
assert resp.status_code == 401
|
|
|
|
def test_proxy_blocks_cloud_metadata(self):
|
|
resp = client.post("/api/v1/ai/proxy", headers=auth_headers(),
|
|
json={"url": "http://169.254.169.254/latest/meta-data/", "body": {}})
|
|
assert resp.status_code == 403
|
|
|
|
def test_proxy_blocks_private_ip_not_configured(self):
|
|
resp = client.post("/api/v1/ai/proxy", headers=auth_headers(),
|
|
json={"url": "http://10.0.0.5/", "body": {}})
|
|
assert resp.status_code == 403
|
|
|
|
def test_proxy_rejects_non_http_scheme(self):
|
|
resp = client.post("/api/v1/ai/proxy", headers=auth_headers(),
|
|
json={"url": "file:///etc/passwd", "body": {}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_proxy_allows_configured_localhost_provider(self):
|
|
# localhost:11434 is in the default AI provider list; it must pass the
|
|
# SSRF check (and then fail to connect in this environment -> 502).
|
|
resp = client.post("/api/v1/ai/proxy", headers=auth_headers(),
|
|
json={"url": "http://localhost:11434/v1/chat/completions", "body": {}})
|
|
assert resp.status_code == 502
|
|
|
|
|
|
# ── 2. Path traversal ──
|
|
|
|
class TestPathTraversal:
|
|
def test_render_output_filename_sanitized(self):
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
project = {
|
|
"metadata": {"bpm": 120, "time_signature_numerator": 4},
|
|
"main_session": {"length_bars": 1, "tracks": []},
|
|
"section_store": {},
|
|
}
|
|
resp = client.post("/api/v1/plugins/render", headers=auth_headers(),
|
|
json={"project_json": project, "output_filename": "/tmp/evil_traversal.wav"})
|
|
# Absolute paths must be reduced to a basename inside PROCESSED_DIR.
|
|
assert resp.status_code == 200, resp.text
|
|
out_path = resp.json()["path"]
|
|
assert os.path.dirname(out_path) == settings.PROCESSED_DIR
|
|
assert os.path.basename(out_path) == "evil_traversal.wav"
|
|
assert os.path.isfile(out_path)
|
|
|
|
def test_audio_download_rejects_traversal(self):
|
|
resp = client.get("/api/v1/audio/download/..%2F..%2Fapp%2Fconfig.py")
|
|
assert resp.status_code == 404
|
|
|
|
def test_ai_scan_rejects_traversal_file_id(self):
|
|
resp = client.post("/api/v1/audio/ai-scan",
|
|
json={"track_id": "1", "file_id": "../../app/config.py"})
|
|
# Traversal must NOT read the file: falls through to the demo branch.
|
|
assert resp.status_code == 200
|
|
assert resp.json()["success"] is True
|
|
|
|
|
|
# ── 3. Filesystem exposure via media endpoints ──
|
|
|
|
class TestMediaAuth:
|
|
# Use a fresh client (no cookies from earlier logins) to prove 401.
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_client(self):
|
|
self.fresh = TestClient(app)
|
|
yield
|
|
self.fresh.close()
|
|
|
|
def test_media_computer_requires_auth(self):
|
|
resp = self.fresh.get("/api/v1/media/computer")
|
|
assert resp.status_code == 401
|
|
|
|
def test_media_browse_requires_auth(self):
|
|
resp = self.fresh.get("/api/v1/media/browse", params={"path": "/etc"})
|
|
assert resp.status_code == 401
|
|
|
|
def test_media_file_requires_auth(self):
|
|
resp = self.fresh.get("/api/v1/media/file", params={"path": "/etc/passwd"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ── 4. Secret key ──
|
|
|
|
class TestSecretKey:
|
|
def test_secret_key_not_hardcoded_default(self):
|
|
assert core_auth.SECRET_KEY != "sonicforge_secret_key_super_secure_2026"
|
|
assert len(core_auth.SECRET_KEY) >= 32
|
|
|
|
|
|
# ── 5. Quota enforcement on update ──
|
|
|
|
class TestQuotaUpdate:
|
|
def test_update_cloud_project_enforces_quota(self):
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
# Register a fresh user with a small quota (unique name per run so the
|
|
# test is re-runnable against a persistent DB).
|
|
import uuid as _uuid
|
|
uname = f"quota_user_{_uuid.uuid4().hex[:8]}"
|
|
resp = client.post("/api/v1/auth/register", json={
|
|
"username": uname, "email": f"{uname}@studio.com", "password": "quota_pass_123"})
|
|
assert resp.status_code == 200, resp.text
|
|
user_token = resp.json()["access_token"]
|
|
user_headers = {"Authorization": f"Bearer {user_token}"}
|
|
|
|
# Shrink quota to 1 MB via admin API.
|
|
uid = resp.json()["user"]["id"]
|
|
r = client.put(f"/api/v1/admin/quotas/{uid}", headers=auth_headers(),
|
|
json={"storage_limit_mb": 1, "max_tracks": 16})
|
|
assert r.status_code == 200, r.text
|
|
|
|
# Save a small project.
|
|
small = json.dumps({
|
|
"project_id": "p1",
|
|
"metadata": {"title": "small", "bpm": 120, "time_signature_numerator": 4,
|
|
"time_signature_denominator": 4, "sample_rate": 44100},
|
|
"main_session": {"id": "main", "name": "MAIN SESSION", "is_root": True,
|
|
"length_bars": 16.0, "auto_compute_length": True, "tracks": []},
|
|
"section_store": {}})
|
|
r = client.post("/api/v1/projects/cloud", headers=user_headers,
|
|
json={"name": "small", "data_json": small})
|
|
assert r.status_code == 200, r.text
|
|
pid = r.json()["project_id"]
|
|
|
|
# Updating with a payload over the quota must be rejected (was a bypass).
|
|
items = [{
|
|
"type": "AUDIO_ITEM", "id": f"it_{i}", "name": "n",
|
|
"start_bar": 0.0, "duration_bars": 1.0, "clip_start_offset_bars": 0.0,
|
|
"source_data": {"audio_file_url": "", "gain": 1.0},
|
|
} for i in range(20000)]
|
|
huge = json.dumps({
|
|
"project_id": "p1",
|
|
"metadata": {"title": "huge", "bpm": 120, "time_signature_numerator": 4,
|
|
"time_signature_denominator": 4, "sample_rate": 44100},
|
|
"main_session": {"id": "main", "name": "MAIN SESSION", "is_root": True,
|
|
"length_bars": 16.0, "auto_compute_length": True,
|
|
"tracks": [{"id": "t", "name": "x", "type": "AUDIO", "items": items}]},
|
|
"section_store": {}})
|
|
r = client.put(f"/api/v1/projects/cloud/{pid}", headers=user_headers,
|
|
json={"name": "huge", "data_json": huge})
|
|
assert r.status_code == 400, r.text
|
|
assert "Quota" in r.json()["detail"]
|
|
|
|
|
|
# ── 6. Render engine: resampling correctness ──
|
|
|
|
class TestRenderResample:
|
|
def test_audio_item_resampled_to_engine_rate(self, tmp_path):
|
|
from app.core.render_engine import PythonRenderEngine
|
|
# 44.1kHz source, engine at 22.05kHz -> exactly 2x downsampling.
|
|
sr_src = 44100
|
|
t = np.arange(sr_src) / sr_src
|
|
tone = (0.5 * np.sin(2 * np.pi * 440 * t)).astype(np.float32)
|
|
src_path = os.path.join(settings.UPLOADS_DIR, "resample_test_tone.wav")
|
|
sf.write(src_path, tone, sr_src)
|
|
|
|
engine = PythonRenderEngine(sample_rate=22050)
|
|
session = {
|
|
"tracks": [{
|
|
"type": "AUDIO",
|
|
"volume_db": 0.0, "pan": 0.0, "mute": False,
|
|
"items": [{
|
|
"type": "AUDIO_ITEM",
|
|
"start_bar": 0.0, "duration_bars": 4.0,
|
|
"clip_start_offset_bars": 0.0,
|
|
"source_data": {"audio_file_url": "/static/audio/uploads/resample_test_tone.wav", "gain": 1.0},
|
|
}],
|
|
}]
|
|
}
|
|
buf = engine.render_session_container(session, {}, bpm=120.0, time_sig_num=4,
|
|
total_samples=engine.sample_rate * 2)
|
|
# A 1s 440Hz tone must actually render energy (previously the SR
|
|
# mismatch silently skipped the audio).
|
|
assert np.max(np.abs(buf)) > 0.01
|
|
# Duration should be ~1 second at the engine rate, not 2.
|
|
nonzero = np.where(np.abs(buf[0]) > 1e-4)[0]
|
|
assert len(nonzero) > 0
|
|
assert (nonzero[-1] - nonzero[0]) < int(engine.sample_rate * 1.3)
|
|
try:
|
|
os.remove(src_path)
|
|
except OSError:
|
|
pass
|