FEAT: thêm nút bypass cho track strip để bypass không qua mastering panel

tests/test_security_hardening.py

@@ -0,0 +1,234 @@
+"""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