358 lines
15 KiB
Python
358 lines
15 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 time
|
|
|
|
import httpx
|
|
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 -> 502). Bug #5: máy có Ollama
|
|
# chạy ở localhost:11434 trả 200 → test fail giả. Mock connection fail
|
|
# để test deterministic, không phụ thuộc service ngoài.
|
|
import app.api.v1.ai_proxy as ai_proxy
|
|
|
|
class _FakeConnectError(httpx.ConnectError):
|
|
def __init__(self, *a, **k):
|
|
super().__init__("mock connection refused", request=httpx.Request("POST", "http://localhost:11434/"))
|
|
|
|
class _FakeAsyncClient(httpx.AsyncClient):
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def post(self, *a, **k):
|
|
raise _FakeConnectError()
|
|
|
|
monkeypatch = pytest.MonkeyPatch()
|
|
monkeypatch.setattr(ai_proxy.httpx, "AsyncClient", _FakeAsyncClient)
|
|
try:
|
|
resp = client.post("/api/v1/ai/proxy", headers=auth_headers(),
|
|
json={"url": "http://localhost:11434/v1/chat/completions", "body": {}})
|
|
finally:
|
|
monkeypatch.undo()
|
|
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
|
|
|
|
# ── 7. Bug fixes 2026-08-10 audit round 2 ──
|
|
|
|
class TestBugFixes:
|
|
def test_static_mount_blocks_dotfiles(self):
|
|
# Bug #1: /static/audio/.secret_key, sonicforge.db, temp/autosave.json
|
|
# từng serve 200 không cần auth — giờ phải 404.
|
|
assert client.get("/static/audio/.secret_key").status_code == 404
|
|
assert client.get("/static/audio/sonicforge.db").status_code == 404
|
|
assert client.get("/static/audio/temp/autosave.json").status_code == 404
|
|
assert client.get("/static/audio/../.secret_key").status_code in (404, 400)
|
|
|
|
def test_static_uploads_still_served(self):
|
|
# File audio hợp lệ trong uploads vẫn phải serve được (không vỡ UI).
|
|
fid = f"user_test_{os.urandom(4).hex()}.wav"
|
|
p = os.path.join(settings.UPLOADS_DIR, fid)
|
|
with open(p, "wb") as f:
|
|
f.write(b"RIFFxxxxWAVE")
|
|
try:
|
|
assert client.get(f"/static/audio/uploads/{fid}").status_code == 200
|
|
finally:
|
|
os.remove(p)
|
|
|
|
def test_delete_user_file_blocks_traversal(self):
|
|
# Bug #2: `user_<id>_../../x` không được xóa file ngoài storage.
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
prof = client.get("/api/v1/auth/profile", headers={"Authorization": f"Bearer {token}"})
|
|
uid = prof.json()["id"]
|
|
marker = os.path.join(settings.STORAGE_DIR, "pwn_marker_%s.txt" % os.urandom(4).hex())
|
|
with open(marker, "w") as f:
|
|
f.write("x")
|
|
try:
|
|
resp = client.delete(f"/api/v1/audio/my-files/user_{uid}_../../{os.path.basename(marker)}",
|
|
headers={"Authorization": f"Bearer {token}"})
|
|
assert resp.status_code in (403, 404)
|
|
assert os.path.exists(marker), "traversal delete phải bị chặn"
|
|
finally:
|
|
if os.path.exists(marker):
|
|
os.remove(marker)
|
|
|
|
def test_upload_rejects_non_audio_extension(self):
|
|
# Bug #4: upload .exe bị từ chối.
|
|
resp = client.post("/api/v1/audio/upload",
|
|
files={"file": ("evil.exe", b"MZ\x90\x00", "application/octet-stream")})
|
|
assert resp.status_code == 400
|
|
|
|
def test_change_password_rejects_weak(self):
|
|
# Bug #7: change-password áp cùng policy độ mạnh như register.
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
resp = client.post("/api/v1/auth/change-password",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"old_password": "admin123", "new_password": "a"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_cleanup_keeps_referenced_files(self):
|
|
# Bug #3: cleanup_expired_files_task không xóa file được project tham chiếu.
|
|
from app.tasks.worker import cleanup_expired_files_task
|
|
from app.models.user import get_db_connection
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
prof = client.get("/api/v1/auth/profile", headers={"Authorization": f"Bearer {token}"})
|
|
uid = prof.json()["id"]
|
|
ref_id = f"user_{uid}_ref_{os.urandom(4).hex()}.wav"
|
|
stray_id = f"user_{uid}_stray_{os.urandom(4).hex()}.wav"
|
|
ref_path = os.path.join(settings.PROCESSED_DIR, ref_id)
|
|
stray_path = os.path.join(settings.PROCESSED_DIR, stray_id)
|
|
old = time.time() - 999999
|
|
for p in (ref_path, stray_path):
|
|
with open(p, "wb") as f:
|
|
f.write(b"RIFFxxxxWAVE")
|
|
os.utime(p, (old, old))
|
|
pid = f"cleanup_test_{os.urandom(4).hex()}"
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
proj = {"tracks": [{"serverFileId": ref_id}]}
|
|
cursor.execute("INSERT INTO projects (id, user_id, name, data_json, is_temp, size_bytes, updated_at) VALUES (?,?,?,?,0,?,?)",
|
|
(pid, uid, "cleanup test", json.dumps(proj), 0, old))
|
|
conn.commit()
|
|
conn.close()
|
|
try:
|
|
result = cleanup_expired_files_task(max_age_hours=0)
|
|
assert os.path.exists(ref_path), "file được project tham chiếu phải giữ"
|
|
assert not os.path.exists(stray_path), "file không tham chiếu phải bị dọn"
|
|
assert result["referenced_files_kept"] >= 1
|
|
finally:
|
|
for p in (ref_path, stray_path):
|
|
if os.path.exists(p):
|
|
os.remove(p)
|
|
conn = get_db_connection()
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM projects WHERE id = ?", (pid,))
|
|
conn.commit()
|
|
conn.close()
|